diff --git a/app/modules/plex/plex.py b/app/modules/plex/plex.py index b7bdcf688..2f1b1a195 100644 --- a/app/modules/plex/plex.py +++ b/app/modules/plex/plex.py @@ -4,6 +4,7 @@ from pathlib import Path from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Union from urllib.parse import quote_plus +from plexapi.base import PlexPartialObject from plexapi.exceptions import NotFound from plexapi.myplex import MyPlexAccount from plexapi.server import PlexServer @@ -26,6 +27,9 @@ from app.schemas.types import MediaSource, MediaType PLEX_LIBRARY_SYNC_PAGE_SIZE = 50 PLEX_SYNC_TIMEOUT_MAX_ATTEMPTS = 3 PLEX_SYNC_TIMEOUT_RETRY_DELAY_SECONDS = 1 +PLEX_SYNC_EXCLUDE_ELEMENTS = ( + "Genre,Country,Rating,Collection,Director,Writer,Role,Producer,Similar,Style,Mood,Format" +) class Plex: @@ -708,7 +712,7 @@ class Plex: def get_items(self, parent: Union[str, int], start_index: Optional[int] = 0, limit: Optional[int] = -1) \ -> Generator[MediaServerItem | None, Any, None]: """ - 分页获取媒体服务器项目列表;超时会有限重试,重试耗尽时中止同步。 + 分页读取包含媒体路径和 Guid 的列表;读取失败时中止同步以保护旧缓存。 :param parent: 媒体库ID,用于标识要获取的媒体库 :param start_index: 起始索引,用于分页获取数据。默认为 0,即从第一个项目开始获取 @@ -716,8 +720,10 @@ class Plex: :return: 返回一个生成器对象,用于逐步获取媒体服务器中的项目 """ - if not parent or not self._plex: - return None + if not parent: + raise ValueError("Plex 媒体库 ID 不能为空") + if not self._plex: + raise ConnectionError("Plex 服务器未连接") plex = self._plex try: section = self._retry_timeout( @@ -725,7 +731,7 @@ class Plex: f"获取媒体库 {parent}", ) if not section: - return None + raise ValueError(f"Plex 媒体库 {parent} 不存在") if limit is None or limit == -1: fetch_all = True @@ -736,12 +742,19 @@ class Plex: return None page_size = limit page_start = max(start_index or 0, 0) + # section.all() 的默认列表会排除 Media;读取 locations 会让局部对象逐条请求详情。 + search_key = section._buildSearchKey(libtype=section.TYPE) while True: items = self._retry_timeout( - lambda: section.all( + lambda: section.fetchItems( + search_key, container_start=page_start, container_size=page_size, maxresults=page_size, + params={ + "excludeElements": PLEX_SYNC_EXCLUDE_ELEMENTS, + "skipRefresh": 1, + }, ), f"读取媒体库 {parent} 的第 {page_start} 项分页", ) @@ -751,16 +764,17 @@ class Plex: for item in items: if not item: continue + if isinstance(item, PlexPartialObject): + # 缺失的可选字段不得触发详情重载或 Plex 对网盘文件的重新读取。 + item._autoReload = False try: media_item = self._retry_timeout( lambda: self.__build_media_server_item(item), f"处理媒体项目 {getattr(item, 'key', '')}", ) except Exception as err: - if self._is_timeout_exception(err): - raise - logger.error(f"处理媒体项目时出错:{str(err)}, 跳过此项目") - continue + logger.error(f"处理媒体项目时出错,同步中止:{str(err)}") + raise if media_item is not None: yield media_item @@ -768,10 +782,8 @@ class Plex: break page_start += len(items) except Exception as err: - if self._is_timeout_exception(err): - raise - logger.error(f"获取媒体库列表出错:{str(err)}") - return None + logger.error(f"同步读取 Plex 媒体库 {parent} 出错:{str(err)}") + raise @staticmethod def _is_timeout_exception(error: Exception) -> bool: diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index c5a14eff6..d08bff1b8 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -704,11 +704,12 @@ }, { "bindings": [ + "from:plexapi.base.PlexPartialObject as PlexPartialObject", "from:plexapi.exceptions.NotFound as NotFound", "from:plexapi.myplex.MyPlexAccount as MyPlexAccount", "from:plexapi.server.PlexServer as PlexServer" ], - "fingerprint": "76c1334863dc6c6623ce6ad3415bec5ff92656f51c31e4d982285898f193e57e", + "fingerprint": "55baa4290bb62468d504e954add3921c0068f3a3c77aa3db92667ae9022909ca", "kind": "network_sdk", "source": "app.modules.plex.plex", "target": "plexapi", diff --git a/tests/fixtures/architecture/dependency-policy.json b/tests/fixtures/architecture/dependency-policy.json index 278d2758c..2e4715ab2 100644 --- a/tests/fixtures/architecture/dependency-policy.json +++ b/tests/fixtures/architecture/dependency-policy.json @@ -143,7 +143,7 @@ {"source": "app.modules.feishu.feishu", "target": "lark_oapi", "kind": "network_sdk", "fingerprint": "99d16968a59932b5980f6d42ea2bbc8fa35ccdb1a27933beec29b4bdc093d78b"}, {"source": "app.modules.filemanager.storages.smb", "target": "smbclient", "kind": "network_sdk", "fingerprint": "62e0585282ef206ac81b2bf9e93a58423fb058685edc0ef4ee592453bbf30376"}, {"source": "app.modules.filemanager.storages.u115", "target": "oss2", "kind": "network_sdk", "fingerprint": "f7b89c8ae6dad2603f0a9e0caaa159769aef5b3581d7e728f62445b979366eae"}, - {"source": "app.modules.plex.plex", "target": "plexapi", "kind": "network_sdk", "fingerprint": "76c1334863dc6c6623ce6ad3415bec5ff92656f51c31e4d982285898f193e57e"}, + {"source": "app.modules.plex.plex", "target": "plexapi", "kind": "network_sdk", "fingerprint": "55baa4290bb62468d504e954add3921c0068f3a3c77aa3db92667ae9022909ca"}, {"source": "app.modules.qbittorrent.qbittorrent", "target": "qbittorrentapi", "kind": "network_sdk", "fingerprint": "b2f5a27f0c54cf95ed42fe99848fd6c9c0641caca8bc4ff8006a411cb427066d"}, {"source": "app.modules.slack.slack", "target": "slack_bolt", "kind": "network_sdk", "fingerprint": "be5dcb032ece8d8627abeb243f98143aaf60f26751ab6e5e098813d4048419e6"}, {"source": "app.modules.slack.slack", "target": "slack_sdk", "kind": "network_sdk", "fingerprint": "7559f31e4172ad3bbbaf161e1164ea48b997c56d05c77b49b82626298c39aa14"}, diff --git a/tests/test_architecture_egress.py b/tests/test_architecture_egress.py index 23241f518..8616d7f8b 100644 --- a/tests/test_architecture_egress.py +++ b/tests/test_architecture_egress.py @@ -124,7 +124,7 @@ FROZEN_EGRESS_FINGERPRINT_BY_EDGE = { ("app.modules.filemanager.storages.smb", "smbprotocol"): "f02686afd99c59820dffa7b4c2627a0be1ad62980c98a9cd33b7564697842b22", ("app.modules.filemanager.storages.u115", "oss2"): "f7b89c8ae6dad2603f0a9e0caaa159769aef5b3581d7e728f62445b979366eae", ("app.modules.jellyfin.jellyfin", "requests"): "5c46d09ca9a4bcc0bae21ca5d554ef09baa3f901c562b27c5f5ee1439c7746b5", - ("app.modules.plex.plex", "plexapi"): "76c1334863dc6c6623ce6ad3415bec5ff92656f51c31e4d982285898f193e57e", + ("app.modules.plex.plex", "plexapi"): "55baa4290bb62468d504e954add3921c0068f3a3c77aa3db92667ae9022909ca", ("app.modules.qbittorrent.qbittorrent", "qbittorrentapi"): "b2f5a27f0c54cf95ed42fe99848fd6c9c0641caca8bc4ff8006a411cb427066d", ("app.modules.qqbot.gateway", "websocket"): "6f6b7d61f3a95e620e67c450d544f0aa077f087a188d98f32e4501d94c6b37ae", ("app.modules.rtorrent.rtorrent", "http.client"): "6adc93b3bc479dfe81197554c7977930a60abc51714f14ef29298eb40730aaf7", diff --git a/tests/test_mediaserver_sync_incremental.py b/tests/test_mediaserver_sync_incremental.py index 6b783e235..60e587db4 100644 --- a/tests/test_mediaserver_sync_incremental.py +++ b/tests/test_mediaserver_sync_incremental.py @@ -370,8 +370,9 @@ def test_sync_targets_one_server_without_excluding_other_enabled_servers(monkeyp assert excluded_server_calls == [["plex-a", "plex-b"]] -def test_sync_preserves_cached_rows_when_plex_item_timeout_aborts_library(database): - """Plex 条目读取超时中止同步时,应保留上一轮缓存而不执行陈旧数据清理。""" +@pytest.mark.parametrize("failure", [TimeoutError("read timeout"), ConnectionError("connection refused")]) +def test_sync_preserves_cached_rows_when_plex_read_fails(database, failure): + """Plex 读取中断时,应保留上一轮缓存且不报告同步完成。""" with database() as session: session.add( MediaServerItem( @@ -390,8 +391,8 @@ def test_sync_preserves_cached_rows_when_plex_item_timeout_aborts_library(databa chain.media_count = lambda _server: 1 chain.items_count = lambda **_kwargs: pytest.fail("整服统计存在时不应逐库计数") - def items_with_timeout(**_kwargs): - """模拟 Plex 已返回部分条目后,分页请求最终仍超时。""" + def items_with_failure(**_kwargs): + """模拟 Plex 已返回部分条目后,下一次分页读取失败。""" yield schemas.MediaServerItem( server="plex", library="movies", @@ -399,11 +400,12 @@ def test_sync_preserves_cached_rows_when_plex_item_timeout_aborts_library(databa item_type="Movie", title="已读取媒体", ) - raise TimeoutError("Plex read timeout") + raise failure - chain.items = items_with_timeout + chain.items = items_with_failure chain.episodes = lambda *_args, **_kwargs: [] chain.media_server_repository = TransactionalMediaServerRepository(database) + progress = [] with ( patch.object( @@ -411,9 +413,9 @@ def test_sync_preserves_cached_rows_when_plex_item_timeout_aborts_library(databa "get_mediaserver_configs", return_value=[SimpleNamespace(name="plex", enabled=True, sync_libraries=["all"])], ), - pytest.raises(TimeoutError, match="Plex read timeout"), + pytest.raises(type(failure), match=str(failure)), ): - chain.sync() + chain.sync(progress_callback=lambda **kwargs: progress.append(kwargs)) with database() as session: item_ids = { @@ -422,6 +424,7 @@ def test_sync_preserves_cached_rows_when_plex_item_timeout_aborts_library(databa } assert item_ids == {"not-reached-before-timeout", "received-before-timeout"} + assert not any("同步完成" in entry.get("text", "") for entry in progress) def test_sync_partial_commit_preserves_stale_rows_until_next_run( diff --git a/tests/test_plex_sync_timeout.py b/tests/test_plex_sync_timeout.py index 14b752303..185afda62 100644 --- a/tests/test_plex_sync_timeout.py +++ b/tests/test_plex_sync_timeout.py @@ -1,7 +1,10 @@ from types import SimpleNamespace from unittest.mock import Mock, call, patch +from xml.etree.ElementTree import fromstring import pytest +from plexapi.library import MovieSection +from plexapi.video import Movie # pylint: disable=unused-import from app.modules.plex import plex as plex_module from app.modules.plex.plex import Plex @@ -11,13 +14,16 @@ class _ReadTimeout(Exception): """模拟 PlexAPI 底层 HTTP 客户端的 ReadTimeout。""" -def _make_client(section: Mock) -> Plex: +def _make_client(section: Mock | MovieSection) -> Plex: """构造不建立真实网络连接的 Plex 客户端。""" client = Plex.__new__(Plex) client._plex = SimpleNamespace( library=SimpleNamespace(sectionByID=Mock(return_value=section)) ) client._Plex__build_media_server_item = lambda item: item + if isinstance(section, Mock): + section.TYPE = "movie" + section._buildSearchKey.return_value = "/library/sections/1/all?includeGuids=1&type=1" return client @@ -76,30 +82,42 @@ def test_plex_sync_reads_full_library_in_pages_and_retries_timeouts(monkeypatch) monkeypatch.setattr(plex_module.time, "sleep", lambda _seconds: None) items = [SimpleNamespace(key=f"item-{index}") for index in range(3)] section = Mock() - section.all.side_effect = [_ReadTimeout("read timeout"), items[:2], items[2:]] + section.fetchItems.side_effect = [_ReadTimeout("read timeout"), items[:2], items[2:]] client = _make_client(section) result = list(client.get_items("1")) assert result == items - assert section.all.call_args_list == [ - call(container_start=0, container_size=2, maxresults=2), - call(container_start=0, container_size=2, maxresults=2), - call(container_start=2, container_size=2, maxresults=2), + options = dict(params={ + "excludeElements": plex_module.PLEX_SYNC_EXCLUDE_ELEMENTS, + "skipRefresh": 1, + }) + assert section.fetchItems.call_args_list == [ + call(section._buildSearchKey.return_value, container_start=0, container_size=2, + maxresults=2, **options), + call(section._buildSearchKey.return_value, container_start=0, container_size=2, + maxresults=2, **options), + call(section._buildSearchKey.return_value, container_start=2, container_size=2, + maxresults=2, **options), ] + section._buildSearchKey.assert_called_once_with(libtype="movie") def test_plex_sync_honors_explicit_page_start_and_limit(): """显式分页调用应只读取请求的起点和条目数。""" items = [SimpleNamespace(key=f"item-{index}") for index in range(2)] section = Mock() - section.all.return_value = items + section.fetchItems.return_value = items client = _make_client(section) result = list(client.get_items("1", start_index=8, limit=2)) assert result == items - section.all.assert_called_once_with(container_start=8, container_size=2, maxresults=2) + section.fetchItems.assert_called_once_with( + section._buildSearchKey.return_value, + container_start=8, container_size=2, maxresults=2, + params={"excludeElements": plex_module.PLEX_SYNC_EXCLUDE_ELEMENTS, "skipRefresh": 1}, + ) def test_plex_sync_retries_item_conversion_timeout(monkeypatch): @@ -108,7 +126,7 @@ def test_plex_sync_retries_item_conversion_timeout(monkeypatch): monkeypatch.setattr(plex_module.time, "sleep", lambda _seconds: None) item = SimpleNamespace(key="item-1") section = Mock() - section.all.return_value = [item] + section.fetchItems.return_value = [item] client = _make_client(section) convert = Mock(side_effect=[_ReadTimeout("read timeout"), "converted-item"]) monkeypatch.setattr(client, "_Plex__build_media_server_item", convert) @@ -123,10 +141,73 @@ def test_plex_sync_raises_after_all_page_timeout_retries(monkeypatch): """分页超时重试耗尽时应中止整库同步,不能静默跳过页面。""" monkeypatch.setattr(plex_module.time, "sleep", lambda _seconds: None) section = Mock() - section.all.side_effect = _ReadTimeout("read timeout") + section.fetchItems.side_effect = _ReadTimeout("read timeout") client = _make_client(section) with pytest.raises(_ReadTimeout, match="read timeout"): list(client.get_items("1")) - assert section.all.call_count == plex_module.PLEX_SYNC_TIMEOUT_MAX_ATTEMPTS + assert section.fetchItems.call_count == plex_module.PLEX_SYNC_TIMEOUT_MAX_ATTEMPTS + + +def test_plex_sync_raises_on_connection_failure_after_partial_page(monkeypatch): + """连接拒绝后不得把已读部分视作完整媒体库。""" + monkeypatch.setattr(plex_module, "PLEX_LIBRARY_SYNC_PAGE_SIZE", 1) + section = Mock() + section.fetchItems.side_effect = [[SimpleNamespace(key="first")], ConnectionError("refused")] + client = _make_client(section) + with pytest.raises(ConnectionError, match="refused"): + list(client.get_items("1")) + + +def test_plex_sync_raises_on_item_conversion_failure(): + """条目转换失败时不能跳过记录后继续执行陈旧清理。""" + section = Mock() + section.fetchItems.return_value = [SimpleNamespace(key="first")] + client = _make_client(section) + client._Plex__build_media_server_item = Mock(side_effect=ConnectionError("refused")) + with pytest.raises(ConnectionError, match="refused"): + list(client.get_items("1")) + + +def test_plex_sync_raises_when_server_or_library_is_unavailable(): + """服务器失联或媒体库消失时不得返回空库并触发旧数据清理。""" + section = Mock() + client = _make_client(section) + client._plex = None + with pytest.raises(ConnectionError, match="未连接"): + list(client.get_items("1")) + + client = _make_client(section) + client._plex.library.sectionByID.return_value = None + with pytest.raises(ValueError, match="不存在"): + list(client.get_items("1")) + + +def test_plex_sync_uses_list_metadata_without_reloading_each_movie(): + """列表中的文件路径和 Guid 足够生成缓存条目,不应发起逐条详情请求。""" + server = Mock() + server.query.return_value = fromstring( + '' + '' + ) + section = MovieSection( + server, fromstring(''), + "/library/sections", + ) + client = _make_client(section) + del client._Plex__build_media_server_item + + result = list(client.get_items("1")) + + assert len(result) == 1 + assert result[0].path == "/cloud/Demo.mkv" + assert str(result[0].media_id) == "123" + server.query.assert_called_once_with( + "/library/sections/1/all?includeGuids=1&type=1", + headers={"X-Plex-Container-Start": "0", "X-Plex-Container-Size": "50"}, + params={"excludeElements": plex_module.PLEX_SYNC_EXCLUDE_ELEMENTS, "skipRefresh": 1}, + )