nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-11-05 06:39:33 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: preserve media library cache on incomplete enumeration (#7145)

Browse Source
This commit is contained in:
Aqr-K
2026-10-23 06:53:14 +08:00
committed by GitHub
parent f3c1449c8b
commit d666ae9d86
17 changed files with 580 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
app/application/mediaserver.py
View File

@@ -516,15 +516,12 @@ def format_emby_family_item(
playback_position_ticks = user_data.get("PlaybackPositionTicks")
resume = playback_position_ticks and playback_position_ticks > 0
last_played_date = user_data.get("LastPlayedDate")
if last_played_date is not None and "." in last_played_date:
last_played_date = last_played_date.split(".")[0]
user_state = _SchemaMediaServerItemUserState(
played=user_data.get("Played"),
resume=resume,
last_played_date=datetime.strptime(
last_played_date,
"%Y-%m-%dT%H:%M:%S",
).strftime("%Y-%m-%d %H:%M:%S") if last_played_date else None,
last_played_date=datetime.fromisoformat(last_played_date).strftime(
"%Y-%m-%d %H:%M:%S"
) if last_played_date else None,
play_count=user_data.get("PlayCount"),
percentage=user_data.get("PlayedPercentage"),
)

14
app/chain/download/existence.py
View File

@@ -161,12 +161,14 @@ class DownloadExistenceOwner(_DownloadOwnerBase):
query: dict[str, Any],
) -> Optional[ExistMediaInfo]:
"""
按媒体服务器隔离缓存条目，再执行存在性检查。
按媒体服务器隔离缓存条目，并合并各服务器已确认存在的剧集。
"""
server_names = [str(config.name) for config in get_mediaserver_configs() if config.name]
if not server_names:
itemid = self.media_server_repository.get_item_id(**query)
return self.media_exists(mediainfo=mediainfo, itemid=itemid, query=MEDIA_SERVER_EXISTENCE_QUERY)
combined: Optional[ExistMediaInfo] = None
seasons: dict[int, set[int]] = {}
for server in server_names:
itemid = self.media_server_repository.get_item_id(
server=server,
@@ -176,8 +178,14 @@ class DownloadExistenceOwner(_DownloadOwnerBase):
mediainfo=mediainfo, itemid=itemid, server=server, query=MEDIA_SERVER_EXISTENCE_QUERY
)
if exists:
return exists
return None
if mediainfo.type != MediaType.TV:
return exists
combined = combined or exists
for season, episodes in (exists.seasons or {}).items():
seasons.setdefault(season, set()).update(episodes)
return combined.model_copy(update={
"seasons": {season: sorted(episodes) for season, episodes in seasons.items()},
}) if combined else None
@staticmethod
def _append_no_exists(

17
app/modules/_base/mediaserver.py
View File

@@ -5,7 +5,7 @@
定时重连与连接测试。服务器差异（认证 API、存在性检查端点、连接探测方式）
通过类属性与钩子方法保留在各模块。
"""
from typing import Any, Iterable, Optional, Tuple, cast
from typing import Any, Iterable, Mapping, Optional, Tuple, cast
from app.application.mediaserver import MusicMediaServerHelper
from app.domain.context import MediaInfo, MusicInfo
@@ -19,6 +19,21 @@ from app.schemas.mediaserver import MediaServerField, MediaServerQueryOptions
from app.schemas.types import ChainEventType, MediaType
def emby_episode_numbers(item: Mapping[str, Any]) -> range:
"""
展开 Emby 系分集条目的起止集号，缺少结束集号或倒序时按单集处理。
:param item: 含 IndexNumber 与可空 IndexNumberEnd 的分集响应
:return: 包含合并文件所覆盖全部集号的区间，未提供起始集号时为空
"""
number = item.get("IndexNumber")
if number is None:
return range(0)
start = int(number)
end = item.get("IndexNumberEnd")
return range(start, max(start, int(end) if end is not None else start) + 1)
def emby_query_params(query: Optional[MediaServerQueryOptions]) -> dict[str, str]:
"""
把显式投影翻译为 Emby/Jellyfin 已公开的列表查询控制参数。

10
app/modules/emby/emby.py
View File

@@ -13,7 +13,7 @@ from requests import Response
from app.adapters.network.http import RequestUtils
from app.application.mediaserver import MediaServerIdentityHelper, format_emby_family_item
from app.foundation.url import UrlUtils
from app.modules._base.mediaserver import emby_query_params
from app.modules._base.mediaserver import emby_episode_numbers, emby_query_params
from app.runtime.log import logger
from app.runtime.settings import get_runtime_setting
from app.schemas.dashboard import Statistic as _SchemaStatistic
@@ -584,6 +584,7 @@ class Emby:
查询电视剧身份及各季已经存在的集号。
已有缓存标识失效时按标题及年份重新查找；仅统计服务端实际存在且具有季号、集号的条目。
合并文件按 IndexNumber 至 IndexNumberEnd 的闭区间计入全部集数。
:param item_id: Emby 条目标识
:param title: 缺少条目标识或缓存标识失效时用于回退搜索的剧名
@@ -651,7 +652,7 @@ class Emby:
continue
if season_index not in season_episodes:
season_episodes[season_index] = []
season_episodes[season_index].append(episode_index)
season_episodes[season_index].extend(emby_episode_numbers(res_item))
return item_id, season_episodes
except Exception as e:
logger.error("连接Shows/Id/Episodes出错：" + str(e))
@@ -664,6 +665,8 @@ class Emby:
"""
读取指定季的分集条目标识。
合并文件覆盖的各个集号指向同一个条目标识。
:param item_id: Emby 条目标识
:param season: 目标季号
:param query: 可选查询内容投影，省略时保留原有请求和返回内容
@@ -688,7 +691,8 @@ class Emby:
episode_id = res_item.get("Id")
if episode_index is None or not episode_id:
continue
episode_ids[int(episode_index)] = str(episode_id)
for number in emby_episode_numbers(res_item):
episode_ids[number] = str(episode_id)
return episode_ids
except Exception as e:
logger.error(f"获取 Emby 季集条目 ID 出错：{str(e)}")

10
app/modules/jellyfin/jellyfin.py
View File

@@ -11,7 +11,7 @@ from requests import Response
from app.adapters.network.http import RequestUtils
from app.application.mediaserver import MediaServerIdentityHelper, format_emby_family_item
from app.foundation.url import UrlUtils
from app.modules._base.mediaserver import emby_query_params
from app.modules._base.mediaserver import emby_episode_numbers, emby_query_params
from app.runtime.cache import MemoryBackend
from app.runtime.log import logger
from app.runtime.settings import get_runtime_setting
@@ -646,6 +646,7 @@ class Jellyfin:
查询电视剧身份及各季已经存在的集号。
已有缓存标识失效时按标题及年份重新查找；仅统计服务端实际存在且具有季号、集号的条目。
合并文件按 IndexNumber 至 IndexNumberEnd 的闭区间计入全部集数。
:param item_id: Jellyfin 条目标识
:param title: 缺少条目标识或缓存标识失效时用于回退搜索的剧名
@@ -717,7 +718,7 @@ class Jellyfin:
continue
if not season_episodes.get(season_index):
season_episodes[season_index] = []
season_episodes[season_index].append(episode_index)
season_episodes[season_index].extend(emby_episode_numbers(res_item))
return item_id, season_episodes
except Exception as e:
logger.error("连接Shows/Id/Episodes出错：" + str(e))
@@ -730,6 +731,8 @@ class Jellyfin:
"""
读取指定季的分集条目标识。
合并文件覆盖的各个集号指向同一个条目标识。
:param item_id: Jellyfin 条目标识
:param season: 目标季号
:param query: 可选查询内容投影，省略时保留原有请求和返回内容
@@ -754,7 +757,8 @@ class Jellyfin:
episode_id = res_item.get("Id")
if episode_index is None or not episode_id:
continue
episode_ids[int(episode_index)] = str(episode_id)
for number in emby_episode_numbers(res_item):
episode_ids[number] = str(episode_id)
return episode_ids
except Exception as e:
logger.error(f"获取 Jellyfin 季集条目 ID 出错：{str(e)}")

32
app/modules/navidrome/navidrome.py
View File

@@ -132,9 +132,7 @@ class Navidrome:
def authenticate(self, username: str, password: str) -> Optional[str]:
"""
临时使用给定凭据验证 Navidrome 用户。
验证期间会修改当前实例凭据，并在结束后恢复；此方法未提供并发认证隔离。
使用独立客户端验证给定凭据，不改变服务账号或其他并发认证上下文。
:param username: 待验证的用户名
:param password: 待验证的密码
@@ -142,12 +140,7 @@ class Navidrome:
"""
if not username or not password or not self._host:
return None
original_username, original_password = self._username, self._password
try:
self._username, self._password = username, password
return username if self._call("ping") else None
finally:
self._username, self._password = original_username, original_password
return Navidrome(host=self._host, username=username, password=password).user
def get_user_count(self) -> int:
"""
@@ -233,19 +226,25 @@ class Navidrome:
return f"{url}/getCoverArt?{urlencode(self._params(id=cover_id, size=512))}"
def _albums(
self, list_type: str = "alphabeticalByName", size: int = 500
self, list_type: str = "alphabeticalByName", size: int = 500, *, strict: bool = False
) -> List[dict[str, Any]]:
"""
分页读取专辑，直至短页、空页或接口失败。
:param list_type: Subsonic 专辑排序方式，默认按名称排序
:param size: 每页请求数量，默认 500
:param strict: 同步读取时要求完整结果，失败抛出 ConnectionError
:return: 已读取的专辑列表；中途失败会返回此前已读取的部分结果
"""
albums: List[dict[str, Any]] = []
offset = 0
while True:
payload = self._call("getAlbumList2", type=list_type, size=size, offset=offset)
if strict:
if payload is None or not isinstance(payload.get("albumList2"), dict):
raise ConnectionError("读取 Navidrome 专辑列表失败")
if "album" in payload["albumList2"] and not isinstance(payload["albumList2"]["album"], list):
raise ConnectionError("Navidrome 专辑列表响应格式错误")
page = ((payload or {}).get("albumList2") or {}).get("album") or []
if not page:
break
@@ -311,18 +310,19 @@ class Navidrome:
:param limit: 返回数量；-1 或 None 表示从起始位置返回到末尾
:param query: 可选查询内容投影，省略时保留原有请求和返回内容
:yields: 已转换为统一媒体模型的专辑条目
:raises ConnectionError: 任意一页读取失败，不能将部分结果作为完整媒体库
"""
if query is not None and query.fields is not None:
for album in self._iter_albums(start_index, limit):
for album in self._iter_albums(start_index, limit, strict=True):
yield self._album_to_item(album)
return
albums = self._albums()
albums = self._albums(strict=True)
end = None if limit is None or limit == -1 else start_index + limit
for album in albums[start_index:end]:
yield self._album_to_item(album)
def _iter_albums(
self, start_index: int, limit: Optional[int], *, list_type: str = "alphabeticalByName"
self, start_index: int, limit: Optional[int], *, list_type: str = "alphabeticalByName", strict: bool = False
) -> Generator[dict[str, Any], None, None]:
"""
直接使用 Subsonic 的偏移和数量控制逐页读取，避免先取全量再切片。
@@ -330,6 +330,7 @@ class Navidrome:
:param start_index: 首个专辑偏移，负数按零处理
:param limit: 本次最大数量；None 或 -1 持续翻页，其它非正值不发请求
:param list_type: Subsonic 专辑排序方式
:param strict: 同步读取时要求完整结果，失败抛出 ConnectionError
:yields: 服务器返回的原始专辑
"""
offset = max(0, start_index)
@@ -337,6 +338,11 @@ class Navidrome:
while remaining is None or remaining > 0:
size = 500 if remaining is None else min(500, remaining)
payload = self._call("getAlbumList2", type=list_type, size=size, offset=offset)
if strict:
if payload is None or not isinstance(payload.get("albumList2"), dict):
raise ConnectionError("读取 Navidrome 专辑列表失败")
if "album" in payload["albumList2"] and not isinstance(payload["albumList2"]["album"], list):
raise ConnectionError("Navidrome 专辑列表响应格式错误")
albums = ((payload or {}).get("albumList2") or {}).get("album") or []
yield from albums
if len(albums) < size:

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

@@ -2,6 +2,7 @@
Plex 客户端，负责媒体库同步、媒体身份查询、播放链接及 Webhook 转换。
"""
import hashlib
import json
import time
from pathlib import Path
@@ -211,8 +212,21 @@ class Plex:
[Callable[..., Optional[List[str]]]], Callable[..., Optional[List[str]]]
] = cached(maxsize=32, ttl=86400)
@_image_cache
def __get_library_images(self, library_key: str, mtype: int) -> Optional[List[str]]:
"""
按服务地址与令牌摘要隔离媒体库封面缓存。
:param library_key: Plex 媒体库标识
:param mtype: Plex 搜索类型值
:return: 当前服务器凭据对应的媒体库封面
"""
identity = hashlib.sha256(f"{self._host}\0{self._token}".encode()).hexdigest()
return self.__get_cached_library_images(library_key, mtype, identity)
@_image_cache
def __get_cached_library_images(
self, library_key: str, mtype: int, server_identity: str,
) -> Optional[List[str]]:
"""
分页读取最近新增媒体，选取最多 4 张不同的库封面。
@@ -220,6 +234,7 @@ class Plex:
:param library_key: Plex 媒体库标识
:param mtype: Plex 搜索类型值：1 为电影，2 为剧集，8 为音乐
:param server_identity: 服务地址与令牌摘要，只用于隔离缓存，不保存原始凭据
:return: 带服务令牌的本地图片 URL 列表；SDK 未连接时返回 None
"""
if not self._plex:
@@ -1153,7 +1168,8 @@ class Plex:
message.get('Metadata', {}).get('title'),
"(" + str(message.get('Metadata', {}).get('year')) + ")")
eventItem.item_id = message.get('Metadata', {}).get('key')
if len(message.get('Metadata', {}).get('summary')) > 100:
if (message.get('Metadata', {}).get('summary')
and len(message.get('Metadata', {}).get('summary')) > 100):
eventItem.overview = str(message.get('Metadata', {}).get('summary'))[:100] + "..."
else:
eventItem.overview = message.get('Metadata', {}).get('summary')

4
app/modules/trimemedia/api.py
View File

@@ -562,8 +562,10 @@ class Api:
post["exclude_grouped_video"] = 1
if (res := self.request("/item/list", data=post)) and res.success:
if not res.data:
if res.data is None:
return []
if not isinstance(res.data, dict) or not isinstance(res.data.get("list", []), list):
return None
return [self.__build_item(info) for info in res.data.get("list", [])]
return None

24
app/modules/trimemedia/trimemedia.py
View File

@@ -314,7 +314,7 @@ class TrimeMedia:
使用独立临时 API 验证用户名和密码。
验证结束后会登出并关闭临时会话；返回令牌仅用于认证结果，不保证之后仍可访问服务。
当前辅助认证探测不传递实例的访问码，开启访问码的服务可能无法通过此入口。
新会话沿用实例访问码，先校验访问权限，再验证待登录用户。
:param username: 待验证的用户名
:param password: 待验证的原始密码
@@ -324,7 +324,7 @@ class TrimeMedia:
return None
if not self.is_configured():
return None
if result := self.__create_api(self._host):
if result := self.__create_api(self._host, self._access_code):
try:
return result.api.login(username, password)
finally:
@@ -684,29 +684,29 @@ class TrimeMedia:
"""
查询一页媒体库内容，并递归展开目录中的电影和电视剧。
未认证时不产生条目；底层查询失败按空列表结束当前层，不向上传播专门的同步失败异常。
未认证或底层查询失败时中止迭代，避免将失败误认为成功的空媒体库。
:param parent: 媒体库或父目录 guid
:param start_index: 当前实现将该值加 1 作为页码；并非条目偏移，默认 0 请求第一页
:param limit: 传给服务端的每页数量；None 或 -1 传为 -1，递归子目录使用默认不限数量
:yields: 统一电影或电视剧条目
:raises ConnectionError: 未认证或任意一层目录读取失败
"""
if not self.is_authenticated():
return None
raise ConnectionError("飞牛影视服务尚未认证")
assert self._api is not None
if (page_size := limit) is None:
page_size = -1
if start_index is None:
raise TypeError("unsupported operand type(s) for +: 'NoneType' and 'int'")
items = (
self._api.item_list(
guid=parent,
page=start_index + 1,
page_size=page_size,
types=[fnapi.Type.MOVIE, fnapi.Type.TV, fnapi.Type.DIRECTORY],
)
or []
items = self._api.item_list(
guid=parent,
page=start_index + 1,
page_size=page_size,
types=[fnapi.Type.MOVIE, fnapi.Type.TV, fnapi.Type.DIRECTORY],
)
if items is None:
raise ConnectionError("读取飞牛影视媒体库失败")
for item in items:
if item.type == fnapi.Type.DIRECTORY:
yield from self.get_items(parent=item.guid)

40
app/modules/ugreen/ugreen.py
View File

@@ -28,6 +28,16 @@ from app.schemas.mediaserver import WebhookEventInfo as _SchemaWebhookEventInfo
from app.schemas.types import MediaSource, MediaType, SystemConfigKey
def _media_service_items(data: Mapping[str, Any]) -> list[Mapping[str, Any]]:
"""
校验媒体库服务列表，保留合法空数组并拒绝未知响应结构。
"""
items = data.get("Items")
if not isinstance(items, list):
raise ConnectionError("绿联媒体库服务列表响应格式错误")
return items
class Ugreen:
"""
绿联影视媒体服务器客户端。
@@ -1622,14 +1632,15 @@ class Ugreen:
"""
广度优先分页遍历海报墙目录，最多访问 20000 个不同路径。
重复路径只访问一次。单个目录请求失败时结束该目录，继续已有队列，不抛出专门的同步失败异常。
重复路径只访问一次。任意目录或分页失败都中止，不能将部分结果视为完整媒体库。
:param root_path: 遍历起始目录
:param page_size: 每个目录的分页大小，默认 100
:yields: 服务端 video_arr 中的有效视频字典
:raises ConnectionError: 目录读取失败或超过完整遍历的路径数量上限
"""
if not self._api or not root_path:
return
raise ConnectionError("绿联内部服务尚未连接或媒体库路径为空")
queue = deque([root_path])
visited: set[str] = set()
@@ -1650,6 +1661,10 @@ class Ugreen:
sort_type=1,
order_type=1,
)
if data is None:
raise ConnectionError("读取绿联海报墙目录失败")
if any(field in data and not isinstance(data[field], list) for field in ("video_arr", "folder_arr")):
raise ConnectionError("绿联海报墙目录响应格式错误")
if not data:
break
@@ -1667,6 +1682,8 @@ class Ugreen:
if data.get("is_last_page"):
break
page += 1
if queue:
raise ConnectionError("绿联媒体库目录数量超过遍历上限")
def __iter_media_service_items(
self,
@@ -1676,8 +1693,12 @@ class Ugreen:
) -> Generator[_SchemaMediaServerItem, None, None]:
"""
分页遍历媒体库服务用户可见的电影、剧集和音乐专辑。
:raises ConnectionError: 未认证、分页失败或列表结构错误，调用方须保留旧缓存
"""
if not self.is_authenticated() or not self._media_api or not parent or self.__is_library_blocked(str(parent)):
if not self.is_authenticated() or not self._media_api or not parent:
raise ConnectionError("绿联媒体库服务尚未连接或媒体库标识为空")
if self.__is_library_blocked(str(parent)):
return
remaining = None if limit in (None, -1) else max(0, limit or 0)
@@ -1700,9 +1721,9 @@ class Ugreen:
fields=fields,
)
if data is None:
return
page_items = data.get("Items")
if not isinstance(page_items, list) or not page_items:
raise ConnectionError("读取绿联媒体库服务条目失败")
page_items = _media_service_items(data)
if not page_items:
return
for raw_item in page_items:
@@ -1776,18 +1797,19 @@ class Ugreen:
内部模式按目录遍历并只返回电影、剧集，媒体库模式还支持音乐专辑。
当前内部模式对 limit=0 仍可能产生首个条目，媒体库模式不产生条目。
底层读取失败可能提前结束迭代并保留已经产出的部分结果。
底层读取失败抛出异常，调用方不得清理未读取到的旧缓存。
:param parent: 媒体库标识
:param start_index: 跳过的条目数量，负值或空值按 0 处理
:param limit: 期望返回的条目上限，None 或 -1 表示不限；调用方应提供正数或不限标记
:yields: 统一媒体条目
:raises ConnectionError: 服务未连接、媒体库路径缺失或任意目录分页失败
"""
if self.__is_media_service_mode():
yield from self.__iter_media_service_items(parent, start_index, limit)
return None
if not self.is_authenticated() or not self._api:
return None
raise ConnectionError("绿联内部服务尚未认证")
library_id = str(parent)
if not self._library_paths:
@@ -1795,7 +1817,7 @@ class Ugreen:
root_path = self._library_paths.get(library_id)
if not root_path:
return None
raise ConnectionError("绿联媒体库路径不可用")
skip = max(0, start_index or 0)
remain = -1 if limit is None or limit == -1 else max(0, limit)

11
app/modules/vyo/vyo.py
View File

@@ -211,15 +211,16 @@ class Vyo:
使用固定页大小遍历媒体库中的电影和电视剧。
按 100 条分页将偏移换算为页号，并在本地丢弃页内偏移。
配置不完整或读取失败时结束迭代；无法转换的条目仍占用数量配额。
配置不完整或读取失败时抛出异常；无法转换的条目仍占用数量配额。
:param parent: 媒体库标识
:param start_index: 起始条目偏移，空值或负数按 0 处理
:param limit: 最多检查的原始条目数；None 或 -1 不限，其他非正值不产生条目
:yields: 转换成功的统一媒体条目
:raises ConnectionError: 配置不完整、分页失败或列表响应结构错误
"""
if not parent or not self.is_configured():
return
raise ConnectionError("Vyo 服务配置不完整或媒体库标识为空")
# 页大小固定，页码才能稳定映射到偏移量；条数上限在产出侧裁剪
skip = max(0, start_index or 0)
remaining = None if limit is None or limit == -1 else max(0, limit)
@@ -233,8 +234,10 @@ class Vyo:
page_size=self.PAGE_LIMIT,
)
if result is None:
return
rows = result.get("items") or []
raise ConnectionError("读取 Vyo 媒体库失败")
rows = result.get("items")
if not isinstance(rows, list):
raise ConnectionError("Vyo 媒体库列表响应格式错误")
for row in rows[drop:]:
if remaining is not None and remaining <= 0:
return

30
app/modules/zspace/zspace.py
View File

@@ -45,6 +45,16 @@ def _user_id(value: object) -> Optional[Union[str, int]]:
raise TypeError("极影视用户标识格式错误")
def _library_items(result: Mapping[str, Any]) -> list[dict[str, Any]]:
"""
校验完整枚举的列表字段，非法结构不能代表成功空库。
"""
items = result.get("Items")
if not isinstance(items, list):
raise ConnectionError("极影视媒体库列表响应格式错误")
return items
class ZSpace:
"""
通过极影视的 Emby 兼容接口提供媒体检索、登录和播放信息。
@@ -929,15 +939,16 @@ class ZSpace:
分页遍历媒体库，展开合集并补查缺少身份或路径的条目。
仅产出电影、电视剧和音乐专辑。合集递归展开，因此产出数量可能超过单页限制。
未配置或分页请求失败会结束迭代，已产出的结果不会撤回。
未配置或任意分页请求失败会抛出异常，调用方不得清理未读到的缓存。
:param parent: 媒体库或合集标识
:param start_index: 零基起始偏移，空值或负数按 0 处理
:param limit: None 或 -1 按每页 100 条读完整个库；其他值只请求一页，原值传给服务端
:yields: 统一媒体条目；底层转换未成功时可能为 None
:raises ConnectionError: 服务未连接、分页失败或列表响应结构错误
"""
if not parent or not self._host or not self._apikey or not self.user:
return None
raise ConnectionError("极影视服务尚未连接或媒体库标识为空")
url = f"{self._host}emby/Users/{self.user}/Items"
fetch_all = limit is None or limit == -1
page_size = DEFAULT_ITEMS_PAGE_SIZE if limit is None or limit == -1 else limit
@@ -955,14 +966,15 @@ class ZSpace:
try:
res = self.__request_utils().get_res(url, params=params)
if not res or res.status_code != 200:
return None
raise ConnectionError("读取极影视媒体库失败")
result = res.json() or {}
items = result.get("Items") or []
items = _library_items(result)
for item in items:
if not item:
continue
if item.get("Type") == "BoxSet" and item.get("Id"):
for sub_item in self.get_items(parent=item.get("Id")):
item_id = item.get("Id")
if item.get("Type") == "BoxSet" and item_id:
for sub_item in self.get_items(parent=item_id):
if sub_item:
yield sub_item
continue
@@ -977,15 +989,15 @@ class ZSpace:
or not item.get("ProductionYear")
or not item.get("Path")
)
if needs_detail and item.get("Id"):
detail_item = self.get_iteminfo(item.get("Id"))
if needs_detail and item_id:
detail_item = self.get_iteminfo(item_id)
if detail_item:
yield detail_item
continue
yield self.__format_item_info(item)
except Exception as e:
logger.error(f"连接Users/Items出错：{e}")
return None
raise
if not fetch_all:
break

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

@@ -641,6 +641,9 @@
"uses": []
},
{
"source": "app.modules.plex.plex",
"target": "plexapi",
"kind": "network_sdk",
"bindings": [
"from:plexapi.audio.Album as Album",
"from:plexapi.audio.Artist as Artist",
@@ -656,14 +659,10 @@
"from:plexapi.video.Season as Season",
"from:plexapi.video.Show as Show"
],
"fingerprint": "0fe42fa932618997d4f589a5217a36b45f7ac2947212a371bdfbbc55d9cf1def",
"kind": "network_sdk",
"source": "app.modules.plex.plex",
"target": "plexapi",
"uses": [
"Plex.__fetch_item|call:exceptions.NotFound",
"Plex.__fetch_item|call:fetchItem",
"Plex.__get_library_images|call:fetchItems",
"Plex.__get_cached_library_images|call:fetchItems",
"Plex.__init__|call:library.sections",
"Plex.__init__|call:server.PlexServer",
"Plex.__search_show|call:library.search",
@@ -687,7 +686,8 @@
"Plex.refresh_library_by_items|call:library.update",
"Plex.refresh_library_by_items|call:query",
"Plex.refresh_root_library|call:library.update"
]
],
"fingerprint": "6109de36f963b5c6193ed60941a88856cbee9c37303dedf13da3a3be94478543"
},
{
"bindings": [

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

@@ -141,7 +141,7 @@
{"source": "app.modules.discord.discord", "target": "discord", "kind": "network_sdk", "fingerprint": "a46a1d6da767a51eb4e1085de5fb936490fc8360c2f3a814635d408468c1b150"},
{"source": "app.modules.filemanager.storages.smb", "target": "smbclient", "kind": "network_sdk", "fingerprint": "37619f050b03598fa559c6ae2df94bda6adf7cffd1cc42d9db9059f1fb41af47"},
{"source": "app.modules.filemanager.storages.u115", "target": "oss2", "kind": "network_sdk", "fingerprint": "f7b89c8ae6dad2603f0a9e0caaa159769aef5b3581d7e728f62445b979366eae"},
{"source": "app.modules.plex.plex", "target": "plexapi", "kind": "network_sdk", "fingerprint": "0fe42fa932618997d4f589a5217a36b45f7ac2947212a371bdfbbc55d9cf1def"},
{"source": "app.modules.plex.plex", "target": "plexapi", "kind": "network_sdk", "fingerprint": "6109de36f963b5c6193ed60941a88856cbee9c37303dedf13da3a3be94478543"},
{"source": "app.modules.qbittorrent.qbittorrent", "target": "qbittorrentapi", "kind": "network_sdk", "fingerprint": "b2f5a27f0c54cf95ed42fe99848fd6c9c0641caca8bc4ff8006a411cb427066d"},
{"source": "app.modules.slack.slack", "target": "slack_bolt", "kind": "network_sdk", "fingerprint": "d970999c7c77ca5e93ec8a2773fa0b3db10df63ecce780b5fd52ca038734666a"},
{"source": "app.modules.slack.slack", "target": "slack_sdk", "kind": "network_sdk", "fingerprint": "7559f31e4172ad3bbbaf161e1164ea48b997c56d05c77b49b82626298c39aa14"},

2
tests/test_architecture_egress.py
View File

@@ -130,7 +130,7 @@ FROZEN_EGRESS_FINGERPRINT_BY_EDGE = {
("app.modules.jellyfin.jellyfin", "requests"): "5c46d09ca9a4bcc0bae21ca5d554ef09baa3f901c562b27c5f5ee1439c7746b5",
# 补充现有 SDK 返回类型导入；未增加客户端或请求路径。
# 新增 LibrarySection 类型绑定以描述音乐查询；SDK 外连操作集合保持一致。
("app.modules.plex.plex", "plexapi"): "0fe42fa932618997d4f589a5217a36b45f7ac2947212a371bdfbbc55d9cf1def",
("app.modules.plex.plex", "plexapi"): "6109de36f963b5c6193ed60941a88856cbee9c37303dedf13da3a3be94478543",
("app.modules.plex.plex", "requests"): "e49c0856fd6c99a3409850898aac7e6e3fa611d4e91546634c178fbfd244e8da",
("app.modules.qbittorrent.qbittorrent", "qbittorrentapi"): "b2f5a27f0c54cf95ed42fe99848fd6c9c0641caca8bc4ff8006a411cb427066d",
("app.modules.qqbot.gateway", "websocket"): "6f6b7d61f3a95e620e67c450d544f0aa077f087a188d98f32e4501d94c6b37ae",

412
tests/test_mediaserver_review_regressions.py Normal file
View File

@@ -0,0 +1,412 @@
"""
媒体服务器审查问题的离线回归，覆盖认证、缺集、协议字段与同步失败保护。
"""
import hashlib
import json
from concurrent.futures import ThreadPoolExecutor
from threading import Barrier
from types import SimpleNamespace
from unittest.mock import Mock
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.adapters.network.http import RequestUtils
from app.application.mediaserver import MediaServerSyncItem, format_emby_family_item
from app.chain.download.existence import DownloadExistenceOwner
from app.chain.mediaserver import MediaServerChain
from app.db.adapters.mediaserver import TransactionalMediaServerRepository
from app.db.models.mediaserver import MediaServerItem as CachedItem
from app.domain.context import MediaInfo
from app.domain.meta.metavideo import MetaVideo
from app.modules.emby.emby import Emby
from app.modules.jellyfin.jellyfin import Jellyfin
from app.modules.navidrome.navidrome import Navidrome
from app.modules.plex.plex import Plex
from app.modules.trimemedia.api import Api as TrimeApi
from app.modules.trimemedia.api import User as TrimeUser
from app.modules.trimemedia.trimemedia import TrimeMedia
from app.modules.ugreen.api import Api as UgreenApi
from app.modules.ugreen.api import UgreenMediaApi
from app.modules.ugreen.ugreen import Ugreen
from app.modules.vyo.api import Api as VyoApi
from app.modules.vyo.vyo import Vyo
from app.modules.zspace.zspace import ZSpace
from app.schemas.mediaserver import ExistMediaInfo, MediaServerLibrary, MediaServerQueryOptions
from app.schemas.types import MediaSource, MediaType
def _response(payload, status=200):
"""
创建不会访问网络的 HTTP 响应替身。
"""
return Mock(status_code=status, ok=status < 400, json=Mock(return_value=payload))
def test_navidrome_concurrent_authentication_preserves_each_identity(monkeypatch):
"""
并发辅助认证不得借用他人密码，也不得改变服务账号。
"""
client = Navidrome.__new__(Navidrome)
client._host = "https://navidrome.invalid/"
client._username, client._password = "configured", "service-password"
barrier = Barrier(2)
original_params = Navidrome._params
seen = []
def synchronized_params(instance, **kwargs):
"""
让两个线程同时构造请求，放大共享凭据覆盖窗口。
"""
barrier.wait(timeout=5)
params = original_params(instance, **kwargs)
barrier.wait(timeout=5)
return params
def request(_instance, _url, params):
"""
仅认可 valid 用户的独立密码签名。
"""
seen.append(params["u"])
valid = params["u"] == "valid" and params["t"] == hashlib.md5(
f"right{params['s']}".encode()
).hexdigest()
return _response({"subsonic-response": {"status": "ok" if valid else "failed"}})
monkeypatch.setattr(Navidrome, "_params", synchronized_params)
monkeypatch.setattr(RequestUtils, "get_res", request)
with ThreadPoolExecutor(max_workers=2) as pool:
victim = pool.submit(client.authenticate, "victim", "wrong")
valid = pool.submit(client.authenticate, "valid", "right")
assert victim.result(timeout=10) is None
assert valid.result(timeout=10) == "valid"
assert sorted(seen) == ["valid", "victim"]
assert (client._username, client._password) == ("configured", "service-password")
@pytest.mark.parametrize("played_at", [
"2026-10-01T12:34:56Z", "2026-10-01T12:34:56+00:00",
"2026-10-01T12:34:56.1234567Z", "2026-10-01T12:34:56.5+08:00",
"2026-10-01T12:34:56",
])
def test_emby_family_accepts_iso_playback_dates(played_at):
"""
ISO 日期的小数秒和时区变化不应导致整个媒体条目被丢弃。
"""
item = format_emby_family_item({
"Id": "film", "Name": "电影", "Type": "Movie",
"UserData": {"Played": True, "LastPlayedDate": played_at},
}, server="emby")
assert item is not None
assert item.user_state.last_played_date == "2026-10-01 12:34:56"
@pytest.mark.parametrize("client_class", [Emby, Jellyfin])
@pytest.mark.parametrize("end,expected", [(3, [1, 2, 3]), (None, [1]), (0, [1])])
def test_emby_family_combined_episode_includes_end(client_class, end, expected, monkeypatch):
"""
合并剧集的每个集号都应存在，并映射到同一个物理条目标识。
"""
client = client_class.__new__(client_class)
client._host, client._apikey, client.user = "https://media.invalid/", "key", "user"
monkeypatch.setattr(client, "get_iteminfo", lambda *_args, **_kwargs: Mock())
monkeypatch.setattr("app.application.mediaserver.MediaServerIdentityHelper.is_compatible", lambda *_args: True)
monkeypatch.setattr(RequestUtils, "get_res", lambda *_args, **_kwargs: _response({"Items": [{
"Id": "episode", "ParentIndexNumber": 1, "IndexNumber": 1, "IndexNumberEnd": end,
}]}))
assert client.get_tv_episodes(item_id="series")[1] == {1: expected}
assert client.get_season_episode_ids("series", 1) == {number: "episode" for number in expected}
@pytest.mark.parametrize("server_order", [("partial", "complete"), ("complete", "partial"), ("odd", "even")])
def test_tv_existence_merges_all_servers_independently_of_order(server_order, monkeypatch):
"""
多个服务器上的完整或互补集数都应参与缺集计算。
"""
owner = DownloadExistenceOwner.__new__(DownloadExistenceOwner)
owner.media_server_repository = Mock(get_item_id=Mock(return_value=None))
episodes = {"partial": [1], "complete": [1, 2], "odd": [1], "even": [2]}
owner.media_exists = lambda **kwargs: ExistMediaInfo(type=MediaType.TV, seasons={1: episodes[kwargs["server"]]})
monkeypatch.setattr("app.chain.download.existence.get_mediaserver_configs",
lambda: [SimpleNamespace(name=name) for name in server_order])
media = MediaInfo()
media.type = MediaType.TV
media.media_source, media.media_id = MediaSource.TMDB, "123"
media.seasons = {1: [1, 2]}
result = owner._media_exists_with_server_cache(media, {})
assert result.seasons == {1: [1, 2]}
assert owner.get_no_exists_info(MetaVideo("示例 S01"), media) == (True, {})
@pytest.fixture
def plex_images_cache():
"""
在测试前后清理媒体库封面区域，防止缓存影响其他测试。
"""
function = getattr(Plex, "_Plex__get_cached_library_images", Plex._Plex__get_library_images)
function.cache_clear()
yield
function.cache_clear()
def test_plex_library_images_do_not_share_server_or_token(plex_images_cache):
"""
相同库标识的多个服务器以及令牌轮换应使用各自图片缓存。
"""
clients = []
for host, token in [("a", "token-a"), ("b", "token-b"), ("a", "rotated")]:
client = Plex.__new__(Plex)
client._host, client._token = f"https://{host}.invalid/", token
client._plex = Mock()
client._plex.fetchItems.return_value = [SimpleNamespace(type="movie", thumb="/poster")]
clients.append(client)
for client in clients:
assert client._Plex__get_library_images("review-unique-library", 1) == [
f"{client._host}poster?X-Plex-Token={client._token}"
]
client._Plex__get_library_images("review-unique-library", 1)
assert client._plex.fetchItems.call_count == 1
@pytest.mark.parametrize("summary", [None, "", "简介", "长" * 101])
def test_plex_webhook_accepts_nullable_summary(summary, monkeypatch):
"""
无简介电影仍可解析播放事件，长简介保持截断规则。
"""
client = Plex.__new__(Plex)
monkeypatch.setattr(client, "get_remote_image_by_id", lambda **_kwargs: None)
event = client.get_webhook_message({"payload": json.dumps({
"event": "media.play", "Metadata": {"type": "movie", "title": "电影", "key": "1", "summary": summary},
})})
assert event is not None
assert event.overview == (summary[:100] + "..." if summary and len(summary) > 100 else summary)
def test_trimemedia_auxiliary_login_revalidates_access_code(monkeypatch):
"""
辅助认证的新会话必须使用配置访问码，且认证后关闭独立会话。
"""
client = TrimeMedia.__new__(TrimeMedia)
client._host, client._username, client._password = "https://trime.invalid/v", "service", "password"
client._access_code = "device-code"
verified_sessions = set()
requests = []
def get_response(session, url, **_kwargs):
"""
访问码校验为当前会话授权，不向其他会话传播。
"""
requests.append(url)
if url.endswith("/c/device-code"):
verified_sessions.add(id(session))
return _response({})
return None
def request(session, **kwargs):
"""
只有通过访问码校验的临时会话可访问版本与登录接口。
"""
if id(session) not in verified_sessions:
return None
if kwargs["url"].endswith("/sys/version"):
return _response({"code": 0, "data": {"version": "1", "mediasrvVersion": "1"}})
return _response({"code": 0, "data": {"token": "user-token"}})
monkeypatch.setattr(RequestUtils, "get_res", get_response)
monkeypatch.setattr(RequestUtils, "request", request)
assert client.authenticate("user", "correct") == "user-token"
assert requests == ["https://trime.invalid/c/device-code"] * 2
@pytest.fixture
def sync_repository(tmp_path):
"""
为每次同步场景创建独占临时 SQLite 缓存，预存一条旧媒体。
"""
engine = create_engine(f"sqlite:///{tmp_path / 'library.db'}")
factory = sessionmaker(bind=engine)
CachedItem.__table__.create(bind=engine)
repository = TransactionalMediaServerRepository(factory)
repository.upsert(MediaServerSyncItem(
server="test", library="library", item_id="old", item_type="电影", title="旧媒体",
original_title=None, year=None, media_source=None, media_id=None, path=None,
seasoninfo=(), note_json=None, lst_mod_date="2026-10-01 00:00:00",
))
yield repository, factory
engine.dispose()
def _sync_client(kind, monkeypatch, payload):
"""
使用真实客户端和协议 API，仅替换 HTTP 传输边界。
"""
calls = []
def respond(*args, **kwargs):
"""
为分页场景按请求顺序返回响应，其余场景复用固定响应。
"""
calls.append((args, kwargs))
return payload if payload is None or hasattr(payload, "status_code") else payload(*args, **kwargs)
monkeypatch.setattr(RequestUtils, "get_res", respond)
monkeypatch.setattr(RequestUtils, "request", respond)
if kind == "navidrome":
client = Navidrome.__new__(Navidrome)
client._host, client._username, client._password, client.user = "https://nav.invalid/", "u", "p", "u"
elif kind == "zspace":
client = ZSpace.__new__(ZSpace)
client._host, client._apikey, client.user = "https://z.invalid/", "key", "u"
elif kind == "vyo":
client = Vyo.__new__(Vyo)
client._host, client._apikey = "https://vyo.invalid/", "key"
client._api = VyoApi(client._host, client._apikey)
elif kind == "trime":
client = TrimeMedia.__new__(TrimeMedia)
client._host, client._username, client._password = "https://trime.invalid/", "u", "p"
client._api = TrimeApi(client._host, "key")
client._api._token, client._userinfo = "token", TrimeUser("u", "u")
else:
client = Ugreen.__new__(Ugreen)
client._host, client._username, client._password = "https://ugreen.invalid/", "u", "p"
client._sync_libraries = []
if kind == "ugreen-media":
client._connection_mode = "media_service"
client._media_api = UgreenMediaApi(client._host)
client._media_api._token, client._media_api._user_id = "token", "u"
else:
client._connection_mode = "internal"
client._api = UgreenApi(client._host)
client._api._token, client._userinfo = "token", {"id": "u"}
client._api._crypto = SimpleNamespace(
build_encrypted_request=lambda **kwargs: SimpleNamespace(
url=kwargs["url"], headers={}, params=kwargs["params"], json=None, aes_key="fake-key"),
decrypt_response=lambda data, _key: data,
)
client._library_paths = {"library": "/media"}
return client, calls
_EMPTY_PAYLOADS = {
"navidrome": {"subsonic-response": {"status": "ok", "albumList2": {}}},
"zspace": {"Items": [], "TotalRecordCount": 0},
"vyo": {"success": True, "data": {"items": [], "total": 0}},
"trime": {"code": 0, "data": {"list": []}},
"ugreen-media": {"Items": [], "TotalRecordCount": 0},
"ugreen-internal": {"code": 200, "data": {"video_arr": [], "folder_arr": [], "is_last_page": True}},
}
_FAILED_PAYLOADS = {
"navidrome": {"subsonic-response": {"status": "failed", "error": {"code": 40}}},
"zspace": {},
"vyo": {"success": False, "message": "unauthorized"},
"trime": {"code": -15, "msg": "unauthorized"},
"ugreen-media": {},
"ugreen-internal": {"code": 401, "msg": "unauthorized"},
}
_MALFORMED_PAYLOADS = {
"navidrome": {"subsonic-response": {"status": "ok", "albumList2": {"album": {}}}},
"zspace": {"Items": {}},
"vyo": {"success": True, "data": {"items": {}}},
"trime": {"code": 0, "data": {"list": {}}},
"ugreen-media": {"Items": {}},
"ugreen-internal": {"code": 200, "data": {"video_arr": {}, "folder_arr": [], "is_last_page": True}},
}
@pytest.mark.parametrize("kind", list(_EMPTY_PAYLOADS))
@pytest.mark.parametrize("state", ["network", "rejected", "malformed", "empty"])
def test_remote_library_failure_preserves_sqlite_cache(kind, state, monkeypatch, sync_repository):
"""
真实同步链只在确认成功的空库后删除旧缓存，网络失败必须保留旧记录。
"""
repository, factory = sync_repository
successful_empty = state == "empty"
payload = None if state == "network" else _response(
_EMPTY_PAYLOADS[kind] if successful_empty else (
_MALFORMED_PAYLOADS[kind] if state == "malformed" else _FAILED_PAYLOADS[kind]
),
status=403 if state == "rejected" and kind in {"zspace", "ugreen-media"} else 200,
)
client, calls = _sync_client(kind, monkeypatch, payload)
chain = MediaServerChain.__new__(MediaServerChain)
chain.items = lambda **kwargs: (
client.get_items(query=kwargs["query"]) if kind == "navidrome" else client.get_items(kwargs["library_id"])
)
def operation():
"""
对真实短事务缓存执行完整服务器枚举。
"""
return chain._sync_server_libraries(
"test", [MediaServerLibrary(id="library", name="媒体库")], {"library": None}, repository,
"2026-10-23 00:00:00", None, 1, 1, None, 0,
)
if successful_empty:
assert operation() == (0, 0)
else:
with pytest.raises(ConnectionError):
operation()
with factory() as session:
assert session.query(CachedItem).count() == (0 if successful_empty else 1)
assert len(calls) == 1
@pytest.mark.parametrize("kind", list(_EMPTY_PAYLOADS))
@pytest.mark.parametrize("failure", ["network", "malformed"])
def test_partial_library_failure_preserves_older_sqlite_rows(kind, failure, monkeypatch, sync_repository):
"""
后续页或子目录失败时，即使已写入部分新媒体，也不能清理尚未枚举的旧媒体。
"""
movie = {"Id": "new", "Type": "Movie", "Name": "电影", "ProductionYear": 2026,
"Path": "/media/new", "ProviderIds": {"Tmdb": "123"}}
first_pages = {
"navidrome": {"subsonic-response": {"status": "ok", "albumList2": {
"album": [{"id": f"new-{index}", "name": "专辑"} for index in range(500)],
}}},
"zspace": {"Items": [movie], "TotalRecordCount": 2},
"vyo": {"success": True, "data": {"items": [
{"id": f"new-{index}", "kind": "Movie", "title": "电影"} for index in range(100)
]}},
"trime": {"code": 0, "data": {"list": [
{"guid": "new", "type": "Movie", "title": "电影"},
{"guid": "folder", "type": "Directory"},
]}},
"ugreen-media": {"Items": [dict(movie, Id=f"new-{index}") for index in range(100)],
"TotalRecordCount": 101},
"ugreen-internal": {"code": 200, "data": {
"video_arr": [{"ug_video_info_id": "new", "type": 1, "name": "电影"}],
"folder_arr": [{"path": "/media/subfolder"}], "is_last_page": True,
}},
}
responses = iter([_response(first_pages[kind]),
_response(_MALFORMED_PAYLOADS[kind]) if failure == "malformed" else None])
client, calls = _sync_client(kind, monkeypatch, lambda *_args, **_kwargs: next(responses))
chain = MediaServerChain.__new__(MediaServerChain)
chain.items = lambda **kwargs: (
client.get_items(query=kwargs["query"]) if kind == "navidrome" else client.get_items(kwargs["library_id"])
)
repository, factory = sync_repository
with pytest.raises(ConnectionError):
chain._sync_server_libraries(
"test", [MediaServerLibrary(id="library", name="媒体库")], {"library": None}, repository,
"2026-10-23 00:00:00", None, 1, 1, None, 0,
)
with factory() as session:
assert session.query(CachedItem).filter(CachedItem.item_id == "old").count() == 1
assert session.query(CachedItem).count() > 1
assert len(calls) == 2
@pytest.mark.parametrize("projected", [False, True])
def test_navidrome_partial_page_failure_is_not_successful_enumeration(projected, monkeypatch):
"""
两种查询模式都必须在后续页失败时中止，不能返回半份完整库。
"""
client = Navidrome.__new__(Navidrome)
responses = iter([{"albumList2": {"album": [{"id": str(index)} for index in range(500)]}}, None])
monkeypatch.setattr(client, "_call", lambda *_args, **_kwargs: next(responses))
with pytest.raises(ConnectionError):
list(client.get_items(**({"query": MediaServerQueryOptions(fields=frozenset())} if projected else {})))

6
tests/test_vyo_module.py
View File

@@ -192,7 +192,11 @@ def test_get_items_stops_when_request_fails():
return _paged_items(rows)(params, None)
client = _client({"/items": flaky})
assert len([*client.get_items("lib-1")]) == 100
items = client.get_items("lib-1")
first_page = [next(items) for _ in range(100)]
assert len(first_page) == 100
with pytest.raises(ConnectionError):
next(items)
# ── 媒体库与统计 ────────────────────────────────────────────────────