diff --git a/app/chain/download.py b/app/chain/download.py index 18ae12d1..2c6e29c1 100644 --- a/app/chain/download.py +++ b/app/chain/download.py @@ -1,11 +1,13 @@ import base64 import copy +import hashlib import json import re import shutil import time from pathlib import Path from typing import List, Optional, Tuple, Set, Dict, Union +from urllib.parse import parse_qs, urlparse from app import schemas from app.chain import ChainBase @@ -16,6 +18,7 @@ from app.core.context import MediaInfo, SubtitleInfo, TorrentInfo, Context from app.core.event import eventmanager, Event from app.core.meta import MetaBase from app.core.metainfo import MetaInfo +from app.db.downloadfailure_oper import DownloadFailureOper from app.db.downloadhistory_oper import DownloadHistoryOper from app.db.mediaserver_oper import MediaServerOper from app.helper.directory import DirectoryHelper, validate_download_save_path @@ -31,6 +34,21 @@ from app.utils.string import StringUtils from app.utils.system import SystemUtils +DOWNLOAD_FAILURE_RESOURCE_TTL_SECONDS = 24 * 60 * 60 +DOWNLOAD_FAILURE_TRANSIENT_TTL_SECONDS = 60 * 60 +DOWNLOAD_FAILURE_RESOURCE_ERROR_KEYWORDS = ( + "无法读取种子文件", + "下载种子内容为空", + "无法获取下载地址", + "种子下载失败", + "torrent not found", + "not found", + "404", + "deleted", + "invalid torrent", +) + + class DownloadChain(ChainBase): """ 下载处理链 @@ -365,6 +383,183 @@ class DownloadChain(ChainBase): except Exception as err: logger.error(f"提交下载成功后处理后台任务失败：{str(err)}") + @staticmethod + def _is_subscribe_source(source: Optional[str]) -> bool: + """ + 判断下载来源是否为订阅任务。 + """ + return bool(source and str(source).startswith("Subscribe|")) + + @staticmethod + def _format_failure_episodes(meta: Optional[MetaBase]) -> Optional[str]: + """ + 从识别元数据中格式化用于失败记录的集数。 + """ + if not meta: + return None + if getattr(meta, "episode", None): + return meta.episode + episode_list = getattr(meta, "episode_list", None) + if episode_list: + return StringUtils.format_ep(list(episode_list)) + return None + + @staticmethod + def _torrent_resource_key(torrent: Optional[TorrentInfo]) -> str: + """ + 生成不保存敏感下载链接的种子资源键。 + """ + if not torrent: + return "" + for attr_name in ("torrent_id", "info_hash"): + value = getattr(torrent, attr_name, None) + if value: + return str(value) + + for attr_name in ("page_url", "enclosure"): + url = getattr(torrent, attr_name, None) + if not url: + continue + match = re.search(r"\[(.*?)](.*)", str(url)) + if match: + url = match.group(2) + parsed = urlparse(str(url)) + params = parse_qs(parsed.query) + for param_name in ("id", "torrentid", "torrent_id", "tid", "hash"): + values = params.get(param_name) + if values: + return f"{parsed.netloc}:{param_name}={values[0]}" + if parsed.netloc and parsed.path: + return f"{parsed.netloc}{parsed.path}" + + title = getattr(torrent, "title", "") or "" + size = getattr(torrent, "size", "") or "" + return f"title={title}|size={size}" + + @classmethod + def _build_download_failure_fingerprint(cls, context: Context) -> Optional[str]: + """ + 根据媒体和种子资源信息生成失败冷却指纹。 + """ + media = getattr(context, "media_info", None) + torrent = getattr(context, "torrent_info", None) + if not media or not torrent: + return None + + media_type = getattr(getattr(media, "type", None), "value", getattr(media, "type", None)) + media_key = ( + getattr(media, "tmdb_id", None) + or getattr(media, "douban_id", None) + or getattr(media, "imdb_id", None) + or getattr(media, "tvdb_id", None) + or f"{getattr(media, 'title', '')}:{getattr(media, 'year', '')}" + ) + meta = getattr(context, "meta_info", None) + site = getattr(torrent, "site", None) or getattr(torrent, "site_name", None) + payload = { + "media_type": str(media_type or ""), + "media_key": str(media_key or ""), + "season": str(getattr(meta, "season", None) or getattr(media, "season", None) or ""), + "episodes": cls._format_failure_episodes(meta) or "", + "site": str(site or ""), + "resource": cls._torrent_resource_key(torrent), + } + if not payload["media_type"] or not payload["media_key"] or not payload["resource"]: + return None + raw_text = json.dumps(payload, ensure_ascii=False, sort_keys=True) + return hashlib.sha256(raw_text.encode("utf-8")).hexdigest() + + @staticmethod + def _download_failure_ttl(error_msg: Optional[str]) -> int: + """ + 按失败原因确定资源冷却时间。 + """ + error_text = str(error_msg or "").lower() + if any(keyword in error_text for keyword in DOWNLOAD_FAILURE_RESOURCE_ERROR_KEYWORDS): + return DOWNLOAD_FAILURE_RESOURCE_TTL_SECONDS + return DOWNLOAD_FAILURE_TRANSIENT_TTL_SECONDS + + def _record_download_failure( + self, + context: Context, + error_msg: Optional[str], + downloader: Optional[str] = None, + source: Optional[str] = None, + episodes: Optional[Set[int]] = None, + ) -> Optional[str]: + """ + 记录资源级下载失败，并返回本次失败指纹。 + """ + fingerprint = self._build_download_failure_fingerprint(context) + if not fingerprint: + return None + + now_timestamp = time.time() + now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(now_timestamp)) + next_retry_at = time.strftime( + "%Y-%m-%d %H:%M:%S", + time.localtime(now_timestamp + self._download_failure_ttl(error_msg)), + ) + media = context.media_info + meta = context.meta_info + torrent = context.torrent_info + site = getattr(torrent, "site", None) + try: + DownloadFailureOper().record_failure( + fingerprint=fingerprint, + now_time=now_time, + next_retry_at=next_retry_at, + type=getattr(getattr(media, "type", None), "value", getattr(media, "type", None)), + title=getattr(media, "title", None), + year=getattr(media, "year", None), + tmdbid=getattr(media, "tmdb_id", None), + doubanid=getattr(media, "douban_id", None), + seasons=getattr(meta, "season", None), + episodes=StringUtils.format_ep(list(episodes)) if episodes else self._format_failure_episodes(meta), + site=site if isinstance(site, int) else None, + site_name=getattr(torrent, "site_name", None), + torrent_id=self._torrent_resource_key(torrent), + torrent_name=getattr(torrent, "title", None), + torrent_size=getattr(torrent, "size", None), + downloader=downloader, + source=str(source)[:1000] if source else None, + error_message=str(error_msg or "")[:1000], + ) + except Exception as err: + logger.error(f"记录下载失败冷却失败：{str(err)}") + return fingerprint + + def _active_download_failure_fingerprints( + self, + contexts: List[Context], + source: Optional[str], + ) -> Set[str]: + """ + 查询当前订阅候选中仍处于冷却期的失败指纹。 + """ + if not self._is_subscribe_source(source): + return set() + fingerprints = [ + fingerprint + for fingerprint in [ + self._build_download_failure_fingerprint(context) + for context in contexts or [] + ] + if fingerprint + ] + if not fingerprints: + return set() + now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + try: + return set( + DownloadFailureOper() + .get_active_by_fingerprints(fingerprints=fingerprints, now_time=now_time) + .keys() + ) + except Exception as err: + logger.error(f"查询下载失败冷却失败：{str(err)}") + return set() + def download_torrent(self, torrent: TorrentInfo, channel: MessageChannel = None, source: Optional[str] = None, @@ -578,6 +773,13 @@ class DownloadChain(ChainBase): torrent_content = cache_backend.get(torrent_file.as_posix(), region="torrents") if not torrent_content: + self._record_download_failure( + context=context, + error_msg="下载种子内容为空", + downloader=downloader or _site_downloader, + source=source, + episodes=episodes, + ) return (None, "下载种子内容为空") if return_detail else None # 获取种子文件的文件夹名和文件清单 @@ -732,6 +934,13 @@ class DownloadChain(ChainBase): # 下载失败 logger.error(f"{_media.title_year} 添加下载任务失败：" f"{_torrent.title} - {_torrent.enclosure}，{error_msg}") + self._record_download_failure( + context=context, + error_msg=error_msg, + downloader=_downloader or downloader or _site_downloader, + source=source, + episodes=episodes, + ) # 只发送给对应渠道和用户 self.post_message(Notification( channel=channel, @@ -897,6 +1106,28 @@ class DownloadChain(ChainBase): # 仅排序，不提前按媒体控重；下载失败时需要继续尝试同组后续候选。 contexts = TorrentHelper().sort_torrents(contexts) + active_failure_fingerprints = self._active_download_failure_fingerprints( + contexts=contexts, + source=source, + ) + + def __is_context_in_failure_cooldown(_context: Context) -> bool: + """ + 判断候选资源是否仍处于失败冷却期。 + """ + fingerprint = self._build_download_failure_fingerprint(_context) + if fingerprint and fingerprint in active_failure_fingerprints: + logger.info(f"{_context.torrent_info.title} 近期添加下载失败，暂时跳过该资源") + return True + return False + + def __remember_context_failure(_context: Context) -> None: + """ + 将本轮失败候选加入内存冷却集合，避免同一批次重复尝试。 + """ + fingerprint = self._build_download_failure_fingerprint(_context) + if fingerprint: + active_failure_fingerprints.add(fingerprint) # 如果是电影，直接下载 downloaded_movies = set() @@ -904,6 +1135,8 @@ class DownloadChain(ChainBase): if global_vars.is_system_stopped: break if context.media_info.type == MediaType.MOVIE: + if __is_context_in_failure_cooldown(context): + continue movie_key = __get_movie_download_key(context) if movie_key in downloaded_movies: continue @@ -915,6 +1148,8 @@ class DownloadChain(ChainBase): logger.info(f"{context.torrent_info.title} 添加下载成功") downloaded_list.append(context) downloaded_movies.add(movie_key) + else: + __remember_context_failure(context) # 电视剧整季匹配 if no_exists: @@ -959,6 +1194,8 @@ class DownloadChain(ChainBase): # 不重复添加 if context in downloaded_list: continue + if __is_context_in_failure_cooldown(context): + continue # 种子季是需要季或者子集 if set(torrent_season).issubset(set(need_season)): complete_coverage_matched = False @@ -968,6 +1205,13 @@ class DownloadChain(ChainBase): 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} 下载地址是磁力链，无法确定种子文件集数") @@ -1039,6 +1283,8 @@ class DownloadChain(ChainBase): if not need_season: # 全部下载完成 break + else: + __remember_context_failure(context) # 电视剧季内的集匹配 if no_exists: logger.info(f"开始电视剧完整集匹配：{no_exists}") @@ -1079,6 +1325,8 @@ class DownloadChain(ChainBase): # 不重复添加 if context in downloaded_list: continue + if __is_context_in_failure_cooldown(context): + continue # 种子季 torrent_season = meta.season_list # 只处理单季含集的种子 @@ -1121,6 +1369,8 @@ class DownloadChain(ChainBase): _sea=need_season, _current=torrent_episodes) logger.info(f"季 {need_season} 剩余需要集：{need_episodes}") + else: + __remember_context_failure(context) # 仍然缺失的剧集，从整季中选择需要的集数文件下载，仅支持QB和TR if no_exists: @@ -1163,6 +1413,8 @@ class DownloadChain(ChainBase): # 不重复添加 if context in downloaded_list: continue + if __is_context_in_failure_cooldown(context): + continue # 没有需要集后退出 if not need_episodes: break @@ -1181,6 +1433,13 @@ class DownloadChain(ChainBase): 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} 下载地址是磁力链，无法解析种子文件集数") @@ -1209,6 +1468,7 @@ class DownloadChain(ChainBase): custom_words=custom_words ) if not download_id: + __remember_context_failure(context) continue # 下载成功 logger.info(f"{torrent.title} 添加下载成功") diff --git a/app/db/downloadfailure_oper.py b/app/db/downloadfailure_oper.py new file mode 100644 index 00000000..42faa3d2 --- /dev/null +++ b/app/db/downloadfailure_oper.py @@ -0,0 +1,61 @@ +from typing import Dict, List, Optional + +from app.db import DbOper +from app.db.models.downloadfailure import DownloadFailure + + +class DownloadFailureOper(DbOper): + """ + 下载失败冷却记录管理。 + """ + + def get_active_by_fingerprints( + self, + fingerprints: List[str], + now_time: str, + ) -> Dict[str, DownloadFailure]: + """ + 批量按指纹查询仍在冷却期的失败记录。 + """ + failures = DownloadFailure.get_active_by_fingerprints( + self._db, + fingerprints=fingerprints, + now_time=now_time, + ) + return { + failure.fingerprint: failure + for failure in failures + if failure and failure.fingerprint + } + + def record_failure( + self, + fingerprint: str, + now_time: str, + next_retry_at: str, + **kwargs: object, + ) -> DownloadFailure: + """ + 新增或更新资源失败记录。 + """ + return DownloadFailure.record_failure( + self._db, + fingerprint=fingerprint, + now_time=now_time, + next_retry_at=next_retry_at, + **kwargs, + ) + + def delete_expired( + self, + before_time: str, + limit: Optional[int] = 500, + ) -> int: + """ + 删除已过期较久的失败记录。 + """ + return DownloadFailure.delete_expired( + self._db, + before_time=before_time, + limit=limit, + ) diff --git a/app/db/models/__init__.py b/app/db/models/__init__.py index 81b7227b..72166dc3 100644 --- a/app/db/models/__init__.py +++ b/app/db/models/__init__.py @@ -1,4 +1,5 @@ from .agentchat import AgentChat +from .downloadfailure import DownloadFailure from .downloadhistory import DownloadHistory, DownloadFiles from .mediaserver import MediaServerItem from .message import Message diff --git a/app/db/models/downloadfailure.py b/app/db/models/downloadfailure.py new file mode 100644 index 00000000..b90e5bda --- /dev/null +++ b/app/db/models/downloadfailure.py @@ -0,0 +1,137 @@ +from typing import List, Optional + +from sqlalchemy import Column, Float, Index, Integer, String +from sqlalchemy.orm import Session + +from app.db import Base, db_query, db_update, get_id_column + + +class DownloadFailure(Base): + """ + 下载失败冷却记录。 + """ + + id = get_id_column() + # 资源失败指纹 + fingerprint = Column(String, nullable=False) + # 类型 电影/电视剧 + type = Column(String) + # 标题 + title = Column(String) + # 年份 + year = Column(String) + # TMDBID + tmdbid = Column(Integer) + # 豆瓣ID + doubanid = Column(String) + # Sxx + seasons = Column(String) + # Exx + episodes = Column(String) + # 站点ID + site = Column(Integer) + # 站点名称 + site_name = Column(String) + # 种子资源键 + torrent_id = Column(String) + # 种子名称 + torrent_name = Column(String) + # 种子大小 + torrent_size = Column(Float) + # 下载器 + downloader = Column(String) + # 下载来源 + source = Column(String) + # 失败原因 + error_message = Column(String) + # 重试次数 + retry_count = Column(Integer, default=0) + # 首次失败时间 + first_failed_at = Column(String) + # 最近失败时间 + last_failed_at = Column(String) + # 下次允许重试时间 + next_retry_at = Column(String) + + __table_args__ = ( + Index("ux_downloadfailure_fingerprint", "fingerprint", unique=True), + Index("ix_downloadfailure_next_retry_at", "next_retry_at"), + Index("ix_downloadfailure_media_site", "type", "tmdbid", "doubanid", "site"), + ) + + @classmethod + @db_query + def get_active_by_fingerprints( + cls, + db: Session, + fingerprints: List[str], + now_time: str, + ) -> List["DownloadFailure"]: + """ + 按指纹批量查询仍处于冷却期的失败记录。 + """ + normalized = list(dict.fromkeys([fingerprint for fingerprint in fingerprints if fingerprint])) + if not normalized: + return [] + return ( + db.query(cls) + .filter(cls.fingerprint.in_(normalized), cls.next_retry_at > now_time) + .all() + ) + + @classmethod + @db_update + def record_failure( + cls, + db: Session, + fingerprint: str, + now_time: str, + next_retry_at: str, + **kwargs: object, + ) -> "DownloadFailure": + """ + 新增或更新资源失败记录。 + """ + failure = db.query(cls).filter(cls.fingerprint == fingerprint).first() + payload = { + **kwargs, + "fingerprint": fingerprint, + "last_failed_at": now_time, + "next_retry_at": next_retry_at, + } + if failure: + payload["retry_count"] = (failure.retry_count or 0) + 1 + for key, value in payload.items(): + setattr(failure, key, value) + return failure + + failure = cls( + **payload, + retry_count=1, + first_failed_at=now_time, + ) + db.add(failure) + return failure + + @classmethod + @db_update + def delete_expired( + cls, + db: Session, + before_time: str, + limit: Optional[int] = 500, + ) -> int: + """ + 分批清理已过期较久的失败冷却记录。 + """ + ids = [ + row[0] + for row in db.query(cls.id) + .filter(cls.next_retry_at < before_time) + .order_by(cls.id.asc()) + .limit(limit) + .all() + ] + if not ids: + return 0 + return db.query(cls).filter(cls.id.in_(ids)).delete(synchronize_session=False) diff --git a/database/versions/b7d4a9c2e6f1_2_2_11.py b/database/versions/b7d4a9c2e6f1_2_2_11.py new file mode 100644 index 00000000..33877f62 --- /dev/null +++ b/database/versions/b7d4a9c2e6f1_2_2_11.py @@ -0,0 +1,80 @@ +"""2.2.11 +新增下载失败资源冷却表 + +Revision ID: b7d4a9c2e6f1 +Revises: 8ab72c49d1e3 +Create Date: 2026-07-07 +""" + +from alembic import op +import sqlalchemy as sa + +revision = "b7d4a9c2e6f1" +down_revision = "8ab72c49d1e3" +branch_labels = None +depends_on = None + + +def _has_table(inspector: sa.Inspector, table_name: str) -> bool: + """检查数据表是否已存在。""" + return table_name in inspector.get_table_names() + + +def upgrade() -> None: + """升级数据库结构。""" + inspector = sa.inspect(op.get_bind()) + if _has_table(inspector, "downloadfailure"): + return + + op.create_table( + "downloadfailure", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("fingerprint", sa.String(), nullable=False), + sa.Column("type", sa.String(), nullable=True), + sa.Column("title", sa.String(), nullable=True), + sa.Column("year", sa.String(), nullable=True), + sa.Column("tmdbid", sa.Integer(), nullable=True), + sa.Column("doubanid", sa.String(), nullable=True), + sa.Column("seasons", sa.String(), nullable=True), + sa.Column("episodes", sa.String(), nullable=True), + sa.Column("site", sa.Integer(), nullable=True), + sa.Column("site_name", sa.String(), nullable=True), + sa.Column("torrent_id", sa.String(), nullable=True), + sa.Column("torrent_name", sa.String(), nullable=True), + sa.Column("torrent_size", sa.Float(), nullable=True), + sa.Column("downloader", sa.String(), nullable=True), + sa.Column("source", sa.String(), nullable=True), + sa.Column("error_message", sa.String(), nullable=True), + sa.Column("retry_count", sa.Integer(), nullable=True), + sa.Column("first_failed_at", sa.String(), nullable=True), + sa.Column("last_failed_at", sa.String(), nullable=True), + sa.Column("next_retry_at", sa.String(), nullable=True), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + "ux_downloadfailure_fingerprint", + "downloadfailure", + ["fingerprint"], + unique=True, + ) + op.create_index( + "ix_downloadfailure_next_retry_at", + "downloadfailure", + ["next_retry_at"], + ) + op.create_index( + "ix_downloadfailure_media_site", + "downloadfailure", + ["type", "tmdbid", "doubanid", "site"], + ) + + +def downgrade() -> None: + """回滚数据库结构。""" + inspector = sa.inspect(op.get_bind()) + if not _has_table(inspector, "downloadfailure"): + return + op.drop_index("ix_downloadfailure_media_site", table_name="downloadfailure") + op.drop_index("ix_downloadfailure_next_retry_at", table_name="downloadfailure") + op.drop_index("ux_downloadfailure_fingerprint", table_name="downloadfailure") + op.drop_table("downloadfailure") diff --git a/tests/test_download_chain.py b/tests/test_download_chain.py index 2105908e..13fc6eed 100644 --- a/tests/test_download_chain.py +++ b/tests/test_download_chain.py @@ -568,6 +568,147 @@ def test_batch_download_threads_custom_words_to_download_single(monkeypatch): assert chain.download_single.call_args.kwargs["custom_words"] == custom_words +def test_download_single_records_failure_cooldown_when_downloader_rejects(monkeypatch): + """ + 下载器拒绝种子且没有返回 hash 时，应记录资源级失败冷却。 + """ + captured = {} + + class _CapturingDownloadFailureOper: + """ + 捕获下载失败冷却记录，避免测试写入数据库。 + """ + + def record_failure(self, **kwargs: object) -> SimpleNamespace: + """ + 保存写入字段供断言使用。 + """ + captured.update(kwargs) + return SimpleNamespace(id=1) + + monkeypatch.setattr( + "app.helper.directory.DirectoryHelper.get_download_dirs", + lambda _self: _download_dirs(), + ) + monkeypatch.setattr(download_module, "TorrentHelper", _FakeTorrentHelper) + monkeypatch.setattr(download_module, "DownloadFailureOper", _CapturingDownloadFailureOper) + monkeypatch.setattr(download_module.eventmanager, "send_event", lambda *args, **kwargs: None) + + chain = DownloadChain.__new__(DownloadChain) + error_msg = "添加种子任务失败：无法读取种子文件" + chain.download = MagicMock(return_value=("qb", None, "Original", error_msg)) + chain.post_message = MagicMock() + + context = Context( + meta_info=MetaInfo("Demo Movie 2026"), + media_info=MediaInfo( + type=MediaType.MOVIE, + title="Demo Movie", + year="2026", + tmdb_id=1, + genre_ids=[18], + ), + torrent_info=TorrentInfo( + site=12, + site_name="AGSVPT", + title="Demo Movie 2026 1080p", + enclosure="https://example.com/download.php?id=484660", + size=1024, + ), + ) + + download_id, returned_error = chain.download_single( + context=context, + torrent_content=b"torrent-content", + save_path="/downloads", + source="Subscribe|{}", + return_detail=True, + ) + + assert download_id is None + assert returned_error == error_msg + assert captured["fingerprint"] == DownloadChain._build_download_failure_fingerprint(context) + assert captured["torrent_id"] == "example.com:id=484660" + assert captured["site"] == 12 + assert captured["error_message"] == error_msg + assert captured["next_retry_at"] > captured["now_time"] + + +def test_batch_download_skips_failed_subscription_resource_and_tries_next(monkeypatch): + """ + 订阅自动下载应跳过冷却中的失败资源，但继续尝试同媒体的后续候选。 + """ + _FakeBatchTorrentHelper.episodes = [] + monkeypatch.setattr(download_module, "TorrentHelper", _FakeBatchTorrentHelper) + monkeypatch.setattr(download_module.eventmanager, "send_event", lambda *args, **kwargs: None) + + first_context = SimpleNamespace( + media_info=SimpleNamespace( + type=MediaType.MOVIE, + title="Demo Movie", + year="2026", + title_year="Demo Movie (2026)", + tmdb_id=1, + douban_id=None, + ), + meta_info=SimpleNamespace(season=None, episode=None, episode_list=[], season_episode=""), + torrent_info=SimpleNamespace( + site=12, + site_name="AGSVPT", + title="Demo Movie Bad", + torrent_id="484660", + size=1024, + ), + ) + second_context = SimpleNamespace( + media_info=SimpleNamespace( + type=MediaType.MOVIE, + title="Demo Movie", + year="2026", + title_year="Demo Movie (2026)", + tmdb_id=1, + douban_id=None, + ), + meta_info=SimpleNamespace(season=None, episode=None, episode_list=[], season_episode=""), + torrent_info=SimpleNamespace( + site=13, + site_name="OtherSite", + title="Demo Movie Good", + torrent_id="999999", + size=2048, + ), + ) + failed_fingerprint = DownloadChain._build_download_failure_fingerprint(first_context) + + class _ActiveDownloadFailureOper: + """ + 返回第一个候选的活跃失败冷却记录。 + """ + + def get_active_by_fingerprints(self, fingerprints: list[str], now_time: str) -> dict: + """ + 模拟数据库批量查询活跃失败记录。 + """ + assert now_time + assert failed_fingerprint in fingerprints + return {failed_fingerprint: SimpleNamespace(fingerprint=failed_fingerprint)} + + monkeypatch.setattr(download_module, "DownloadFailureOper", _ActiveDownloadFailureOper) + + chain = DownloadChain.__new__(DownloadChain) + chain.download_single = MagicMock(return_value="hash") + + downloads, lefts = chain.batch_download( + contexts=[first_context, second_context], + source="Subscribe|{}", + ) + + assert downloads == [second_context] + assert lefts is None + chain.download_single.assert_called_once() + assert chain.download_single.call_args.args[0] is second_context + + def test_batch_download_accepts_complete_coverage_when_files_cover_target_range(monkeypatch): """ 自定义起始集场景按目标范围覆盖判断，100-143 可满足 start=100、total=143。