From 9eae3b9b834e731d19cdd305cb2f710c85aaee19 Mon Sep 17 00:00:00 2001 From: freeman Date: Wed, 30 Sep 2026 21:50:47 +0800 Subject: [PATCH] refactor(download): split batch matching into explicit phases --- app/chain/download/batch.py | 763 ++++++++---------- docs/rules/05-architecture.md | 11 + .../architecture/complexity-baseline.json | 2 - tests/test_download_batch_phases.py | 319 ++++++++ 4 files changed, 654 insertions(+), 441 deletions(-) create mode 100644 tests/test_download_batch_phases.py diff --git a/app/chain/download/batch.py b/app/chain/download/batch.py index 46453cf3c..21dc835c4 100644 --- a/app/chain/download/batch.py +++ b/app/chain/download/batch.py @@ -1,7 +1,7 @@ """批量下载编排 owner。""" import copy -from typing import Callable, Dict, List, Optional, Set, Tuple, cast +from typing import TYPE_CHECKING, Callable, Dict, List, Optional, Set, Tuple, TypedDict, cast from app.application.download import selection as _selection from app.application.download.admission import SubscriptionDownloadGovernance @@ -19,6 +19,9 @@ from app.schemas.types import ( NotificationChannel, ) +if TYPE_CHECKING: + from app.application.download.failures import DownloadFailureSnapshot + def _new_torrent_helper() -> TorrentHelper: """构造保留动态初始化行为但具有静态返回类型的种子助手。""" @@ -91,452 +94,334 @@ class DownloadBatchOwner(_DownloadOwnerBase): :param governance: 订阅取消与下载器副作用边界 :return: 已下载资源列表及剩余缺集，键格式为 no_exists[source:id] """ - no_exists_was_none = no_exists is None - if no_exists is None: - no_exists = {} - # 已下载的项目 - downloaded_list: List[Context] = [] - custom_word_list = custom_words.splitlines() if custom_words else None - - # 缺集记账与覆盖判定规则已下沉至 app/application/download/selection.py； - # 此处仅保留闭包委托，让编排循环保持原调用形态。 - def __update_seasons( - _mid: str, - _need: List[int], - _current: List[int], - ) -> List[int]: - """更新need_tvs季数，返回剩余季数。""" - return _selection.update_no_exists_seasons(no_exists, _mid, _need, _current) - - def __update_episodes( - _mid: str, - _sea: int, - _need: List[int], - _current: Set[int], - ) -> List[int]: - """更新need_tvs集数，返回剩余集数。""" - return _selection.update_no_exists_episodes(no_exists, _mid, _sea, _need, _current) - - def __get_season_episodes(_mid: str, season: int) -> int: - """获取需要的季的集数。""" - return _selection.get_season_episodes(no_exists, _mid, season) - - def __get_no_exist_media(_mid: str, season: int) -> Optional[NotExistMediaInfo]: - """获取指定媒体和季的缺失信息。""" - return _selection.get_no_exist_media(no_exists, _mid, season) - - def __get_required_episodes(_mid: str, season: int) -> Set[int]: - """获取整季候选必须覆盖的目标集范围。""" - return _selection.get_required_episodes(no_exists, _mid, season) - - def __requires_complete_coverage(_tv: Optional[NotExistMediaInfo]) -> bool: - """判断当前缺失范围是否要求候选资源完整覆盖目标范围。""" - return _selection.requires_complete_coverage(_tv) - - def __apply_allowed_episodes( - _need_episodes: Set[int] | List[int], - _context: Context, - ) -> Set[int]: - """根据候选允许集裁剪 need_episodes，返回真正可下载的剧集集合。""" - return _selection.apply_allowed_episodes(set(_need_episodes), _context) - - # 仅排序，不提前按媒体控重；下载失败时需要继续尝试同组后续候选。 - contexts, active_failure_records = self._prepare_batch_download_contexts( - contexts=contexts, - downloader=downloader, - source=source, + missing = no_exists if no_exists is not None else {} + options = _BatchDownloadOptions( + save_path=save_path, channel=channel, source=source, userid=userid, + username=username, downloader=downloader, custom_words=custom_words, + governance=governance, ) + batch = _BatchDownloadRun(self, contexts, missing, options) + batch.run() + return batch.downloaded, None if no_exists is None else missing - def __is_context_in_failure_cooldown(_context: Context) -> bool: - """ - 判断候选资源是否仍处于失败冷却期。 - """ - 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], + +class _BatchDownloadOptions(TypedDict): + """各阶段原样传递给单资源下载的调用参数，包括订阅副作用治理边界。""" + + save_path: Optional[str] + channel: Optional[NotificationChannel] + source: Optional[str] + userid: Optional[str] + username: Optional[str] + downloader: Optional[str] + custom_words: Optional[str] + governance: Optional[SubscriptionDownloadGovernance] + + +class _BatchDownloadRun: + """单次批量调用的状态；候选、失败冷却与缺集记账不保存在共享 Chain 上。""" + + def __init__( + self, owner: _DownloadOwnerBase, contexts: List[Context], + no_exists: Dict[str, Dict[int, NotExistMediaInfo]], options: _BatchDownloadOptions, + ) -> None: + self.owner = owner + self.contexts = contexts + self.no_exists = no_exists + self.options = options + self.downloaded: List[Context] = [] + self.failures: Dict[str, Optional[DownloadFailureSnapshot]] = {} + words = options["custom_words"] + self.custom_words = words.splitlines() if words else None + + def run(self) -> None: + """只准备一次候选，依次匹配电影/音乐、整季、完整集和拆包补集。""" + self.contexts, self.failures = self.owner._prepare_batch_download_contexts( + contexts=self.contexts, + downloader=self.options["downloader"], source=self.options["source"], + ) + self.owner._download_movie_music_candidates( + contexts=self.contexts, downloaded_list=self.downloaded, + active_failure_records=self.failures, **self.options, + ) + if self.no_exists: + self._download_whole_seasons() + if self.no_exists: + self._download_episode_packs() + if self.no_exists: + self._download_partial_packs() + logger.info(f"成功下载种子数：{len(self.downloaded)}，剩余未下载的剧集：{self.no_exists}") + + def _in_failure_cooldown(self, context: Context) -> bool: + fingerprint = self.owner._build_download_failure_fingerprint(context) + if fingerprint and fingerprint in self.failures: + self.owner._log_download_failure_cooldown(context, self.failures[fingerprint]) + return True + return False + + def _remember_failure(self, context: Context) -> None: + """按当前识别集数记录失败，后续阶段不得重复提交本轮失败候选。""" + fingerprint = self.owner._build_download_failure_fingerprint(context) + if fingerprint: + self.failures[fingerprint] = None + + def _read_torrent( + self, context: Context, *, partial: bool = False, + ) -> Optional[Tuple[bytes, List[int]]]: + """读取真实文件集数；空内容进入冷却，无法检查的磁力链只跳过当前阶段。""" + torrent = context.torrent_info + meta = context.meta_info + assert meta is not None + logger.info(f"开始下载种子 {torrent.title} ...") + content, _, files = self.owner.download_torrent(torrent) + if not content: + log_failure = logger.info if partial else logger.warn + log_failure(f"{torrent.title} 种子下载失败！") + self.owner._record_download_failure( + context=context, error_msg="下载种子内容为空", + downloader=self.options["downloader"], source=self.options["source"], + ) + self._remember_failure(context) + return None + if isinstance(content, str): + reason = "无法解析" if partial else "无法确定" + logger.warn(f"{meta.org_string} 下载地址是磁力链，{reason}种子文件集数") + return None + episodes = cast(List[int], _new_torrent_helper().get_torrent_episodes( + files, custom_words=self.custom_words, + )) + if partial: + logger.info(f"{torrent.site_name} - {meta.org_string} 解析种子文件集数：{episodes}") + else: + logger.info(f"{meta.org_string} 解析种子文件集数为 {episodes}") + return content, episodes + + def _download_whole_seasons(self) -> None: + logger.info(f"开始匹配电视剧整季：{self.no_exists}") + needed: Dict[str, List[int]] = {} + for media_key, missing_seasons in self.no_exists.items(): + for info in missing_seasons.values(): + if info and not info.episodes: + needed.setdefault(media_key, []).append(info.season if info.season is not None else 1) + logger.info(f"缺失整季：{needed}") + for media_key, need_seasons in needed.items(): + for context in self.contexts: + if runtime_stop_state.is_system_stopped: + break + seasons = self._whole_season_candidate(context, media_key, need_seasons) + if not seasons or not self._submit_whole_season(context, media_key, seasons): + continue + need_seasons = _selection.update_no_exists_seasons( + self.no_exists, media_key, need_seasons, seasons, ) - return True + logger.info(f"{media_key} 剩余需要季：{need_seasons}") + if not need_seasons: + break + + def _whole_season_candidate( + self, context: Context, media_key: str, needed: List[int], + ) -> List[int]: + media, meta = context.media_info, context.meta_info + if media is None or meta is None or media.type != MediaType.TV: + return [] + seasons = meta.season_list or [1] + if meta.episode_list or not self.owner._matches_media_identity(media, media_key): + return [] + if context in self.downloaded or self._in_failure_cooldown(context): + return [] + if not set(seasons).issubset(set(needed)): + return [] + required = _selection.get_required_episodes(self.no_exists, media_key, seasons[0]) + if not _selection.allows_full_season(seasons, required, context): + return [] + return seasons + + def _inspect_whole_season( + self, context: Context, media_key: str, season: int, + ) -> Optional[Tuple[bytes, bool]]: + """整季不合格也保留识别集数，供后续完整集阶段继续判断该候选。""" + parsed = self._read_torrent(context) + if parsed is None: + return None + content, episodes = parsed + if not episodes or not _selection.allows_full_pack(set(episodes), context): + return None + meta = context.meta_info + assert meta is not None + meta.set_episodes(begin=min(episodes), end=max(episodes)) + info = _selection.get_no_exist_media(self.no_exists, media_key, season) + required = _selection.get_required_episodes(self.no_exists, media_key, season) \ + if _selection.requires_complete_coverage(info) else set() + total = _selection.get_season_episodes(self.no_exists, media_key, season) + complete = bool(required) and required.issubset(set(episodes)) + if complete: + logger.info( + f"{meta.org_string} 解析文件集数已完整覆盖目标范围：" + f"{episode_rules.format_ranges(sorted(required))}") + if required and not complete: + missing = sorted(required.difference(episodes)) + logger.info( + f"{meta.org_string} 解析文件集数未覆盖目标范围，" + f"缺少 {episode_rules.format_ranges(missing)}，先放弃这个种子") + return None + if not required and total and len(episodes) < total: + logger.info(f"{meta.org_string} 解析文件集数发现不是完整合集，先放弃这个种子") + return None + return content, complete + + def _submit_whole_season( + self, context: Context, media_key: str, seasons: List[int], + ) -> bool: + complete = False + if len(seasons) == 1: + inspected = self._inspect_whole_season(context, media_key, seasons[0]) + if inspected is None: + return False + content, complete = inspected + logger.info(f"开始下载 {context.torrent_info.title} ...") + download_id = self.owner.download_single( + context=context, torrent_content=content, **self.options, + ) + else: + logger.info(f"开始下载 {context.torrent_info.title} ...") + download_id = self.owner.download_single(context, **self.options) + if not download_id: + self._remember_failure(context) return False + if complete: + context.confirmed_full_coverage = True + logger.info(f"{context.torrent_info.title} 添加下载成功") + self.downloaded.append(context) + return True - def __remember_context_failure(_context: Context) -> None: - """ - 将本轮失败候选加入内存冷却集合，避免同一批次重复尝试。 - """ - fingerprint = self._build_download_failure_fingerprint(_context) - if fingerprint: - active_failure_records[fingerprint] = None + def _download_episode_packs(self) -> None: + logger.info(f"开始电视剧完整集匹配：{self.no_exists}") + for media_key in list(self.no_exists): + seasons = self.no_exists.get(media_key) + if not seasons: + continue + # 单季成功会重写缺集字典；每个媒体仍按阶段开始时的季快照遍历。 + for season, info in copy.deepcopy(seasons).items(): + self._download_season_episode_packs(media_key, season, info) - self._download_movie_music_candidates( - contexts=contexts, - downloaded_list=downloaded_list, - active_failure_records=active_failure_records, - save_path=save_path, - channel=channel, - source=source, - userid=userid, - username=username, - downloader=downloader, - custom_words=custom_words, governance=governance, - ) + def _download_season_episode_packs( + self, media_key: str, season: int, info: NotExistMediaInfo, + ) -> None: + needed = info.episodes or list(range(info.start_episode or 1, (info.total_episode or 0) + 1)) + for context in self.contexts: + if runtime_stop_state.is_system_stopped: + break + episodes = self._episode_pack_candidate(context, media_key, season, needed, info) + if not episodes: + continue + meta = context.meta_info + assert meta is not None + logger.info(f"开始下载 {meta.title} ...") + if not self.owner.download_single(context, **self.options): + self._remember_failure(context) + continue + if _selection.requires_complete_coverage(info): + context.confirmed_full_coverage = True + logger.info(f"{meta.title} 添加下载成功") + self.downloaded.append(context) + needed = _selection.update_no_exists_episodes( + self.no_exists, media_key, season, needed, episodes, + ) + logger.info(f"季 {season} 剩余需要集：{needed}") - # 电视剧整季匹配 - if no_exists: - logger.info(f"开始匹配电视剧整季：{no_exists}") - # 先把整季缺失的拿出来，看是否刚好有所有季都满足的种子 {source:id: [seasons]} - need_seasons: Dict[str, List[int]] = {} - for need_mid, missing_seasons in no_exists.items(): - for tv in missing_seasons.values(): - if not tv: - continue - # 季列表为空的，代表全季缺失 - if not tv.episodes: - if not need_seasons.get(need_mid): - need_seasons[need_mid] = [] - need_seasons[need_mid].append(tv.season if tv.season is not None else 1) - logger.info(f"缺失整季：{need_seasons}") - # 查找整季包含的种子，只处理整季没集的种子或者是集数超过季的种子 - for need_mid, need_season in need_seasons.items(): - # 循环种子 - for context in contexts: - if runtime_stop_state.is_system_stopped: - break - # 媒体信息 - media = context.media_info - # 识别元数据 - meta = context.meta_info - # 种子信息 - torrent = context.torrent_info - if media is None or meta is None: - continue - # 排除电视剧 - if media.type != MediaType.TV: - continue - # 种子的季清单 - torrent_season = meta.season_list - # 没有季的默认为第1季 - if not torrent_season: - torrent_season = [1] - # 种子有集的不要 - if meta.episode_list: - continue - # 匹配TMDBID - if self._matches_media_identity(media, need_mid): - # 不重复添加 - if context in downloaded_list: - continue - if __is_context_in_failure_cooldown(context): - continue - # 种子季是需要季或者子集 - if set(torrent_season).issubset(set(need_season)): - # 分集洗版的整包只有全部目标集均可升级时才可整体下载；多季包无法用单季允许集证明安全。 - required_episodes = __get_required_episodes(need_mid, torrent_season[0]) - if not _selection.allows_full_season(torrent_season, required_episodes, context): - continue - complete_coverage_matched = False - if len(torrent_season) == 1: - # 只有一季的可能是命名错误，需要打开种子鉴别，只有实际集数大于等于总集数才下载 - logger.info(f"开始下载种子 {torrent.title} ...") - content, _, torrent_files = self.download_torrent(torrent) - if not content: - logger.warn(f"{torrent.title} 种子下载失败！") - self._record_download_failure( - context=context, - error_msg="下载种子内容为空", - downloader=downloader, - source=source, - ) - __remember_context_failure(context) - continue - if isinstance(content, str): - logger.warn(f"{meta.org_string} 下载地址是磁力链，无法确定种子文件集数") - continue - torrent_episodes = cast( - List[int], - _new_torrent_helper().get_torrent_episodes( - torrent_files, - custom_words=custom_word_list, - ), - ) - logger.info(f"{meta.org_string} 解析种子文件集数为 {torrent_episodes}") - if not torrent_episodes: - continue - torrent_episodes_set = set(torrent_episodes) - if not _selection.allows_full_pack(torrent_episodes_set, context): - continue - # 更新集数范围 - begin_ep = min(torrent_episodes) - end_ep = max(torrent_episodes) - meta.set_episodes(begin=begin_ep, end=end_ep) - # 需要目标集范围；完整覆盖场景必须覆盖范围内每一集，不能只按数量判断。 - need_tv_info = __get_no_exist_media(need_mid, torrent_season[0]) - required_episodes = __get_required_episodes(need_mid, torrent_season[0]) \ - if __requires_complete_coverage(need_tv_info) else set() - need_total = __get_season_episodes(need_mid, torrent_season[0]) - complete_coverage_matched = bool(required_episodes) \ - and required_episodes.issubset(torrent_episodes_set) - if complete_coverage_matched: - logger.info( - f"{meta.org_string} 解析文件集数已完整覆盖目标范围：" - f"{episode_rules.format_ranges(sorted(required_episodes))}") - if required_episodes and not complete_coverage_matched: - missing_episodes = sorted(required_episodes.difference(torrent_episodes_set)) - logger.info( - f"{meta.org_string} 解析文件集数未覆盖目标范围，" - f"缺少 {episode_rules.format_ranges(missing_episodes)}，先放弃这个种子") - continue - if not required_episodes and need_total and len(torrent_episodes) < need_total: - logger.info( - f"{meta.org_string} 解析文件集数发现不是完整合集，先放弃这个种子") - continue - else: - # 下载 - logger.info(f"开始下载 {torrent.title} ...") - download_id = self.download_single( - context=context, - torrent_content=content, - save_path=save_path, - channel=channel, - source=source, - userid=userid, - username=username, - downloader=downloader, - custom_words=custom_words, governance=governance, - ) - else: - # 下载 - logger.info(f"开始下载 {torrent.title} ...") - download_id = self.download_single(context, save_path=save_path, - channel=channel, source=source, - userid=userid, username=username, - downloader=downloader, - custom_words=custom_words, governance=governance) + def _episode_pack_candidate( + self, context: Context, media_key: str, season: int, + needed: List[int], info: NotExistMediaInfo, + ) -> Set[int]: + media, meta = context.media_info, context.meta_info + if media is None or meta is None or media.type != MediaType.TV: + return set() + if not self.owner._matches_media_identity(media, media_key): + return set() + if context in self.downloaded or self._in_failure_cooldown(context): + return set() + if len(meta.season_list) != 1 or meta.season_list[0] != season: + return set() + episodes = set(meta.episode_list) + if not episodes: + return set() + effective = _selection.apply_allowed_episodes(set(needed), context) + if not effective: + return set() + if _selection.requires_complete_coverage(info): + required = _selection.get_required_episodes(self.no_exists, media_key, season) + matches = bool(required) and required.issubset(episodes) \ + and _selection.allows_full_pack(episodes, context) + else: + matches = episodes.issubset(effective) + return episodes if matches else set() - if download_id: - # 下载成功 - if complete_coverage_matched: - context.confirmed_full_coverage = True - logger.info(f"{torrent.title} 添加下载成功") - downloaded_list.append(context) - # 更新仍需季集 - need_season = __update_seasons(_mid=need_mid, - _need=need_season, - _current=torrent_season) - logger.info(f"{need_mid} 剩余需要季：{need_season}") - if not need_season: - # 全部下载完成 - break - else: - __remember_context_failure(context) - # 电视剧季内的集匹配 - if no_exists: - logger.info(f"开始电视剧完整集匹配：{no_exists}") - # TMDBID列表 - media_keys = list(no_exists) - for need_mid in media_keys: - # dict[season, [NotExistMediaInfo]] - season_map = no_exists.get(need_mid) - if not season_map: + def _download_partial_packs(self) -> None: + logger.info(f"开始电视剧多集拆包匹配：{self.no_exists}") + for media_key in list(self.no_exists): + seasons = self.no_exists.get(media_key) + if not seasons: + continue + for season in list(seasons): + info = seasons.get(season) + if info is None or _selection.requires_complete_coverage(info) or not info.episodes: continue - season_map_copy = copy.deepcopy(season_map) - # 循环每一季 - for sea, missing_info in season_map_copy.items(): - # 当前需要季 - season_number = sea - # 当前需要集 - need_episodes = missing_info.episodes - # TMDB总集数 - total_episode = missing_info.total_episode or 0 - # 需要开始集 - start_episode = missing_info.start_episode or 1 - # 缺失整季的转化为缺失集进行比较 - if not need_episodes: - need_episodes = list(range(start_episode, total_episode + 1)) - # 循环种子 - for context in contexts: - if runtime_stop_state.is_system_stopped: - break - # 媒体信息 - media = context.media_info - # 识别元数据 - meta = context.meta_info - # 非剧集不处理 - if media is None or meta is None or media.type != MediaType.TV: - continue - # 匹配TMDB - if self._matches_media_identity(media, need_mid): - # 不重复添加 - if context in downloaded_list: - continue - if __is_context_in_failure_cooldown(context): - continue - # 种子季 - torrent_season = meta.season_list - # 只处理单季含集的种子 - if len(torrent_season) != 1 or torrent_season[0] != season_number: - continue - # 种子集列表 - candidate_episodes: Set[int] = set(meta.episode_list) - # 整季的不处理 - if not candidate_episodes: - continue - # 上游对本候选施加的允许集（如洗版按集允许列表）裁剪本季缺集，得到真正可下载范围。 - effective_need = __apply_allowed_episodes(need_episodes, context) - if not effective_need: - continue - if __requires_complete_coverage(missing_info): - # 完整覆盖任务要求候选集数覆盖目标范围，允许资源包含范围外的额外集。 - required_episodes = __get_required_episodes(need_mid, season_number) - match_episodes = bool(required_episodes) \ - and required_episodes.issubset(candidate_episodes) \ - and _selection.allows_full_pack(candidate_episodes, context) - else: - # 普通缺集下载保持原语义：候选自身必须是所需集的子集。 - match_episodes = candidate_episodes.issubset(effective_need) - if match_episodes: - # 下载 - logger.info(f"开始下载 {meta.title} ...") - download_id = self.download_single(context, save_path=save_path, - channel=channel, source=source, - userid=userid, username=username, - downloader=downloader, - custom_words=custom_words, governance=governance) - if download_id: - # 下载成功 - if __requires_complete_coverage(missing_info): - context.confirmed_full_coverage = True - logger.info(f"{meta.title} 添加下载成功") - downloaded_list.append(context) - # 更新仍需集数 - need_episodes = __update_episodes(_mid=need_mid, - _need=need_episodes, - _sea=season_number, - _current=candidate_episodes) - logger.info(f"季 {season_number} 剩余需要集：{need_episodes}") - else: - __remember_context_failure(context) + self._download_season_partial_packs(media_key, season, info.episodes) - # 仍然缺失的剧集，从整季中选择需要的集数文件下载，仅支持QB和TR - if no_exists: - logger.info(f"开始电视剧多集拆包匹配：{no_exists}") - # TMDBID列表 - no_exists_list = list(no_exists) - for need_mid in no_exists_list: - # dict[season, [NotExistMediaInfo]] - remaining_seasons = no_exists.get(need_mid) - if not remaining_seasons: - continue - # 需要季列表 - season_numbers = list(remaining_seasons) - # 循环需要季 - for sea in season_numbers: - # NotExistMediaInfo - remaining_info = remaining_seasons.get(sea) - if remaining_info is None: - continue - # 当前需要季 - season_number = sea - # 当前需要集 - need_episodes = remaining_info.episodes - if __requires_complete_coverage(remaining_info): - continue - # 没有集的不处理 - if not need_episodes: - continue - # 循环种子 - for context in contexts: - if runtime_stop_state.is_system_stopped: - break - # 媒体信息 - media = context.media_info - # 识别元数据 - meta = context.meta_info - # 种子信息 - torrent = context.torrent_info - # 非剧集不处理 - if media is None or meta is None or media.type != MediaType.TV: - continue - # 不重复添加 - if context in downloaded_list: - continue - if __is_context_in_failure_cooldown(context): - continue - # 没有需要集后退出 - if not need_episodes: - break - # 上游对本候选施加的允许集（如洗版按集允许列表）裁剪本季缺集，得到真正可下载范围。 - effective_need = __apply_allowed_episodes(need_episodes, context) - if not effective_need: - continue - # 选中一个单季整季的或单季包括需要的所有集的 - if self._matches_media_identity(media, need_mid) \ - and (not meta.episode_list - or set(meta.episode_list).intersection(effective_need)) \ - and len(meta.season_list) == 1 \ - and meta.season_list[0] == season_number: - # 检查种子看是否有需要的集 - logger.info(f"开始下载种子 {torrent.title} ...") - content, _, torrent_files = self.download_torrent(torrent) - if not content: - logger.info(f"{torrent.title} 种子下载失败！") - self._record_download_failure( - context=context, - error_msg="下载种子内容为空", - downloader=downloader, - source=source, - ) - __remember_context_failure(context) - continue - if isinstance(content, str): - logger.warn(f"{meta.org_string} 下载地址是磁力链，无法解析种子文件集数") - continue - # 种子全部集 - torrent_episodes = cast( - List[int], - _new_torrent_helper().get_torrent_episodes( - torrent_files, - custom_words=custom_word_list, - ), - ) - logger.info(f"{torrent.site_name} - {meta.org_string} 解析种子文件集数：{torrent_episodes}") - # 选中的集 - selected_episodes = set(torrent_episodes).intersection(effective_need) - if not selected_episodes: - logger.info(f"{torrent.site_name} - {torrent.title} 没有需要的集，跳过...") - continue - logger.info(f"{torrent.site_name} - {torrent.title} 选中集数：{selected_episodes}") - # 添加下载 - logger.info(f"开始下载 {torrent.title} ...") - download_id = self.download_single( - context=context, - torrent_content=content, - episodes=selected_episodes, - save_path=save_path, - channel=channel, - source=source, - userid=userid, - username=username, - downloader=downloader, - custom_words=custom_words, governance=governance, - ) - if not download_id: - __remember_context_failure(context) - continue - # 下载成功 - logger.info(f"{torrent.title} 添加下载成功") - downloaded_list.append(context) - # 更新种子集数范围 - begin_ep = min(torrent_episodes) - end_ep = max(torrent_episodes) - meta.set_episodes(begin=begin_ep, end=end_ep) - # 更新仍需集数 - need_episodes = __update_episodes(_mid=need_mid, - _need=need_episodes, - _sea=season_number, - _current=selected_episodes) - logger.info(f"季 {season_number} 剩余需要集：{need_episodes}") + def _download_season_partial_packs( + self, media_key: str, season: int, needed: List[int], + ) -> None: + for context in self.contexts: + if runtime_stop_state.is_system_stopped: + break + media, meta = context.media_info, context.meta_info + if media is None or meta is None or media.type != MediaType.TV: + continue + # 拆包阶段沿用先检查重复/冷却、再检查缺集与媒体身份的顺序。 + if context in self.downloaded or self._in_failure_cooldown(context): + continue + if not needed: + break + effective = _selection.apply_allowed_episodes(set(needed), context) + if not effective or not self._matches_partial_pack(context, media_key, season, effective): + continue + selected = self._submit_partial_pack(context, effective) + if selected: + needed = _selection.update_no_exists_episodes( + self.no_exists, media_key, season, needed, selected, + ) + logger.info(f"季 {season} 剩余需要集：{needed}") - # 返回下载的资源，剩下没下完的 - logger.info(f"成功下载种子数：{len(downloaded_list)}，剩余未下载的剧集：{no_exists}") - return downloaded_list, None if no_exists_was_none else no_exists + def _matches_partial_pack( + self, context: Context, media_key: str, season: int, effective: Set[int], + ) -> bool: + meta = context.meta_info + assert meta is not None + return self.owner._matches_media_identity(context.media_info, media_key) \ + and (not meta.episode_list or bool(set(meta.episode_list).intersection(effective))) \ + and len(meta.season_list) == 1 and meta.season_list[0] == season + + def _submit_partial_pack(self, context: Context, effective: Set[int]) -> Set[int]: + """只在成功提交选中文件后更新识别集数，失败时保留原候选供换源记账。""" + parsed = self._read_torrent(context, partial=True) + if parsed is None: + return set() + content, episodes = parsed + torrent = context.torrent_info + selected = set(episodes).intersection(effective) + if not selected: + logger.info(f"{torrent.site_name} - {torrent.title} 没有需要的集，跳过...") + return set() + logger.info(f"{torrent.site_name} - {torrent.title} 选中集数：{selected}") + logger.info(f"开始下载 {torrent.title} ...") + if not self.owner.download_single( + context=context, torrent_content=content, episodes=selected, **self.options, + ): + self._remember_failure(context) + return set() + logger.info(f"{torrent.title} 添加下载成功") + self.downloaded.append(context) + meta = context.meta_info + assert meta is not None + meta.set_episodes(begin=min(episodes), end=max(episodes)) + return selected diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index 21985c9da..9c2f4b899 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -662,6 +662,17 @@ remain. Download history, file rows and the durable Outbox intent commit in one transaction; notifications, background post-processing and immediate event publication run only after that commit succeeds. +`batch.py` keeps its candidate state in a private per-invocation runner, never on +the shared Chain. After one selection/sort pass, it executes movie/music, whole +season, labelled episode-pack and partial-pack phases in that order. The phases +share failure cooldown and update missing seasons/episodes only after successful +submission, using the rules in `app.application.download.selection`. Whole-season +file inspection retains its metadata update even when coverage is insufficient, +so later phases can still evaluate the candidate's identified episode range. +Partial-pack inspection updates that range only after successful submission. +The public `batch_download` signature and the supplied missing-map identity remain +unchanged. + Search orchestration is owned by the same-named `app.chain.search` package. Its root lazily exposes only the stable `SearchChain`; `facade.py` preserves the direct `SearchChain -> ChainBase` MRO, event identity and the three exact private diff --git a/tests/fixtures/architecture/complexity-baseline.json b/tests/fixtures/architecture/complexity-baseline.json index 2c31ef976..9382a3d47 100644 --- a/tests/fixtures/architecture/complexity-baseline.json +++ b/tests/fixtures/architecture/complexity-baseline.json @@ -44,7 +44,6 @@ "app/application/torrent/download.py:TorrentHelper.match_torrent": 16, "app/application/transfer/models.py:TransferPlanCheckpoint.__post_init__": 18, "app/chain/_messaging.py:_notification_route_steps": 17, - "app/chain/download/batch.py:DownloadBatchOwner._execute_batch_download": 79, "app/chain/download/existence.py:DownloadExistenceOwner.get_no_exists_info": 23, "app/chain/download/subtitle.py:DownloadSubtitleOwner._save_subtitle_response": 16, "app/chain/media/album.py:_album_context_with_resource": 20, @@ -150,7 +149,6 @@ "app/application/rss.py:RssHelper._parse_impl": 7, "app/application/torrent/download.py:TorrentHelper.download_torrent": 6, "app/chain/_messaging.py:_notification_route_steps": 6, - "app/chain/download/batch.py:DownloadBatchOwner._execute_batch_download": 8, "app/chain/scraping.py:ScrapingChain._download_and_save_image": 7, "app/chain/scraping.py:ScrapingChain.scrape_metadata_event": 6, "app/chain/subscribe/match.py:SubscribeMatchOwner._execute_match": 7, diff --git a/tests/test_download_batch_phases.py b/tests/test_download_batch_phases.py new file mode 100644 index 000000000..2a35094d3 --- /dev/null +++ b/tests/test_download_batch_phases.py @@ -0,0 +1,319 @@ +"""保护批量下载各阶段之间的状态传递与副作用顺序。""" + +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +import app.chain.download.batch as download_batch +import app.chain.download.selection as download_selection +from app.chain.download import DownloadChain +from app.domain.context import Context, MediaInfo, TorrentInfo +from app.domain.metainfo import MetaInfo +from app.schemas.mediaserver import NotExistMediaInfo +from app.schemas.types import MediaSource, MediaType, NotificationChannel + + +def _context( + name: str, + media_id: str = "1", + episodes: tuple[int, ...] = (), + season: int = 1, + media_type: MediaType = MediaType.TV, +) -> Context: + meta = MetaInfo(f"{name} S{season:02d}") + if episodes: + meta.set_episodes(begin=min(episodes), end=max(episodes)) + return Context( + meta_info=meta, + media_info=MediaInfo( + title=name, type=media_type, media_source=MediaSource.TMDB, media_id=media_id, + ), + torrent_info=TorrentInfo(title=name, site_name="test", site=1), + ) + + +def _missing( + episodes: tuple[int, ...] = (), + total: int = 4, + season: int = 1, + complete: bool = False, +) -> NotExistMediaInfo: + return NotExistMediaInfo( + season=season, episodes=list(episodes), total_episode=total, + start_episode=1, require_complete_coverage=complete, + ) + + +@pytest.fixture +def batch(monkeypatch): + """只替换事件、排序、持久化及下载边界，保留真实阶段选择和失败指纹。""" + chain = object.__new__(DownloadChain) + chain.eventmanager = SimpleNamespace(send_event=MagicMock(return_value=None)) + chain._active_download_failure_fingerprints = MagicMock(return_value={}) + chain._record_download_failure = MagicMock() + chain._log_download_failure_cooldown = MagicMock() + chain.download_single = MagicMock(return_value="download-id") + chain.download_torrent = MagicMock(return_value=(b"torrent", None, [1, 2, 3, 4])) + helper = SimpleNamespace( + sort_torrents=MagicMock(side_effect=lambda contexts: contexts), + get_torrent_episodes=MagicMock(side_effect=lambda files, **_kwargs: files), + ) + stop = SimpleNamespace(is_system_stopped=False) + monkeypatch.setattr(download_batch, "TorrentHelper", lambda: helper) + monkeypatch.setattr(download_selection, "TorrentHelper", lambda: helper) + monkeypatch.setattr(download_batch, "runtime_stop_state", stop) + monkeypatch.setattr(download_selection, "runtime_stop_state", stop) + return SimpleNamespace(chain=chain, helper=helper, stop=stop) + + +def _submitted(chain) -> list[Context]: + return [ + call.args[0] if call.args else call.kwargs["context"] + for call in chain.download_single.call_args_list + ] + + +def test_batch_preserves_phase_order_and_submission_options(batch): + partial = _context("partial", media_id="3") + labelled = _context("labelled", media_id="2", episodes=(2,)) + whole = _context("whole") + movie = _context("movie", media_type=MediaType.MOVIE) + music = _context("music", media_type=MediaType.MUSIC) + missing = { + "tmdb:3": {1: _missing((3,))}, + "tmdb:2": {1: _missing((2,))}, + "tmdb:1": {1: _missing()}, + } + options = { + "save_path": "smb:/server/media", + "channel": NotificationChannel.Telegram, + "source": "Subscribe|42", + "userid": "user-id", + "username": "plugin-name", + "downloader": "test-downloader", + "custom_words": "first\nsecond", + "governance": object(), + } + result, remaining = batch.chain.batch_download( + [partial, labelled, whole, movie, music], missing, **options, + ) + + assert result == [movie, music, whole, labelled, partial] + assert _submitted(batch.chain) == result + assert remaining is missing + assert remaining == {} + assert batch.chain.eventmanager.send_event.call_count == 1 + assert batch.helper.sort_torrents.call_count == 1 + assert batch.chain._active_download_failure_fingerprints.call_count == 1 + for call in batch.chain.download_single.call_args_list: + assert {key: call.kwargs[key] for key in options} == options + calls = batch.chain.download_single.call_args_list + assert "torrent_content" not in calls[0].kwargs + assert calls[2].kwargs["torrent_content"] == b"torrent" + assert "torrent_content" not in calls[3].kwargs + assert calls[4].kwargs["episodes"] == {3} + assert all( + call.kwargs["custom_words"] == ["first", "second"] + for call in batch.helper.get_torrent_episodes.call_args_list + ) + + +def test_incomplete_season_becomes_labelled_candidate_after_file_inspection(batch): + candidate = _context("incomplete") + missing = {"tmdb:1": {1: _missing()}} + batch.chain.download_torrent.return_value = (b"torrent", None, [1, 2]) + + result, remaining = batch.chain.batch_download([candidate], missing) + + assert result == [candidate] + assert remaining is missing + assert remaining["tmdb:1"][1].episodes == [3, 4] + assert candidate.meta_info.episode_list == [1, 2] + batch.chain.download_torrent.assert_called_once() + batch.chain.download_single.assert_called_once() + assert "torrent_content" not in batch.chain.download_single.call_args.kwargs + batch.chain._record_download_failure.assert_not_called() + + +def test_failed_season_submission_stays_in_cooldown_after_metadata_changes(batch): + whole = _context("whole") + fallback = _context("fallback", episodes=(1,)) + missing = {"tmdb:1": {1: _missing(total=3)}} + batch.chain.download_torrent.return_value = (b"torrent", None, [1, 2, 3]) + batch.chain.download_single.side_effect = [None, "fallback-id"] + + result, remaining = batch.chain.batch_download([whole, fallback], missing) + + assert result == [fallback] + assert _submitted(batch.chain) == [whole, fallback] + assert remaining["tmdb:1"][1].episodes == [2, 3] + assert whole.meta_info.episode_list == [1, 2, 3] + batch.chain.download_torrent.assert_called_once() + assert batch.chain._log_download_failure_cooldown.call_count == 2 + + +def test_empty_torrent_is_recorded_once_and_not_retried_in_later_phases(batch): + whole = _context("empty") + fallback = _context("fallback", episodes=(1,)) + missing = {"tmdb:1": {1: _missing(total=3)}} + batch.chain.download_torrent.return_value = (None, "failed", []) + + result, remaining = batch.chain.batch_download( + [whole, fallback], missing, source="Subscribe|42", downloader="test", + ) + + assert result == [fallback] + assert remaining["tmdb:1"][1].episodes == [2, 3] + batch.chain.download_torrent.assert_called_once() + batch.chain._record_download_failure.assert_called_once_with( + context=whole, error_msg="下载种子内容为空", downloader="test", source="Subscribe|42", + ) + + +def test_uninspectable_magnet_is_not_cooled_down_as_a_download_failure(batch): + whole = _context("magnet") + fallback = _context("fallback", episodes=(1,)) + missing = {"tmdb:1": {1: _missing(total=3)}} + batch.chain.download_torrent.return_value = ("magnet:?xt=test", None, []) + + result, remaining = batch.chain.batch_download([whole, fallback], missing) + + assert result == [fallback] + assert remaining["tmdb:1"][1].episodes == [2, 3] + assert batch.chain.download_torrent.call_count == 2 + batch.chain._record_download_failure.assert_not_called() + batch.chain._log_download_failure_cooldown.assert_not_called() + batch.helper.get_torrent_episodes.assert_not_called() + + +def test_partial_packs_only_settle_successful_allowed_episode_selections(batch): + failed = _context("failed", episodes=(1, 4)) + first = _context("first") + first.allowed_episodes = {1, 3} + second = _context("second") + missing = {"tmdb:1": {1: _missing((1, 2, 3))}} + batch.chain.download_torrent.side_effect = [ + (b"failed", None, [1, 2, 4]), + (b"first", None, [1, 3, 4]), + (b"second", None, [2, 4]), + ] + needs_at_submission = [] + + def submit(**kwargs): + needs_at_submission.append(list(missing["tmdb:1"][1].episodes)) + return None if kwargs["context"] is failed else "id" + + batch.chain.download_single.side_effect = submit + result, remaining = batch.chain.batch_download([failed, first, second], missing) + + assert result == [first, second] + assert remaining is missing + assert remaining == {} + assert needs_at_submission == [[1, 2, 3], [1, 2, 3], [2]] + assert [ + call.kwargs["episodes"] for call in batch.chain.download_single.call_args_list + ] == [{1, 2}, {1, 3}, {2}] + assert first.meta_info.episode_list == [1, 2, 3, 4] + assert second.meta_info.episode_list == [2, 3, 4] + assert failed.meta_info.episode_list == [1, 2, 3, 4] + + +@pytest.mark.parametrize("missing", [None, {}]) +def test_batch_preserves_optional_missing_result_and_resets_per_call_state(batch, missing): + candidate = _context("movie", media_type=MediaType.MOVIE) + batch.chain.download_single.side_effect = [None, "retry-id"] + + first, first_remaining = batch.chain.batch_download([candidate], missing) + second, second_remaining = batch.chain.batch_download([candidate], missing) + + assert first == [] + assert second == [candidate] + assert first_remaining is missing + assert second_remaining is missing + assert batch.chain.download_single.call_count == 2 + + +def test_reentrant_batch_call_keeps_outer_candidates_and_accounting(batch): + """下载回调触发另一批次时，两次调用不能共用候选、结果或缺集状态。""" + movie = _context("outer-movie", media_type=MediaType.MOVIE) + nested = _context("nested-movie", media_type=MediaType.MOVIE) + whole = _context("whole") + missing = {"tmdb:1": {1: _missing()}} + nested_results = [] + + def submit(*args, **kwargs): + context = args[0] if args else kwargs["context"] + if context is movie: + nested_results.append(batch.chain.batch_download([nested])) + return "id" + + batch.chain.download_single.side_effect = submit + result, remaining = batch.chain.batch_download([whole, movie], missing) + + assert result == [movie, whole] + assert nested_results == [([nested], None)] + assert remaining is missing + assert remaining == {} + assert _submitted(batch.chain) == [movie, nested, whole] + + +def test_stop_during_movie_submission_prevents_later_phase_side_effects(batch): + movie = _context("movie", media_type=MediaType.MOVIE) + whole = _context("whole") + missing = {"tmdb:1": {1: _missing()}} + + def submit(*_args, **_kwargs): + batch.stop.is_system_stopped = True + return "movie-id" + + batch.chain.download_single.side_effect = submit + result, remaining = batch.chain.batch_download([whole, movie], missing) + + assert result == [movie] + assert remaining == {"tmdb:1": {1: _missing()}} + batch.chain.download_single.assert_called_once() + batch.chain.download_torrent.assert_not_called() + + +def test_multiseason_pack_submits_without_reading_torrent_files(batch): + candidate = _context("multiseason", season=0) + candidate.meta_info.end_season = 1 + missing = {"tmdb:1": {0: _missing(season=0), 1: _missing()}} + + result, remaining = batch.chain.batch_download([candidate], missing) + + assert result == [candidate] + assert remaining is missing + assert remaining == {} + batch.chain.download_torrent.assert_not_called() + assert batch.chain.download_single.call_args.args == (candidate,) + assert candidate.confirmed_full_coverage is False + + +def test_partial_failure_does_not_rewrite_candidate_episode_range(batch): + candidate = _context("partial") + missing = {"tmdb:1": {1: _missing((2,))}} + batch.chain.download_single.return_value = None + batch.chain.download_torrent.return_value = (b"torrent", None, [1, 2, 3]) + + result, remaining = batch.chain.batch_download([candidate], missing) + + assert result == [] + assert remaining["tmdb:1"][1].episodes == [2] + assert candidate.meta_info.episode_list == [] + batch.chain.download_single.assert_called_once() + + +def test_failed_complete_coverage_submission_does_not_confirm_or_settle(batch): + candidate = _context("complete") + missing = {"tmdb:1": {1: _missing(complete=True)}} + batch.chain.download_single.return_value = None + + result, remaining = batch.chain.batch_download([candidate], missing) + + assert result == [] + assert remaining["tmdb:1"][1].require_complete_coverage is True + assert remaining["tmdb:1"][1].episodes == [] + assert candidate.confirmed_full_coverage is False + batch.chain.download_single.assert_called_once()