fix: cool down failed subscription resources

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} 添加下载成功")

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,
+        )

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

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)

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")

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。