From 6f9c0756c1a749d79b786bf559127fbc01b53a0e Mon Sep 17 00:00:00 2001 From: jxxghp Date: Mon, 12 Oct 2026 13:34:41 +0800 Subject: [PATCH] fix(download): reuse torrent content before resolving tickets (#6967) --- app/chain/download/batch.py | 18 ++- app/chain/download/submission.py | 74 +++++++++---- docs/testing.md | 7 ++ tests/test_download_resource_cache.py | 154 ++++++++++++++++++++++++++ tests/test_torrent_link_refresh.py | 7 +- 5 files changed, 235 insertions(+), 25 deletions(-) create mode 100644 tests/test_download_resource_cache.py diff --git a/app/chain/download/batch.py b/app/chain/download/batch.py index d2e055be1..b49f6acf1 100644 --- a/app/chain/download/batch.py +++ b/app/chain/download/batch.py @@ -131,6 +131,7 @@ class _BatchDownloadRun: no_exists: Dict[str, Dict[int, NotExistMediaInfo]], options: _BatchDownloadOptions, allowed_episodes: Optional[Set[int]], ) -> None: + """隔离本批候选、集数限制及失败记账，避免共享下载链串扰。""" self.owner = owner self.contexts = contexts self.no_exists = no_exists @@ -165,6 +166,7 @@ class _BatchDownloadRun: 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]) @@ -180,17 +182,18 @@ class _BatchDownloadRun: 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) + errors: List[str] = [] + content, _, files = self.owner.download_torrent(torrent, on_failure=errors.append) 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="下载种子内容为空", + context=context, error_msg=errors[-1] if errors else "下载种子内容为空", downloader=self.options["downloader"], source=self.options["source"], ) self._remember_failure(context) @@ -209,6 +212,7 @@ class _BatchDownloadRun: 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(): @@ -233,6 +237,7 @@ class _BatchDownloadRun: 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 [] @@ -291,6 +296,7 @@ class _BatchDownloadRun: 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]) @@ -314,6 +320,7 @@ class _BatchDownloadRun: return True 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) @@ -326,6 +333,7 @@ class _BatchDownloadRun: 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: @@ -352,6 +360,7 @@ class _BatchDownloadRun: 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() @@ -376,6 +385,7 @@ class _BatchDownloadRun: return episodes if matches else set() 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) @@ -390,6 +400,7 @@ class _BatchDownloadRun: 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 @@ -414,6 +425,7 @@ class _BatchDownloadRun: 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) \ diff --git a/app/chain/download/submission.py b/app/chain/download/submission.py index 02817b5f2..aacb55f35 100644 --- a/app/chain/download/submission.py +++ b/app/chain/download/submission.py @@ -1,6 +1,7 @@ """种子获取与单任务提交 owner。""" import base64 +import hashlib import json import re from copy import copy @@ -225,6 +226,7 @@ class _DownloadResourceOwner(_DownloadOwnerBase): ) -> Tuple[Optional[Union[str, bytes]], str, list[str]]: """ 下载种子文件，如果是磁力链，会返回磁力链接本身 + 换票种子按站点资源和原始换票凭据复用本地内容，命中后不再请求临时链接。 :param on_failure: 将本次具体失败原因交还准备流程，保持返回元组与并发调用隔离 :return: 种子内容，种子目录名，种子文件清单 """ @@ -234,27 +236,10 @@ class _DownloadResourceOwner(_DownloadOwnerBase): if torrent.enclosure.startswith("magnet:"): return torrent.enclosure, "", [] torrent = self._with_site_credentials(torrent) - # Cookie - site_cookie: Optional[str] = torrent.site_cookie - indirect_download = torrent.enclosure.startswith("[") - if indirect_download: - # 需要解码获取下载地址 - torrent_url = self._resolve_indirect_download_url( - url=torrent.enclosure, - ua=torrent.site_ua, - cookie=site_cookie, - ) - # 涉及解析地址的不使用Cookie下载种子，否则MT会出错 - site_cookie = None - else: - torrent_url = torrent.enclosure - if not torrent_url: - logger.error(f"{torrent.title} 无法获取下载地址！") + result = self._get_torrent_content(torrent) + if result is None: return None, "", [] - # 下载种子文件 - _, content, download_folder, files, error_msg = self._download_torrent_content( - torrent, torrent_url, site_cookie, indirect_download - ) + content, download_folder, files, error_msg = result if isinstance(content, str): # 磁力链 @@ -276,6 +261,55 @@ class _DownloadResourceOwner(_DownloadOwnerBase): # 返回 种子文件路径，种子目录名，种子文件清单 return content, download_folder or "", files or [] + @staticmethod + def _indirect_torrent_cache_key(torrent: TorrentInfo) -> Optional[str]: + """以资源身份和换票凭据隔离缓存，临时链接不参与键，凭据不写入文件名。""" + identity = torrent_rules.resource_identity(torrent) + if not torrent.enclosure.startswith("[") or identity is None: + return None + scope = json.dumps((identity, torrent.enclosure, torrent.site_cookie)) + return f"resource-{hashlib.sha256(scope.encode('utf-8')).hexdigest()}.torrent" + + def _get_torrent_content(self, torrent: TorrentInfo) -> Optional[Tuple[ + Optional[Union[str, bytes]], Optional[str], Optional[list[str]], Optional[str], + ]]: + """先读取已验证的换票种子内容；缓存损坏时重新换票，失败与磁力链不缓存。""" + cache_key = self._indirect_torrent_cache_key(torrent) + cache = FileCache(local_only=True) if cache_key else None + if cache is not None and cache_key is not None: + cached = cache.get(cache_key, region="torrents") + if isinstance(cached, bytes) and cached: + helper: TorrentHelper = cast(Any, TorrentHelper)() + cached_folder, cached_files = helper.get_fileinfo_from_torrent_content(cached) + if cached_files: + return cached, cached_folder, cached_files, "" + cache.delete(cache_key, region="torrents") + # Cookie + site_cookie: Optional[str] = torrent.site_cookie + indirect_download = torrent.enclosure.startswith("[") + if indirect_download: + # 需要解码获取下载地址 + torrent_url = self._resolve_indirect_download_url( + url=torrent.enclosure, + ua=torrent.site_ua, + cookie=site_cookie, + ) + # 涉及解析地址的不使用Cookie下载种子，否则MT会出错 + site_cookie = None + else: + torrent_url = torrent.enclosure + if not torrent_url: + logger.error(f"{torrent.title} 无法获取下载地址！") + return None + # 下载种子文件 + _, content, download_folder, files, error_msg = self._download_torrent_content( + torrent, torrent_url, site_cookie, indirect_download + ) + + if cache is not None and cache_key is not None and isinstance(content, bytes) and files: + cache.set(cache_key, content, region="torrents") + return content, download_folder, files, error_msg + def _download_torrent_content( self, torrent: TorrentInfo, diff --git a/docs/testing.md b/docs/testing.md index 4910d6258..e7c866295 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -112,6 +112,13 @@ with patch.object(SomeModule, "fetch", new=AsyncMock(return_value=FAKE)): 交给持久化冷却，避免统一的空内容错误覆盖它；签名链接遇到明确限流也不刷新重试。 此策略不追溯修改已经写入的失败记录。 +换票资源的跨批次复用见 `tests/test_download_resource_cache.py`：种子成功解析后， +本地 `torrents` 缓存以站点资源身份、原始换票请求和 Cookie 的摘要保存内容，命中后 +不再换票。缓存沿用种子目录的清理周期；损坏或清理后重新获取，缺少稳定身份时不共享。 +测试覆盖十二轮无需要集或无法识别集数、缺集变化后的即时选中、站点与凭据隔离， +以及完整包和拆包保留每日限流原因并采用 24 小时冷却。缓存不保存缺集判断， +当前识别词和允许集数仍在每批重新应用。 + ## 自隔离：用了什么，就还原什么 用例若修改了**进程级状态**——`sys.modules` 桩、单例（`Singleton._instances`）、`lru_cache`、环境变量、`settings` 字段——必须在用例或模块结束时还原。pytest 一次性导入全部测试模块，未还原的污染会扩散到后续用例，产生“单独跑过、一起跑挂”的测不准现象。 diff --git a/tests/test_download_resource_cache.py b/tests/test_download_resource_cache.py new file mode 100644 index 000000000..1606e9fdd --- /dev/null +++ b/tests/test_download_resource_cache.py @@ -0,0 +1,154 @@ +"""换票资源跨批次复用内容与真实限流冷却的离线回归。""" + +import json +from dataclasses import replace +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +from requests import Response + +import app.application.torrent.download as torrent_download +import app.chain.download.submission as submission +from app.adapters.cache.backends import FileBackend +from app.chain.download import DownloadChain +from app.chain.download.batch import _BatchDownloadRun +from app.chain.download.failure import DownloadFailureOwner +from app.domain.context import Context, MediaInfo, TorrentInfo +from app.domain.metainfo import MetaInfo +from app.schemas.types import MediaType + + +def _content(filename: str) -> bytes: + """构造可由真实解析器读取的单文件种子。""" + name = filename.encode() + return b"d4:infod4:name" + str(len(name)).encode() + b":" + name + b"6:lengthi1eee" + + +def _response(content: bytes) -> Response: + """构造 HTTP 成功响应，内容可以是种子或站点限流 JSON。""" + response = Response() + response.status_code = 200 + response._content = content + response.encoding = "utf-8" + return response + + +def _context(**fields) -> Context: + """构造同站同资源的多集包候选，识别结果保持空集数。""" + values = dict( + site=1, site_name="馒头", title="Series S05", + page_url="https://tracker.example/detail/793424", + enclosure="[encoded]https://tracker.example/token", site_cookie="session=first", + ) + values.update(fields) + return Context( + meta_info=MetaInfo("Series S05"), + media_info=MediaInfo(title="Series", type=MediaType.TV), + torrent_info=TorrentInfo(**values), + ) + + +def _batch(chain) -> _BatchDownloadRun: + """每轮重新创建批次，验证复用不依赖上轮批次状态。""" + options = dict( + save_path=None, channel=None, source="Subscribe|48", userid=None, + username=None, downloader="test", custom_words=None, governance=None, + ) + return _BatchDownloadRun(chain, [], {}, options, None) + + +@pytest.fixture +def boundary(monkeypatch, tmp_path): + """仅替换 HTTP、换票与持久化边界，使用真实种子解析及临时文件缓存。""" + chain = object.__new__(DownloadChain) + chain.post_message = Mock() + chain.download_single = Mock(return_value="download-id") + chain._record_download_failure = Mock() + chain._resolve_indirect_download_url = Mock(side_effect=[ + f"https://tracker.example/download?id=793424&t={number}&sign=ticket{number}" + for number in range(30) + ]) + cache = FileBackend(tmp_path) + factories = [Mock(return_value=cache), Mock(return_value=cache)] + monkeypatch.setattr(submission, "FileCache", factories[0]) + monkeypatch.setattr(torrent_download, "FileCache", factories[1]) + port = Mock() + port.request.return_value = _response(_content("Series.S05E001.mkv")) + monkeypatch.setattr(torrent_download, "_require_torrent_port", Mock(return_value=port)) + return SimpleNamespace(chain=chain, cache=cache, port=port, factories=factories) + + +@pytest.mark.parametrize("filename", ["Series.S05E001.mkv", "Series.mkv"]) +def test_repeated_partial_pack_rounds_fetch_content_once(boundary, filename): + """无需要集和无法识别集数均可跨十二轮复用，避免耗尽每日配额。""" + boundary.port.request.return_value = _response(_content(filename)) + for _ in range(12): + assert _batch(boundary.chain)._submit_partial_pack(_context(), {196, 197}) == set() + boundary.port.request.assert_called_once() + boundary.chain._resolve_indirect_download_url.assert_called_once() + boundary.chain.download_single.assert_not_called() + boundary.chain._record_download_failure.assert_not_called() + boundary.chain.post_message.assert_not_called() + assert all(factory.call_args.kwargs == {"local_only": True} for factory in boundary.factories) + + +def test_new_missing_episode_uses_cached_content(boundary): + """缓存正向文件事实，缺集变化后立即选中需要的集，不记资源失败。""" + assert _batch(boundary.chain)._submit_partial_pack(_context(), {196}) == set() + context = _context() + assert _batch(boundary.chain)._submit_partial_pack(context, {1, 196}) == {1} + boundary.port.request.assert_called_once() + assert boundary.chain.download_single.call_args.kwargs["episodes"] == {1} + assert boundary.chain.download_single.call_args.kwargs["torrent_content"] == _content("Series.S05E001.mkv") + boundary.chain._record_download_failure.assert_not_called() + + +@pytest.mark.parametrize("fields", [ + {"site": 2}, {"page_url": "https://tracker.example/detail/917009"}, + {"site_cookie": "session=second"}, {"enclosure": "[other-key]https://tracker.example/token"}, +]) +def test_resource_and_credentials_isolate_cached_content(boundary, fields): + """同种子 ID 的其他站点、其他种子与新换票凭据都必须重新取种子。""" + first = _context().torrent_info + boundary.chain.download_torrent(first) + boundary.chain.download_torrent(replace(first, **fields)) + assert boundary.port.request.call_count == 2 + assert boundary.chain._resolve_indirect_download_url.call_count == 2 + + +def test_corrupt_or_cleared_resource_cache_refetches_content(boundary): + """损坏的资源缓存回退到换票，清理种子缓存后也不再复用。""" + torrent = _context().torrent_info + key = boundary.chain._indirect_torrent_cache_key(torrent) + boundary.cache.set(key, b"invalid", region="torrents") + assert boundary.chain.download_torrent(torrent)[0] == _content("Series.S05E001.mkv") + boundary.cache.clear(region="torrents") + assert boundary.chain.download_torrent(torrent)[0] == _content("Series.S05E001.mkv") + assert boundary.port.request.call_count == 2 + + +@pytest.mark.parametrize("partial", [False, True]) +def test_batch_keeps_daily_limit_reason_and_cooldown(boundary, partial): + """完整包和拆包均记录真实每日限流，交给既有策略冷却 24 小时。""" + reason = "相同種子當天最多下載10次" + boundary.port.request.return_value = _response(json.dumps({"message": reason}).encode()) + context = _context() + assert _batch(boundary.chain)._read_torrent(context, partial=partial) is None + recorded = boundary.chain._record_download_failure.call_args.kwargs + assert recorded["error_msg"] == f"站点限流：{reason}" + assert recorded["context"] is context + assert recorded["source"] == "Subscribe|48" + assert DownloadFailureOwner._download_failure_ttl(recorded["error_msg"]) == 86400 + boundary.port.request.assert_called_once() + key = boundary.chain._indirect_torrent_cache_key(context.torrent_info) + assert boundary.cache.get(key, region="torrents") is None + + +def test_resource_without_stable_identity_does_not_share_cache(boundary): + """无法确认种子身份时只保留既有 URL 缓存，不按标题共享资源内容。""" + torrent = _context(page_url=None).torrent_info + for _ in range(2): + boundary.chain.download_torrent(torrent) + assert boundary.port.request.call_count == 2 + assert boundary.chain._indirect_torrent_cache_key(torrent) is None diff --git a/tests/test_torrent_link_refresh.py b/tests/test_torrent_link_refresh.py index 4c929369e..338ebaf83 100644 --- a/tests/test_torrent_link_refresh.py +++ b/tests/test_torrent_link_refresh.py @@ -128,6 +128,9 @@ def download_chain(monkeypatch): helper = Mock() helper.download_torrent.return_value = FAILED monkeypatch.setattr(submission, "TorrentHelper", Mock(return_value=helper)) + cache = Mock() + cache.get.return_value = None + monkeypatch.setattr(submission, "FileCache", Mock(return_value=cache)) sites = Mock() sites.get_indexers.return_value = [SITE] monkeypatch.setattr(submission, "SitesHelper", Mock(return_value=sites), raising=False) @@ -386,8 +389,8 @@ def test_no_cache_rule_only_matches_explicit_nonempty_keywords(monkeypatch, keys assert chain._is_no_cache_site(domain) is expected -def test_cached_mteam_request_obtains_new_token_for_each_download(download_chain, monkeypatch): - """同一缓存换票请求连续下载时，每次均从接口获取新的临时链接。""" +def test_uncached_mteam_content_obtains_new_token_for_each_download(download_chain, monkeypatch): + """种子内容缓存未命中时，同一换票请求每次获取新链接，不复用旧签名。""" chain, helper = download_chain request = {"method": "post", "params": {"id": "39"}, "result": "data"} encoded = base64.b64encode(json.dumps(request).encode()).decode()