diff --git a/app/application/chain/data.py b/app/application/chain/data.py index 2714285f8..9707c6d5a 100644 --- a/app/application/chain/data.py +++ b/app/application/chain/data.py @@ -10,10 +10,14 @@ from collections.abc import Callable from dataclasses import dataclass from typing import Any, Optional +from app.application.download.failures import DownloadFailureRepository +from app.application.mediaserver import MediaServerRepository from app.application.transfer.execution import TransferExecutionRepository from app.application.transfer.workflow import TransferAdmissionRepository OperFactory = Callable[[], Any] +DownloadFailureRepositoryFactory = Callable[[], DownloadFailureRepository] +MediaServerRepositoryFactory = Callable[[], MediaServerRepository] TransferAdmissionRepositoryFactory = Callable[[], TransferAdmissionRepository] TransferExecutionRepositoryFactory = Callable[[], TransferExecutionRepository] @@ -28,8 +32,8 @@ class ChainDataPorts: transfer_history: OperFactory transfer_pending: TransferAdmissionRepositoryFactory transfer_execution: TransferExecutionRepositoryFactory - media_server: OperFactory - download_failure: OperFactory + media_server: MediaServerRepositoryFactory + download_failure: DownloadFailureRepositoryFactory user: OperFactory @@ -44,8 +48,8 @@ def configure_chain_data_ports( transfer_history: OperFactory, transfer_pending: TransferAdmissionRepositoryFactory, transfer_execution: TransferExecutionRepositoryFactory, - media_server: OperFactory, - download_failure: OperFactory, + media_server: MediaServerRepositoryFactory, + download_failure: DownloadFailureRepositoryFactory, user: OperFactory, ) -> None: """由启动组合根登记显式命名的 Chain 数据端口实现。""" @@ -100,13 +104,13 @@ def get_chain_transfer_execution_port() -> TransferExecutionRepository: return get_chain_data_ports().transfer_execution() -def get_chain_media_server_port() -> Any: - """创建媒体服务器数据端口实例。""" +def get_chain_media_server_port() -> MediaServerRepository: + """创建类型化的媒体服务器本地缓存端口实例。""" return get_chain_data_ports().media_server() -def get_chain_download_failure_port() -> Any: - """创建下载失败数据端口实例。""" +def get_chain_download_failure_port() -> DownloadFailureRepository: + """创建类型化的下载失败冷却持久化端口实例。""" return get_chain_data_ports().download_failure() diff --git a/app/application/download/failures.py b/app/application/download/failures.py new file mode 100644 index 000000000..9ec11a5b0 --- /dev/null +++ b/app/application/download/failures.py @@ -0,0 +1,57 @@ +"""下载失败冷却记录的类型化应用契约。""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional, Protocol, Union + +from app.schemas.types import MediaSource + + +@dataclass(frozen=True, slots=True) +class DownloadFailureWrite: + """一次下载失败冷却写入所需的完整稳定数据。""" + + fingerprint: str + failed_at: str + next_retry_at: str + media_type: Optional[str] = None + title: Optional[str] = None + year: Optional[str] = None + media_source: Optional[MediaSource] = None + media_id: Optional[str] = None + seasons: Optional[str] = None + episodes: Optional[str] = None + site: Optional[int] = None + site_name: Optional[str] = None + torrent_id: Optional[str] = None + torrent_name: Optional[str] = None + torrent_size: Optional[Union[float, int]] = None + downloader: Optional[str] = None + source: Optional[str] = None + error_message: Optional[str] = None + + +@dataclass(frozen=True, slots=True) +class DownloadFailureSnapshot: + """脱离数据库会话后供资源冷却判断使用的只读快照。""" + + fingerprint: str + error_message: Optional[str] + next_retry_at: Optional[str] + + +class DownloadFailureRepository(Protocol): + """下载链读写失败冷却状态所需的最小持久化端口。""" + + def get_active_by_fingerprints( + self, + fingerprints: list[str], + now_time: str, + ) -> dict[str, DownloadFailureSnapshot]: + """返回仍处于冷却期的不可变失败快照。""" + ... + + def record_failure(self, failure: DownloadFailureWrite) -> None: + """持久化一次失败事实，完成提交后不暴露数据库记录。""" + ... diff --git a/app/application/mediaserver.py b/app/application/mediaserver.py index f245bf1d4..60083f8c6 100644 --- a/app/application/mediaserver.py +++ b/app/application/mediaserver.py @@ -1,16 +1,17 @@ +import json import re from collections.abc import Iterable, Mapping +from dataclasses import dataclass from datetime import datetime -from typing import Any, Optional, Protocol +from typing import Any, Optional, Protocol, Union -from app.schemas.mediaserver import MediaServerItem as _SchemaMediaServerItem -from app.schemas.mediaserver import MediaServerItemUserState as _SchemaMediaServerItemUserState +from app.application.service import ServiceBaseHelper from app.domain.context import MusicInfo from app.runtime.log import logger from app.schemas.media import normalize_media_source, resolve_media_identity -from app.application.service import ServiceBaseHelper -from app.schemas.system import MediaServerConf -from app.schemas.system import ServiceInfo +from app.schemas.mediaserver import MediaServerItem as _SchemaMediaServerItem +from app.schemas.mediaserver import MediaServerItemUserState as _SchemaMediaServerItemUserState +from app.schemas.system import MediaServerConf, ServiceInfo from app.schemas.types import ( MUSIC_ENTITY_ALBUM, MediaSource, @@ -19,16 +20,106 @@ from app.schemas.types import ( ) +@dataclass(frozen=True, slots=True) +class MediaServerSyncItem: + """媒体库同步写端使用的脱离远端响应与数据库会话的冻结条目。""" + + server: str + library: Optional[str] + item_id: str + item_type: Optional[str] + title: Optional[str] + original_title: Optional[str] + year: Optional[Union[str, int]] + media_source: Optional[MediaSource] + media_id: Optional[str] + path: Optional[str] + seasoninfo: tuple[tuple[int, tuple[int, ...]], ...] + note_json: Optional[str] + lst_mod_date: str + + @classmethod + def from_item( + cls, + item: _SchemaMediaServerItem, + *, + item_type: Optional[str], + seasoninfo: Mapping[int, Optional[list[int]]], + sync_time: str, + ) -> "MediaServerSyncItem": + """冻结远端媒体条目和本轮剧集快照，供短事务持久化。""" + return cls( + server=str(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, + title=item.title, + original_title=item.original_title, + year=item.year, + media_source=item.media_source, + media_id=item.media_id, + path=item.path, + seasoninfo=tuple( + (season, tuple(episodes or ())) + for season, episodes in seasoninfo.items() + ), + note_json=( + json.dumps(item.note, ensure_ascii=False) + if item.note is not None + else None + ), + lst_mod_date=sync_time, + ) + + +class MediaServerRepository(Protocol): + """Chain 查询与同步媒体服务器本地缓存所需的最小持久化端口。""" + + def get_item_id( + self, + *, + title: Optional[str] = None, + year: Optional[Union[str, int]] = None, + mtype: Optional[str] = None, + media_source: Optional[MediaSource] = None, + media_id: Optional[str] = None, + season: Optional[int] = None, + ) -> Optional[str]: + """返回匹配条目的服务器 item_id，未命中时返回 None。""" + ... + + def upsert(self, item: MediaServerSyncItem) -> bool: + """在短事务中新增或更新一个冻结媒体条目。""" + ... + + def delete_stale(self, *, server: str, sync_time: str) -> int: + """删除指定服务器本轮同步未更新的本地条目。""" + ... + + def delete_excluded_servers(self, servers: list[str]) -> int: + """删除已停用或已移除服务器的本地条目。""" + ... + + class AsyncMediaServerQueryRepository(Protocol): """媒体服务器本地条目查询所需的异步持久化端口。""" - async def async_exists(self, **kwargs: Any) -> Any | None: - """按标题或统一媒体身份查找已同步条目。""" + async def async_get_item_id( + self, + *, + title: Optional[str] = None, + year: Optional[Union[str, int]] = None, + mtype: Optional[str] = None, + media_source: Optional[MediaSource] = None, + media_id: Optional[str] = None, + season: Optional[int] = None, + ) -> Optional[str]: + """按标题或统一媒体身份返回已同步条目的标量 ID。""" ... class MediaServerQueryService: - """封装媒体服务器本地存在性查询与 ORM 投影。""" + """封装媒体服务器本地存在性标量查询。""" def __init__(self, repository: AsyncMediaServerQueryRepository): """使用显式媒体服务器查询端口初始化服务。""" @@ -38,14 +129,14 @@ class MediaServerQueryService: self, *, title: Optional[str] = None, - year: Optional[str] = None, + year: Optional[Union[str, int]] = None, mtype: Optional[str] = None, media_source: Optional[MediaSource] = None, media_id: Optional[str] = None, season: Optional[int] = None, ) -> Optional[str]: """返回匹配条目的服务器 item_id，未命中时返回 None。""" - item = await self._repository.async_exists( + return await self._repository.async_get_item_id( title=title, year=year, mtype=mtype, @@ -53,7 +144,6 @@ class MediaServerQueryService: media_id=media_id, season=season, ) - return item.item_id if item else None class MediaServerIdentityHelper: diff --git a/app/chain/download.py b/app/chain/download.py index e90dc8dd3..e55822d8c 100644 --- a/app/chain/download.py +++ b/app/chain/download.py @@ -6,7 +6,7 @@ import re import shutil import time from pathlib import Path -from typing import TYPE_CHECKING, Dict, List, Optional, Set, Tuple, Union +from typing import Dict, List, Optional, Set, Tuple, Union from urllib.parse import parse_qs, urlencode, urljoin, urlparse from app.adapters.network.http import RequestUtils @@ -19,6 +19,11 @@ from app.application.chain.data import ( from app.application.configuration import get_chain_runtime_config_snapshot from app.application.directory import DirectoryHelper, validate_download_save_path from app.application.download import selection as _selection +from app.application.download.failures import ( + DownloadFailureRepository, + DownloadFailureSnapshot, + DownloadFailureWrite, +) from app.application.download.tasks import DownloadTaskService from app.application.torrent import TorrentHelper from app.chain import ChainBase @@ -63,12 +68,6 @@ from app.schemas.types import ( ) from app.schemas.workflow import FileItem as _SchemaFileItem -if TYPE_CHECKING: - from typing import Any - - DownloadFailure = Any - - DOWNLOAD_FAILURE_RESOURCE_TTL_SECONDS = 24 * 60 * 60 DOWNLOAD_FAILURE_TRANSIENT_TTL_SECONDS = 60 * 60 DOWNLOAD_FAILURE_RESOURCE_ERROR_KEYWORDS = ( @@ -880,25 +879,40 @@ class DownloadChain(ChainBase): torrent = context.torrent_info site = getattr(torrent, "site", None) try: - get_chain_download_failure_port().record_failure( - fingerprint=fingerprint, - now_time=now_time, - next_retry_at=next_retry_at, - type=getattr(getattr(media, "type", None), "value", getattr(media, "type", None)), - title=getattr(media, "title", None), - year=getattr(media, "year", None), - media_source=media_source, - media_id=media_id, - seasons=getattr(meta, "season", None), - episodes=episode_rules.format_ranges(list(episodes)) if episodes else self._format_failure_episodes(meta), - site=site if isinstance(site, int) else None, - site_name=getattr(torrent, "site_name", None), - torrent_id=self._torrent_resource_key(torrent), - torrent_name=getattr(torrent, "title", None), - torrent_size=getattr(torrent, "size", None), - downloader=downloader, - source=str(source)[:1000] if source else None, - error_message=str(error_msg or "")[:1000], + repository: DownloadFailureRepository = get_chain_download_failure_port() + repository.record_failure( + DownloadFailureWrite( + fingerprint=fingerprint, + failed_at=now_time, + next_retry_at=next_retry_at, + media_type=getattr( + getattr(media, "type", None), + "value", + getattr(media, "type", None), + ), + title=getattr(media, "title", None), + year=getattr(media, "year", None), + media_source=media_source, + media_id=media_id, + seasons=( + str(getattr(meta, "season", None)) + if getattr(meta, "season", None) is not None + else None + ), + episodes=( + episode_rules.format_ranges(list(episodes)) + if episodes + else self._format_failure_episodes(meta) + ), + site=site if isinstance(site, int) else None, + site_name=getattr(torrent, "site_name", None), + torrent_id=self._torrent_resource_key(torrent), + torrent_name=getattr(torrent, "title", None), + torrent_size=getattr(torrent, "size", None), + downloader=downloader, + source=str(source)[:1000] if source else None, + error_message=str(error_msg or "")[:1000], + ) ) except Exception as err: logger.error(f"记录下载失败冷却失败：{str(err)}") @@ -908,7 +922,7 @@ class DownloadChain(ChainBase): self, contexts: List[Context], source: Optional[str], - ) -> Dict[str, "DownloadFailure"]: + ) -> Dict[str, DownloadFailureSnapshot]: """ 查询当前订阅候选中仍处于冷却期的失败记录，返回指纹到失败记录的映射。 """ @@ -926,7 +940,8 @@ class DownloadChain(ChainBase): return {} now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) try: - return get_chain_download_failure_port().get_active_by_fingerprints( + repository: DownloadFailureRepository = get_chain_download_failure_port() + return repository.get_active_by_fingerprints( fingerprints=fingerprints, now_time=now_time, ) except Exception as err: @@ -936,7 +951,7 @@ class DownloadChain(ChainBase): @staticmethod def _log_download_failure_cooldown( context: Context, - failure: Optional["DownloadFailure"], + failure: Optional[DownloadFailureSnapshot], ) -> None: """记录候选资源处于失败冷却期时的跳过原因和下次重试时间。""" reason = getattr(failure, "error_message", None) or "未知原因" @@ -957,7 +972,7 @@ class DownloadChain(ChainBase): contexts: List[Context], downloader: Optional[str], source: Optional[str], - ) -> Tuple[List[Context], Dict[str, "DownloadFailure"]]: + ) -> Tuple[List[Context], Dict[str, Optional[DownloadFailureSnapshot]]]: """ 执行批量下载前的资源选择、排序和失败冷却准备。 @@ -979,16 +994,20 @@ class DownloadChain(ChainBase): ) contexts = event_data.updated_contexts contexts = TorrentHelper().sort_torrents(contexts) - return contexts, self._active_download_failure_fingerprints( - contexts=contexts, - source=source, - ) + active_failures: Dict[str, Optional[DownloadFailureSnapshot]] = { + fingerprint: failure + for fingerprint, failure in self._active_download_failure_fingerprints( + contexts=contexts, + source=source, + ).items() + } + return contexts, active_failures def _download_movie_music_candidates( self, contexts: List[Context], downloaded_list: List[Context], - active_failure_records: Dict[str, "DownloadFailure"], + active_failure_records: Dict[str, Optional[DownloadFailureSnapshot]], save_path: Optional[str], channel: Optional[NotificationChannel], source: Optional[str], diff --git a/app/chain/mediaserver.py b/app/chain/mediaserver.py index 838b1dfdc..fb4466d49 100644 --- a/app/chain/mediaserver.py +++ b/app/chain/mediaserver.py @@ -3,7 +3,11 @@ from datetime import datetime from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Union from app.application.chain.data import get_chain_media_server_port -from app.application.mediaserver import get_mediaserver_configs +from app.application.mediaserver import ( + MediaServerRepository, + MediaServerSyncItem, + get_mediaserver_configs, +) from app.application.security.url import SecurityUtils from app.chain import ChainBase from app.runtime.log import logger @@ -271,6 +275,7 @@ class MediaServerChain(ChainBase): self, mediaservers: List[Any], server: Optional[str], + repository: MediaServerRepository, ) -> Tuple[List[Any], int, Dict[str, Any], Optional[int]]: """ 准备启用媒体服务器的同步库和进度总量。 @@ -281,8 +286,7 @@ class MediaServerChain(ChainBase): item.name for item in mediaservers if item and item.enabled and item.name ] - dboper = get_chain_media_server_port() - dboper.delete_excluded_servers(enabled_servers) + repository.delete_excluded_servers(enabled_servers) selected_servers = [ item for item in mediaservers if item and item.enabled and (not server or item.name == server) @@ -332,7 +336,7 @@ class MediaServerChain(ChainBase): server_name: str, selected_libraries: List[Any], library_media_counts: Dict[str, Optional[int]], - dboper: Any, + repository: MediaServerRepository, sync_time: str, progress_callback: Optional[Callable[..., None]], server_index: int, @@ -363,14 +367,17 @@ class MediaServerChain(ChainBase): item_type = self._normalize_item_type(item.item_type) if item_type == MediaType.TV.value: for episode in self.episodes(server_name, item.item_id) or []: - seasoninfo[episode.season] = episode.episodes - item_dict = item.model_dump() - item_dict.update({ - "seasoninfo": seasoninfo, - "item_type": item_type, - "lst_mod_date": sync_time, - }) - dboper.upsert(**item_dict) + if episode.season is None: + continue + seasoninfo[episode.season] = episode.episodes or [] + repository.upsert( + MediaServerSyncItem.from_item( + item, + item_type=item_type, + seasoninfo=seasoninfo, + sync_time=sync_time, + ) + ) if progress_callback: if global_media_total: progress_value = min(global_media_finished / global_media_total, 1) * 100 @@ -423,7 +430,10 @@ class MediaServerChain(ChainBase): "media_finished": global_media_finished, }, ) - stale_count = dboper.delete_stale(server=server_name, sync_time=sync_time) + stale_count = repository.delete_stale( + server=server_name, + sync_time=sync_time, + ) logger.info(f"媒体服务器 {server_name} 清理陈旧数据完成，删除数量：{stale_count}") return total_count, global_media_finished @@ -463,9 +473,9 @@ class MediaServerChain(ChainBase): with lock: # 汇总统计 total_count = 0 - dboper = get_chain_media_server_port() + repository = get_chain_media_server_port() mediaservers, total_servers, server_sync_contexts, global_media_total = ( - self._prepare_sync_contexts(mediaservers, server) + self._prepare_sync_contexts(mediaservers, server, repository) ) if progress_callback: progress_callback( @@ -543,7 +553,7 @@ class MediaServerChain(ChainBase): server_name=server_name, selected_libraries=selected_libraries, library_media_counts=library_media_counts, - dboper=dboper, + repository=repository, sync_time=sync_time, progress_callback=progress_callback, server_index=server_index, diff --git a/app/db/adapters/download.py b/app/db/adapters/download.py index 7eddae833..383480bcc 100644 --- a/app/db/adapters/download.py +++ b/app/db/adapters/download.py @@ -3,8 +3,13 @@ from __future__ import annotations from collections.abc import Callable -from typing import Any, cast +from sqlalchemy.orm import Session + +from app.application.download.failures import ( + DownloadFailureSnapshot, + DownloadFailureWrite, +) from app.db.oper.downloadfailure import DownloadFailureOper from app.db.uow import SqlAlchemyUnitOfWork @@ -12,7 +17,7 @@ from app.db.uow import SqlAlchemyUnitOfWork class TransactionalDownloadFailureRepository: """为 Chain 下载失败读写创建短生命周期会话并显式收口事务。""" - def __init__(self, session_factory: Callable[[], Any]) -> None: + def __init__(self, session_factory: Callable[[], Session]) -> None: """保存由启动组合根提供的同步会话工厂。""" self._session_factory = session_factory @@ -20,36 +25,48 @@ class TransactionalDownloadFailureRepository: self, fingerprints: list[str], now_time: str, - ) -> dict[str, Any]: + ) -> dict[str, DownloadFailureSnapshot]: """在独立只读会话中查询仍处于冷却期的失败记录。""" with self._session_factory() as session: - return cast( - dict[str, Any], - DownloadFailureOper(db=session).get_active_by_fingerprints( - fingerprints=fingerprints, - now_time=now_time, - ), + records = DownloadFailureOper(db=session).get_active_by_fingerprints( + fingerprints=fingerprints, + now_time=now_time, ) + return { + fingerprint: DownloadFailureSnapshot( + fingerprint=record.fingerprint, + error_message=record.error_message, + next_retry_at=record.next_retry_at, + ) + for fingerprint, record in records.items() + } - def record_failure( - self, - fingerprint: str, - now_time: str, - next_retry_at: str, - **kwargs: object, - ) -> Any: + def record_failure(self, failure: DownloadFailureWrite) -> None: """在一个显式 UoW 中新增或更新下载失败记录。""" with self._session_factory() as session: transaction = SqlAlchemyUnitOfWork(session) try: - failure = DownloadFailureOper(db=session).record_failure( - fingerprint=fingerprint, - now_time=now_time, - next_retry_at=next_retry_at, - **kwargs, + DownloadFailureOper(db=session).record_failure( + fingerprint=failure.fingerprint, + now_time=failure.failed_at, + next_retry_at=failure.next_retry_at, + type=failure.media_type, + title=failure.title, + year=failure.year, + media_source=failure.media_source, + media_id=failure.media_id, + seasons=failure.seasons, + episodes=failure.episodes, + site=failure.site, + site_name=failure.site_name, + torrent_id=failure.torrent_id, + torrent_name=failure.torrent_name, + torrent_size=failure.torrent_size, + downloader=failure.downloader, + source=failure.source, + error_message=failure.error_message, ) transaction.commit() - return failure except Exception: transaction.rollback() raise diff --git a/app/db/adapters/mediaserver.py b/app/db/adapters/mediaserver.py new file mode 100644 index 000000000..f5cca389c --- /dev/null +++ b/app/db/adapters/mediaserver.py @@ -0,0 +1,110 @@ +"""媒体服务器本地缓存的显式短会话与事务适配器。""" + +from __future__ import annotations + +import json +from collections.abc import Callable +from typing import Optional, TypeVar, Union + +from sqlalchemy.orm import Session + +from app.application.mediaserver import MediaServerSyncItem +from app.db.oper.mediaserver import MediaServerOper +from app.db.uow import SqlAlchemyUnitOfWork +from app.schemas.types import MediaSource + +T = TypeVar("T") + + +class TransactionalMediaServerRepository: + """让每次媒体库缓存查询或变更各自拥有一个短生命周期 Session。""" + + def __init__(self, session_factory: Callable[[], Session]) -> None: + """保存由启动组合根提供的同步会话工厂。""" + self._session_factory = session_factory + + def _read(self, operation: Callable[[MediaServerOper], T]) -> T: + """在独立只读会话中执行媒体库缓存查询。""" + with self._session_factory() as session: + return operation(MediaServerOper(db=session)) + + def _write(self, operation: Callable[[MediaServerOper], T]) -> T: + """在独立 UoW 中执行并提交一项媒体库缓存变更。""" + with self._session_factory() as session: + unit_of_work = SqlAlchemyUnitOfWork(session) + try: + result = operation(MediaServerOper(db=session)) + unit_of_work.commit() + return result + except Exception: + unit_of_work.rollback() + raise + + def get_item_id( + self, + *, + title: Optional[str] = None, + year: Optional[Union[str, int]] = None, + mtype: Optional[str] = None, + media_source: Optional[MediaSource] = None, + media_id: Optional[str] = None, + season: Optional[int] = None, + ) -> Optional[str]: + """在短会话中返回匹配条目的服务器 item_id。""" + item_id: Optional[str] = self._read( + lambda repository: repository.get_item_id( + title=title, + year=year, + mtype=mtype, + media_source=media_source, + media_id=media_id, + season=season, + ) + ) + return item_id + + def upsert(self, item: MediaServerSyncItem) -> bool: + """在单个短事务中新增或更新一个媒体库同步条目。""" + payload = { + "server": item.server, + "library": item.library, + "item_id": item.item_id, + "item_type": item.item_type, + "title": item.title, + "original_title": item.original_title, + "year": item.year, + "media_source": item.media_source, + "media_id": item.media_id, + "path": item.path, + "seasoninfo": { + season: list(episodes) + for season, episodes in item.seasoninfo + }, + "note": ( + json.loads(item.note_json) + if item.note_json is not None + else None + ), + "lst_mod_date": item.lst_mod_date, + } + created: bool = self._write( + lambda repository: repository.upsert(**payload) + ) + return created + + def delete_stale(self, *, server: str, sync_time: str) -> int: + """在短事务中删除指定服务器本轮未更新的条目。""" + deleted: int = self._write( + lambda repository: repository.delete_stale( + server=server, + sync_time=sync_time, + ), + ) + return deleted + + def delete_excluded_servers(self, servers: list[str]) -> int: + """在短事务中删除已停用或已移除服务器的条目。""" + deleted: int = self._write( + lambda repository: repository.delete_excluded_servers(servers) + ) + return deleted diff --git a/app/db/oper/mediaserver.py b/app/db/oper/mediaserver.py index 72ab46c6a..720cc949e 100644 --- a/app/db/oper/mediaserver.py +++ b/app/db/oper/mediaserver.py @@ -29,6 +29,12 @@ class MediaServerOper(DbOper): if hasattr(MediaServerItem, k) and k != "id" } + @staticmethod + def _contains_season(item: MediaServerItem, season: int) -> bool: + """兼容 JSON 往返后变为字符串的 seasoninfo 键。""" + seasoninfo = item.seasoninfo or {} + return season in seasoninfo or str(season) in seasoninfo + def add(self, **kwargs) -> bool: """ 新增媒体服务器数据 @@ -132,10 +138,7 @@ class MediaServerOper(DbOper): if kwargs.get("season") is not None: # 判断季是否存在 - if not item.seasoninfo: - return None - seasoninfo = item.seasoninfo or {} - if kwargs.get("season") not in seasoninfo.keys(): + if not self._contains_season(item, kwargs["season"]): return None return item @@ -169,10 +172,7 @@ class MediaServerOper(DbOper): if kwargs.get("season") is not None: # 判断季是否存在 - if not item.seasoninfo: - return None - seasoninfo = item.seasoninfo or {} - if kwargs.get("season") not in seasoninfo.keys(): + if not self._contains_season(item, kwargs["season"]): return None return item diff --git a/app/startup/initializers/modules.py b/app/startup/initializers/modules.py index 02aad1b9b..9085830d1 100644 --- a/app/startup/initializers/modules.py +++ b/app/startup/initializers/modules.py @@ -114,6 +114,7 @@ from app.application.workflow import ( from app.command import CommandChain from app.db.adapters.chain import TransactionalChainDurableEventWriter from app.db.adapters.download import TransactionalDownloadFailureRepository +from app.db.adapters.mediaserver import TransactionalMediaServerRepository from app.db.adapters.outbox import SqlAlchemyAsyncOutboxStager, SqlAlchemyOutboxRepository from app.db.adapters.query import SqlAlchemyDataQueryAdapter from app.db.adapters.site import TransactionalSiteRepository @@ -892,7 +893,7 @@ async def init_modules() -> HostRuntime: transfer_execution=lambda: TransactionalTransferExecutionRepository( SessionFactory ), - media_server=lambda: MediaServerOper(), + media_server=lambda: TransactionalMediaServerRepository(SessionFactory), download_failure=lambda: TransactionalDownloadFailureRepository( SessionFactory ), diff --git a/docs/architecture-optimization-checklist.md b/docs/architecture-optimization-checklist.md index fe0bee642..81a776c06 100644 --- a/docs/architecture-optimization-checklist.md +++ b/docs/architecture-optimization-checklist.md @@ -69,7 +69,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 850 / 6,944 | `dependency-baseline.json` 当前快照 | +| 宿主 Python 模块 / 内部依赖边 | 852 / 6,962 | `dependency-baseline.json` 当前快照 | | 非平凡 SCC | 2 | 新增 Chain 包根环；另一个是隔离的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | @@ -77,9 +77,9 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement | | Python 源码量 | 约 271,400 行 | 60 个文件超过 1,000 行，14 个超过 2,000 行 | | 长方法 | 281 个超过 80 行 | 67 个超过 150 行，23 个超过 250 行；大量是私有方法 | -| 全量 mypy 历史债务 | 11,809 / 596 文件 | strict frontier 当前覆盖 41 个文件，本批迁移路径的类型债务已清零 | -| Ruff 历史诊断 | 875 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | -| 覆盖率低水位 | Application 78.89%，Domain 79.29% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | +| 全量 mypy 历史债务 | 11,808 / 596 文件 | strict frontier 当前覆盖 41 个文件，本批迁移路径的类型债务已清零 | +| Ruff 历史诊断 | 872 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | +| 覆盖率低水位 | Application 78.95%，Domain 79.29% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | ### 3.3 热点文件 @@ -291,6 +291,8 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` `WorkflowOper` 只保留显式 Session query/stage，旧无 Session 五方法只存在于 SDK Legacy/Compat。 - [x] 删除 Chain registry 中零消费者 `*PortProxy`/动态转发和 `ChainRuntimeContext.data_ports` 伪注入；Workflow 执行服务只在 Application owner 配置一次，不再重复注册到 `ChainDataPorts`。 +- [x] DownloadFailure/MediaServer 两个 registry 字段改用冻结 DTO 与 typed Repository factory； + ORM 不越过短 Session，媒体库远端枚举期间不持有事务，旧 Oper/Compat 与公开 Chain ABI 保持不变。 - [ ] `ChainDataPorts`/`AgentDataPorts` 可暂时保留为兼容聚合器，但字段必须显式、可类型检查。 - [ ] 以一个业务纵切面迁移并验证后，再迁移下一组，禁止一次替换所有 Oper。 - [ ] 增加 AST 门禁，禁止向 `ChainDataPorts`、`AgentDataPorts` 和新的 canonical use-case service diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index a366a3661..ffa181b24 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -697,15 +697,15 @@ flowchart LR SDK 导出（若公开）、`docs/rules/05-architecture.md` 与上述架构测试。 - 延迟导入不被接受为隐藏循环依赖的手段。 -### 10.1 2026-08-27 当前收口状态与后续边界 +### 10.1 2026-08-28 当前收口状态与后续边界 当前宿主架构基线（排除 `app/plugins/**`）如下；数字来自 `tests/fixtures/architecture/`，更新基线前必须先审查语义变化： | 指标 | 当前值 | |---|---:| -| Python 模块 | 850 | -| 内部导入边 | 6,944 | +| Python 模块 | 852 | +| 内部导入边 | 6,962 | | 非平凡 SCC | 2（`ARCH-107` 临时 Chain 包根环；精确 containment 的 TMDB 移植包环） | | Direct egress | 66（12 条待迁移债务，54 条精确 containment） | | Module Contract V2 spec | 217（其中 215 个进入 `run_module` 观察面） | diff --git a/docs/architecture-refactor-roadmap.md b/docs/architecture-refactor-roadmap.md index db155a3d3..b061cb042 100644 --- a/docs/architecture-refactor-roadmap.md +++ b/docs/architecture-refactor-roadmap.md @@ -102,7 +102,7 @@ canonical 主程序；兼容只经统一 Compat/SDK 门面提供。 | S1-L3.1 Workflow typed execution | `DELIVERED` | S1-L2 | `17d8be2af`、`b33b29876`：Chain 直连类型化事务服务且单次执行只取一个 port；canonical Oper 删除旧 writer/无 Session 写方法，旧 ABI 只在 `_legacy/workflow.py` 与 Compat overlay；Unit Tests `33103913838`、Pylint `33103913935` 全绿，Application 覆盖率低水位提升至 `78.79%` | | S1-L3.2 Chain registry/DI | `ACTIVE` | S1-L3.1 | 显式类型化 factory，删除 PortProxy 与失效的双重注入，构造器注入真实控制调用 | | S1-L3.2.1 Registry hygiene | `DELIVERED` | S1-L3.1 | `ac7a20132`：删除零消费者 PortProxy/动态转发和 `ChainRuntimeContext.data_ports` 伪注入；Workflow 退出 Chain registry，只保留 Application owner 单一配置入口；Unit Tests `33120205586`、Pylint `33120205581` 全绿 | -| S1-L3.3 DownloadFailure/MediaServer | `PLANNED` | S1-L3.2 | 两组窄 DTO/Port/adapter 清零 raw Oper，不跨远端 I/O 持有 Session | +| S1-L3.3 DownloadFailure/MediaServer | `VERIFIED` | S1-L3.2 | 两组 raw factory 已替换为冻结 DTO/typed Port；失败冷却在 Session 内投影，媒体库查询只返回标量且每个 upsert/cleanup 独立短事务，远端枚举不持有 Session；旧 Oper 与插件可见 Chain ABI 保持不变 | | S1-L3.4 User | `PLANNED` | S1-L3.3 | 认证、偏好与渠道绑定投影冻结快照，User Chain/Agent 不接收 ORM | | S1-L3.5 History | `PLANNED` | S1-L3.4 | Download/Transfer history 统一 typed query/mutation，删除下载历史双事务 fail-open | | S1-L3.6 Site | `PLANNED` | S1-L3.5 | 复用 Site query/health，补齐同步 typed command，Session 内完成 DTO 投影 | @@ -153,7 +153,7 @@ canonical 主程序；兼容只经统一 Compat/SDK 门面提供。 | S4-L2 Event strict contract | `PLANNED` | S0-L2.6,S1-L6 | 宿主事件输入/输出按风险 strict，诊断例外只属于第三方插件兼容 | | S4-L3 Complexity v2 | `PLANNED` | S3 | 私有方法、class/file、圈复杂度进入门禁；所有超限通过职责拆分归零 | | S4-L4 全量 mypy 清零 | `PLANNED` | S3,S4-L1,S4-L2 | `mypy-baseline.json` 归零并删除债务接受路径，全宿主 strict 类型通过 | -| S4-L5 Ruff 治理债务清零 | `PLANNED` | S3 | 当前受控 875 条诊断归零，规则集扩展经过独立审查且新增诊断为零 | +| S4-L5 Ruff 治理债务清零 | `PLANNED` | S3 | 当前受控 872 条诊断归零，规则集扩展经过独立审查且新增诊断为零 | | S4-L6 Coverage/并发/质量证据 | `PLANNED` | S3,S4-L1,S4-L2 | 高风险包纳入 coverage；raw concurrency 分类清零；Module Quality 有真实 evidence test | ### S5：Plugin、Agent、Domain、Startup 与最终收口 diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index ca0762deb..fd8685081 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -63,7 +63,7 @@ to make the directory tree look symmetrical. | `app/application/*.py` | Established single-module application services and compatibility facades | | `app/application/subscription/` | Subscription use cases: `write.py` owns media-to-row translation and the write port; `contract.py` owns shared metadata/media-key projection; query, mutation, deletion, identity and search stay in their single-word modules | | `app/application/search/` | Search state and later search-plan use cases | -| `app/application/download/` | Download task querying/control and later submission use cases | +| `app/application/download/` | Download task querying/control and selection use cases; `failures.py` owns the frozen failure-cooldown write/query DTOs and persistence Port | | `app/application/music/` | Multi-source music catalog orchestration | | `app/application/chain/` | Injectable Chain runtime capabilities: `context.py` owns the runtime dependency aggregate, `data.py` owns named persistence ports, and `events.py` owns durable event write contracts plus replayable payload conversion | | `app/application/agentdata.py` | Named Agent data ports; canonical Agent consumers use `get_agent_*_port()` and do not alias legacy proxies to Oper classes | @@ -143,6 +143,9 @@ Chain consumers temporarily use the named `get_chain_*_port()` functions from The retired migration-time `*PortProxy` classes and dynamic `__getattr__` forwarding must not be recreated; they had no host, SDK or plugin consumers. Workflow execution uses its owning `app.application.workflow` service directly and must not be registered again in `ChainDataPorts`. +Download-failure cooldown and media-server cache consumers use their typed Application DTO/Port +contracts. Their DB adapters project ORM values before Session close and commit each local write +in a separate short UoW, so remote media enumeration never holds a database transaction. Agent orchestration, memory and tool implementations follow the same rule via the named `get_agent_*_port()` functions from `app/application/agentdata.py`. The legacy Agent `*Port` proxy classes remain import-compatible boundaries and @@ -671,6 +674,9 @@ driven workflow registration. | `app/application/agent.py` | Agent orchestration facade (`get_agent_manager` / `get_prompt_manager` / capability queries / prompt builders); lightweight providers register through `app/startup/initializers/agent.py`, with no static `application -> agent` edge | | `app/agent/runtime_loader.py` | Agent-specific capability discovery and canonical entrypoint/service materialization; reuses the generic Capability Runtime while keeping Agent ownership under `app/agent/` | | `app/application/subscription/write.py` | Subscription media translation and sync/async write-port orchestration | +| `app/application/download/failures.py` | Frozen download-failure cooldown write/query DTOs and Chain persistence Port | +| `app/db/adapters/download.py` | Short-session download-failure snapshot and mutation adapter | +| `app/db/adapters/mediaserver.py` | Per-operation media-server cache query/upsert/cleanup transaction adapter | | `app/application/outbox.py` | Durable intent, topic handler and Outbox repository contracts | | `app/db/adapters/outbox.py` | SQLAlchemy Outbox persistence, claim/lease and retry state adapter | | `app/application/chain/events.py` | Chain durable-event write port, settlement projection and replayable payload conversion | diff --git a/tests/conftest.py b/tests/conftest.py index d7078c0cd..1446e2c5d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -223,6 +223,7 @@ def configure_plugin_system_services(): configure_agent_task_execution, ) from app.db.adapters.download import TransactionalDownloadFailureRepository + from app.db.adapters.mediaserver import TransactionalMediaServerRepository from app.db.adapters.site import TransactionalSiteRepository from app.db.adapters.subscription import TransactionalSubscribeWriter from app.db.adapters.transaction import TransactionalWriteRunner @@ -312,7 +313,7 @@ def configure_plugin_system_services(): transfer_execution=lambda: TransactionalTransferExecutionRepository( SessionFactory ), - media_server=lambda: MediaServerOper(), + media_server=lambda: TransactionalMediaServerRepository(SessionFactory), download_failure=lambda: TransactionalDownloadFailureRepository( SessionFactory ), diff --git a/tests/fixtures/architecture/coverage-baseline.json b/tests/fixtures/architecture/coverage-baseline.json index 4114fab87..39714e741 100644 --- a/tests/fixtures/architecture/coverage-baseline.json +++ b/tests/fixtures/architecture/coverage-baseline.json @@ -1,8 +1,8 @@ { "application": { - "covered_lines": 10084, - "percent": 78.89, - "statements": 12782 + "covered_lines": 10133, + "percent": 78.95, + "statements": 12834 }, "domain": { "covered_lines": 3392, diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index d9e0c7111..53a4230a0 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1441,8 +1441,8 @@ "runtime_only": true } }, - "edge_count": 6944, - "edge_sha256": "9f6be750c55150ef9e061f54ade997c328561b7950f6ec44bce6ada373d01f3e", + "edge_count": 6962, + "edge_sha256": "d78cc1aa6f3837310c7460d2ea17873222b4b50708012ae2f3db249d8197f77e", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -3984,6 +3984,9 @@ "app.application.chain.context -> app.runtime", "app.application.chain.context -> app.runtime.stop", "app.application.chain.data -> app.application", + "app.application.chain.data -> app.application.download", + "app.application.chain.data -> app.application.download.failures", + "app.application.chain.data -> app.application.mediaserver", "app.application.chain.data -> app.application.transfer", "app.application.chain.data -> app.application.transfer.execution", "app.application.chain.data -> app.application.transfer.workflow", @@ -4020,6 +4023,8 @@ "app.application.directory -> app.schemas.file", "app.application.directory -> app.schemas.system", "app.application.directory -> app.schemas.types", + "app.application.download.failures -> app.schemas", + "app.application.download.failures -> app.schemas.types", "app.application.download.selection -> app.domain", "app.application.download.selection -> app.domain.context", "app.application.download.selection -> app.schemas", @@ -4640,6 +4645,7 @@ "app.chain.download -> app.application.configuration", "app.chain.download -> app.application.directory", "app.chain.download -> app.application.download", + "app.chain.download -> app.application.download.failures", "app.chain.download -> app.application.download.selection", "app.chain.download -> app.application.download.tasks", "app.chain.download -> app.application.torrent", @@ -5134,10 +5140,21 @@ "app.db.adapters.chain -> app.db.oper.transferpending", "app.db.adapters.chain -> app.db.oper.transfersettlementreceipt", "app.db.adapters.chain -> app.db.uow", + "app.db.adapters.download -> app.application", + "app.db.adapters.download -> app.application.download", + "app.db.adapters.download -> app.application.download.failures", "app.db.adapters.download -> app.db", "app.db.adapters.download -> app.db.oper", "app.db.adapters.download -> app.db.oper.downloadfailure", "app.db.adapters.download -> app.db.uow", + "app.db.adapters.mediaserver -> app.application", + "app.db.adapters.mediaserver -> app.application.mediaserver", + "app.db.adapters.mediaserver -> app.db", + "app.db.adapters.mediaserver -> app.db.oper", + "app.db.adapters.mediaserver -> app.db.oper.mediaserver", + "app.db.adapters.mediaserver -> app.db.uow", + "app.db.adapters.mediaserver -> app.schemas", + "app.db.adapters.mediaserver -> app.schemas.types", "app.db.adapters.outbox -> app.application", "app.db.adapters.outbox -> app.application.outbox", "app.db.adapters.outbox -> app.db", @@ -8020,6 +8037,7 @@ "app.startup.initializers.modules -> app.db.adapters", "app.startup.initializers.modules -> app.db.adapters.chain", "app.startup.initializers.modules -> app.db.adapters.download", + "app.startup.initializers.modules -> app.db.adapters.mediaserver", "app.startup.initializers.modules -> app.db.adapters.outbox", "app.startup.initializers.modules -> app.db.adapters.pluginidentity", "app.startup.initializers.modules -> app.db.adapters.plugininstallation", @@ -8389,7 +8407,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 850, + "module_count": 852, "modules": [ "app", "app.adapters", @@ -8652,6 +8670,7 @@ "app.application.database", "app.application.directory", "app.application.download", + "app.application.download.failures", "app.application.download.selection", "app.application.download.tasks", "app.application.downloader", @@ -8783,6 +8802,7 @@ "app.db.adapters", "app.db.adapters.chain", "app.db.adapters.download", + "app.db.adapters.mediaserver", "app.db.adapters.outbox", "app.db.adapters.pluginidentity", "app.db.adapters.plugininstallation", diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index fe44137ab..94ac1fdbe 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -1422,7 +1422,7 @@ "call-overload": 1, "comparison-overlap": 1, "misc": 1, - "no-any-return": 4, + "no-any-return": 3, "no-redef": 1, "no-untyped-call": 7, "no-untyped-def": 6, diff --git a/tests/fixtures/architecture/ruff-baseline.json b/tests/fixtures/architecture/ruff-baseline.json index eb1c97429..5b72fd7f1 100644 --- a/tests/fixtures/architecture/ruff-baseline.json +++ b/tests/fixtures/architecture/ruff-baseline.json @@ -312,9 +312,6 @@ "app/application/maintenance.py": { "I001": 1 }, - "app/application/mediaserver.py": { - "I001": 1 - }, "app/application/messaging/agent.py": { "I001": 1 }, @@ -1276,9 +1273,6 @@ "tests/test_douban_recognition.py": { "I001": 1 }, - "tests/test_download_chain.py": { - "I001": 1 - }, "tests/test_download_paths_endpoint.py": { "I001": 1 }, @@ -1410,9 +1404,6 @@ "tests/test_mediaserver_image_signing.py": { "I001": 1 }, - "tests/test_mediaserver_sync_incremental.py": { - "I001": 1 - }, "tests/test_mediaserver_sync_scheduler.py": { "I001": 1 }, diff --git a/tests/fixtures/architecture/startup-performance-baseline.json b/tests/fixtures/architecture/startup-performance-baseline.json index bd807e6c0..bd9c36aff 100644 --- a/tests/fixtures/architecture/startup-performance-baseline.json +++ b/tests/fixtures/architecture/startup-performance-baseline.json @@ -1,41 +1,41 @@ { "schema_version": 2, - "generated_at": "2026-08-27T13:33:52.081319+00:00", + "generated_at": "2026-08-27T23:28:48.301843+00:00", "platform": "macOS-26.5.2-arm64-arm-64bit-Mach-O", "python": "3.14.3", "repeat": 3, "targets": { "app.startup.lifecycle": { - "loaded_app_module_count": 392, - "max_ms": 1035.341, - "median_ms": 993.245, - "min_ms": 991.998, + "loaded_app_module_count": 394, + "max_ms": 1298.973, + "median_ms": 999.845, + "min_ms": 994.938, "samples_ms": [ - 1035.341, - 991.998, - 993.245 + 1298.973, + 999.845, + 994.938 ] }, "app.factory": { - "loaded_app_module_count": 404, - "max_ms": 1006.172, - "median_ms": 1005.88, - "min_ms": 1003.841, + "loaded_app_module_count": 406, + "max_ms": 1079.626, + "median_ms": 1019.979, + "min_ms": 1014.804, "samples_ms": [ - 1005.88, - 1006.172, - 1003.841 + 1019.979, + 1014.804, + 1079.626 ] }, "app.main": { - "loaded_app_module_count": 406, - "max_ms": 1085.652, - "median_ms": 1055.994, - "min_ms": 1043.212, + "loaded_app_module_count": 408, + "max_ms": 1075.788, + "median_ms": 1060.558, + "min_ms": 1057.105, "samples_ms": [ - 1055.994, - 1043.212, - 1085.652 + 1075.788, + 1060.558, + 1057.105 ] } }, @@ -47,56 +47,25 @@ { "mode": "normal", "enabled_component_count": 25, - "startup_ms": 0.702, - "full_lifespan_ms": 1.54, + "startup_ms": 0.674, + "full_lifespan_ms": 1.563, "stage_ms": { - "后台任务登记器": 0.09, - "数据库准备": 0.048, - "HTTP 基础能力": 0.03, - "领域依赖装配": 0.033, - "数据库引擎预热": 0.031, - "数据库连接预算": 0.031, - "路由": 0.028, - "模块服务": 0.022, - "插件备份恢复": 0.021, - "插件": 0.025, - "定时器": 0.023, - "监控器": 0.021, - "待处理整理回放": 0.025, - "命令服务": 0.021, - "工作流": 0.023, - "插件同步与启动收尾": 0.024 - }, - "threads_before": 2, - "threads_started": 2, - "threads_after": 2, - "tasks_before": 1, - "tasks_started": 1, - "tasks_after": 1, - "database_connections_started": 0 - }, - { - "mode": "normal", - "enabled_component_count": 25, - "startup_ms": 0.667, - "full_lifespan_ms": 1.559, - "stage_ms": { - "后台任务登记器": 0.078, - "数据库准备": 0.037, - "HTTP 基础能力": 0.027, - "领域依赖装配": 0.03, - "数据库引擎预热": 0.027, + "后台任务登记器": 0.081, + "数据库准备": 0.039, + "HTTP 基础能力": 0.031, + "领域依赖装配": 0.029, + "数据库引擎预热": 0.028, "数据库连接预算": 0.024, "路由": 0.026, "模块服务": 0.026, - "插件备份恢复": 0.026, + "插件备份恢复": 0.024, "插件": 0.025, - "定时器": 0.024, - "监控器": 0.024, - "待处理整理回放": 0.027, + "定时器": 0.026, + "监控器": 0.023, + "待处理整理回放": 0.021, "命令服务": 0.026, "工作流": 0.023, - "插件同步与启动收尾": 0.021 + "插件同步与启动收尾": 0.025 }, "threads_before": 2, "threads_started": 2, @@ -109,24 +78,55 @@ { "mode": "normal", "enabled_component_count": 25, - "startup_ms": 0.735, - "full_lifespan_ms": 1.57, + "startup_ms": 0.662, + "full_lifespan_ms": 1.526, "stage_ms": { - "后台任务登记器": 0.103, - "数据库准备": 0.041, + "后台任务登记器": 0.073, + "数据库准备": 0.036, + "HTTP 基础能力": 0.034, + "领域依赖装配": 0.028, + "数据库引擎预热": 0.024, + "数据库连接预算": 0.025, + "路由": 0.025, + "模块服务": 0.024, + "插件备份恢复": 0.027, + "插件": 0.024, + "定时器": 0.025, + "监控器": 0.024, + "待处理整理回放": 0.023, + "命令服务": 0.023, + "工作流": 0.023, + "插件同步与启动收尾": 0.023 + }, + "threads_before": 2, + "threads_started": 2, + "threads_after": 2, + "tasks_before": 1, + "tasks_started": 1, + "tasks_after": 1, + "database_connections_started": 0 + }, + { + "mode": "normal", + "enabled_component_count": 25, + "startup_ms": 0.651, + "full_lifespan_ms": 1.495, + "stage_ms": { + "后台任务登记器": 0.077, + "数据库准备": 0.038, "HTTP 基础能力": 0.031, - "领域依赖装配": 0.032, - "数据库引擎预热": 0.028, - "数据库连接预算": 0.026, - "路由": 0.027, - "模块服务": 0.026, - "插件备份恢复": 0.026, + "领域依赖装配": 0.028, + "数据库引擎预热": 0.025, + "数据库连接预算": 0.024, + "路由": 0.026, + "模块服务": 0.025, + "插件备份恢复": 0.023, "插件": 0.025, - "定时器": 0.024, - "监控器": 0.025, + "定时器": 0.026, + "监控器": 0.024, "待处理整理回放": 0.024, - "命令服务": 0.024, - "工作流": 0.024, + "命令服务": 0.021, + "工作流": 0.022, "插件同步与启动收尾": 0.024 }, "threads_before": 2, @@ -138,8 +138,8 @@ "database_connections_started": 0 } ], - "median_startup_ms": 0.702, - "median_full_lifespan_ms": 1.559, + "median_startup_ms": 0.662, + "median_full_lifespan_ms": 1.526, "enabled_component_count": 25, "enabled_components": [ "后台任务登记器", @@ -174,66 +174,66 @@ { "mode": "safe", "enabled_component_count": 13, - "startup_ms": 0.527, - "full_lifespan_ms": 0.956, + "startup_ms": 0.485, + "full_lifespan_ms": 0.915, + "stage_ms": { + "后台任务登记器": 0.08, + "数据库准备": 0.038, + "HTTP 基础能力": 0.03, + "领域依赖装配": 0.031, + "数据库引擎预热": 0.026, + "数据库连接预算": 0.023, + "路由": 0.023, + "模块服务": 0.025, + "插件同步与启动收尾": 0.022 + }, + "threads_before": 2, + "threads_started": 2, + "threads_after": 2, + "tasks_before": 1, + "tasks_started": 1, + "tasks_after": 1, + "database_connections_started": 0 + }, + { + "mode": "safe", + "enabled_component_count": 13, + "startup_ms": 0.548, + "full_lifespan_ms": 0.942, "stage_ms": { "后台任务登记器": 0.091, - "数据库准备": 0.042, - "HTTP 基础能力": 0.03, - "领域依赖装配": 0.03, - "数据库引擎预热": 0.028, - "数据库连接预算": 0.027, - "路由": 0.026, - "模块服务": 0.026, - "插件同步与启动收尾": 0.025 - }, - "threads_before": 2, - "threads_started": 2, - "threads_after": 2, - "tasks_before": 1, - "tasks_started": 1, - "tasks_after": 1, - "database_connections_started": 0 - }, - { - "mode": "safe", - "enabled_component_count": 13, - "startup_ms": 0.523, - "full_lifespan_ms": 0.954, - "stage_ms": { - "后台任务登记器": 0.082, - "数据库准备": 0.045, - "HTTP 基础能力": 0.036, - "领域依赖装配": 0.033, - "数据库引擎预热": 0.026, - "数据库连接预算": 0.024, - "路由": 0.023, - "模块服务": 0.024, - "插件同步与启动收尾": 0.023 - }, - "threads_before": 2, - "threads_started": 2, - "threads_after": 2, - "tasks_before": 1, - "tasks_started": 1, - "tasks_after": 1, - "database_connections_started": 0 - }, - { - "mode": "safe", - "enabled_component_count": 13, - "startup_ms": 0.515, - "full_lifespan_ms": 0.934, - "stage_ms": { - "后台任务登记器": 0.084, - "数据库准备": 0.044, - "HTTP 基础能力": 0.029, - "领域依赖装配": 0.027, - "数据库引擎预热": 0.031, - "数据库连接预算": 0.028, - "路由": 0.023, + "数据库准备": 0.04, + "HTTP 基础能力": 0.042, + "领域依赖装配": 0.029, + "数据库引擎预热": 0.023, + "数据库连接预算": 0.029, + "路由": 0.024, "模块服务": 0.022, - "插件同步与启动收尾": 0.025 + "插件同步与启动收尾": 0.021 + }, + "threads_before": 2, + "threads_started": 2, + "threads_after": 2, + "tasks_before": 1, + "tasks_started": 1, + "tasks_after": 1, + "database_connections_started": 0 + }, + { + "mode": "safe", + "enabled_component_count": 13, + "startup_ms": 0.496, + "full_lifespan_ms": 0.923, + "stage_ms": { + "后台任务登记器": 0.077, + "数据库准备": 0.035, + "HTTP 基础能力": 0.035, + "领域依赖装配": 0.026, + "数据库引擎预热": 0.025, + "数据库连接预算": 0.024, + "路由": 0.024, + "模块服务": 0.024, + "插件同步与启动收尾": 0.026 }, "threads_before": 2, "threads_started": 2, @@ -244,8 +244,8 @@ "database_connections_started": 0 } ], - "median_startup_ms": 0.523, - "median_full_lifespan_ms": 0.954, + "median_startup_ms": 0.496, + "median_full_lifespan_ms": 0.923, "enabled_component_count": 13, "enabled_components": [ "后台任务登记器", diff --git a/tests/test_architecture_dependencies.py b/tests/test_architecture_dependencies.py index 51b6ad743..8eb502cdf 100644 --- a/tests/test_architecture_dependencies.py +++ b/tests/test_architecture_dependencies.py @@ -444,6 +444,53 @@ def test_chain_registry_has_no_dynamic_proxies_or_dead_context_injection(): assert "data_ports=" not in startup_source +def test_download_failure_and_mediaserver_ports_are_typed_and_detached(): + """下载失败与媒体库缓存端口不得退回 raw Oper、Any 或 ORM 投影。""" + data_path = APP_ROOT / "application" / "chain" / "data.py" + data_tree = ast.parse( + data_path.read_text(encoding="utf-8"), + filename=str(data_path), + ) + data_class = next( + node + for node in data_tree.body + if isinstance(node, ast.ClassDef) and node.name == "ChainDataPorts" + ) + annotations = { + node.target.id: ast.unparse(node.annotation) + for node in data_class.body + if isinstance(node, ast.AnnAssign) + and isinstance(node.target, ast.Name) + } + returns = { + node.name: ast.unparse(node.returns) + for node in data_tree.body + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and node.returns is not None + } + assert annotations["download_failure"] == "DownloadFailureRepositoryFactory" + assert annotations["media_server"] == "MediaServerRepositoryFactory" + assert returns["get_chain_download_failure_port"] == "DownloadFailureRepository" + assert returns["get_chain_media_server_port"] == "MediaServerRepository" + + download_source = (APP_ROOT / "chain" / "download.py").read_text( + encoding="utf-8" + ) + mediaserver_source = (APP_ROOT / "chain" / "mediaserver.py").read_text( + encoding="utf-8" + ) + application_source = ( + APP_ROOT / "application" / "mediaserver.py" + ).read_text(encoding="utf-8") + startup_source = ( + APP_ROOT / "startup" / "initializers" / "modules.py" + ).read_text(encoding="utf-8") + assert "DownloadFailure = Any" not in download_source + assert "dboper" not in mediaserver_source + assert "async def async_get_item_id(" in application_source + assert "TransactionalMediaServerRepository(SessionFactory)" in startup_source + + def test_canonical_workflow_oper_has_no_legacy_writer_or_duplicate_exports(): """工作流旧写入口只能存在于 SDK Legacy facade。""" oper_path = APP_ROOT / "db" / "oper" / "workflow.py" diff --git a/tests/test_db_mediaserver_queries.py b/tests/test_db_mediaserver_queries.py index de7da61bb..466a14144 100644 --- a/tests/test_db_mediaserver_queries.py +++ b/tests/test_db_mediaserver_queries.py @@ -158,6 +158,32 @@ def test_exists_by_title_matches_async_twin(db): assert (sync_found is None) == (async_found is None) +def test_oper_season_lookup_accepts_json_string_keys(db): + """跨 Session 后的 JSON 季号键为字符串，查询仍应按整数季号命中。""" + item = _item("emby", "season-json", item_type="电视剧", media_id="season-1") + item.seasoninfo = {"1": [1, 2]} + db.add(item) + + assert MediaServerOper(db.session).get_item_id( + media_source=MediaSource.TMDB, + media_id="season-1", + mtype="电视剧", + season=1, + ) == "season-json" + + async def check() -> None: + """验证异步查询与同步查询采用相同的 JSON 季号兼容规则。""" + async with async_session_scope() as session: + assert await MediaServerOper(session).async_get_item_id( + media_source=MediaSource.TMDB, + media_id="season-1", + mtype="电视剧", + season=1, + ) == "season-json" + + asyncio.run(check()) + + def test_empty_clears_only_the_given_server(db): """ 指定服务器时只清空该服务器的条目，不给则清空全表。 diff --git a/tests/test_download_chain.py b/tests/test_download_chain.py index 29568c01f..49a1f7499 100644 --- a/tests/test_download_chain.py +++ b/tests/test_download_chain.py @@ -1,3 +1,4 @@ +from dataclasses import asdict from pathlib import Path from types import SimpleNamespace from unittest.mock import MagicMock @@ -5,10 +6,14 @@ from unittest.mock import MagicMock import pytest import app.chain.download as download_module +from app.application.download.failures import ( + DownloadFailureSnapshot, + DownloadFailureWrite, +) from app.chain.download import DownloadChain -from app.runtime.config import settings from app.domain.context import Context, MediaInfo, SubtitleInfo, TorrentInfo from app.domain.metainfo import MetaInfo +from app.runtime.config import settings from app.schemas.file import FileItem from app.schemas.mediaserver import NotExistMediaInfo from app.schemas.system import TransferDirectoryConf @@ -804,12 +809,11 @@ def test_download_single_records_failure_cooldown_when_downloader_rejects(monkey 捕获下载失败冷却记录，避免测试写入数据库。 """ - def record_failure(self, **kwargs: object) -> SimpleNamespace: + def record_failure(self, failure: DownloadFailureWrite) -> None: """ 保存写入字段供断言使用。 """ - captured.update(kwargs) - return SimpleNamespace(id=1) + captured.update(asdict(failure)) monkeypatch.setattr( "app.application.directory.DirectoryHelper.get_download_dirs", @@ -858,11 +862,95 @@ def test_download_single_records_failure_cooldown_when_downloader_rejects(monkey assert download_id is None assert returned_error == error_msg - assert captured["fingerprint"] == DownloadChain._build_download_failure_fingerprint(context) + assert captured["fingerprint"] == DownloadChain._build_download_failure_fingerprint( + context + ) assert captured["torrent_id"] == "example.com:id=484660" assert captured["site"] == 12 assert captured["error_message"] == error_msg - assert captured["next_retry_at"] > captured["now_time"] + assert captured["next_retry_at"] > captured["failed_at"] + + +def test_download_failure_query_skips_non_subscribe_sources(monkeypatch): + """非订阅下载不得读取失败冷却，避免改变手工下载行为。""" + get_repository = MagicMock(side_effect=AssertionError("unexpected query")) + monkeypatch.setattr( + download_module, + "get_chain_download_failure_port", + get_repository, + ) + + chain = DownloadChain.__new__(DownloadChain) + + assert chain._active_download_failure_fingerprints( + contexts=[_build_tv_context()], + source="Manual", + ) == {} + get_repository.assert_not_called() + + +def test_download_failure_query_failure_is_fail_open(monkeypatch): + """冷却查询失败时继续下载候选，并记录一次明确错误。""" + + class _FailingDownloadFailureRepository: + """模拟失败的下载冷却查询端口。""" + + def get_active_by_fingerprints( + self, + fingerprints: list[str], + now_time: str, + ) -> dict[str, DownloadFailureSnapshot]: + """拒绝查询以验证 Chain 的 fail-open 语义。""" + assert fingerprints + assert now_time + raise RuntimeError("query failed") + + error = MagicMock() + monkeypatch.setattr( + download_module, + "get_chain_download_failure_port", + _FailingDownloadFailureRepository, + ) + monkeypatch.setattr(download_module.logger, "error", error) + + chain = DownloadChain.__new__(DownloadChain) + + assert chain._active_download_failure_fingerprints( + contexts=[_build_tv_context()], + source="Subscribe|{}", + ) == {} + error.assert_called_once_with("查询下载失败冷却失败：query failed") + + +def test_download_failure_write_failure_is_fail_open(monkeypatch): + """冷却写入失败不得覆盖原下载结果，并记录一次明确错误。""" + + class _FailingDownloadFailureRepository: + """模拟失败的下载冷却写入端口。""" + + def record_failure(self, failure: DownloadFailureWrite) -> None: + """拒绝写入以验证 Chain 的 fail-open 语义。""" + assert failure.fingerprint + raise RuntimeError("write failed") + + error = MagicMock() + monkeypatch.setattr( + download_module, + "get_chain_download_failure_port", + _FailingDownloadFailureRepository, + ) + monkeypatch.setattr(download_module.logger, "error", error) + context = _build_tv_context() + chain = DownloadChain.__new__(DownloadChain) + + fingerprint = chain._record_download_failure( + context=context, + error_msg="下载失败", + source="Subscribe|{}", + ) + + assert fingerprint == DownloadChain._build_download_failure_fingerprint(context) + error.assert_called_once_with("记录下载失败冷却失败：write failed") def test_download_failure_fingerprint_distinguishes_special_season_zero(): @@ -943,17 +1031,23 @@ def test_batch_download_skips_failed_subscription_resource_and_tries_next(monkey 返回第一个候选的活跃失败冷却记录。 """ - def get_active_by_fingerprints(self, fingerprints: list[str], now_time: str) -> dict: + def get_active_by_fingerprints( + self, + fingerprints: list[str], + now_time: str, + ) -> dict[str, DownloadFailureSnapshot]: """ 模拟数据库批量查询活跃失败记录。 """ assert now_time assert failed_fingerprint in fingerprints - return {failed_fingerprint: SimpleNamespace( - fingerprint=failed_fingerprint, - error_message="无法读取种子文件", - next_retry_at="2026-01-02 03:04:05", - )} + return { + failed_fingerprint: DownloadFailureSnapshot( + fingerprint=failed_fingerprint, + error_message="无法读取种子文件", + next_retry_at="2026-01-02 03:04:05", + ) + } monkeypatch.setattr( download_module, diff --git a/tests/test_mediaserver_sync_incremental.py b/tests/test_mediaserver_sync_incremental.py index ceb221a5a..1a1f1116b 100644 --- a/tests/test_mediaserver_sync_incremental.py +++ b/tests/test_mediaserver_sync_incremental.py @@ -8,9 +8,10 @@ from sqlalchemy.orm import sessionmaker from app import schemas from app.chain import mediaserver as MEDIA_SERVER_CHAIN_MODULE from app.chain.mediaserver import MediaServerChain +from app.db.adapters.mediaserver import TransactionalMediaServerRepository from app.db.base import Base -from app.db.oper.mediaserver import MediaServerOper from app.db.models.mediaserver import MediaServerItem +from app.db.oper.mediaserver import MediaServerOper from app.runtime.config import global_vars @@ -100,18 +101,16 @@ def test_sync_persists_music_without_querying_tv_episodes(database): ) chain.episodes = lambda *_args, **_kwargs: pytest.fail("音乐条目不应查询电视剧分集") - with database() as session: - with patch.object( - MEDIA_SERVER_CHAIN_MODULE, - "get_chain_media_server_port", - lambda: MediaServerOper(session), - ), patch.object( - MEDIA_SERVER_CHAIN_MODULE, - "get_mediaserver_configs", - return_value=[SimpleNamespace(name="navidrome", enabled=True, sync_libraries=["all"])], - ): - chain.sync() - session.commit() + with patch.object( + MEDIA_SERVER_CHAIN_MODULE, + "get_chain_media_server_port", + lambda: TransactionalMediaServerRepository(database), + ), patch.object( + MEDIA_SERVER_CHAIN_MODULE, + "get_mediaserver_configs", + return_value=[SimpleNamespace(name="navidrome", enabled=True, sync_libraries=["all"])], + ): + chain.sync() with database() as db: item = db.query(MediaServerItem).one() @@ -121,6 +120,43 @@ def test_sync_persists_music_without_querying_tv_episodes(database): assert item.seasoninfo == {} +def test_sync_normalizes_incomplete_tv_episode_rows(database): + """TV 同步应丢弃无季号记录，并把缺失集列表规范为空列表。""" + chain = object.__new__(MediaServerChain) + chain.librarys = lambda _server: [SimpleNamespace(id="shows", name="剧集库")] + chain.media_count = lambda _server: 1 + chain.items_count = lambda **_kwargs: 1 + chain.items = lambda **_kwargs: iter([ + schemas.MediaServerItem( + server="plex", + library="shows", + item_id="show-1", + item_type="Series", + title="测试剧集", + ) + ]) + chain.episodes = lambda *_args, **_kwargs: [ + schemas.MediaServerSeasonInfo(season=None, episodes=[99]), + schemas.MediaServerSeasonInfo(season=1, episodes=None), + ] + + with patch.object( + MEDIA_SERVER_CHAIN_MODULE, + "get_chain_media_server_port", + lambda: TransactionalMediaServerRepository(database), + ), patch.object( + MEDIA_SERVER_CHAIN_MODULE, + "get_mediaserver_configs", + return_value=[SimpleNamespace(name="plex", enabled=True, sync_libraries=["all"])], + ): + chain.sync() + + with database() as session: + item = session.query(MediaServerItem).one() + + assert item.seasoninfo == {"1": []} + + def test_sync_updates_rows_and_removes_stale_entries(database): """同步应更新已存在条目，并清理未再出现或已移除服务的数据。""" old_sync_time = "2026-05-01 00:00:00" @@ -193,18 +229,16 @@ def test_sync_updates_rows_and_removes_stale_entries(database): ) chain.episodes = lambda *_args, **_kwargs: [] - with database() as session: - with patch.object( - MEDIA_SERVER_CHAIN_MODULE, - "get_chain_media_server_port", - lambda: MediaServerOper(session), - ), patch.object( - MEDIA_SERVER_CHAIN_MODULE, - "get_mediaserver_configs", - return_value=[SimpleNamespace(name="plex", enabled=True, sync_libraries=["movies"])], - ): - chain.sync() - session.commit() + with patch.object( + MEDIA_SERVER_CHAIN_MODULE, + "get_chain_media_server_port", + lambda: TransactionalMediaServerRepository(database), + ), patch.object( + MEDIA_SERVER_CHAIN_MODULE, + "get_mediaserver_configs", + return_value=[SimpleNamespace(name="plex", enabled=True, sync_libraries=["movies"])], + ): + chain.sync() with database() as db: items = ( @@ -277,23 +311,21 @@ def test_sync_queries_counts_before_items_and_reports_media_progress(database): chain.items = items chain.episodes = lambda *_args, **_kwargs: [] - with database() as session: - with patch.object( - MEDIA_SERVER_CHAIN_MODULE, - "get_chain_media_server_port", - lambda: MediaServerOper(session), - ), patch.object( - MEDIA_SERVER_CHAIN_MODULE, - "get_mediaserver_configs", - return_value=[ - SimpleNamespace(name="plex-a", enabled=True, sync_libraries=["all"]), - SimpleNamespace(name="plex-b", enabled=True, sync_libraries=["all"]), - ], - ): - chain.sync( - progress_callback=lambda **kwargs: progress_snapshots.append(kwargs) - ) - session.commit() + with patch.object( + MEDIA_SERVER_CHAIN_MODULE, + "get_chain_media_server_port", + lambda: TransactionalMediaServerRepository(database), + ), patch.object( + MEDIA_SERVER_CHAIN_MODULE, + "get_mediaserver_configs", + return_value=[ + SimpleNamespace(name="plex-a", enabled=True, sync_libraries=["all"]), + SimpleNamespace(name="plex-b", enabled=True, sync_libraries=["all"]), + ], + ): + chain.sync( + progress_callback=lambda **kwargs: progress_snapshots.append(kwargs) + ) assert events == [ "count:plex-a", @@ -354,6 +386,101 @@ def test_sync_targets_one_server_without_excluding_other_enabled_servers(monkeyp assert excluded_server_calls == [["plex-a", "plex-b"]] +def test_sync_partial_commit_preserves_stale_rows_until_next_run( + database, + monkeypatch, +): + """前序条目提交后发生失败时保留 stale，并由下一轮成功同步收敛。""" + old_sync_time = "2026-05-01 00:00:00" + with database() as session: + session.add(MediaServerItem( + server="plex", + library="movies", + item_id="stale", + item_type="电影", + title="陈旧条目", + lst_mod_date=old_sync_time, + )) + session.commit() + + chain = object.__new__(MediaServerChain) + stale_calls = [] + items = [ + schemas.MediaServerItem( + server="plex", + library="movies", + item_id=f"movie-{index}", + item_type="Movie", + title=f"测试电影 {index}", + ) + for index in (1, 2) + ] + transactional = TransactionalMediaServerRepository(database) + + class FailingMediaServerRepository: + """提交首个条目后让第二个条目失败，并记录不应发生的清理。""" + + def __init__(self): + """初始化本轮已尝试写入计数。""" + self.upsert_count = 0 + + def delete_excluded_servers(self, servers): + """委托真实短事务清理已移除服务器。""" + return transactional.delete_excluded_servers(servers) + + def upsert(self, item): + """提交首条记录，在第二条写入前模拟数据库不可用。""" + self.upsert_count += 1 + if self.upsert_count == 2: + raise RuntimeError("database unavailable") + return transactional.upsert(item) + + def delete_stale(self, **kwargs): + """记录不应发生的 stale 清理。""" + stale_calls.append(kwargs) + return transactional.delete_stale(**kwargs) + + chain.librarys = lambda _server: [SimpleNamespace(id="movies", name="电影库")] + chain.media_count = lambda _server: len(items) + chain.items_count = lambda **_kwargs: len(items) + chain.items = lambda **_kwargs: iter(items) + chain.episodes = lambda *_args, **_kwargs: [] + repository = FailingMediaServerRepository() + monkeypatch.setattr( + MEDIA_SERVER_CHAIN_MODULE, + "get_chain_media_server_port", + lambda: repository, + ) + monkeypatch.setattr( + MEDIA_SERVER_CHAIN_MODULE, + "get_mediaserver_configs", + lambda **_kwargs: [ + SimpleNamespace(name="plex", enabled=True, sync_libraries=["all"]) + ], + ) + + with pytest.raises(RuntimeError, match="database unavailable"): + chain.sync() + + assert stale_calls == [] + with database() as session: + assert { + item.item_id for item in session.query(MediaServerItem).all() + } == {"movie-1", "stale"} + + monkeypatch.setattr( + MEDIA_SERVER_CHAIN_MODULE, + "get_chain_media_server_port", + lambda: transactional, + ) + chain.sync() + + with database() as session: + assert { + item.item_id for item in session.query(MediaServerItem).all() + } == {"movie-1", "movie-2"} + + def test_sync_stops_without_emitting_completion_after_stop_signal(monkeypatch): """系统停止发生在逐库同步期间时，不应再发送服务器或全局完成进度。""" chain = object.__new__(MediaServerChain) @@ -379,7 +506,12 @@ def test_sync_stops_without_emitting_completion_after_stop_signal(monkeypatch): monkeypatch.setattr( chain, "_prepare_sync_contexts", - lambda _servers, _server: ([server], 1, {"plex": ([], {})}, 0), + lambda _servers, _server, _repository: ( + [server], + 1, + {"plex": ([], {})}, + 0, + ), ) monkeypatch.setattr(chain, "_sync_server_libraries", stop_during_sync) monkeypatch.setattr( diff --git a/tests/test_transactional_download_failure.py b/tests/test_transactional_download_failure.py index 89522b4f3..426d25582 100644 --- a/tests/test_transactional_download_failure.py +++ b/tests/test_transactional_download_failure.py @@ -1,33 +1,138 @@ """下载失败冷却切片显式 UoW 的回归测试。""" +from dataclasses import FrozenInstanceError +from types import SimpleNamespace from unittest.mock import MagicMock, patch import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from app.application.download.failures import ( + DownloadFailureSnapshot, + DownloadFailureWrite, +) from app.db.adapters.download import TransactionalDownloadFailureRepository +from app.db.base import Base +from app.db.models.downloadfailure import DownloadFailure +from app.db.session import SessionFactory -def _session_factory(session: MagicMock): +def _session_factory(session: MagicMock) -> MagicMock: """返回支持 context manager 的固定会话工厂。""" session.__enter__.return_value = session session.__exit__.return_value = False return MagicMock(return_value=session) -def test_record_failure_commits_explicit_unit_of_work() -> None: - """写入成功后只由适配器显式提交一次事务。""" +def _failure_write() -> DownloadFailureWrite: + """构造事务测试共用的最小下载失败写入 DTO。""" + return DownloadFailureWrite( + fingerprint="fp", + failed_at="now", + next_retry_at="next", + title="片名", + ) + + +@pytest.fixture +def real_session_factory(tmp_path): + """创建隔离的下载失败冷却数据库与会话工厂。""" + engine = create_engine(f"sqlite:///{tmp_path / 'download-failure.db'}") + factory = sessionmaker(bind=engine) + Base.metadata.create_all(bind=engine) + yield factory + engine.dispose() + + +def test_real_session_round_trip_returns_detached_snapshot(real_session_factory) -> None: + """真实写入提交后，查询结果在 Session 关闭后仍是可读冻结快照。""" + repository = TransactionalDownloadFailureRepository(real_session_factory) + + assert repository.record_failure(_failure_write()) is None + result = repository.get_active_by_fingerprints(["fp"], "before-next") + + assert result == { + "fp": DownloadFailureSnapshot( + fingerprint="fp", + error_message=None, + next_retry_at="next", + ) + } + with pytest.raises(FrozenInstanceError): + result["fp"].next_retry_at = "changed" + + +def test_active_failures_are_projected_to_detached_frozen_snapshots() -> None: + """只读适配器必须在会话关闭前投影，不得返回 ORM 记录。""" session = MagicMock() + record = SimpleNamespace( + fingerprint="fp", + error_message="无法读取种子文件", + next_retry_at="next", + ) oper = MagicMock() - oper.record_failure.return_value = object() + oper.get_active_by_fingerprints.return_value = {"fp": record} repository = TransactionalDownloadFailureRepository(_session_factory(session)) with patch( "app.db.adapters.download.DownloadFailureOper", return_value=oper, ): - result = repository.record_failure("fp", "now", "next", title="片名") + result = repository.get_active_by_fingerprints(["fp"], "now") - assert result is oper.record_failure.return_value + snapshot = result["fp"] + assert snapshot == DownloadFailureSnapshot( + fingerprint="fp", + error_message="无法读取种子文件", + next_retry_at="next", + ) + assert snapshot is not record + with pytest.raises(FrozenInstanceError): + setattr(snapshot, "error_message", "changed") + session.__exit__.assert_called_once() + + +def test_sqlite_snapshot_survives_repository_session_close(db) -> None: + """真实 SQLite 查询关闭适配器 Session 后仍可完整读取快照。""" + db.watermark(DownloadFailure) + repository = TransactionalDownloadFailureRepository(SessionFactory) + repository.record_failure(DownloadFailureWrite( + fingerprint="typed-detached-snapshot", + failed_at="2026-08-28 00:00:00", + next_retry_at="2026-08-29 00:00:00", + error_message="无法读取种子文件", + )) + + snapshots = repository.get_active_by_fingerprints( + ["typed-detached-snapshot"], + "2026-08-28 12:00:00", + ) + + assert snapshots["typed-detached-snapshot"] == DownloadFailureSnapshot( + fingerprint="typed-detached-snapshot", + error_message="无法读取种子文件", + next_retry_at="2026-08-29 00:00:00", + ) + + +def test_record_failure_commits_explicit_unit_of_work() -> None: + """写入成功后只由适配器显式提交一次事务。""" + session = MagicMock() + oper = MagicMock() + repository = TransactionalDownloadFailureRepository(_session_factory(session)) + + with patch( + "app.db.adapters.download.DownloadFailureOper", + return_value=oper, + ): + result = repository.record_failure(_failure_write()) + + assert result is None + assert oper.record_failure.call_args.kwargs["fingerprint"] == "fp" + assert oper.record_failure.call_args.kwargs["now_time"] == "now" + assert oper.record_failure.call_args.kwargs["next_retry_at"] == "next" + assert oper.record_failure.call_args.kwargs["title"] == "片名" session.commit.assert_called_once_with() session.rollback.assert_not_called() @@ -43,7 +148,23 @@ def test_record_failure_rolls_back_explicit_unit_of_work() -> None: "app.db.adapters.download.DownloadFailureOper", return_value=oper, ), pytest.raises(ValueError, match="duplicate"): - repository.record_failure("fp", "now", "next") + repository.record_failure(_failure_write()) session.rollback.assert_called_once_with() session.commit.assert_not_called() + + +def test_record_failure_rolls_back_commit_failure() -> None: + """提交失败时也必须回滚并传播提交异常。""" + session = MagicMock() + session.commit.side_effect = RuntimeError("commit failed") + repository = TransactionalDownloadFailureRepository(_session_factory(session)) + + with patch("app.db.adapters.download.DownloadFailureOper"), pytest.raises( + RuntimeError, + match="commit failed", + ): + repository.record_failure(_failure_write()) + + session.commit.assert_called_once_with() + session.rollback.assert_called_once_with() diff --git a/tests/test_transactional_mediaserver.py b/tests/test_transactional_mediaserver.py new file mode 100644 index 000000000..92d395440 --- /dev/null +++ b/tests/test_transactional_mediaserver.py @@ -0,0 +1,183 @@ +"""媒体服务器本地缓存短事务适配器的回归测试。""" + +import json +from dataclasses import FrozenInstanceError +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from app.application.mediaserver import MediaServerQueryService, MediaServerSyncItem +from app.db.adapters.mediaserver import TransactionalMediaServerRepository +from app.db.base import Base +from app.db.models.mediaserver import MediaServerItem +from app.schemas.mediaserver import MediaServerItem as MediaServerItemSchema +from app.schemas.types import MediaSource + + +@pytest.fixture +def session_factory(tmp_path): + """创建隔离的媒体服务器缓存数据库与会话工厂。""" + engine = create_engine(f"sqlite:///{tmp_path / 'mediaserver-adapter.db'}") + factory = sessionmaker(bind=engine) + Base.metadata.create_all(bind=engine) + yield factory + engine.dispose() + + +def _sync_item(*, title: str, sync_time: str) -> MediaServerSyncItem: + """构造一个可重复 upsert 的冻结媒体库条目。""" + return MediaServerSyncItem( + server="plex", + library="movies", + item_id="item-1", + item_type="电影", + title=title, + original_title=None, + year="2026", + media_source=MediaSource.TMDB, + media_id="1001", + path=f"/media/{title}.mkv", + seasoninfo=((1, (1, 2)),), + note_json=None, + lst_mod_date=sync_time, + ) + + +def test_upsert_commits_insert_and_update_without_external_session( + session_factory, + monkeypatch, +) -> None: + """每次 upsert 应在单一短 Session 中完成查询、写入和提交。""" + from app.db import base as db_base + + monkeypatch.setattr( + db_base, + "run_sync_transaction", + lambda _operation: (_ for _ in ()).throw( + AssertionError("显式 Session 不应创建兼容事务") + ), + ) + repository = TransactionalMediaServerRepository(session_factory) + + assert repository.upsert( + _sync_item(title="旧标题", sync_time="2026-08-28 10:00:00") + ) + assert not repository.upsert( + _sync_item(title="新标题", sync_time="2026-08-28 11:00:00") + ) + + with session_factory() as session: + items = session.query(MediaServerItem).all() + + assert len(items) == 1 + assert items[0].title == "新标题" + assert items[0].path == "/media/新标题.mkv" + assert items[0].seasoninfo == {"1": [1, 2]} + + +def test_get_item_id_returns_scalar_after_short_session_closes(session_factory) -> None: + """查询端口只返回标量 ID，不把 ORM 条目带出短 Session。""" + repository = TransactionalMediaServerRepository(session_factory) + repository.upsert( + _sync_item(title="查询标题", sync_time="2026-08-28 10:00:00") + ) + + item_id = repository.get_item_id( + media_source=MediaSource.TMDB, + media_id="1001", + mtype="电影", + season=1, + ) + + assert item_id == "item-1" + + +def test_sync_item_deeply_detaches_mutable_remote_values() -> None: + """冻结 DTO 必须规范化季集并断开与远端 note 的可变引用。""" + note = {"nested": ["original"]} + source = MediaServerItemSchema( + server="plex", + library="shows", + item_id="show-1", + item_type="Series", + title="剧集", + note=note, + ) + + snapshot = MediaServerSyncItem.from_item( + source, + item_type="电视剧", + seasoninfo={1: None, 2: [1, 2]}, + sync_time="2026-08-28 10:00:00", + ) + note["nested"].append("changed") + + assert snapshot.seasoninfo == ((1, ()), (2, (1, 2))) + assert json.loads(snapshot.note_json or "null") == {"nested": ["original"]} + with pytest.raises(FrozenInstanceError): + snapshot.note_json = None + + +def test_upsert_rolls_back_and_preserves_original_error() -> None: + """单条写入异常时必须回滚短事务并继续抛出原始异常。""" + session = MagicMock() + session.__enter__.return_value = session + session.__exit__.return_value = False + repository = TransactionalMediaServerRepository(MagicMock(return_value=session)) + + with patch( + "app.db.adapters.mediaserver.MediaServerOper.upsert", + side_effect=ValueError("invalid item"), + ), pytest.raises(ValueError, match="invalid item"): + repository.upsert( + _sync_item(title="失败标题", sync_time="2026-08-28 10:00:00") + ) + + session.rollback.assert_called_once_with() + session.commit.assert_not_called() + + +def test_upsert_rolls_back_commit_failure() -> None: + """提交阶段失败也必须回滚短事务并传播原始异常。""" + session = MagicMock() + session.__enter__.return_value = session + session.__exit__.return_value = False + session.commit.side_effect = RuntimeError("commit failed") + repository = TransactionalMediaServerRepository(MagicMock(return_value=session)) + + with patch( + "app.db.adapters.mediaserver.MediaServerOper.upsert", + return_value=True, + ), pytest.raises(RuntimeError, match="commit failed"): + repository.upsert( + _sync_item(title="提交失败", sync_time="2026-08-28 10:00:00") + ) + + session.commit.assert_called_once_with() + session.rollback.assert_called_once_with() + + +@pytest.mark.asyncio +async def test_query_service_consumes_scalar_repository_result() -> None: + """Application 查询服务只消费标量 ID，不读取 ORM 属性。""" + repository = MagicMock() + repository.async_get_item_id = AsyncMock(return_value="item-1") + service = MediaServerQueryService(repository) + + result = await service.find_item_id( + media_source=MediaSource.TMDB, + media_id="1001", + mtype="电影", + ) + + assert result == "item-1" + repository.async_get_item_id.assert_awaited_once_with( + title=None, + year=None, + mtype="电影", + media_source=MediaSource.TMDB, + media_id="1001", + season=None, + )