From 16f7d65640361b2c2633fecefc3f9c9d26f9f157 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 22 Aug 2026 22:31:24 +0800 Subject: [PATCH] refactor: complete chain entrypoint slicing --- app/chain/download.py | 70 +++++++-- app/chain/mediaserver.py | 2 + app/chain/subscribe.py | 144 ++++++++++++------ .../backend-architecture-next-stage.md | 7 + .../architecture/complexity-baseline.json | 6 +- tests/test_mediaserver_sync_incremental.py | 40 +++++ 6 files changed, 206 insertions(+), 63 deletions(-) diff --git a/app/chain/download.py b/app/chain/download.py index 8af3875ab..6bda0f18d 100644 --- a/app/chain/download.py +++ b/app/chain/download.py @@ -925,6 +925,25 @@ class DownloadChain(ChainBase): logger.error(f"查询下载失败冷却失败：{str(err)}") return {} + @staticmethod + def _log_download_failure_cooldown( + context: Context, + failure: Optional["DownloadFailure"], + ) -> None: + """记录候选资源处于失败冷却期时的跳过原因和下次重试时间。""" + reason = getattr(failure, "error_message", None) or "未知原因" + retry_at = getattr(failure, "next_retry_at", None) + if retry_at: + logger.info( + f"{context.torrent_info.title} 近期添加下载失败（失败原因：{reason}），" + f"暂时跳过该资源，将于 {retry_at} 后重试" + ) + else: + logger.info( + f"{context.torrent_info.title} 近期添加下载失败（失败原因：{reason}），" + "暂时跳过该资源" + ) + def _prepare_batch_download_contexts( self, contexts: List[Context], @@ -987,6 +1006,10 @@ class DownloadChain(ChainBase): continue fingerprint = self._build_download_failure_fingerprint(context) if fingerprint and fingerprint in active_failure_records: + self._log_download_failure_cooldown( + context, + active_failure_records[fingerprint], + ) continue if media_type == MediaType.MOVIE: download_key = context.media_info.title_year @@ -1289,6 +1312,40 @@ class DownloadChain(ChainBase): return_detail: bool = False, custom_words: Optional[str] = None) -> Union[Optional[str], Tuple[Optional[str], Optional[str]]]: """ + 下载单个资源并发送结果通知。 + + 保持下载链、消息入口和插件使用的公开签名，实际流程委托给内部执行阶段。 + """ + return self._execute_download_single( + context=context, + torrent_file=torrent_file, + torrent_content=torrent_content, + episodes=episodes, + channel=channel, + source=source, + downloader=downloader, + save_path=save_path, + userid=userid, + username=username, + label=label, + return_detail=return_detail, + custom_words=custom_words, + ) + + def _execute_download_single(self, context: Context, + torrent_file: Path = None, + torrent_content: Optional[Union[str, bytes]] = None, + episodes: Set[int] = None, + channel: NotificationChannel = None, + source: Optional[str] = None, + downloader: Optional[str] = None, + save_path: Optional[str] = None, + userid: Union[str, int] = None, + username: Optional[str] = None, + label: Optional[str] = None, + return_detail: bool = False, + custom_words: Optional[str] = None) -> Union[Optional[str], Tuple[Optional[str], Optional[str]]]: + """ 下载及发送通知 :param context: 资源上下文 :param torrent_file: 种子文件路径 @@ -1623,15 +1680,10 @@ class DownloadChain(ChainBase): """ fingerprint = self._build_download_failure_fingerprint(_context) if fingerprint and fingerprint in active_failure_records: - _failure = active_failure_records[fingerprint] - _reason = getattr(_failure, "error_message", None) or "未知原因" - _retry_at = getattr(_failure, "next_retry_at", None) - if _retry_at: - logger.info(f"{_context.torrent_info.title} 近期添加下载失败（失败原因：{_reason}），" - f"暂时跳过该资源，将于 {_retry_at} 后重试") - else: - logger.info(f"{_context.torrent_info.title} 近期添加下载失败（失败原因：{_reason}），" - f"暂时跳过该资源") + self._log_download_failure_cooldown( + _context, + active_failure_records[fingerprint], + ) return True return False diff --git a/app/chain/mediaserver.py b/app/chain/mediaserver.py index 25fe88264..650dbc9e7 100644 --- a/app/chain/mediaserver.py +++ b/app/chain/mediaserver.py @@ -554,6 +554,8 @@ class MediaServerChain(ChainBase): global_media_total=global_media_total, global_media_finished=global_media_finished, ) + if global_vars.is_system_stopped: + return total_count += server_count logger.info(f"媒体服务器 {server_name} 数据同步完成，总同步数量：{total_count}") if progress_callback: diff --git a/app/chain/subscribe.py b/app/chain/subscribe.py index 4a9d2ce20..149ce93f4 100644 --- a/app/chain/subscribe.py +++ b/app/chain/subscribe.py @@ -1498,6 +1498,25 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): state: Optional[str] = 'N', manual: Optional[bool] = False, progress_callback: Optional[Callable[..., None]] = None, + ) -> None: + """ + 执行订阅搜索。 + + 保持定时任务、API 和插件使用的公开签名，搜索实现委托给内部执行阶段。 + """ + return self._execute_search( + sid=sid, + state=state, + manual=manual, + progress_callback=progress_callback, + ) + + def _execute_search( + self, + sid: Optional[int] = None, + state: Optional[str] = 'N', + manual: Optional[bool] = False, + progress_callback: Optional[Callable[..., None]] = None, ) -> None: """ 订阅搜索 @@ -1928,10 +1947,85 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): self.get_states_for_search('R') ) + def _prepare_match_torrents( + self, + torrents: Dict[str, List[Context]], + ) -> Dict[str, List[Context]]: + """预识别待匹配资源，并保留原上下文供后续订阅复用。""" + processed_torrents: Dict[str, List[Context]] = {} + for domain, contexts in torrents.items(): + if global_vars.is_system_stopped: + break + processed_torrents[domain] = [] + for context in contexts: + if global_vars.is_system_stopped: + break + if context.torrent_info and getattr(context.torrent_info, "category", None) in ( + MediaType.MUSIC, + MediaType.MUSIC.value, + ): + # 音乐 RSS 使用订阅目标做实体匹配，不应进入影视识别并累计失败次数。 + processed_torrents[domain].append(context) + continue + if ( + not context.media_info + or not resolve_media_identity(media=context.media_info)[1] + ) and context.media_recognize_fail_count < 3: + logger.debug( + f'尝试重新识别种子：{context.torrent_info.title}，当前失败次数：' + f'{context.media_recognize_fail_count}/3' + ) + re_mediainfo = MediaChain().recognize_by_meta( + context.meta_info, + obtain_images=False, + ) + if re_mediainfo: + re_mediainfo.clear() + context.media_info = re_mediainfo + context.match_source = self.__get_media_id_match_source(re_mediainfo) + context.candidate_recognized = bool( + resolve_media_identity(media=re_mediainfo)[1] + ) + context.media_info_is_target = False + context.media_recognize_fail_count = 0 + logger.debug(f'种子 {context.torrent_info.title} 重新识别成功') + else: + context.media_recognize_fail_count += 1 + logger.debug( + f'种子 {context.torrent_info.title} 媒体识别失败，失败次数：' + f'{context.media_recognize_fail_count}/3' + ) + elif context.media_recognize_fail_count >= 3: + logger.debug(f'种子 {context.torrent_info.title} 已达到最大识别失败次数(3次)，跳过识别') + processed_torrents[domain].append(context) + return processed_torrents + def match( self, torrents: Dict[str, List[Context]], progress_callback: Optional[Callable[..., None]] = None, + ) -> None: + """ + 从缓存中匹配订阅，并自动下载。 + + 该入口保持订阅刷新、定时任务和插件调用的稳定签名，具体匹配流程由内部阶段执行。 + """ + if not torrents: + logger.warn('没有缓存资源，无法匹配订阅') + if progress_callback: + progress_callback(value=100, text="没有缓存资源，跳过订阅匹配") + return + if progress_callback: + progress_callback(value=0, text="正在预处理订阅资源 ...") + return self._execute_match( + torrents=torrents, + progress_callback=progress_callback, + ) + + def _execute_match( + self, + torrents: Dict[str, List[Context]], + progress_callback: Optional[Callable[..., None]] = None, ) -> None: """ 从缓存中匹配订阅，并自动下载 @@ -1954,55 +2048,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): if not lock_acquired: return - # 预识别所有未识别的种子 - processed_torrents: Dict[str, List[Context]] = {} - for domain, contexts in torrents.items(): - if global_vars.is_system_stopped: - break - processed_torrents[domain] = [] - for context in contexts: - if global_vars.is_system_stopped: - break - if context.torrent_info and getattr(context.torrent_info, "category", None) in ( - MediaType.MUSIC, - MediaType.MUSIC.value, - ): - # 音乐 RSS 使用订阅目标做实体匹配，不应进入影视识别并累计失败次数。 - processed_torrents[domain].append(context) - continue - # 如果种子未识别且失败次数未超过3次，尝试识别 - if ( - not context.media_info - or not resolve_media_identity(media=context.media_info)[1] - ) and context.media_recognize_fail_count < 3: - logger.debug( - f'尝试重新识别种子：{context.torrent_info.title}，当前失败次数：{context.media_recognize_fail_count}/3') - re_mediainfo = MediaChain().recognize_by_meta( - context.meta_info, - obtain_images=False, - ) - if re_mediainfo: - # 清理多余信息 - re_mediainfo.clear() - # 更新种子缓存 - context.media_info = re_mediainfo - context.match_source = self.__get_media_id_match_source(re_mediainfo) - context.candidate_recognized = bool( - resolve_media_identity(media=re_mediainfo)[1] - ) - context.media_info_is_target = False - # 重置失败次数 - context.media_recognize_fail_count = 0 - logger.debug(f'种子 {context.torrent_info.title} 重新识别成功') - else: - # 识别失败，增加失败次数 - context.media_recognize_fail_count += 1 - logger.debug( - f'种子 {context.torrent_info.title} 媒体识别失败，失败次数：{context.media_recognize_fail_count}/3') - elif context.media_recognize_fail_count >= 3: - logger.debug(f'种子 {context.torrent_info.title} 已达到最大识别失败次数(3次)，跳过识别') - # 添加已预处理 - processed_torrents[domain].append(context) + processed_torrents = self._prepare_match_torrents(torrents) # 所有订阅 subscribes = SubscribeOper().list(self.get_states_for_search('R')) diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index affcf51bb..3779256c3 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -995,6 +995,13 @@ MFA/Passkey 专项测试与架构门禁通过，密钥类配置仍保留在安 随后将 `TransferChain.do_transfer` 的公开入口收口为稳定兼容 Facade，先提取媒体身份规范化阶段，保留显式 `media_source/media_id` 校验、识别失败文案和所有原有调用参数；整理专项 80 项测试通过，复杂度基线移除该入口， 后续继续拆分其批次规划与执行阶段。 +2026-08-22 继续完成入口垂直切片：`DownloadChain.download_single`、`SubscribeChain.search` 和 +`SubscribeChain.match` 均改为稳定兼容 Facade，分别委托下载执行、搜索执行、资源预处理和订阅匹配阶段； +保留原参数、对象类型、锁、进度回调、停止信号、候选过滤、失败冷却日志和下载结算语义。 +`MediaServerChain.sync` 补回停止信号后的立即退出，避免系统停止后继续发送服务器/全局完成进度。 +下载、订阅、媒体服务器及 durable/outbox 专项共 370 项测试通过，复杂度基线移除上述三个订阅/下载入口。 +当前仍不把普通用户通知和 MoviePilot Server 外部统计标记为 durable：它们尚未与业务写入和 outbox intent +绑定在同一事务，继续保持 post-commit 的准确边界。 #### ARCH-272：异步阻塞检测 diff --git a/tests/fixtures/architecture/complexity-baseline.json b/tests/fixtures/architecture/complexity-baseline.json index c7567378d..bb2e6af38 100644 --- a/tests/fixtures/architecture/complexity-baseline.json +++ b/tests/fixtures/architecture/complexity-baseline.json @@ -16,9 +16,5 @@ "app/application/rss.py:RssHelper.parse": 206, "app/application/security/cookie.py:CookieHelper.get_site_cookie_ua": 221 }, - "chain_public": { - "app/chain/download.py:DownloadChain.download_single": 167, - "app/chain/subscribe.py:SubscribeChain.match": 415, - "app/chain/subscribe.py:SubscribeChain.search": 246 - } + "chain_public": {} } diff --git a/tests/test_mediaserver_sync_incremental.py b/tests/test_mediaserver_sync_incremental.py index 807721bc7..7eef264db 100644 --- a/tests/test_mediaserver_sync_incremental.py +++ b/tests/test_mediaserver_sync_incremental.py @@ -11,6 +11,7 @@ from app.chain.mediaserver import MediaServerChain from app.db import Base from app.db.oper.mediaserver import MediaServerOper from app.db.models.mediaserver import MediaServerItem +from app.runtime.config import global_vars @pytest.fixture @@ -329,3 +330,42 @@ def test_sync_targets_one_server_without_excluding_other_enabled_servers(monkeyp assert library_calls == ["plex-a"] assert excluded_server_calls == [["plex-a", "plex-b"]] + + +def test_sync_stops_without_emitting_completion_after_stop_signal(monkeypatch): + """系统停止发生在逐库同步期间时，不应再发送服务器或全局完成进度。""" + chain = object.__new__(MediaServerChain) + server = SimpleNamespace(name="plex", enabled=True) + progress = [] + + class FakeMediaServerOper: + """提供同步阶段所需的最小数据库端口。""" + + def delete_excluded_servers(self, _servers): + """忽略测试中的媒体服务器清理。""" + + def stop_during_sync(**_kwargs): + """模拟读取媒体条目时收到系统停止信号。""" + global_vars.stop_system() + return 0, 0 + + monkeypatch.setattr(MEDIA_SERVER_CHAIN_MODULE, "MediaServerOper", FakeMediaServerOper) + monkeypatch.setattr( + chain, + "_prepare_sync_contexts", + lambda _servers, _server: ([server], 1, {"plex": ([], {})}, 0), + ) + monkeypatch.setattr(chain, "_sync_server_libraries", stop_during_sync) + monkeypatch.setattr( + MEDIA_SERVER_CHAIN_MODULE.ServiceConfigHelper, + "get_mediaserver_configs", + lambda: [server], + ) + global_vars.STOP_EVENT.clear() + try: + chain.sync(progress_callback=lambda **kwargs: progress.append(kwargs)) + finally: + global_vars.STOP_EVENT.clear() + + texts = [item.get("text") for item in progress] + assert not any(text and "同步完成" in text for text in texts)