import urllib.parse import xml.etree.ElementTree as ET import requests from urllib.parse import urlparse from core import AppLogger class DLNAController: """ DLNA 媒体渲染器控制器 只负责控制已发现的设备，不包含发现功能 """ def __init__(self, device_info): """ 初始化控制器 Args: device_info: 设备信息字典，必须包含: - 'location': 设备描述文档的URL - 'friendly_name': 设备友好名称 (可选) - 'ip': 设备IP地址 (可选) """ if 'location' not in device_info: raise ValueError("device_info 必须包含 'location' 字段") self.device_info = device_info self.location = device_info['location'] self.base_url = None self.av_transport_url = None self.rendering_control_url = None self.connection_manager_url = None # 解析设备信息 self._parse_device_info() self.logger = AppLogger('DLNAController') def _parse_device_info(self): """ 解析设备描述文档，提取控制URL """ try: # 获取基础URL parsed = urlparse(self.location) self.base_url = f"{parsed.scheme}://{parsed.netloc}" # 获取设备描述文档 response = requests.get(self.location, timeout=5) response.raise_for_status() # 解析XML root = ET.fromstring(response.content) ns = {'upnp': 'urn:schemas-upnp-org:device-1-0'} # 查找服务列表 service_list = root.find('.//upnp:device/upnp:serviceList', ns) if service_list is None: raise ValueError("设备描述中未找到服务列表") # 遍历服务，提取控制URL for service in service_list.findall('upnp:service', ns): service_type_elem = service.find('upnp:serviceType', ns) control_url_elem = service.find('upnp:controlURL', ns) if service_type_elem is None or control_url_elem is None: continue service_type = service_type_elem.text control_url = control_url_elem.text # 确保URL是完整的 if control_url.startswith('/'): control_url = f"{self.base_url}{control_url}" if 'AVTransport' in service_type: self.av_transport_url = control_url elif 'RenderingControl' in service_type: self.rendering_control_url = control_url elif 'ConnectionManager' in service_type: self.connection_manager_url = control_url # 验证必要的服务是否存在 if not self.av_transport_url: raise ValueError("设备不支持 AVTransport 服务") except requests.exceptions.RequestException as e: raise ConnectionError(f"无法连接设备: {e}") except ET.ParseError as e: raise ValueError(f"设备描述文档格式错误: {e}") except Exception as e: raise RuntimeError(f"解析设备信息失败: {e}") def _send_soap_request(self, service_url, action, body): """ 发送SOAP请求到设备 Args: service_url: 服务控制URL action: SOAP Action body: SOAP Body内容 Returns: tuple: (success, response_xml) - 成功标志和响应XML """ soap_envelope = f""" {body} """ headers = { 'Content-Type': 'text/xml; charset="utf-8"', 'SOAPAction': f'"{action}"', 'User-Agent': 'Python DLNA Controller' } try: response = requests.post(service_url, data=soap_envelope, headers=headers, timeout=5) response.raise_for_status() return True, response.text except requests.exceptions.Timeout: return False, "请求超时" except requests.exceptions.ConnectionError: return False, "连接失败" except Exception as e: return False, str(e) # ================== AVTransport 服务方法 ================== def set_av_transport_uri(self, media_url, title = "", metadata=""): """ 设置播放URI Args: media_url: 媒体文件的URL title: 媒体文件名称 metadata: 媒体元数据（可选） Returns: bool: 是否成功 """ if not self.av_transport_url: return False if title and not metadata: # 转义title中的XML特殊字符 title_escaped = title.replace('&', '&').replace('<', '<').replace('>', '>').replace('"', '"') # 构建简单的DIDL-Lite元数据 metadata = f""" {title_escaped} """ # 转义XML特殊字符 media_url_escaped = media_url.replace('&', '&').replace('<', '<').replace('>', '>') media_url_escaped = urllib.parse.quote(media_url_escaped,safe=':/') metadata_escaped = metadata.replace('&', '&').replace('<', '<').replace('>', '>') soap_body = f""" 0 {media_url_escaped} {metadata_escaped} """ action = "urn:schemas-upnp-org:service:AVTransport:1#SetAVTransportURI" success, response = self._send_soap_request(self.av_transport_url, action, soap_body) self.logger.info(f"设置播放URI：{media_url_escaped}") self.logger.info(f"设置播放URI：{success}：{response}") return success def play(self, speed = 1.0): """ 开始播放 Args: speed: 播放速度，默认为1 Returns: bool: 是否成功 """ if not self.av_transport_url: return False soap_body = f""" 0 {speed} """ action = "urn:schemas-upnp-org:service:AVTransport:1#Play" success, response = self._send_soap_request(self.av_transport_url, action, soap_body) self.logger.info(f"播放结果：{success}：{response}") return success def pause(self): """ 暂停播放 Returns: bool: 是否成功 """ if not self.av_transport_url: return False soap_body = """ 0 """ action = "urn:schemas-upnp-org:service:AVTransport:1#Pause" success, response = self._send_soap_request(self.av_transport_url, action, soap_body) self.logger.info(f"暂停播放结果：{success}：{response}") return success def stop(self): """ 停止播放 Returns: bool: 是否成功 """ if not self.av_transport_url: return False soap_body = """ 0 """ action = "urn:schemas-upnp-org:service:AVTransport:1#Stop" success, response = self._send_soap_request(self.av_transport_url, action, soap_body) self.logger.info(f"停止播放结果：{success}：{response}") return success def seek(self, unit, target): """ 跳转到指定位置 Args: unit: 时间单位，如 "REL_TIME"（相对时间）或 "TRACK_NR"（曲目号） target: 目标位置，如 "00:05:30" 或 "2"（第2首） Returns: bool: 是否成功 """ if not self.av_transport_url: return False target_escaped = target.replace('&', '&').replace('<', '<').replace('>', '>') soap_body = f""" 0 {unit} {target_escaped} """ action = "urn:schemas-upnp-org:service:AVTransport:1#Seek" success, response = self._send_soap_request(self.av_transport_url, action, soap_body) self.logger.info(f"跳转到指定位置结果：{success}：{response}") return success def get_transport_info(self): """ 获取传输状态信息 Returns: dict: 传输状态信息，包含: - state: 当前传输状态 - status: 当前传输状态 - speed: 当前播放速度 None: 获取失败 """ if not self.av_transport_url: return None soap_body = """ 0 """ action = "urn:schemas-upnp-org:service:AVTransport:1#GetTransportInfo" success, response = self._send_soap_request(self.av_transport_url, action, soap_body) if not success: return None try: # 解析响应XML root = ET.fromstring(response) ns = {'s': 'http://schemas.xmlsoap.org/soap/envelope/', 'u': 'urn:schemas-upnp-org:service:AVTransport:1'} # 查找Body中的响应 body = root.find('.//s:Body', ns) if body is None: return None # 获取状态信息 current_transport_state = body.find('.//CurrentTransportState') current_transport_status = body.find('.//CurrentTransportStatus') current_speed = body.find('.//CurrentSpeed') return { 'state': current_transport_state.text if current_transport_state is not None else '未知', 'status': current_transport_status.text if current_transport_status is not None else '未知', 'speed': current_speed.text if current_speed is not None else '未知' } except ET.ParseError: return None def get_position_info(self): """ 获取当前位置信息 Returns: dict: 位置信息，包含: - track: 当前曲目 - track_duration: 曲目时长 - track_meta_data: 曲目元数据 - track_uri: 曲目URI - rel_time: 相对时间 - abs_time: 绝对时间 - rel_count: 相对计数 - abs_count: 绝对计数 None: 获取失败 """ if not self.av_transport_url: return None soap_body = """ 0 """ action = "urn:schemas-upnp-org:service:AVTransport:1#GetPositionInfo" success, response = self._send_soap_request(self.av_transport_url, action, soap_body) if not success: return None try: root = ET.fromstring(response) ns = {'s': 'http://schemas.xmlsoap.org/soap/envelope/', 'u': 'urn:schemas-upnp-org:service:AVTransport:1'} body = root.find('.//s:Body', ns) if body is None: return None # 提取位置信息 info = {} fields = [ 'Track', 'TrackDuration', 'TrackMetaData', 'TrackURI', 'RelTime', 'AbsTime', 'RelCount', 'AbsCount' ] for field in fields: elem = body.find(f'.//{field}') info[field.lower()] = elem.text if elem is not None else '' return info except ET.ParseError: return None # ================== RenderingControl 服务方法 ================== def set_volume(self, volume, channel='Master'): """ 设置音量 Args: volume: 音量值 (0-100) channel: 声道，默认为 'Master' Returns: bool: 是否成功 """ if not self.rendering_control_url: return False # 确保音量在0-100范围内 volume = max(0, min(100, int(volume))) soap_body = f""" 0 {channel} {volume} """ action = "urn:schemas-upnp-org:service:RenderingControl:1#SetVolume" success, response = self._send_soap_request(self.rendering_control_url, action, soap_body) self.logger.info(f"音量设置结果：{success}：{response}") return success def get_volume(self, channel='Master'): """ 获取当前音量 Args: channel: 声道，默认为 'Master' Returns: int: 音量值 (0-100)，None表示获取失败 """ if not self.rendering_control_url: return None soap_body = f""" 0 {channel} """ action = "urn:schemas-upnp-org:service:RenderingControl:1#GetVolume" success, response = self._send_soap_request(self.rendering_control_url, action, soap_body) if not success: return None try: root = ET.fromstring(response) ns = {'s': 'http://schemas.xmlsoap.org/soap/envelope/', 'u': 'urn:schemas-upnp-org:service:RenderingControl:1'} body = root.find('.//s:Body', ns) if body is None: return None current_volume = body.find('.//CurrentVolume') if current_volume is not None and current_volume.text: return int(current_volume.text) except (ET.ParseError, ValueError): pass return None def set_mute(self, mute, channel='Master'): """ 设置静音 Args: mute: 是否静音 (True/False) channel: 声道，默认为 'Master' Returns: bool: 是否成功 """ if not self.rendering_control_url: return False desired_mute = "1" if mute else "0" soap_body = f""" 0 {channel} {desired_mute} """ action = "urn:schemas-upnp-org:service:RenderingControl:1#SetMute" success, response = self._send_soap_request(self.rendering_control_url, action, soap_body) self.logger.info(f"静音设置结果：{success}：{response}") return success def get_mute(self, channel='Master'): """ 获取静音状态 Args: channel: 声道，默认为 'Master' Returns: bool: 是否静音，None表示获取失败 """ if not self.rendering_control_url: return None soap_body = f""" 0 {channel} """ action = "urn:schemas-upnp-org:service:RenderingControl:1#GetMute" success, response = self._send_soap_request(self.rendering_control_url, action, soap_body) if not success: return None try: root = ET.fromstring(response) ns = {'s': 'http://schemas.xmlsoap.org/soap/envelope/', 'u': 'urn:schemas-upnp-org:service:RenderingControl:1'} body = root.find('.//s:Body', ns) if body is None: return None current_mute = body.find('.//CurrentMute') if current_mute is not None and current_mute.text: return current_mute.text == "1" except ET.ParseError: pass return None # ================== ConnectionManager 服务方法 ================== def get_protocol_info(self): """ 获取设备支持的协议信息 Returns: dict: 协议信息，包含: - source: 源协议信息 - sink: 接收协议信息 None: 获取失败 """ if not self.connection_manager_url: return None soap_body = """ """ action = "urn:schemas-upnp-org:service:ConnectionManager:1#GetProtocolInfo" success, response = self._send_soap_request(self.connection_manager_url, action, soap_body) if not success: return None try: root = ET.fromstring(response) ns = {'s': 'http://schemas.xmlsoap.org/soap/envelope/', 'u': 'urn:schemas-upnp-org:service:ConnectionManager:1'} body = root.find('.//s:Body', ns) if body is None: return None source = body.find('.//Source') sink = body.find('.//Sink') return { 'source': source.text if source is not None else '', 'sink': sink.text if sink is not None else '' } except ET.ParseError: return None