From 9d534d19549f8db87dbcfc089e87f13c2b2580fc Mon Sep 17 00:00:00 2001 From: jxxghp Date: Thu, 22 Oct 2026 07:36:18 +0800 Subject: [PATCH] fix(search): restore complete packs before initial parts --- app/chain/search/scan.py | 37 ++++++-- app/chain/subscribe/search.py | 3 +- docs/testing.md | 5 + tests/test_search_scan.py | 121 +++++++++++++++++++++++- tests/test_subscribe_paged_search.py | 134 +++++++++++++++++++++++++++ 5 files changed, 287 insertions(+), 13 deletions(-) diff --git a/app/chain/search/scan.py b/app/chain/search/scan.py index abe28514b..1038399cf 100644 --- a/app/chain/search/scan.py +++ b/app/chain/search/scan.py @@ -213,9 +213,12 @@ class SearchScan: snapshot: Optional[SearchSessionSnapshot] = None, target_scope: Optional[tuple[int, int]] = None, page_limit: Optional[int] = None, + prefer_complete: bool = False, ) -> None: """ 创建或恢复一轮搜索，候选与匹配前史由检查点一起持有。 + + :param prefer_complete: 提交分集前先准备覆盖全部目标的候选；由订阅当前整包优先策略决定 """ if not isinstance(plan.mediainfo, MediaInfo): raise ValueError("分页影视搜索不适用于音乐") @@ -230,6 +233,7 @@ class SearchScan: for missing in seasons.values()))) self.target_scope = target_scope self.page_limit = page_limit + self.prefer_complete = prefer_complete self.ended = False self.restart = False self.collection = SearchCollection(targets) @@ -284,6 +288,7 @@ class SearchScan: if self.plan.rule_groups_signature is not None else self.plan.rule_groups, "filter": self.plan.filter_params, "sites": sorted(self.sources.sites), "full": self.full, "target_scope": self.target_scope, + "prefer_complete": self.prefer_complete, "raw_title": False, "multiple_names": self.owner.runtime_config.search_multiple_name, "identifiers": configuration.get(SystemConfigKey.CustomIdentifiers), "release_groups": configuration.get(SystemConfigKey.CustomReleaseGroups), @@ -630,10 +635,16 @@ class SearchScan: def _prepared_batch(self, contexts: list[Context], ready: set[str]) -> tuple[list[Context], set[str]]: """ - 保留排序优先级，先提交可用前缀，再刷新下一个临时票据所在原页。 + 整包优先时集中恢复覆盖全部目标的候选，其余场景沿用可用前缀分批。 """ - batch: list[Context] = [] - goals: set[str] = set() + ordered_keys = [cast(str, item.torrent_info.search_resource_id) + for item in contexts if item.torrent_info is not None] + if self.prefer_complete: + # 只调整整包与分集的先后，两类内部仍保留规则排序；不能在更早的 + # 缺票据分集处切批，导致后面的整包根本没有机会参与下载选择。 + contexts = sorted(contexts, key=lambda item: item.torrent_info is None or self._coverage( + cast(str, item.torrent_info.search_resource_id)) != ready) + batch: dict[str, Context] = {} for context in contexts: if context.torrent_info is None: continue @@ -644,21 +655,29 @@ class SearchScan: context = self.live.get(key, context) if context.torrent_info is None: continue + coverage = self._coverage(key) if not context.torrent_info.enclosure: - if batch: + # 全部整包候选须一起交给下载层校验；否则前一个不完整整包可能 + # 在后一个仍待恢复票据时，就被下载层按分集回退并提交。 + complete_pack = self.prefer_complete and coverage == ready + if batch and not complete_pack: break restored = self._recover_candidate(key) - if restored is None or not self._filter([restored]): + if (restored is None or restored.torrent_info is None + or not restored.torrent_info.enclosure or not self._filter([restored])): # 恢复后的促销、标签等信息可能已变化，须重新通过过滤，不能沿用旧对象的筛选结论。 self.attempted.add(key) continue context = restored - coverage = self._coverage(key) + coverage = self._coverage(key) if not coverage or not coverage <= ready: continue - batch.append(context) - goals.update(coverage) - return batch, goals + batch[key] = context + # 恢复顺序只用于保证整包能参与选择，下载层回退时仍须沿用原有规则排序。 + prepared = [self.live.get(key, batch[key]) for key in ordered_keys if key in batch and key in self.candidates] + goals = {target for item in prepared if item.torrent_info is not None + for target in self._coverage(cast(str, item.torrent_info.search_resource_id))} + return prepared, goals def choose(self, ready: set[str], submit: Callable[[list[Context], set[str]], set[str]]) -> None: """ diff --git a/app/chain/subscribe/search.py b/app/chain/subscribe/search.py index 21012fa73..fd1505bfe 100644 --- a/app/chain/subscribe/search.py +++ b/app/chain/subscribe/search.py @@ -233,7 +233,8 @@ def _process_paged_subscription(owner: _SubscribeOwnerBase, subscribe: Subscript scan = SearchScan(owner=searchchain, plan=plan, full=strategy == "full" or prefer_initial_pack, page_limit=1 if strategy == "single_page" else None, execution=execution_context, repository=repository, snapshot=snapshot, - target_scope=(subscribe.start_episode or 1, subscribe.total_episode or 0)) + target_scope=(subscribe.start_episode or 1, subscribe.total_episode or 0), + prefer_complete=prefer_initial_pack) if execution_context: execution_context.incremental_search = True diff --git a/docs/testing.md b/docs/testing.md index 869a908e2..3c060f726 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -147,6 +147,11 @@ with patch.object(SomeModule, "fetch", new=AsyncMock(return_value=FAKE)): 回退。已有下载记录或媒体库已有目标集时沿用补集行为，提交前重新检查当前订阅；不会开启洗版 或写入洗版优先级。测试覆盖显式缺集列表、后页整包、不完整包和搜索期间进度变化，均隔离外部服务。 +整包优先也约束候选分批：已收集的整包即使排序低于分集、检查点恢复后缺少临时下载地址， +也须先恢复地址并共同进入下载选择；多个整包不完整、下载失败或恢复后已不可用时才回退分集。 +`tests/test_search_scan.py` 与 `tests/test_subscribe_paged_search.py` 在离线边界验证恢复、过滤和 +实际下载提交的先后顺序，同时保留普通补集按可用前缀提交的行为。 + 全集洗版（`best_version=1`、`best_version_full=1`）的主动搜索始终使用洗版规则组， 包括尚无下载记录或仍有缺集的订阅；未通过洗版规则的整包不会回落到普通订阅规则组。 订阅显式指定的规则组仍优先。分集洗版中从未下载的剧集继续使用普通订阅规则。 diff --git a/tests/test_search_scan.py b/tests/test_search_scan.py index 4a46b9d52..e9a6758d5 100644 --- a/tests/test_search_scan.py +++ b/tests/test_search_scan.py @@ -758,6 +758,67 @@ def test_opaque_ticket_recovery_submits_first_page_before_waiting_and_does_not_r assert calls == [0, 1] +@pytest.mark.parametrize("part_needs_recovery", [False, True]) +def test_whole_season_recovers_late_complete_pack_before_submitting_parts(owner, monkeypatch, part_needs_recovery): + """ + 完整搜索中整季包的临时地址恢复必须先于前页分集提交。 + """ + from app.application.search.session import SearchSessionSnapshot, encode_search_state + from app.application.site.observation import SiteSearchObservation + from app.chain.search.scan import ScanPage + + monkeypatch.setattr(TorrentHelper, "sort_torrents", staticmethod( + lambda items: sorted(items, key=lambda item: item.torrent_info.description))) + parts = [torrent(1), torrent(2)] + if part_needs_recovery: + parts[1].enclosure = "https://site.example/download?ticket=part-do-not-store" + pack = TorrentInfo( + site=1, + description="pack", + title="Show S01 Complete 2026 1080p", + enclosure="https://site.example/download?ticket=do-not-store", + ) + pages(owner, monkeypatch, {0: parts, 1: [pack]}) + initial = SearchScan(owner=owner, plan=plan([1, 2]), full=True, prefer_complete=True) + source = next(key for key in initial.queries if not key.startswith("plugin:")) + for page in range(3): + assert initial.step(source) + assert initial.step("plugin:Show") + + def restore(_method, **params): + """ + 从检查点恢复公开下载地址，临时票据交给原页重新获取。 + """ + return TorrentInfo(**params["record"]) + + monkeypatch.setattr(owner, "run_module", restore) + recovered = TorrentInfo( + site=1, + description="pack", + title="Show S01 Complete 2026 1080p", + enclosure="https://site.example/download?id=pack", + ) + fetches = [] + + def fetch(_sources, key, page, **_params): + """ + 只为整季候选重新请求原页并返回可下载地址。 + """ + fetches.append((key, page)) + return ScanPage([recovered], SiteSearchObservation(True, "success", raw_count=1)) + + monkeypatch.setattr(SearchSources, "fetch", fetch) + snapshot = SearchSessionSnapshot("round", 0, encode_search_state(initial.state())) + scan = SearchScan(owner=owner, plan=plan([1, 2]), full=True, snapshot=snapshot, prefer_complete=True) + submitted = [] + scan.choose({"1:1", "1:2"}, lambda items, goals: submitted.append( + ({item.torrent_info.description for item in items}, goals)) or goals) + + assert fetches == [(source, 1)] + expected = {"1-1", "pack"} if part_needs_recovery else {"1-1", "2-1", "pack"} + assert submitted == [(expected, {"1:1", "1:2"})] + + def test_one_unidentifiable_resource_does_not_block_the_page(owner, monkeypatch): """ 一条无法识别的资源不阻断同页其他候选处理。 @@ -1150,7 +1211,7 @@ def test_candidate_whose_page_keeps_failing_is_dropped_after_limited_retries(own assert not scan.collection.remaining -def _restored_ticket_scan(owner, monkeypatch, first_page, refetched): +def _restored_ticket_scan(owner, monkeypatch, first_page, refetched, *, prefer_complete=False): """ 第 0 页的候选带临时票据（不入检查点），恢复后由 refetched 模拟重取原页的结果。 """ @@ -1161,7 +1222,7 @@ def _restored_ticket_scan(owner, monkeypatch, first_page, refetched): for item in first_page: item.enclosure = "https://site.example/download?ticket=do-not-store" pages(owner, monkeypatch, {0: first_page}) - initial = SearchScan(owner=owner, plan=plan([1, 2]), full=True) + initial = SearchScan(owner=owner, plan=plan([1, 2]), full=True, prefer_complete=prefer_complete) source = next(key for key in initial.queries if not key.startswith("plugin:")) initial.step(source) initial.step(source) @@ -1173,7 +1234,61 @@ def _restored_ticket_scan(owner, monkeypatch, first_page, refetched): return ScanPage(refetched, SiteSearchObservation(True, "success", raw_count=len(refetched))) monkeypatch.setattr(SearchSources, "fetch", fetch) snapshot = SearchSessionSnapshot("round", 0, encode_search_state(initial.state())) - return SearchScan(owner=owner, plan=plan([1, 2]), full=True, snapshot=snapshot), fetched + return SearchScan(owner=owner, plan=plan([1, 2]), full=True, prefer_complete=prefer_complete, + snapshot=snapshot), fetched + + +@pytest.mark.parametrize("outcome", ["cooldown", "error"]) +@pytest.mark.parametrize("recovery_succeeds", [False, True]) +def test_complete_pack_recovery_waits_across_checkpoints_before_partial_fallback( + owner, monkeypatch, scan_clock, outcome, recovery_succeeds, +): + """ + 整包恢复等待期间不提交分集；检查点保留重试次数，恢复成功或达到上限后才能推进。 + """ + pack = TorrentInfo(site=1, description="pack", title="Show S01 Complete 2026 1080p") + scan, _ = _restored_ticket_scan(owner, monkeypatch, [torrent(1), pack], [], prefer_complete=True) + part_key = next(key for key, saved in scan.candidates.items() if saved["torrent"]["description"] == "1-1") + scan.candidates[part_key]["torrent"]["enclosure"] = "https://site.example/download?id=1" + calls = [] + + def fetch(_sources, _key, page, **_params): + """ + 整包原页前两次受限或失败，第三次可恢复，也可继续失败到本轮放弃。 + """ + calls.append(page) + if recovery_succeeds and len(calls) == 3: + restored = TorrentInfo(site=1, description="pack", title="Show S01 Complete 2026 1080p", + enclosure="https://site.example/download?id=pack") + return ScanPage([torrent(1), restored], SiteSearchObservation(True, "success", raw_count=2)) + if outcome == "cooldown": + retry_at = (scan_clock["now"] + timedelta(hours=1)).isoformat(timespec="seconds") + return ScanPage([], SiteSearchObservation(), retry_at, "cooldown") + return ScanPage([], SiteSearchObservation(True, "error", error="timeout")) + + monkeypatch.setattr(SearchSources, "fetch", fetch) + submitted = [] + + def submit(items, goals): + """ + 记录恢复等待结束后首次允许交给下载层的候选。 + """ + submitted.append(({item.torrent_info.description for item in items}, goals)) + return goals + + for version in range(2): + with pytest.raises(SubscriptionSearchDeferred) as error: + scan.run(submit) + assert submitted == [] + snapshot = SearchSessionSnapshot("round", version, encode_search_state(scan.state())) + scan = SearchScan(owner=owner, plan=plan([1, 2]), full=True, prefer_complete=True, snapshot=snapshot) + assert not scan.restart + scan_clock["now"] = datetime.fromisoformat(error.value.retry_at) + timedelta(seconds=1) + + scan.run(submit) + + assert calls == [0, 0, 0] + assert submitted == ([({"1-1", "pack"}, {"1:1", "1:2"})] if recovery_succeeds else [({"1-1"}, {"1:1"})]) @pytest.mark.parametrize("still_listed", [False, True]) diff --git a/tests/test_subscribe_paged_search.py b/tests/test_subscribe_paged_search.py index 4a4c22584..ca7fa6ae2 100644 --- a/tests/test_subscribe_paged_search.py +++ b/tests/test_subscribe_paged_search.py @@ -22,6 +22,7 @@ from app.application.subscription.priority import ( ) from app.chain.download import DownloadChain from app.chain.search.facade import SearchChain +from app.chain.search.result import TorrentHelper from app.chain.subscribe.facade import SubscribeChain from app.chain.subscribe.metadata import SubscriptionSearchTarget from app.chain.subscribe.search import ( @@ -450,6 +451,139 @@ def test_whole_season_downloads_late_complete_pack_before_early_parts(subscripti assert scope.current["value"].lack_episode == (1 if strategy == "single_page" else 0) +@pytest.fixture +def ticket_search(subscription, monkeypatch): + """ + 两站整季候选排序低于分集，首次页面无地址，重取原页后才取得下载地址。 + """ + scope = subscription + current = replace(scope.initial, total_episode=3, manual_total_episode=3) + scope.current["value"] = current + scope.media.seasons = {1: [1, 2, 3]} + target = SubscriptionSearchTarget(current, build_subscribe_meta(current), scope.media, + _missing(scope, current), "tmdb:1") + search = SearchChain() + monkeypatch.setattr(search, "_sync_indexers", lambda _sites: [{"id": 1, "name": "A"}, {"id": 2, "name": "B"}]) + monkeypatch.setattr(search, "get_search_page_size", lambda **_params: 100) + monkeypatch.setattr(search, "search_plugin_torrents", lambda **_params: []) + monkeypatch.setattr(scope.chain, "get_sub_sites", lambda _subscribe: [1, 2]) + monkeypatch.setattr(subscribe_search, "_subscription_search_strategy", lambda _best_version: "single_page") + monkeypatch.setattr(TorrentHelper, "sort_torrents", staticmethod( + lambda items: sorted(items, key=lambda item: -item.torrent_info.pri_order))) + attempts = {1: 0, 2: 0} + restored = {1: "ready", 2: "ready"} + events = [] + + def request(**params): + """ + 保留站点与资源身份，允许逐个模拟整包消失或重取后仍缺地址。 + """ + site = params["site"]["id"] + assert params["page"] == 0 + attempts[site] += 1 + recovering = attempts[site] > 1 + events.append(("recover" if recovering else "search", site)) + parts = [TorrentInfo(site=1, site_name="A", title=f"Example Show S01E{episode:02d}", + enclosure=f"https://site.example/download?id={episode}", pri_order=100) + for episode in (1, 2)] if site == 1 else [] + outcome = restored[site] if recovering else "missing" + packs = [] if outcome == "gone" else [TorrentInfo( + site=site, site_name="A" if site == 1 else "B", title="Example Show S01 Complete", + enclosure=f"https://site.example/download?id=pack{site}" if outcome == "ready" else None, + pri_order=80 if site == 1 else 60, + )] + items = parts + packs + report_site_search_outcome(attempted=True, outcome="success") + report_site_search_page(raw_count=len(items)) + return items + + def prepare(torrent, **_params): + """ + 记录种子解析顺序，让测试控制整包真实文件覆盖范围。 + """ + events.append(("torrent", torrent.site)) + return b"torrent", None, [1, 2, 3] + + def download(context, **_params): + """ + 任何下载副作用前，两站整包都必须已恢复或确认不可用。 + """ + assert attempts == {1: 2, 2: 2} + events.append(("download", context.torrent_info.site)) + return "hash" + + monkeypatch.setattr(search, "search_site_torrents", request) + scope.download.download_torrent.side_effect = prepare + scope.download.download_single.side_effect = download + return SimpleNamespace(scope=scope, search=search, current=current, target=target, + attempts=attempts, restored=restored, events=events) + + +@pytest.mark.parametrize("first_outcome", ["complete", "incomplete", "failed", "gone", "missing"]) +def test_initial_download_prepares_all_complete_packs_before_any_partial_submission(ticket_search, first_outcome): + """ + 低排序整包必须恢复后共同参与选择，前一个不完整或下载失败时先尝试后一个整包。 + """ + search = ticket_search + scope = search.scope + if first_outcome in {"gone", "missing"}: + search.restored[1] = first_outcome + if first_outcome == "incomplete": + prepare = scope.download.download_torrent.side_effect + + def incomplete(torrent, **params): + """ + 第一整包标题完整但实际只有一集，必须继续尝试另一站点整包。 + """ + content, path, episodes = prepare(torrent, **params) + return content, path, [1] if torrent.site == 1 else episodes + + scope.download.download_torrent.side_effect = incomplete + if first_outcome == "failed": + download = scope.download.download_single.side_effect + + def failed(context, **params): + """ + 第一整包下载器提交失败，保留第二整包成功的真实回退路径。 + """ + result = download(context, **params) + return None if context.torrent_info.site == 1 else result + + scope.download.download_single.side_effect = failed + + _process_paged_subscription(scope.chain, search.current, search.search, search.target, + SystemConfigKey.SubscribeFilterRuleGroups, None) + + contexts = [call.args[0] if call.args else call.kwargs["context"] + for call in scope.download.download_single.call_args_list] + assert [context.torrent_info.site for context in contexts] == ( + [1] if first_outcome == "complete" else [1, 2] if first_outcome == "failed" else [2]) + assert all(context.torrent_info.title == "Example Show S01 Complete" for context in contexts) + assert scope.current["value"].lack_episode == 0 + assert scope.current["value"].best_version == search.current.best_version + assert scope.current["value"].current_priority == search.current.current_priority + assert search.events[:4] == [("search", 1), ("search", 2), ("recover", 1), ("recover", 2)] + + +def test_initial_download_falls_back_to_original_partial_order_after_all_packs_disappear(ticket_search): + """ + 两站整包均不可用后按原规则下载分集，并保留未覆盖的第三集。 + """ + search = ticket_search + scope = search.scope + search.restored.update({1: "gone", 2: "gone"}) + + _process_paged_subscription(scope.chain, search.current, search.search, search.target, + SystemConfigKey.SubscribeFilterRuleGroups, None) + + contexts = [call.args[0] if call.args else call.kwargs["context"] + for call in scope.download.download_single.call_args_list] + assert [context.torrent_info.title for context in contexts] == [ + "Example Show S01E01", "Example Show S01E02"] + assert scope.current["value"].lack_episode == 1 + scope.download.download_torrent.assert_not_called() + + @pytest.mark.parametrize("fields,episodes,expected", [ ({}, [1, 2], True), ({"lack_episode": 1}, [1, 2], True),