nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-10-29 11:27:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: avoid Plex sync detail reloads and preserve cache on read failure

Browse Source
This commit is contained in:
jxxghp
2026-09-28 06:57:38 +08:00
parent 4f730418c1
commit e2f925af71
6 changed files with 132 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
app/modules/plex/plex.py
View File

@@ -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:

3
tests/fixtures/architecture/dependency-baseline.json vendored
View File

@@ -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",

2
tests/fixtures/architecture/dependency-policy.json vendored
View File

@@ -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"},

2
tests/test_architecture_egress.py
View File

@@ -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",

19
tests/test_mediaserver_sync_incremental.py
View File

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

103
tests/test_plex_sync_timeout.py
View File

@@ -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(
'<MediaContainer size="1" totalSize="1" librarySectionID="1">'
'<Video type="movie" key="/library/metadata/1" ratingKey="1" '
'title="Demo" year="2026">'
'<Media><Part file="/cloud/Demo.mkv"/></Media>'
'<Guid id="tmdb://123"/></Video></MediaContainer>'
)
section = MovieSection(
server, fromstring('<Directory key="1" type="movie" title="Movies"/>'),
"/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},
)