555 lines
18 KiB
Python
555 lines
18 KiB
Python
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"""<?xml version="1.0" encoding="utf-8"?>
|
||
<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/"
|
||
s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">
|
||
<s:Body>
|
||
{body}
|
||
</s:Body>
|
||
</s:Envelope>"""
|
||
|
||
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"""<DIDL-Lite xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/"
|
||
xmlns:dc="http://purl.org/dc/elements/1.1/">
|
||
<item id="1" parentID="0" restricted="1">
|
||
<dc:title>{title_escaped}</dc:title>
|
||
</item>
|
||
</DIDL-Lite>"""
|
||
|
||
# 转义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"""<u:SetAVTransportURI xmlns:u="urn:schemas-upnp-org:service:AVTransport:1">
|
||
<InstanceID>0</InstanceID>
|
||
<CurrentURI>{media_url_escaped}</CurrentURI>
|
||
<CurrentURIMetaData>{metadata_escaped}</CurrentURIMetaData>
|
||
</u:SetAVTransportURI>"""
|
||
|
||
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"媒体播放地址：{media_url_escaped}")
|
||
self.logger.info(f"DLNA控制响应：{response}")
|
||
|
||
return success
|
||
|
||
def play(self, speed="1"):
|
||
"""
|
||
开始播放
|
||
|
||
Args:
|
||
speed: 播放速度，默认为1
|
||
|
||
Returns:
|
||
bool: 是否成功
|
||
"""
|
||
if not self.av_transport_url:
|
||
return False
|
||
|
||
soap_body = f"""<u:Play xmlns:u="urn:schemas-upnp-org:service:AVTransport:1">
|
||
<InstanceID>0</InstanceID>
|
||
<Speed>{speed}</Speed>
|
||
</u:Play>"""
|
||
|
||
action = "urn:schemas-upnp-org:service:AVTransport:1#Play"
|
||
success, response = self._send_soap_request(self.av_transport_url, action, soap_body)
|
||
|
||
return success
|
||
|
||
def pause(self):
|
||
"""
|
||
暂停播放
|
||
|
||
Returns:
|
||
bool: 是否成功
|
||
"""
|
||
if not self.av_transport_url:
|
||
return False
|
||
|
||
soap_body = """<u:Pause xmlns:u="urn:schemas-upnp-org:service:AVTransport:1">
|
||
<InstanceID>0</InstanceID>
|
||
</u:Pause>"""
|
||
|
||
action = "urn:schemas-upnp-org:service:AVTransport:1#Pause"
|
||
success, response = self._send_soap_request(self.av_transport_url, action, soap_body)
|
||
|
||
return success
|
||
|
||
def stop(self):
|
||
"""
|
||
停止播放
|
||
|
||
Returns:
|
||
bool: 是否成功
|
||
"""
|
||
if not self.av_transport_url:
|
||
return False
|
||
|
||
soap_body = """<u:Stop xmlns:u="urn:schemas-upnp-org:service:AVTransport:1">
|
||
<InstanceID>0</InstanceID>
|
||
</u:Stop>"""
|
||
|
||
action = "urn:schemas-upnp-org:service:AVTransport:1#Stop"
|
||
success, response = self._send_soap_request(self.av_transport_url, action, soap_body)
|
||
|
||
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"""<u:Seek xmlns:u="urn:schemas-upnp-org:service:AVTransport:1">
|
||
<InstanceID>0</InstanceID>
|
||
<Unit>{unit}</Unit>
|
||
<Target>{target_escaped}</Target>
|
||
</u:Seek>"""
|
||
|
||
action = "urn:schemas-upnp-org:service:AVTransport:1#Seek"
|
||
success, response = self._send_soap_request(self.av_transport_url, action, soap_body)
|
||
|
||
return success
|
||
|
||
def get_transport_info(self):
|
||
"""
|
||
获取传输状态信息
|
||
|
||
Returns:
|
||
dict: 传输状态信息，包含:
|
||
- state: 当前传输状态
|
||
- status: 当前传输状态
|
||
- speed: 当前播放速度
|
||
None: 获取失败
|
||
"""
|
||
if not self.av_transport_url:
|
||
return None
|
||
|
||
soap_body = """<u:GetTransportInfo xmlns:u="urn:schemas-upnp-org:service:AVTransport:1">
|
||
<InstanceID>0</InstanceID>
|
||
</u:GetTransportInfo>"""
|
||
|
||
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 = """<u:GetPositionInfo xmlns:u="urn:schemas-upnp-org:service:AVTransport:1">
|
||
<InstanceID>0</InstanceID>
|
||
</u:GetPositionInfo>"""
|
||
|
||
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"""<u:SetVolume xmlns:u="urn:schemas-upnp-org:service:RenderingControl:1">
|
||
<InstanceID>0</InstanceID>
|
||
<Channel>{channel}</Channel>
|
||
<DesiredVolume>{volume}</DesiredVolume>
|
||
</u:SetVolume>"""
|
||
|
||
action = "urn:schemas-upnp-org:service:RenderingControl:1#SetVolume"
|
||
success, response = self._send_soap_request(self.rendering_control_url, action, soap_body)
|
||
|
||
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"""<u:GetVolume xmlns:u="urn:schemas-upnp-org:service:RenderingControl:1">
|
||
<InstanceID>0</InstanceID>
|
||
<Channel>{channel}</Channel>
|
||
</u:GetVolume>"""
|
||
|
||
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"""<u:SetMute xmlns:u="urn:schemas-upnp-org:service:RenderingControl:1">
|
||
<InstanceID>0</InstanceID>
|
||
<Channel>{channel}</Channel>
|
||
<DesiredMute>{desired_mute}</DesiredMute>
|
||
</u:SetMute>"""
|
||
|
||
action = "urn:schemas-upnp-org:service:RenderingControl:1#SetMute"
|
||
success, response = self._send_soap_request(self.rendering_control_url, action, soap_body)
|
||
|
||
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"""<u:GetMute xmlns:u="urn:schemas-upnp-org:service:RenderingControl:1">
|
||
<InstanceID>0</InstanceID>
|
||
<Channel>{channel}</Channel>
|
||
</u:GetMute>"""
|
||
|
||
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 = """<u:GetProtocolInfo xmlns:u="urn:schemas-upnp-org:service:ConnectionManager:1">
|
||
</u:GetProtocolInfo>"""
|
||
|
||
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
|
||
|