refactor(download): split batch matching into explicit phases

This commit is contained in:
freeman
2026-09-30 21:50:47 +08:00
parent 3fc57808d7
commit 9eae3b9b83
4 changed files with 654 additions and 441 deletions

View File

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

View File

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

View File

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

View File

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