fix(mediaserver): 修复媒体库同步清空本地缓存 (#7033)

2026-11-01 12:56:34 +08:00 · 2026-10-16 22:11:23 +08:00
parent 6b50942ebc
commit 3a88d6d57b
6 changed files with 151 additions and 11 deletions
--- a/app/application/mediaserver.py
+++ b/app/application/mediaserver.py
@@ -46,10 +46,16 @@ class MediaServerSyncItem:
        item_type: Optional[str],
        seasoninfo: Mapping[int, Optional[list[int]]],
        sync_time: str,
+        server: Optional[str] = None,
    ) -> "MediaServerSyncItem":
-        """冻结远端媒体条目和本轮剧集快照，供短事务持久化。"""
+        """
+        冻结远端媒体条目和本轮剧集快照，供短事务持久化。
+
+        :param server: 媒体服务器配置名称；模块条目只携带 emby/jellyfin 等类型名，
+            陈旧清理、排除清理与存在性检查均按配置名称隔离缓存，同步写入必须使用同一键。
+        """
        return cls(
-            server=str(item.server or ""),
+            server=str(server or item.server or ""),
            library=str(item.library) if item.library is not None else None,
            item_id=str(item.item_id or ""),
            item_type=item_type,
--- a/app/chain/mediaserver.py
+++ b/app/chain/mediaserver.py
@@ -375,6 +375,7 @@ class MediaServerChain(ChainBase):
                        item_type=item_type,
                        seasoninfo=seasoninfo,
                        sync_time=sync_time,
+                        server=server_name,
                    )
                )
                if progress_callback:
--- a/app/modules/emby/emby.py
+++ b/app/modules/emby/emby.py
@@ -812,9 +812,13 @@ class Emby:
        :param limit: 每次请求的最大项目数，用于分页。如果为 None 或 -1，则表示一次性获取所有数据，默认为 -1

        :return: 返回一个生成器对象，用于逐步获取媒体服务器中的项目
+
+        读取失败时抛出异常中止同步，避免空结果被当作媒体库已清空而删除旧缓存。
        """
-        if not parent or not self._host or not self._apikey:
-            return None
+        if not parent:
+            raise ValueError("Emby 媒体库 ID 不能为空")
+        if not self._host or not self._apikey:
+            raise ConnectionError("Emby 服务器未连接")
        url = f"{self._host}emby/Users/{self.user}/Items"
        params = {
            "ParentId": parent,
@@ -831,7 +835,9 @@ class Emby:
        try:
            res = RequestUtils().get_res(url, params)
            if not res or res.status_code != 200:
-                return None
+                raise ConnectionError(
+                    f"读取 Emby 媒体库 {parent} 失败，状态码：{res.status_code if res is not None else '无响应'}"
+                )
            items = res.json().get("Items") or []
            for item in items:
                if not item:
@@ -842,8 +848,8 @@ class Emby:
                elif item.get("Type") in ["Movie", "Series", "MusicAlbum"]:
                    yield self.__format_item_info(item)
        except Exception as e:
-            logger.error("连接Users/Items出错：" + str(e))
-        return None
+            logger.error(f"同步读取 Emby 媒体库 {parent} 出错：{str(e)}")
+            raise

    def get_webhook_message(self, form: Any, args: dict) -> Optional[_SchemaWebhookEventInfo]:
        """
--- a/app/modules/jellyfin/jellyfin.py
+++ b/app/modules/jellyfin/jellyfin.py
@@ -987,9 +987,13 @@ class Jellyfin:
        :param limit: 每次请求的最大项目数，用于分页。如果为 None 或 -1，则表示一次性获取所有数据，默认为 -1

        :return: 返回一个生成器对象，用于逐步获取媒体服务器中的项目
+
+        读取失败时抛出异常中止同步，避免空结果被当作媒体库已清空而删除旧缓存。
        """
-        if not parent or not self._host or not self._apikey:
-            return None
+        if not parent:
+            raise ValueError("Jellyfin 媒体库 ID 不能为空")
+        if not self._host or not self._apikey:
+            raise ConnectionError("Jellyfin 服务器未连接")
        url = f"{self._host}Users/{self.user}/Items"
        params = {
            "ParentId": parent,
@@ -1006,7 +1010,9 @@ class Jellyfin:
        try:
            res = self._request().get_res(url, params)
            if not res or res.status_code != 200:
-                return None
+                raise ConnectionError(
+                    f"读取 Jellyfin 媒体库 {parent} 失败，状态码：{res.status_code if res is not None else '无响应'}"
+                )
            items = res.json().get("Items") or []
            for item in items:
                if not item:
@@ -1017,7 +1023,8 @@ class Jellyfin:
                elif item.get("Type") in ["Movie", "Series", "MusicAlbum"]:
                    yield self.__format_item_info(item)
        except Exception as e:
-            logger.error("连接Users/Items出错：" + str(e))
+            logger.error(f"同步读取 Jellyfin 媒体库 {parent} 出错：{str(e)}")
+            raise

    def get_data(self, url: str) -> Optional[Response]:
        """
--- a/tests/test_emby_family_sync_items.py
+++ b/tests/test_emby_family_sync_items.py
@@ -0,0 +1,57 @@
+"""Emby / Jellyfin 媒体库同步读取失败时保护旧缓存的回归。"""
+
+from unittest.mock import Mock, patch
+
+import pytest
+
+from app.modules.emby.emby import Emby
+from app.modules.jellyfin.jellyfin import Jellyfin
+
+_REQUEST_TARGETS = {
+    Emby: "app.modules.emby.emby.RequestUtils",
+    Jellyfin: "app.modules.jellyfin.jellyfin.RequestUtils",
+}
+
+
+def _client(client_class: type):
+    """构造跳过远端初始化的媒体服务器客户端。"""
+    instance = client_class.__new__(client_class)
+    instance._host = "http://mediaserver.local/"
+    instance._apikey = "api-key"
+    instance._playhost = None
+    instance._sync_libraries = []
+    instance.user = "user-id"
+    return instance
+
+
+@pytest.mark.parametrize("client_class", [Emby, Jellyfin])
+@pytest.mark.parametrize("response", [None, Mock(status_code=500)])
+def test_get_items_raises_when_library_read_fails(client_class: type, response):
+    """列表请求失败必须中止同步，不能产出空结果让同步链删除该服务器的旧缓存。"""
+    client = _client(client_class)
+    with patch(_REQUEST_TARGETS[client_class]) as request:
+        request.return_value.get_res.return_value = response
+        with pytest.raises(ConnectionError):
+            list(client.get_items("movies"))
+
+
+@pytest.mark.parametrize("client_class", [Emby, Jellyfin])
+def test_get_items_raises_when_nested_folder_read_fails(client_class: type):
+    """子文件夹读取失败同样中止同步，避免只同步到部分条目。"""
+    client = _client(client_class)
+    folder = Mock(status_code=200)
+    folder.json.return_value = {"Items": [{"Id": "folder-1", "Type": "Folder"}]}
+    with patch(_REQUEST_TARGETS[client_class]) as request:
+        request.return_value.get_res.side_effect = [folder, TimeoutError("read timeout")]
+        with pytest.raises(TimeoutError, match="read timeout"):
+            list(client.get_items("movies"))
+
+
+@pytest.mark.parametrize("client_class", [Emby, Jellyfin])
+def test_get_items_raises_when_server_is_not_connected(client_class: type):
+    """未连接的服务器不得被同步为空媒体库。"""
+    client = _client(client_class)
+    client._apikey = None
+
+    with pytest.raises(ConnectionError):
+        list(client.get_items("movies"))
--- a/tests/test_mediaserver_sync_incremental.py
+++ b/tests/test_mediaserver_sync_incremental.py
@@ -430,6 +430,69 @@ def test_sync_preserves_cached_rows_when_plex_read_fails(database, failure):
    assert not any("同步完成" in entry.get("text", "") for entry in progress)


+# 前端新增媒体服务器的默认名称为「服务器N」，与模块条目携带的类型名不同
+_CUSTOM_NAMED_JELLYFIN = [SimpleNamespace(name="服务器1", enabled=True, sync_libraries=["all"])]
+
+
+def _custom_named_jellyfin_chain(database, items):
+    """构造配置名称与模块类型名不同的 Jellyfin 同步链。"""
+    chain = object.__new__(MediaServerChain)
+    chain.librarys = lambda _server: [SimpleNamespace(id="movies", name="电影库")]
+    chain.media_count = lambda _server: 1
+    chain.items_count = lambda **_kwargs: pytest.fail("整服统计存在时不应逐库计数")
+    chain.items = items
+    chain.episodes = lambda *_args, **_kwargs: []
+    chain.media_server_repository = TransactionalMediaServerRepository(database)
+    return chain
+
+
+def _jellyfin_movie_items(**_kwargs):
+    """模拟 Jellyfin 模块返回的条目，server 字段为类型名而非配置名称。"""
+    yield schemas.MediaServerItem(
+        server="jellyfin",
+        library="movies",
+        item_id="movie-1",
+        item_type="Movie",
+        title="已入库电影",
+        year="2026",
+    )
+
+
+def test_sync_stores_rows_under_config_name_for_existence_lookup(database):
+    """缓存必须按配置名称写入，存在性检查按配置名称查询才能命中。"""
+    chain = _custom_named_jellyfin_chain(database, _jellyfin_movie_items)
+
+    with patch.object(MEDIA_SERVER_CHAIN_MODULE, "get_mediaserver_configs", return_value=_CUSTOM_NAMED_JELLYFIN):
+        chain.sync()
+
+    with database() as session:
+        servers = {item.server for item in session.query(MediaServerItem).all()}
+    assert servers == {"服务器1"}
+    assert chain.media_server_repository.get_item_id(
+        server="服务器1", title="已入库电影", year="2026", mtype="电影",
+    ) == "movie-1"
+
+
+def test_sync_keeps_custom_named_server_cache_when_read_fails(database):
+    """配置名称与类型名不同时，后续读取失败也不得清空上一轮缓存。"""
+    chain = _custom_named_jellyfin_chain(database, _jellyfin_movie_items)
+
+    def items_with_timeout(**_kwargs):
+        """模拟条目列表请求超时。"""
+        yield from ()
+        raise TimeoutError("read timeout")
+
+    with patch.object(MEDIA_SERVER_CHAIN_MODULE, "get_mediaserver_configs", return_value=_CUSTOM_NAMED_JELLYFIN):
+        chain.sync()
+        chain.items = items_with_timeout
+        with pytest.raises(TimeoutError, match="read timeout"):
+            chain.sync()
+
+    with database() as session:
+        rows = [(item.server, item.item_id) for item in session.query(MediaServerItem).all()]
+    assert rows == [("服务器1", "movie-1")]
+
+
 def test_sync_partial_commit_preserves_stale_rows_until_next_run(
    database,
    monkeypatch,