diff --git a/app/application/mediaserver.py b/app/application/mediaserver.py index 056f447b5..417bdca3e 100644 --- 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, diff --git a/app/chain/mediaserver.py b/app/chain/mediaserver.py index a2a6d1e6f..19e0807ba 100644 --- 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: diff --git a/app/modules/emby/emby.py b/app/modules/emby/emby.py index db5fb5e1a..35c4ce32e 100644 --- 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]: """ diff --git a/app/modules/jellyfin/jellyfin.py b/app/modules/jellyfin/jellyfin.py index eea991500..cb7df87fa 100644 --- 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]: """ diff --git a/tests/test_emby_family_sync_items.py b/tests/test_emby_family_sync_items.py new file mode 100644 index 000000000..2996bd079 --- /dev/null +++ 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")) diff --git a/tests/test_mediaserver_sync_incremental.py b/tests/test_mediaserver_sync_incremental.py index 1b25cd30c..229297957 100644 --- 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,