fix(download): reuse torrent content before resolving tickets (#6967)

This commit is contained in:
jxxghp
2026-10-12 13:34:41 +08:00
parent 99ef7e508b
commit 6f9c0756c1
5 changed files with 235 additions and 25 deletions

View File

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

View File

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

View File

@@ -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 一次性导入全部测试模块,未还原的污染会扩散到后续用例,产生“单独跑过、一起跑挂”的测不准现象。

View File

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

View File

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