diff --git a/app/api/endpoints/mediaserver.py b/app/api/endpoints/mediaserver.py index 08653682..a9b01f70 100644 --- a/app/api/endpoints/mediaserver.py +++ b/app/api/endpoints/mediaserver.py @@ -38,7 +38,13 @@ def play_item( if item: play_url = media_chain.get_play_url(server=name, item_id=itemid) if play_url: - return schemas.Response(success=True, data={"url": play_url}) + return schemas.Response( + success=True, + data={ + "url": play_url, + "server_type": item.server, + }, + ) return schemas.Response(success=False, message="未找到播放地址") diff --git a/app/modules/emby/emby.py b/app/modules/emby/emby.py index 06ee68e4..f6095f28 100644 --- a/app/modules/emby/emby.py +++ b/app/modules/emby/emby.py @@ -160,6 +160,7 @@ class Emby: else: library_type = MediaType.UNKNOWN.value image = self.__get_local_image_by_id(library.get("Id")) + server_query = f"serverId={self.serverid}&" if self.serverid else "" libraries.append( schemas.MediaServerLibrary( server="emby", @@ -169,7 +170,7 @@ class Emby: type=library_type, image=image, link=f'{self._playhost or self._host}web/index.html' - f'#!/videos?serverId={self.serverid}&parentId={library.get("Id")}', + f'#!/videos?{server_query}parentId={library.get("Id")}', server_type="emby" ) ) @@ -247,19 +248,22 @@ class Emby: """ if not self._host or not self._apikey: return None - url = f"{self._host}System/Info" params = { 'api_key': self._apikey } - try: - res = RequestUtils().get_res(url, params) - if res: - return res.json().get("Id") - else: - logger.error(f"System/Info 未获取到返回数据") - except Exception as e: - - logger.error(f"连接System/Info出错：" + str(e)) + for path in ("System/Info", "emby/System/Info"): + url = f"{self._host}{path}" + try: + res = RequestUtils().get_res(url, params) + if res: + result = res.json() or {} + server_id = result.get("Id") or result.get("ServerId") + if server_id: + return server_id + else: + logger.error(f"{path} 未获取到返回数据") + except Exception as e: + logger.error(f"连接{path}出错：" + str(e)) return None def get_user_count(self) -> int: @@ -1093,8 +1097,9 @@ class Emby: 拼装媒体播放链接 :param item_id: 媒体的的ID """ + server_query = f"&serverId={self.serverid}" if self.serverid else "" return f"{self._playhost or self._host}web/index.html#!" \ - f"/item?id={item_id}&context=home&serverId={self.serverid}" + f"/item?id={item_id}&context=home{server_query}" def get_backdrop_url(self, item_id: str, image_tag: str, remote: Optional[bool] = False) -> str: """ @@ -1180,6 +1185,8 @@ class Emby: image = self.__get_local_image_by_id(item.get("SeriesId")) ret_resume.append(schemas.MediaServerPlayItem( id=item.get("Id"), + item_id=item.get("Id"), + server_id=self.serverid, title=title, subtitle=subtitle, type=item_type, @@ -1234,6 +1241,8 @@ class Emby: image = self.__get_local_image_by_id(item_id=item.get("Id")) ret_latest.append(schemas.MediaServerPlayItem( id=item.get("Id"), + item_id=item.get("Id"), + server_id=self.serverid, title=item.get("Name"), subtitle=str(item.get("ProductionYear")) if item.get("ProductionYear") else None, type=item_type, diff --git a/app/schemas/mediaserver.py b/app/schemas/mediaserver.py index d99a13bd..dd5daf66 100644 --- a/app/schemas/mediaserver.py +++ b/app/schemas/mediaserver.py @@ -171,6 +171,8 @@ class MediaServerPlayItem(BaseModel): 媒体服务器可播放项目信息 """ id: Optional[Union[str, int]] = None + item_id: Optional[Union[str, int]] = None + server_id: Optional[str] = None title: Optional[str] = None subtitle: Optional[str] = None type: Optional[str] = None diff --git a/tests/test_emby_dashboard_links.py b/tests/test_emby_dashboard_links.py new file mode 100644 index 00000000..367e55ed --- /dev/null +++ b/tests/test_emby_dashboard_links.py @@ -0,0 +1,118 @@ +import unittest +from typing import Any +from unittest.mock import Mock, patch + +from app import schemas +from app.api.endpoints.mediaserver import play_item +from app.modules.emby.emby import Emby + + +class _FakeResponse: + """提供 Emby 接口响应的最小 json 封装。""" + + def __init__(self, payload: Any): + """保存测试预置的响应体。""" + self._payload = payload + + def json(self) -> Any: + """返回测试预置的响应体。""" + return self._payload + + +class EmbyDashboardLinksTest(unittest.TestCase): + """验证 Emby 仪表盘条目使用真实媒体服务器标识生成跳转链接。""" + + @staticmethod + def _build_client() -> Emby: + """构造绕过真实初始化的 Emby 实例。""" + client = Emby.__new__(Emby) + client._host = "http://emby.local/" + client._playhost = None + client._apikey = "api-key" + client._sync_libraries = [] + client.user = "user-id" + client.serverid = "server-id" + return client + + def test_get_server_id_falls_back_to_emby_prefixed_system_info(self): + """ + 兼容 Emby 反代只暴露 /emby/System/Info 的场景，避免生成 serverId=None。 + """ + client = self._build_client() + client.serverid = None + + with patch("app.modules.emby.emby.RequestUtils") as request_utils_cls: + request_utils_cls.return_value.get_res.side_effect = [ + None, + _FakeResponse({"Id": "server-id"}), + ] + + server_id = client.get_server_id() + + self.assertEqual(server_id, "server-id") + self.assertEqual( + request_utils_cls.return_value.get_res.call_args_list[0].args[0], + "http://emby.local/System/Info", + ) + self.assertEqual( + request_utils_cls.return_value.get_res.call_args_list[1].args[0], + "http://emby.local/emby/System/Info", + ) + + def test_get_play_url_omits_missing_server_id(self): + """serverId 为空时不应把 None 字符串拼入播放链接。""" + client = self._build_client() + client.serverid = None + + play_url = client.get_play_url("item-id") + + self.assertEqual( + play_url, + "http://emby.local/web/index.html#!/item?id=item-id&context=home", + ) + + def test_get_latest_returns_item_and_server_ids(self): + """最近入库条目需要显式返回 Emby item_id 和 server_id 供前端纠偏链接。""" + client = self._build_client() + client.get_user_library_folders = Mock(return_value=[]) + + with patch("app.modules.emby.emby.RequestUtils") as request_utils_cls: + request_utils_cls.return_value.get_res.return_value = _FakeResponse([ + { + "Id": "emby-item-id", + "Name": "测试电影", + "Type": "Movie", + "ProductionYear": 2026, + } + ]) + + items = client.get_latest() + + self.assertEqual(items[0].id, "emby-item-id") + self.assertEqual(items[0].item_id, "emby-item-id") + self.assertEqual(items[0].server_id, "server-id") + self.assertIn("id=emby-item-id", items[0].link) + self.assertIn("serverId=server-id", items[0].link) + + def test_play_item_returns_server_type(self): + """播放地址接口需要返回 server_type，供前端跳转时选择正确媒体服务器类型。""" + item = schemas.MediaServerItem(server="emby", item_id="emby-item-id") + + with ( + patch("app.api.endpoints.mediaserver.MediaServerHelper") as helper_cls, + patch("app.api.endpoints.mediaserver.MediaServerChain") as chain_cls, + ): + helper_cls.return_value.get_configs.return_value = {"Emby": object()} + chain = chain_cls.return_value + chain.iteminfo.return_value = item + chain.get_play_url.return_value = "http://emby.local/web/index.html#!/item?id=emby-item-id" + + response = play_item("emby-item-id") + + self.assertTrue(response.success) + self.assertEqual(response.data["url"], "http://emby.local/web/index.html#!/item?id=emby-item-id") + self.assertEqual(response.data["server_type"], "emby") + + +if __name__ == "__main__": + unittest.main()