diff --git a/app/chain/download/contract.py b/app/chain/download/contract.py index da0558efb..f1d2bd46f 100644 --- a/app/chain/download/contract.py +++ b/app/chain/download/contract.py @@ -30,6 +30,7 @@ if TYPE_CHECKING: _matches_media_identity: Callable[..., bool] _prepare_batch_download_contexts: Callable[..., Any] _record_download_failure: Callable[..., Any] + _record_duplicate_download_backoff: Callable[..., Any] _record_music_album_track_keys: Callable[..., Any] _resolve_media_download_dir: Callable[..., Any] _settle_download_success: Callable[..., None] diff --git a/app/chain/download/failure.py b/app/chain/download/failure.py index 3e5a9b851..f9d8df233 100644 --- a/app/chain/download/failure.py +++ b/app/chain/download/failure.py @@ -1,5 +1,5 @@ """ -下载失败指纹、冷却和持久化 owner。 +下载资源级失败与退避指纹、冷却和持久化 owner。 """ import hashlib @@ -27,6 +27,10 @@ from app.schemas.media import build_media_key, resolve_media_identity DOWNLOAD_FAILURE_RESOURCE_TTL_SECONDS = 24 * 60 * 60 DOWNLOAD_FAILURE_TRANSIENT_TTL_SECONDS = 60 * 60 +# 重复提交退避只压制单个资源指纹，必须长于订阅重新选源的周期，否则同一个整包仍会被反复接受。 +DOWNLOAD_FAILURE_DUPLICATE_TTL_SECONDS = 7 * 24 * 60 * 60 +# 退避记录写入台账的成因说明，候选跳过日志直接引用它，不能再假设这是一次失败。 +DOWNLOAD_FAILURE_DUPLICATE_MESSAGE = "下载器已存在同一任务且本次无新增文件明细" DOWNLOAD_FAILURE_RESOURCE_ERROR_KEYWORDS = ( "无法读取种子文件", "无法获取下载地址", @@ -44,7 +48,7 @@ DOWNLOAD_FAILURE_RESOURCE_ERROR_KEYWORDS = ( class DownloadFailureOwner(_DownloadOwnerBase): """ - 下载失败指纹、冷却和持久化 owner。 + 下载资源级失败与退避指纹、冷却和持久化 owner。 """ @staticmethod @@ -101,9 +105,15 @@ class DownloadFailureOwner(_DownloadOwnerBase): return f"title={title}|size={size}" @classmethod - def _build_download_failure_fingerprint(cls, context: Context) -> Optional[str]: + def _build_download_failure_fingerprint( + cls, context: Context, *, resource_backoff: bool = False, + ) -> Optional[str]: """ - 根据媒体和种子资源信息生成失败冷却指纹。 + 根据媒体和种子资源信息生成冷却指纹，整包退避不依赖解析后补齐的集数。 + + :param context: 识别出的媒体与资源上下文 + :param resource_backoff: 是否为整包均已记录的资源退避；与普通失败指纹隔离 + :return: 稳定资源指纹；缺少身份时返回 None """ media = getattr(context, "media_info", None) torrent = getattr(context, "torrent_info", None) @@ -124,7 +134,7 @@ class DownloadFailureOwner(_DownloadOwnerBase): "media_type": str(media_type or ""), "media_key": str(media_key or ""), "season": str(season) if season is not None else "", - "episodes": cls._format_failure_episodes(meta) or "", + "episodes": None if resource_backoff else cls._format_failure_episodes(meta) or "", "site": str(site or ""), "resource": cls._torrent_resource_key(torrent), } @@ -153,11 +163,25 @@ class DownloadFailureOwner(_DownloadOwnerBase): downloader: Optional[str] = None, source: Optional[str] = None, episodes: Optional[Set[int]] = None, + ttl_seconds: Optional[int] = None, + resource_backoff: bool = False, ) -> Optional[str]: """ - 记录资源级下载失败，并返回本次失败指纹。 + 记录资源级冷却事实，并返回本次资源指纹。 + + 该台账是资源级退避记录，既承载真实失败，也承载"重复提交但无新增交付"这类 + 没有失败、但不值得立刻再试一次的事实，因此 ``error_msg`` 只是跳过原因说明。 + + :param context: 本次提交的上下文，用于计算资源指纹 + :param error_msg: 写入台账的跳过原因说明 + :param downloader: 本次使用的下载器 + :param source: 下载来源，仅订阅来源的冷却会被读取 + :param episodes: 目标集数，缺省时按识别元数据记录 + :param ttl_seconds: 显式冷却秒数，缺省时按原因文本推断 + :param resource_backoff: 是否使用不含运行时集数的整包退避指纹 + :return: 本次记录的资源指纹；缺少身份时返回 None """ - fingerprint = self._build_download_failure_fingerprint(context) + fingerprint = self._build_download_failure_fingerprint(context, resource_backoff=resource_backoff) if not fingerprint: return None @@ -165,7 +189,9 @@ class DownloadFailureOwner(_DownloadOwnerBase): 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)), + time.localtime(now_timestamp + ( + ttl_seconds if ttl_seconds is not None else self._download_failure_ttl(error_msg) + )), ) media = context.media_info media_source, media_id = resolve_media_identity(media=media) @@ -212,32 +238,74 @@ class DownloadFailureOwner(_DownloadOwnerBase): logger.error(f"记录下载失败冷却失败：{str(err)}") return fingerprint + def _record_duplicate_download_backoff( + self, + context: Context, + downloader: Optional[str] = None, + source: Optional[str] = None, + episodes: Optional[Set[int]] = None, + ) -> Optional[str]: + """ + 为重复接受且未带来任何新增文件的资源记录退避，并返回该资源指纹。 + + 调用方须确认整包全部媒体文件均已记录。退避只压制这一个资源，订阅候选会继续 + 尝试同媒体的其他资源，订阅本身既不会被判定完成也不会被判失败，冷却到期后仍可重试。 + 仅订阅来源需要记录，因为候选过滤只在订阅来源读取该台账。 + + :param context: 本次结算的上下文，用于计算资源指纹 + :param downloader: 本次使用的下载器 + :param source: 下载来源 + :param episodes: 目标集数 + :return: 本次记录的整包资源指纹；非订阅来源返回 None + """ + if not self._is_subscribe_source(source): + return None + return self._record_download_failure( + context=context, + error_msg=DOWNLOAD_FAILURE_DUPLICATE_MESSAGE, + downloader=downloader, + source=source, + episodes=episodes, + ttl_seconds=DOWNLOAD_FAILURE_DUPLICATE_TTL_SECONDS, + resource_backoff=True, + ) + def _active_download_failure_fingerprints( self, contexts: List[Context], source: Optional[str], ) -> Dict[str, DownloadFailureSnapshot]: """ - 查询当前订阅候选中仍处于冷却期的失败记录，返回指纹到失败记录的映射。 + 查询候选的普通失败与整包退避，并映射回当前候选指纹供所有选源阶段消费。 + + :param contexts: 本轮待筛选的资源上下文 + :param source: 本次批量下载的业务来源 + :return: 当前候选指纹到有效冷却记录的映射；非订阅来源不读取台账 """ if not self._is_subscribe_source(source): return {} - fingerprints = [ - fingerprint - for fingerprint in [ - self._build_download_failure_fingerprint(context) - for context in contexts or [] - ] - if fingerprint - ] + backoff_fingerprints = { + fingerprint: self._build_download_failure_fingerprint(context, resource_backoff=True) or fingerprint + for context in contexts or [] + if (fingerprint := self._build_download_failure_fingerprint(context)) + } + fingerprints = list(dict.fromkeys([ + *backoff_fingerprints, + *backoff_fingerprints.values(), + ])) if not fingerprints: return {} now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) try: repository: DownloadFailureRepository = self.download_failure_repository - return repository.get_active_by_fingerprints( + records = repository.get_active_by_fingerprints( fingerprints=fingerprints, now_time=now_time, ) + return { + fingerprint: records.get(fingerprint) or records[backoff] + for fingerprint, backoff in backoff_fingerprints.items() + if fingerprint in records or backoff in records + } except Exception as err: logger.error(f"查询下载失败冷却失败：{str(err)}") return {} @@ -248,17 +316,20 @@ class DownloadFailureOwner(_DownloadOwnerBase): failure: Optional[DownloadFailureSnapshot], ) -> None: """ - 记录候选资源处于失败冷却期时的跳过原因和下次重试时间。 + 记录候选资源处于冷却期时的跳过原因和下次重试时间。 + + 台账里的成因可能是真实失败，也可能是重复提交退避，所以这里只转述记录中的原因， + 不把它断言成"添加下载失败"。 """ reason = getattr(failure, "error_message", None) or "未知原因" retry_at = getattr(failure, "next_retry_at", None) if retry_at: logger.info( - f"{context.torrent_info.title} 近期添加下载失败（失败原因：{reason}），" - f"暂时跳过该资源，将于 {retry_at} 后重试" + f"{context.torrent_info.title} 暂时跳过该资源（原因：{reason}），" + f"将于 {retry_at} 后重试，本次继续尝试其他资源" ) else: logger.info( - f"{context.torrent_info.title} 近期添加下载失败（失败原因：{reason}），" - "暂时跳过该资源" + f"{context.torrent_info.title} 暂时跳过该资源（原因：{reason}），" + "本次继续尝试其他资源" ) diff --git a/app/chain/download/history.py b/app/chain/download/history.py index 2c283e00a..3e29ced16 100644 --- a/app/chain/download/history.py +++ b/app/chain/download/history.py @@ -82,7 +82,8 @@ class DownloadHistoryOwner(_DownloadOwnerBase): """ 提交下载历史、文件明细和 durable 下载事件。 - :param already_exists: 下载器中已存在同一任务时为 True，仅补记历史，不再重复通知与文件明细 + :param already_exists: 下载器中已存在同一任务时为 True，仅补记历史，不再重复通知与文件明细； + 若整包全部媒体文件已记录过，同时为这个资源登记提交退避 """ if layout == "NoSubfolder" or not folder_name: download_path = download_dir / file_list[0] if file_list else download_dir @@ -127,11 +128,6 @@ class DownloadHistoryOwner(_DownloadOwnerBase): files_to_add: list[DownloadFileWrite] = [] custom_word_list = custom_words.splitlines() if custom_words else [] for file in file_list: - if episodes: - # 与订阅选集使用相同识别词，避免偏移前的文件集数被目标集数过滤。 - file_meta = MetaInfo(Path(file).stem, custom_words=custom_word_list) - if not file_meta.begin_episode or file_meta.begin_episode not in episodes: - continue if not Path(file).suffix or Path(file).suffix.lower() not in self.runtime_config.media_extensions: continue files_to_add.append( @@ -145,7 +141,24 @@ class DownloadHistoryOwner(_DownloadOwnerBase): ) ) if already_exists: - files_to_add = self._filter_unrecorded_download_files(files_to_add, download_hash) + retained_files = self._filter_unrecorded_download_files(files_to_add, download_hash) + if files_to_add and not retained_files: + # 必须核对整包文件，不能因本次只选了已记录的一集就阻止以后补下包内其他集。 + self._record_duplicate_download_backoff( + context=context, + downloader=downloader, + source=source, + episodes=episodes, + ) + files_to_add = retained_files + if episodes: + selected_files = [] + for item in files_to_add: + # 与订阅选集使用相同识别词，避免偏移前的文件集数被目标集数过滤。 + file_meta = MetaInfo(Path(cast(str, item.filepath)).stem, custom_words=custom_word_list) + if file_meta.begin_episode and file_meta.begin_episode in episodes: + selected_files.append(item) + files_to_add = selected_files frozen_files = tuple(files_to_add) event_payload = { "hash": download_hash, "context": context, "username": username, diff --git a/docs/testing.md b/docs/testing.md index fe3f8a538..de482b9ea 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -163,6 +163,13 @@ with patch.object(SomeModule, "fetch", new=AsyncMock(return_value=FAKE)): 短期签名或换票资源在下载链中屏蔽原始 enclosure，不将临时请求地址写入无效缓存。 订阅持久化保留具体失败原因，无效种子使用 24 小时资源冷却，避免重启后每小时重复通知。 网络异常、空响应、403/5xx 和限流仍沿用瞬时重试与流控规则，不因本次修改加入内容无效黑名单。 +下载器报告同一任务已存在且整包全部媒体文件明细已记录时，按资源指纹登记 7 天提交退避， +使订阅不再重投同一个已持有的整包；退避只压制该资源，候选仍继续尝试同媒体的其他种子， +订阅不会被判定完成或失败，且仅订阅来源写台账（手动补投不留冷却记录）。冷却跳过日志转述台账 +中的原因，不再把它断言成"添加下载失败"。见 `tests/test_torrent_download_retry.py` 与 +`tests/test_downloader_submission.py`。 +整包退避指纹不包含解析种子后补齐的集数，下一轮搜索仍能命中；普通失败仍按原有集数指纹隔离。 +只重复选择了已记录的部分集数、但包内仍有未记录文件时，不登记整包退避，也不把未选文件写入历史。 换票资源的跨批次复用见 `tests/test_download_resource_cache.py`：种子成功解析后， 本地 `torrents` 缓存以站点资源身份、原始换票请求和 Cookie 的摘要保存内容，命中后 diff --git a/tests/test_downloader_submission.py b/tests/test_downloader_submission.py index f008dc9f4..ac90192bb 100644 --- a/tests/test_downloader_submission.py +++ b/tests/test_downloader_submission.py @@ -381,15 +381,17 @@ def test_downloader_task_identity_decides_reuse_flag(downloader, existing): @pytest.mark.parametrize("durable", [False, True]) +@pytest.mark.parametrize("source", ["Subscribe|1", "MoviePilot|手动下载"]) @pytest.mark.parametrize("existing,expected_files", [ ("matched", [_FILES[1]]), ("all", []), ("empty", _FILES), ("unrelated", _FILES), ], ids=["matched", "all", "empty", "unrelated"]) -def test_reuse_settlement_keeps_history_and_skips_duplicate_effects(durable, existing, expected_files): +def test_reuse_settlement_keeps_history_and_skips_duplicate_effects(durable, source, existing, expected_files): """ - 复用已有任务时仍记录历史并触发整理与事件，但不生成通知意图，也不重复写入已记录的文件明细。 + 复用已有任务时仍记录历史并触发整理与事件，但不生成通知意图、不重复写入已记录的文件明细； + 文件明细全部已记录时才为该资源登记退避，仍有新增文件则不退避。 """ recorded_file, new_file = _FILES records = { @@ -402,6 +404,7 @@ def test_reuse_settlement_keeps_history_and_skips_duplicate_effects(durable, exi media = MediaInfo(type=MediaType.TV, title="一念永恒", tmdb_id=107371) torrent = TorrentInfo(title="A.Will.Eternal.S04.2160p") chain = DownloadChain.__new__(DownloadChain) + chain.download_failure_repository = MagicMock() chain.download_history_repository = MagicMock() chain.download_history_repository.get_files_by_hash.return_value = [ SimpleNamespace(fullpath=path) for path in records @@ -418,7 +421,7 @@ def test_reuse_settlement_keeps_history_and_skips_duplicate_effects(durable, exi folder_name="shared-show", file_list=[recorded_file, new_file], download_dir=Path("/downloads"), layout="Original", downloader="qb", download_hash="reuse-hash", download_episodes=None, episodes=None, - channel=None, source="subscribe", userid=None, username=None, + channel=None, source=source, userid=None, username=None, torrent_content=b"torrent", custom_words=None, already_exists=True, ) @@ -435,6 +438,42 @@ def test_reuse_settlement_keeps_history_and_skips_duplicate_effects(durable, exi assert [item.filepath for item in files] == list(expected_files) chain._build_download_notification.assert_not_called() chain.download_history_repository.get_files_by_hash.assert_called_once_with("reuse-hash", state=1) + record_backoff = chain.download_failure_repository.record_failure + assert record_backoff.called is (existing == "all" and source == "Subscribe|1") + if record_backoff.called: + write = record_backoff.call_args.args[0] + assert write.source == source + assert write.downloader == "qb" + + +def test_reused_episode_subset_keeps_other_torrent_files_available(): + """ + 仅重投已记录的一集时，整包仍含未记录的另一集，不能对整个资源登记退避。 + """ + recorded_file, _new_file = _FILES + meta = MetaInfo(recorded_file) + media = MediaInfo(type=MediaType.TV, title="一念永恒", tmdb_id=107371) + torrent = TorrentInfo(title="A.Will.Eternal.S04.2160p") + chain = DownloadChain.__new__(DownloadChain) + chain.download_history_repository = MagicMock() + chain.download_history_repository.get_files_by_hash.return_value = [ + SimpleNamespace(fullpath=f"/downloads/shared-show/{recorded_file}"), + ] + chain.download_failure_repository = MagicMock() + chain.durable_event_writer = MagicMock() + + chain._settle_download_success( + context=Context(meta_info=meta, media_info=media, torrent_info=torrent), + media=media, meta=meta, torrent=torrent, + folder_name="shared-show", file_list=_FILES, + download_dir=Path("/downloads"), layout="Original", downloader="qb", + download_hash="reuse-hash", download_episodes=None, episodes={5}, + channel=None, source="Subscribe|1", userid=None, username=None, + torrent_content=b"torrent", custom_words=None, already_exists=True, + ) + + chain.download_failure_repository.record_failure.assert_not_called() + assert chain.durable_event_writer.download_added.call_args.kwargs["files"] == () def test_reuse_settlement_without_media_files_skips_file_lookup(): diff --git a/tests/test_torrent_download_retry.py b/tests/test_torrent_download_retry.py index f58e2c98f..5e3325ccb 100644 --- a/tests/test_torrent_download_retry.py +++ b/tests/test_torrent_download_retry.py @@ -5,15 +5,24 @@ import json from datetime import datetime from pathlib import Path +from types import SimpleNamespace from unittest.mock import Mock import pytest from requests import Response import app.application.torrent.download as download +import app.chain.download.failure as failure_owner import app.chain.download.submission as submission from app.chain.download import DownloadChain -from app.chain.download.failure import DownloadFailureOwner +from app.chain.download.failure import ( + DOWNLOAD_FAILURE_DUPLICATE_MESSAGE, + DOWNLOAD_FAILURE_DUPLICATE_TTL_SECONDS, + DownloadFailureOwner, +) +from app.db.adapters.download import TransactionalDownloadFailureRepository +from app.db.models.downloadfailure import DownloadFailure +from app.db.session import SessionFactory from app.domain.context import Context, MediaInfo, TorrentInfo from app.domain.metainfo import MetaInfo from app.schemas.types import MediaType @@ -278,6 +287,116 @@ def test_invalid_content_preparation_preserves_persisted_resource_cooldown(bound - datetime.fromisoformat(failure.failed_at)).total_seconds() == 86400 +def _duplicate_backoff_context(): + """ + 构造可生成稳定资源指纹的重复提交上下文。 + """ + return Context( + torrent_info=TorrentInfo(site=1, site_name="test", title="Series S01", enclosure=URL), + media_info=MediaInfo(title="Series", type=MediaType.TV), + meta_info=MetaInfo("Series S01"), + ) + + +@pytest.mark.parametrize("source,recorded", [("Subscribe|1", True), ("MoviePilot|手动下载", False)]) +def test_duplicate_submission_backoff_is_subscribe_only_and_long(source, recorded): + """ + 重复提交退避只为订阅来源登记台账，并使用覆盖重新选源周期的长冷却与中性成因。 + """ + chain = object.__new__(DownloadChain) + chain.download_failure_repository = Mock() + context = _duplicate_backoff_context() + + fingerprint = chain._record_duplicate_download_backoff( + context=context, downloader="qb", source=source, + ) + + if not recorded: + assert fingerprint is None + chain.download_failure_repository.record_failure.assert_not_called() + return + write = chain.download_failure_repository.record_failure.call_args.args[0] + assert write.error_message == DOWNLOAD_FAILURE_DUPLICATE_MESSAGE + assert write.downloader == "qb" + assert (datetime.fromisoformat(write.next_retry_at) + - datetime.fromisoformat(write.failed_at)).total_seconds() == DOWNLOAD_FAILURE_DUPLICATE_TTL_SECONDS + assert fingerprint == chain._build_download_failure_fingerprint(context, resource_backoff=True) + + +def test_duplicate_backoff_cools_only_that_resource_for_subscription_candidates(db, monkeypatch): + """ + 退避按资源指纹生效：同一媒体的其他种子仍可被候选选中。 + """ + db.watermark(DownloadFailure) + chain = object.__new__(DownloadChain) + chain.download_failure_repository = TransactionalDownloadFailureRepository(SessionFactory) + cooled = _duplicate_backoff_context() + cooled.media_info = MediaInfo(title="Series", type=MediaType.MOVIE) + other = Context( + torrent_info=TorrentInfo(site=1, site_name="test", title="Series S01 1080p", + enclosure="https://tracker.example/download.php?id=40"), + media_info=cooled.media_info, + meta_info=MetaInfo("Series S01"), + ) + fingerprint = chain._record_duplicate_download_backoff( + context=cooled, downloader="qb", source="Subscribe|1", + ) + chain.eventmanager = Mock() + chain.eventmanager.send_event.return_value = None + chain.download_single = Mock(return_value="new-hash") + monkeypatch.setattr(download.TorrentHelper, "sort_torrents", lambda _self, contexts: contexts) + + downloads, _lefts = chain.batch_download(contexts=[cooled, other], source="Subscribe|1") + + assert fingerprint + assert downloads == [other] + chain.download_single.assert_called_once() + assert chain.download_single.call_args.args[0] is other + + +def test_duplicate_backoff_survives_whole_season_episode_expansion(db): + """ + 整季种子解析后的退避，在下一轮尚未解析文件集数时仍须命中同一资源。 + """ + db.watermark(DownloadFailure) + chain = object.__new__(DownloadChain) + chain.download_failure_repository = TransactionalDownloadFailureRepository(SessionFactory) + submitted = _duplicate_backoff_context() + submitted.meta_info.set_episodes(begin=1, end=12) + chain._record_duplicate_download_backoff(context=submitted, source="Subscribe|1") + next_round = _duplicate_backoff_context() + + active = chain._active_download_failure_fingerprints(contexts=[next_round], source="Subscribe|1") + + assert chain._build_download_failure_fingerprint(next_round) in active + + +def test_cooldown_log_keeps_backoff_reason_neutral(monkeypatch): + """ + 候选跳过日志转述台账成因，不把重复提交退避说成添加下载失败。 + """ + logger = Mock() + monkeypatch.setattr(failure_owner, "logger", logger) + + DownloadFailureOwner._log_download_failure_cooldown( + Context(torrent_info=TorrentInfo(title="Series S01")), + SimpleNamespace(error_message=DOWNLOAD_FAILURE_DUPLICATE_MESSAGE, + next_retry_at="2026-10-26 12:00:00"), + ) + + text = logger.info.call_args.args[0] + assert DOWNLOAD_FAILURE_DUPLICATE_MESSAGE in text + assert "将于 2026-10-26 12:00:00 后重试" in text + assert "添加下载失败" not in text + + DownloadFailureOwner._log_download_failure_cooldown( + Context(torrent_info=TorrentInfo(title="Series S01")), + SimpleNamespace(error_message=None, next_retry_at=None), + ) + + assert logger.info.call_args.args[0] == "Series S01 暂时跳过该资源（原因：未知原因），本次继续尝试其他资源" + + def test_redirected_get_retries_at_destination(boundary): """ 重定向目标上的瞬时错误也重试，并保留原始 URL 的成功缓存键。