refactor(subscribe): reuse fresh facts within batch

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)

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(
+                    mediainfo = fresh_fact_lease.get_or_load(
-                        meta=meta,
+                        subscribe,
-                        mtype=meta.type,
+                        lambda: MediaChain().recognize_media(
-                        **subscribe_recognize_kwargs(subscribe),
+                            meta=meta,
-                        episode_group=subscribe.episode_group,
+                            mtype=meta.type,
-                        cache=False,
+                            **subscribe_recognize_kwargs(subscribe),
+                            episode_group=subscribe.episode_group,
+                            cache=False,
+                        ),
                     )
                     if not mediainfo:
                         logger.warn(

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(
+                mediainfo = fresh_fact_lease.get_or_load(
-                    meta=meta,
+                    subscribe,
-                    mtype=meta.type,
+                    lambda: MediaChain().recognize_media(
-                    **subscribe_recognize_kwargs(subscribe),
+                        meta=meta,
-                    episode_group=subscribe.episode_group,
+                        mtype=meta.type,
-                    cache=False,
+                        **subscribe_recognize_kwargs(subscribe),
+                        episode_group=subscribe.episode_group,
+                        cache=False,
+                    ),
                 )
             if not mediainfo:
                 logger.warn(

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-001D` | 引入单轮新鲜事实租约，评估稳定元数据复用、轻量季集查询和单媒体服务器事实合并 | 001C | `completed（2026-09-01）` |
-| `SUB-GOV-001E` | 若 001B–D 后仍有可重复等待热点，再通过隔离压测评估 2/4 worker 的有界准备；提交保持串行 | 001D | `conditional` |
+| `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 场景

tests/fixtures/architecture/dependency-baseline.json

@@ -1089,8 +1089,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 7701,
+  "edge_count": 7711,
-  "edge_sha256": "9427670889cf46d007599fffd52b4b373f46445f939f209942901497765d233f",
+  "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",

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

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]