diff --git a/app/application/subscription/candidates.py b/app/application/subscription/candidates.py index 0cf5e1bef..8774f5afe 100644 --- a/app/application/subscription/candidates.py +++ b/app/application/subscription/candidates.py @@ -113,8 +113,14 @@ class CandidateIndex: results.append(copied) return results - def route_for_match(self, subscribe: SubscriptionSnapshot) -> CandidateGroups: - """保守路由可能命中的候选，且只排除当前 canonical 逻辑必然拒绝的候选。""" + def route_for_match( + self, + subscribe: SubscriptionSnapshot, + *, + domains: Optional[set[str]] = None, + site_ids: Optional[set[int]] = None, + ) -> CandidateGroups: + """保守路由可能命中的候选，并在外部事实查询前应用站点范围。""" if subscribe.custom_words: positions = set(range(len(self._ordered))) else: @@ -133,6 +139,11 @@ class CandidateIndex: for position, (domain, context) in enumerate(self._ordered): if position not in positions: continue + if domains and domain not in domains: + continue + torrent_info = getattr(context, "torrent_info", None) + if site_ids and getattr(torrent_info, "site", None) not in site_ids: + continue if not subscribe.custom_words and ( not self.media_type_matches(context, subscribe) or not self.season_matches(context, subscribe) diff --git a/app/chain/subscribe/completion.py b/app/chain/subscribe/completion.py index 426f170c2..7b21c0628 100644 --- a/app/chain/subscribe/completion.py +++ b/app/chain/subscribe/completion.py @@ -6,6 +6,7 @@ from typing import Any, Dict, List, Optional, Tuple, Union from app.application.subscription.contract import ( SubscriptionSnapshot, + subscribe_media_key, subscribe_media_keys, ) from app.application.subscription.mutation import SubscriptionActor @@ -174,6 +175,22 @@ class SubscribeCompletionOwner(_SubscribeOwnerBase): elif not downloads: logger.info(f"{mediainfo.title_year} 继续洗版 ...") + def reconcile_subscription_completion( + self, + subscribe: SubscriptionSnapshot, + meta: MetaBase, + mediainfo: MediaInfo, + ) -> bool: + """使用已取得的新鲜媒体事实独立对账订阅完成状态。""" + mediakey = subscribe_media_key(subscribe) + completed, _no_exists = self.check_and_handle_existing_media( + subscribe=subscribe, + meta=meta, + mediainfo=mediainfo, + mediakey=mediakey, + ) + return completed + def _SubscribeChain__update_subscribe_note( self, subscribe: SubscriptionSnapshot, diff --git a/app/chain/subscribe/contract.py b/app/chain/subscribe/contract.py index 7814d628b..cc2d1497a 100644 --- a/app/chain/subscribe/contract.py +++ b/app/chain/subscribe/contract.py @@ -79,6 +79,7 @@ if TYPE_CHECKING: """异步发送订阅通知。""" raise NotImplementedError check_and_handle_existing_media: Callable[..., Any] + check_and_reconcile: Callable[..., Any] filter_torrents: Callable[..., Any] finish_subscribe_or_not: Callable[..., Any] get_params: Callable[..., Any] @@ -96,6 +97,7 @@ if TYPE_CHECKING: post_message: Callable[..., Any] remote_list: Callable[..., Any] resolve_subscribe_missing: Callable[..., Any] + reconcile_subscription_completion: Callable[..., Any] _SubscribeOwnerBase = _SubscribeOwnerHost else: diff --git a/app/chain/subscribe/match.py b/app/chain/subscribe/match.py index 195369531..e45e280be 100644 --- a/app/chain/subscribe/match.py +++ b/app/chain/subscribe/match.py @@ -181,6 +181,15 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): domains = [] if subscribe.sites: domains = self.site_repository.get_domains_by_ids(subscribe.sites) + sub_sites = self.get_sub_sites(subscribe) + routed_torrents = candidate_index.route_for_match( + subscribe, + domains=set(domains) if domains else None, + site_ids=set(sub_sites) if sub_sites else None, + ) + if not routed_torrents: + logger.info(f"订阅 {subscribe.name} 本轮没有可能相关的资源，跳过资源匹配准备") + continue # 识别媒体信息 mediainfo: MediaInfo = MediaChain().recognize_media( meta=meta, @@ -220,7 +229,6 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): torrenthelper = TorrentHelper() systemconfig = get_configured_system_config() wordsmatcher = WordsMatcher() - routed_torrents = candidate_index.route_for_match(subscribe) for domain, contexts in routed_torrents.items(): if runtime_stop_state.is_system_stopped: break @@ -237,7 +245,6 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): torrent_info = _context.torrent_info # 不在订阅站点范围的不处理 - sub_sites = self.get_sub_sites(subscribe) if sub_sites and torrent_info.site not in sub_sites: logger.debug(f"{torrent_info.site_name} - {torrent_info.title} 不符合订阅站点要求") continue diff --git a/app/chain/subscribe/refresh.py b/app/chain/subscribe/refresh.py index 4d04723ec..27bfe9fce 100644 --- a/app/chain/subscribe/refresh.py +++ b/app/chain/subscribe/refresh.py @@ -90,11 +90,16 @@ class SubscribeRefreshOwner(_SubscribeOwnerBase): if progress_callback: progress_callback(value=100, text="订阅刷新完成") - def check(self, progress_callback: Optional[Callable[..., None]] = None) -> None: + def check( + self, + progress_callback: Optional[Callable[..., None]] = None, + reconcile_completion: bool = False, + ) -> None: """ 定时检查订阅，更新订阅信息 :param progress_callback: 定时服务进度更新回调 + :param reconcile_completion: 是否复用本次新鲜媒体事实执行独立完成对账 """ # 查询所有订阅 repository = self.subscription_repository @@ -219,6 +224,12 @@ class SubscribeRefreshOwner(_SubscribeOwnerBase): update_data, scene="metadata_refresh", ) + if reconcile_completion and subscribe.state in self.get_states_for_search("R"): + self.reconcile_subscription_completion( + subscribe=subscribe, + meta=meta, + mediainfo=mediainfo, + ) logger.info(f"{subscribe.name} 订阅元数据更新完成") if progress_callback: progress_callback( @@ -229,6 +240,16 @@ class SubscribeRefreshOwner(_SubscribeOwnerBase): if progress_callback: progress_callback(value=100, text="订阅元数据更新完成") + def check_and_reconcile( + self, + progress_callback: Optional[Callable[..., None]] = None, + ) -> None: + """刷新订阅元数据，并复用同一轮新鲜事实执行完成对账。""" + return self.check( + progress_callback=progress_callback, + reconcile_completion=True, + ) + async def cache_calendar( self, progress_callback: Optional[Callable[..., None]] = None, diff --git a/app/startup/initializers/scheduler.py b/app/startup/initializers/scheduler.py index faa0ab01a..00dd7cbd1 100644 --- a/app/startup/initializers/scheduler.py +++ b/app/startup/initializers/scheduler.py @@ -59,7 +59,7 @@ def configure_scheduler_services() -> None: SchedulerServices( sync_cookies=site_chain.sync_cookies, sync_mediaserver=mediaserver_chain.sync, - check_subscribe=subscribe_chain.check, + check_subscribe=subscribe_chain.check_and_reconcile, search_subscribe=subscribe_chain.search, refresh_subscribe=subscribe_chain.refresh, follow_subscribe=subscribe_chain.follow, diff --git a/docs/refactor/moviepilot-subscription-governance-roadmap.md b/docs/refactor/moviepilot-subscription-governance-roadmap.md index a3e12c1d7..2ce633933 100644 --- a/docs/refactor/moviepilot-subscription-governance-roadmap.md +++ b/docs/refactor/moviepilot-subscription-governance-roadmap.md @@ -1,7 +1,7 @@ # MoviePilot 订阅执行治理 > 状态：`active（2026-09-01 已由 MoviePilot v3 接管）` -> 当前叶：`SUB-GOV-001C` +> 当前叶：`SUB-GOV-001D` > 迁移来源：`V3-RDY-009A`、`V3-RDY-009A1`、`V3-RDY-009A1B`、 > `V3-RDY-009A1C`、`V3-RDY-009A1D` > 适用范围：V3 订阅刷新、匹配、兜底搜索、下载提交与用户状态闭环 @@ -229,8 +229,8 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 | `SUB-GOV-000` | 固定历史行为和生产日志；保留 `V3-RDY-009A1A` 已交付的批量请求正确性 | 无 | `completed` | | `SUB-GOV-001A` | 建立日常 Match 正确性和可重放夹具；撤销 `cache=True` 错误候选，只保留经验证不改变语义的失败识别状态回写 | 000 | `completed（2026-09-01）` | | `SUB-GOV-001B` | 区分完整缓存与本轮 delta，建立无损候选索引；先证明命中集合与基线完全一致 | 001A | `completed（2026-09-01）` | -| `SUB-GOV-001C` | 将资源匹配与完成对账拆开；只有受候选影响的订阅进入匹配，完成对账独立保持新鲜语义 | 001B | `in_progress` | -| `SUB-GOV-001D` | 引入单轮新鲜事实租约，评估稳定元数据复用、轻量季集查询和单媒体服务器事实合并 | 001C | `pending` | +| `SUB-GOV-001C` | 将资源匹配与完成对账拆开；只有受候选影响的订阅进入匹配，完成对账独立保持新鲜语义 | 001B | `completed（2026-09-01）` | +| `SUB-GOV-001D` | 引入单轮新鲜事实租约，评估稳定元数据复用、轻量季集查询和单媒体服务器事实合并 | 001C | `in_progress` | | `SUB-GOV-001E` | 若 001B–D 后仍有可重复等待热点，再通过隔离压测评估 2/4 worker 的有界准备；提交保持串行 | 001D | `conditional` | | `SUB-GOV-002A` | 将 24 小时兜底搜索移出 Match/提交长锁，建立批次、订阅 single-flight、可取消等待和恢复游标 | 001C | `pending` | | `SUB-GOV-002B` | 为兜底搜索建立每站点并发、间隔和错误冷却预算；RSS/Spider 保持基线压力 | 002A | `pending` | @@ -239,7 +239,7 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 | `SUB-GOV-003B` | 后端提供订阅及批次业务状态，前端展示排队、匹配、搜索、提交、完成、失败和取消 | 002A, 003A | `pending` | | `SUB-GOV-004` | 多条订阅记录指向同一媒体时的跨记录季集去重和产品规则 | 003A | `pending（最低优先级）` | -当前只激活 `SUB-GOV-001C`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B +当前只激活 `SUB-GOV-001D`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B 只有在前两条链路的身份和终态稳定后实施。每个叶子完成验收并更新本表后，才激活下一个满足依赖的叶子。 ### 6.1 SUB-GOV-001A 验收证据 @@ -264,6 +264,18 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 - 验证：订阅、候选、音乐缓存和架构测试共 `228 passed`，错误级 Pylint 为 0， `scripts/architecture/baseline.py --check-host` 与 `git diff --check` 通过。 +### 6.3 SUB-GOV-001C 验收证据 + +- Match 在媒体识别和媒体库查询前应用候选身份、类型、季、站点域名和站点 ID 门禁；无可能候选时不读取 + TMDB/媒体服务器事实，也不在资源路径隐式完成订阅； +- 新增 `reconcile_subscription_completion()`，复用 canonical 缺失查询与完成路径，不复制完成规则； +- 新增 `check_and_reconcile()`，现有每 6 小时 `subscribe_tmdb` 调度绑定该入口，在同一次 `cache=False` + 新鲜识别中完成元数据更新和独立完成对账；普通 `check()` 兼容入口仍可只更新元数据； +- 固定重放验证明确冲突候选不触发外部事实查询，并验证无候选批次时独立巡检仍将 12→13 更新为 `13/1` + 后进入完成对账； +- 验证：订阅专项 `131 passed`，API/调度器/架构组合 `183 passed`，错误级 Pylint 为 0，Host + 架构基线和 `git diff --check` 通过。 + ## 7. 上线前验证与验收 ### 7.1 场景 diff --git a/tests/test_subscription_governance_replay.py b/tests/test_subscription_governance_replay.py index e9123f6ee..30433b132 100644 --- a/tests/test_subscription_governance_replay.py +++ b/tests/test_subscription_governance_replay.py @@ -176,3 +176,78 @@ def test_match_replays_fresh_episode_completion_contract(case, monkeypatch): assert repository.current.total_episode == case["expected_total_episode"] assert repository.current.lack_episode == case["expected_lack_episode"] assert bool(completions) is case["expected_completed"] + + +def test_match_skips_external_facts_when_index_has_no_possible_candidates(monkeypatch): + """本轮只有明确冲突候选时，不得为订阅调用 TMDB 或媒体服务器准备。""" + subscribe = _build_subscribe(_load_replay_cases()[0]) + repository = _ReplaySubscriptionRepository(subscribe) + candidate = _build_unrelated_candidate() + candidate.meta_info.media_source = MediaSource.TMDB + candidate.meta_info.media_id = "999" + chain = SubscribeChain() + chain.subscription_repository = repository + chain.resolve_subscribe_missing = lambda **_kwargs: pytest.fail("不应查询媒体库缺失事实") + + class _UnexpectedMediaChain: + """任何媒体识别调用都表示候选门禁失效。""" + + def recognize_media(self, **_kwargs): + """阻止无候选订阅读取外部媒体事实。""" + pytest.fail("不应读取 TMDB 新鲜事实") + + def recognize_by_meta(self, *_args, **_kwargs): + """明确候选身份不应重新识别。""" + pytest.fail("不应重新识别明确候选") + + monkeypatch.setattr("app.chain.subscribe.match.MediaChain", _UnexpectedMediaChain) + monkeypatch.setattr( + "app.chain.subscribe.query.get_configured_system_config", + lambda: SimpleNamespace(get=lambda _key: []), + ) + + chain.match({"replay.example": [candidate]}) + + +def test_metadata_reconcile_reuses_fresh_fact_without_candidate_batch(monkeypatch): + """独立元数据巡检应复用同一次新鲜识别执行完成对账。""" + subscribe = _build_subscribe(_load_replay_cases()[1]) + repository = _ReplaySubscriptionRepository(subscribe) + recognition_calls = [] + reconciled = [] + fresh_media = MediaInfo( + media_source=MediaSource.TMDB, + media_id="100", + type=MediaType.TV, + title="增长中的剧集", + year="2026", + seasons={1: list(range(1, 14))}, + ) + + class _ReplayMediaChain: + """为独立完成对账返回固定新鲜媒体事实。""" + + def recognize_media(self, **kwargs) -> MediaInfo: + """记录识别参数并返回 13 集事实。""" + recognition_calls.append(kwargs) + return fresh_media + + chain = SubscribeChain() + chain.subscription_repository = repository + chain._SubscribeChain__apply_subscribe_update = ( + lambda _subscribe, update_data, **_kwargs: repository.update(dict(update_data)) + ) + chain.reconcile_subscription_completion = ( + lambda **kwargs: reconciled.append(kwargs) + ) + monkeypatch.setattr("app.chain.subscribe.refresh.MediaChain", _ReplayMediaChain) + + chain.check_and_reconcile() + + assert len(recognition_calls) == 1 + assert recognition_calls[0]["cache"] is False + assert repository.current.total_episode == 13 + assert repository.current.lack_episode == 1 + assert len(reconciled) == 1 + assert reconciled[0]["subscribe"] == repository.current + assert reconciled[0]["mediainfo"] is fresh_media