From 77e4a80286299ff0bc76a942e9be4dc393e411bd Mon Sep 17 00:00:00 2001 From: jxxghp Date: Tue, 1 Sep 2026 13:11:13 +0800 Subject: [PATCH] refactor(subscribe): reuse fresh facts within batch --- app/application/subscription/facts.py | 66 +++++++++++++ app/chain/subscribe/match.py | 17 ++-- app/chain/subscribe/refresh.py | 17 ++-- ...iepilot-subscription-governance-roadmap.md | 22 ++++- .../architecture/dependency-baseline.json | 17 +++- tests/test_subscription_fresh_fact_lease.py | 99 +++++++++++++++++++ tests/test_subscription_governance_replay.py | 74 +++++++++++++- 7 files changed, 292 insertions(+), 20 deletions(-) create mode 100644 app/application/subscription/facts.py create mode 100644 tests/test_subscription_fresh_fact_lease.py diff --git a/app/application/subscription/facts.py b/app/application/subscription/facts.py new file mode 100644 index 000000000..1bf361e98 --- /dev/null +++ b/app/application/subscription/facts.py @@ -0,0 +1,66 @@ +"""订阅单轮新鲜媒体事实租约。""" + +import copy +from dataclasses import dataclass +from typing import Callable, Optional + +from app.application.subscription.contract import SubscriptionSnapshot +from app.domain.context import MediaInfo +from app.schemas.media import resolve_media_identity +from app.schemas.types import MediaType + + +@dataclass(frozen=True, slots=True) +class FreshFactKey: + """区分媒体身份、类型、季和剧集组的单轮事实键。""" + + media_source: str + media_id: str + media_type: str + season: Optional[int] + episode_group: Optional[str] + + @classmethod + def from_subscribe(cls, subscribe: SubscriptionSnapshot) -> Optional["FreshFactKey"]: + """从明确媒体身份的订阅构造事实键，身份缺失时禁止跨订阅复用。""" + media_source, media_id = resolve_media_identity(media=subscribe) + if not media_source or not media_id: + return None + media_type = subscribe.type.value if isinstance(subscribe.type, MediaType) else subscribe.type + if not media_type: + return None + return cls( + media_source=str(media_source), + media_id=media_id, + media_type=media_type, + season=subscribe.season, + episode_group=subscribe.episode_group, + ) + + +class FreshFactLease: + """在一个批次内合并相同媒体的新鲜识别，并向消费者返回隔离副本。""" + + def __init__(self) -> None: + """初始化仅在当前调用栈存活的事实缓存和命中计数。""" + self._facts: dict[FreshFactKey, Optional[MediaInfo]] = {} + self.loads = 0 + self.hits = 0 + + def get_or_load( + self, + subscribe: SubscriptionSnapshot, + loader: Callable[[], Optional[MediaInfo]], + ) -> Optional[MediaInfo]: + """读取本轮隔离副本；首次仍由 loader 按 `cache=False` 获取新鲜事实。""" + key = FreshFactKey.from_subscribe(subscribe) + if key is None: + self.loads += 1 + return loader() + if key in self._facts: + self.hits += 1 + return copy.deepcopy(self._facts[key]) + self.loads += 1 + fact = loader() + self._facts[key] = copy.deepcopy(fact) + return copy.deepcopy(fact) diff --git a/app/chain/subscribe/match.py b/app/chain/subscribe/match.py index e45e280be..77786eeb0 100644 --- a/app/chain/subscribe/match.py +++ b/app/chain/subscribe/match.py @@ -7,6 +7,7 @@ from typing import Callable, Dict, List, Optional from app.application.configuration import get_configured_system_config from app.application.subscription.candidates import CandidateBatch, CandidateIndex from app.application.subscription.contract import build_subscribe_meta, subscribe_media_key +from app.application.subscription.facts import FreshFactLease from app.application.torrent.download import TorrentHelper from app.chain.media import MediaChain from app.chain.subscribe.contract import _SubscribeOwnerBase @@ -142,6 +143,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): processed_torrents = self._prepare_match_torrents(torrents) candidate_index = CandidateIndex(processed_torrents) + fresh_fact_lease = FreshFactLease() # 所有订阅 subscribes = self.subscription_repository.list(self.get_states_for_search("R")) @@ -191,12 +193,15 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): logger.info(f"订阅 {subscribe.name} 本轮没有可能相关的资源，跳过资源匹配准备") continue # 识别媒体信息 - mediainfo: MediaInfo = MediaChain().recognize_media( - meta=meta, - mtype=meta.type, - **subscribe_recognize_kwargs(subscribe), - episode_group=subscribe.episode_group, - cache=False, + mediainfo = fresh_fact_lease.get_or_load( + subscribe, + lambda: MediaChain().recognize_media( + meta=meta, + mtype=meta.type, + **subscribe_recognize_kwargs(subscribe), + episode_group=subscribe.episode_group, + cache=False, + ), ) if not mediainfo: logger.warn( diff --git a/app/chain/subscribe/refresh.py b/app/chain/subscribe/refresh.py index 27bfe9fce..df7186846 100644 --- a/app/chain/subscribe/refresh.py +++ b/app/chain/subscribe/refresh.py @@ -11,6 +11,7 @@ from app.application.subscription.contract import ( subscribe_media_key, subscribe_media_keys, ) +from app.application.subscription.facts import FreshFactLease from app.chain.download import DownloadChain from app.chain.media import MediaChain from app.chain.subscribe.contract import _SubscribeOwnerBase @@ -104,6 +105,7 @@ class SubscribeRefreshOwner(_SubscribeOwnerBase): # 查询所有订阅 repository = self.subscription_repository subscribes = repository.list() + fresh_fact_lease = FreshFactLease() total_num = len(subscribes) if progress_callback: progress_callback( @@ -135,12 +137,15 @@ class SubscribeRefreshOwner(_SubscribeOwnerBase): if meta.type == MediaType.MUSIC: mediainfo = self._recognize_music_subscribe(subscribe) else: - mediainfo: MediaInfo = MediaChain().recognize_media( - meta=meta, - mtype=meta.type, - **subscribe_recognize_kwargs(subscribe), - episode_group=subscribe.episode_group, - cache=False, + mediainfo = fresh_fact_lease.get_or_load( + subscribe, + lambda: MediaChain().recognize_media( + meta=meta, + mtype=meta.type, + **subscribe_recognize_kwargs(subscribe), + episode_group=subscribe.episode_group, + cache=False, + ), ) if not mediainfo: logger.warn( diff --git a/docs/refactor/moviepilot-subscription-governance-roadmap.md b/docs/refactor/moviepilot-subscription-governance-roadmap.md index 2ce633933..aa9dc78d5 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-001D` +> 当前叶：`SUB-GOV-001E（条件评估）` > 迁移来源：`V3-RDY-009A`、`V3-RDY-009A1`、`V3-RDY-009A1B`、 > `V3-RDY-009A1C`、`V3-RDY-009A1D` > 适用范围：V3 订阅刷新、匹配、兜底搜索、下载提交与用户状态闭环 @@ -230,8 +230,8 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 | `SUB-GOV-001A` | 建立日常 Match 正确性和可重放夹具；撤销 `cache=True` 错误候选，只保留经验证不改变语义的失败识别状态回写 | 000 | `completed（2026-09-01）` | | `SUB-GOV-001B` | 区分完整缓存与本轮 delta，建立无损候选索引；先证明命中集合与基线完全一致 | 001A | `completed（2026-09-01）` | | `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-001D` | 引入单轮新鲜事实租约，评估稳定元数据复用、轻量季集查询和单媒体服务器事实合并 | 001C | `completed（2026-09-01）` | +| `SUB-GOV-001E` | 若 001B–D 后仍有可重复等待热点，再通过隔离压测评估 2/4 worker 的有界准备；提交保持串行 | 001D | `in_progress（条件评估）` | | `SUB-GOV-002A` | 将 24 小时兜底搜索移出 Match/提交长锁，建立批次、订阅 single-flight、可取消等待和恢复游标 | 001C | `pending` | | `SUB-GOV-002B` | 为兜底搜索建立每站点并发、间隔和错误冷却预算；RSS/Spider 保持基线压力 | 002A | `pending` | | `SUB-GOV-002C` | 联合运行日常 Match、手工搜索和兜底搜索，验证公平性、锁等待、站点压力和失败恢复 | 001D, 002B | `pending` | @@ -239,7 +239,7 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 | `SUB-GOV-003B` | 后端提供订阅及批次业务状态，前端展示排队、匹配、搜索、提交、完成、失败和取消 | 002A, 003A | `pending` | | `SUB-GOV-004` | 多条订阅记录指向同一媒体时的跨记录季集去重和产品规则 | 003A | `pending（最低优先级）` | -当前只激活 `SUB-GOV-001D`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B +当前只激活 `SUB-GOV-001E` 的条件评估。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B 只有在前两条链路的身份和终态稳定后实施。每个叶子完成验收并更新本表后，才激活下一个满足依赖的叶子。 ### 6.1 SUB-GOV-001A 验收证据 @@ -276,6 +276,20 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 - 验证：订阅专项 `131 passed`，API/调度器/架构组合 `183 passed`，错误级 Pylint 为 0，Host 架构基线和 `git diff --check` 通过。 +### 6.4 SUB-GOV-001D 验收证据 + +- 新增 `FreshFactLease`，键包含媒体来源、媒体 ID、类型、季和 episode group；缺少明确身份时禁止跨订阅复用； +- Match 与独立元数据巡检在单次调用内合并相同媒体的新鲜识别，首次仍明确使用 `cache=False`，不跨刷新周期 + 保留事实，也合并同轮失败结果，避免同一故障立即重复请求； +- 租约存储和交付均使用隔离副本，后续 `MediaInfo.clear()` 不会污染同批次其他订阅；固定回放证明同媒体同季的 + 两条订阅只触发一次识别并收到不同对象； +- 稳定元数据与轻量季集查询本叶不拆分：当前 canonical provider 合同由一次 `recognize_media(cache=False)` + 同时产出身份、别名和季集事实，拆成 provider 特化请求会扩大合同而没有独立收益证据； +- 单媒体服务器的最终缺集判定未跨订阅缓存：订阅范围、历史和洗版优先级属于各记录业务事实，只有后续 + `SUB-GOV-004` 明确重复订阅产品规则后才允许跨记录合并； +- 验证：订阅与候选回放 `118 passed`，事实租约/治理回放/架构组合 `108 passed`，错误级源代码 Pylint 为 0， + Host 架构基线与 `git diff --check` 通过。 + ## 7. 上线前验证与验收 ### 7.1 场景 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 378760401..c52557d1a 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1089,8 +1089,8 @@ "runtime_only": true } }, - "edge_count": 7701, - "edge_sha256": "9427670889cf46d007599fffd52b4b373f46445f939f209942901497765d233f", + "edge_count": 7711, + "edge_sha256": "f1ff788de28fa0c486e529afa4dc7af5038587f997bf155c5fc434754ec2a4a3", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -3442,6 +3442,14 @@ "app.application.subscription.delete -> app.schemas", "app.application.subscription.delete -> app.schemas.common", "app.application.subscription.delete -> app.schemas.event", + "app.application.subscription.facts -> app.application", + "app.application.subscription.facts -> app.application.subscription", + "app.application.subscription.facts -> app.application.subscription.contract", + "app.application.subscription.facts -> app.domain", + "app.application.subscription.facts -> app.domain.context", + "app.application.subscription.facts -> app.schemas", + "app.application.subscription.facts -> app.schemas.media", + "app.application.subscription.facts -> app.schemas.types", "app.application.subscription.identity -> app.application", "app.application.subscription.identity -> app.application.outbox", "app.application.subscription.identity -> app.application.subscription", @@ -4458,6 +4466,7 @@ "app.chain.subscribe.match -> app.application.subscription", "app.chain.subscribe.match -> app.application.subscription.candidates", "app.chain.subscribe.match -> app.application.subscription.contract", + "app.chain.subscribe.match -> app.application.subscription.facts", "app.chain.subscribe.match -> app.application.torrent", "app.chain.subscribe.match -> app.application.torrent.download", "app.chain.subscribe.match -> app.chain", @@ -4550,6 +4559,7 @@ "app.chain.subscribe.refresh -> app.application", "app.chain.subscribe.refresh -> app.application.subscription", "app.chain.subscribe.refresh -> app.application.subscription.contract", + "app.chain.subscribe.refresh -> app.application.subscription.facts", "app.chain.subscribe.refresh -> app.application.subscription.priority", "app.chain.subscribe.refresh -> app.chain", "app.chain.subscribe.refresh -> app.chain.download", @@ -8794,7 +8804,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 920, + "module_count": 921, "modules": [ "app", "app.adapters", @@ -9085,6 +9095,7 @@ "app.application.subscription.complete", "app.application.subscription.contract", "app.application.subscription.delete", + "app.application.subscription.facts", "app.application.subscription.identity", "app.application.subscription.mutation", "app.application.subscription.priority", diff --git a/tests/test_subscription_fresh_fact_lease.py b/tests/test_subscription_fresh_fact_lease.py new file mode 100644 index 000000000..5658212ca --- /dev/null +++ b/tests/test_subscription_fresh_fact_lease.py @@ -0,0 +1,99 @@ +"""订阅单轮新鲜事实租约测试。""" + +from app.application.subscription.contract import SubscriptionSnapshot +from app.application.subscription.facts import FreshFactKey, FreshFactLease +from app.domain.context import MediaInfo +from app.schemas.types import MediaSource, MediaType + + +def _subscribe(**overrides) -> SubscriptionSnapshot: + """构造具有明确媒体身份的电视剧订阅。""" + values = { + "id": 1, + "name": "租约测试剧", + "type": MediaType.TV.value, + "media_source": MediaSource.TMDB, + "media_id": "100", + "season": 1, + "episode_group": None, + "state": "R", + } + values.update(overrides) + return SubscriptionSnapshot(**values) + + +def test_fresh_fact_key_isolates_season_and_episode_group(): + """相同媒体的不同季或剧集组不得共享动态季集事实。""" + default = FreshFactKey.from_subscribe(_subscribe()) + other_season = FreshFactKey.from_subscribe(_subscribe(season=2)) + other_group = FreshFactKey.from_subscribe(_subscribe(episode_group="group-a")) + + assert default != other_season + assert default != other_group + assert other_season != other_group + + +def test_fresh_fact_lease_loads_once_and_returns_isolated_copies(): + """相同事实本轮只加载一次，消费者清理对象不会污染后续租约命中。""" + lease = FreshFactLease() + subscribe = _subscribe() + calls = [] + + def _load() -> MediaInfo: + """返回带完整季集和别名的可变媒体事实。""" + calls.append(True) + return MediaInfo( + media_source=MediaSource.TMDB, + media_id="100", + type=MediaType.TV, + title="租约测试剧", + seasons={1: [1, 2, 3]}, + names=["Lease Show"], + ) + + first = lease.get_or_load(subscribe, _load) + first.clear() + second = lease.get_or_load(subscribe, _load) + + assert len(calls) == 1 + assert lease.loads == 1 + assert lease.hits == 1 + assert second.seasons == {1: [1, 2, 3]} + assert second.names == ["Lease Show"] + assert first is not second + + +def test_fresh_fact_lease_merges_failed_result_within_round(): + """相同媒体本轮识别失败后不应立即重复请求外部服务。""" + lease = FreshFactLease() + calls = [] + + def _load(): + """记录一次失败的新鲜事实请求。""" + calls.append(True) + return None + + assert lease.get_or_load(_subscribe(), _load) is None + assert lease.get_or_load(_subscribe(id=2), _load) is None + assert len(calls) == 1 + assert lease.loads == 1 + assert lease.hits == 1 + + +def test_fresh_fact_lease_does_not_share_missing_identity(): + """身份缺失订阅必须各自识别，避免仅按标题错误合并。""" + lease = FreshFactLease() + calls = [] + subscribe = _subscribe(media_source=None, media_id=None) + + def _load() -> MediaInfo: + """返回本次标题识别结果。""" + calls.append(True) + return MediaInfo(type=MediaType.TV, title="标题识别结果") + + lease.get_or_load(subscribe, _load) + lease.get_or_load(subscribe, _load) + + assert len(calls) == 2 + assert lease.loads == 2 + assert lease.hits == 0 diff --git a/tests/test_subscription_governance_replay.py b/tests/test_subscription_governance_replay.py index 30433b132..ecc1f9cf6 100644 --- a/tests/test_subscription_governance_replay.py +++ b/tests/test_subscription_governance_replay.py @@ -38,6 +38,17 @@ class _ReplaySubscriptionRepository: return self.current +class _ReplaySubscriptionListRepository: + """提供多条订阅快照，验证批次级执行合同。""" + + def __init__(self, subscribes: list[SubscriptionSnapshot]) -> None: + self.subscribes = subscribes + + def list(self, _state: str = None) -> list[SubscriptionSnapshot]: + """返回当前批次的全部订阅快照。""" + return self.subscribes + + class _ReplayTorrentHelper: """让无关候选稳定停在身份冲突边界。""" @@ -250,4 +261,65 @@ def test_metadata_reconcile_reuses_fresh_fact_without_candidate_batch(monkeypatc assert repository.current.lack_episode == 1 assert len(reconciled) == 1 assert reconciled[0]["subscribe"] == repository.current - assert reconciled[0]["mediainfo"] is fresh_media + assert reconciled[0]["mediainfo"] == fresh_media + assert reconciled[0]["mediainfo"] is not fresh_media + + +def test_match_reuses_fresh_fact_for_same_media_subscriptions(monkeypatch): + """同媒体同季订阅在一个 Match 批次内只读取一次外部新鲜事实。""" + first = _build_subscribe(_load_replay_cases()[1]) + second = replace(first, id=102) + repository = _ReplaySubscriptionListRepository([first, second]) + recognition_calls = [] + received_media = [] + candidate = _build_unrelated_candidate() + candidate.meta_info.media_source = MediaSource.TMDB + candidate.meta_info.media_id = "100" + candidate.media_info.media_id = "100" + candidate.media_info.title = "增长中的剧集" + + class _ReplayMediaChain: + """返回同一可变对象，验证租约向每个订阅交付独立副本。""" + + def recognize_media(self, **kwargs) -> MediaInfo: + """记录一次外部识别并返回固定媒体事实。""" + recognition_calls.append(kwargs) + return MediaInfo( + media_source=MediaSource.TMDB, + media_id="100", + type=MediaType.TV, + title="增长中的剧集", + year="2026", + seasons={1: list(range(1, 14))}, + ) + + @staticmethod + def recognize_by_meta(*_args, **_kwargs) -> MediaInfo: + """候选已有明确身份，本场景不应重新识别。""" + raise AssertionError("明确身份候选不应重新识别") + + chain = SubscribeChain() + chain.subscription_repository = repository + + def _handle_existing(*, mediainfo, **_kwargs): + """记录每个订阅收到的事实对象并提前结束其匹配。""" + received_media.append(mediainfo) + return True, {} + + chain.check_and_handle_existing_media = _handle_existing + monkeypatch.setattr("app.chain.subscribe.match.MediaChain", _ReplayMediaChain) + monkeypatch.setattr( + "app.chain.subscribe.match.get_configured_system_config", + lambda: SimpleNamespace(get=lambda _key: []), + ) + monkeypatch.setattr( + "app.chain.subscribe.query.get_configured_system_config", + lambda: SimpleNamespace(get=lambda _key: []), + ) + + chain.match({"replay.example": [candidate]}) + + assert len(recognition_calls) == 1 + assert recognition_calls[0]["cache"] is False + assert len(received_media) == 2 + assert received_media[0] is not received_media[1]