fix: correct Emby dashboard links

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="未找到播放地址")
 
 

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,

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

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()