refactor(subscribe): introduce candidate batch index

app/application/subscription/candidates.py

@@ -0,0 +1,290 @@
+"""订阅候选批次与无损路由合同。"""
+
+import copy
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Dict, List, Optional
+from uuid import uuid4
+
+from app.application.subscription.contract import SubscriptionSnapshot
+from app.domain.context import Context, MediaInfo
+from app.foundation import text as text_tools
+from app.schemas.media import resolve_media_identity
+from app.schemas.types import MediaType
+
+
+CandidateGroups = Dict[str, List[Context]]
+
+
+@dataclass(slots=True)
+class CandidateBatch:
+    """一次资源获取产生的完整候选、增量候选与重试候选边界。"""
+
+    batch_id: str
+    source: str
+    candidates: CandidateGroups
+    fresh_candidates: CandidateGroups = field(default_factory=dict)
+    retry_candidates: CandidateGroups = field(default_factory=dict)
+    sites: tuple[str, ...] = ()
+    started_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+    finished_at: Optional[datetime] = None
+
+    @classmethod
+    def create(
+        cls,
+        *,
+        source: str,
+        candidates: CandidateGroups,
+        fresh_candidates: Optional[CandidateGroups] = None,
+        retry_candidates: Optional[CandidateGroups] = None,
+        sites: Optional[List[str]] = None,
+        started_at: Optional[datetime] = None,
+    ) -> "CandidateBatch":
+        """构造已完成获取的候选批次。"""
+        return cls(
+            batch_id=uuid4().hex,
+            source=source,
+            candidates=candidates,
+            fresh_candidates=fresh_candidates or {},
+            retry_candidates=retry_candidates or {},
+            sites=tuple(sites or candidates.keys()),
+            started_at=started_at or datetime.now(timezone.utc),
+            finished_at=datetime.now(timezone.utc),
+        )
+
+    @classmethod
+    def from_legacy(cls, candidates: CandidateGroups, source: str = "legacy") -> "CandidateBatch":
+        """把旧入口传入的完整候选包装为无增量声明的兼容批次。"""
+        return cls.create(source=source, candidates=candidates)
+
+    @staticmethod
+    def count(groups: CandidateGroups) -> int:
+        """统计分站点候选总数。"""
+        return sum(len(contexts) for contexts in groups.values())
+
+    def build_index(self) -> "CandidateIndex":
+        """基于完整候选构建一次性无损索引。"""
+        return CandidateIndex(self.candidates)
+
+
+class CandidateIndex:
+    """一次构建并保持原顺序的订阅候选身份索引。"""
+
+    def __init__(self, candidates: CandidateGroups) -> None:
+        """记录候选顺序、明确身份和必须保守处理的候选集合。"""
+        self._ordered: list[tuple[str, Context]] = []
+        self._by_identity: dict[tuple[str, str], set[int]] = {}
+        self._unknown: set[int] = set()
+        self._reconcilable: set[int] = set()
+        self._explicit_identity: dict[int, tuple[str, str]] = {}
+        for domain, contexts in candidates.items():
+            for context in contexts:
+                position = len(self._ordered)
+                self._ordered.append((domain, context))
+                media_identity = self.media_identity(getattr(context, "media_info", None))
+                meta_identity = self.media_identity(getattr(context, "meta_info", None))
+                identities = {identity for identity in (media_identity, meta_identity) if identity}
+                for identity in identities:
+                    self._by_identity.setdefault(identity, set()).add(position)
+                if not media_identity:
+                    # 主识别失败时 canonical Match 仍允许标题兜底，不能因标题标签 ID 提前排除。
+                    self._unknown.add(position)
+                elif meta_identity:
+                    self._explicit_identity[position] = meta_identity
+                else:
+                    # 标题解析未携带显式 ID 的识别冲突仍可能通过同作品证据复核。
+                    self._reconcilable.add(position)
+
+    def select_cache_candidates(
+        self,
+        subscribe: SubscriptionSnapshot,
+        *,
+        allow_title_match: bool = False,
+    ) -> List[Context]:
+        """返回严格身份候选，并按需附加显式标记的标题兜底副本。"""
+        results: List[Context] = []
+        for _domain, context in self._ordered:
+            copied = copy.deepcopy(context)
+            if self.strict_matches(copied, subscribe):
+                results.append(copied)
+                continue
+            if allow_title_match and self.title_matches(copied, subscribe):
+                self.mark_title_candidate(copied, subscribe)
+                results.append(copied)
+        return results
+
+    def route_for_match(self, subscribe: SubscriptionSnapshot) -> CandidateGroups:
+        """保守路由可能命中的候选，且只排除当前 canonical 逻辑必然拒绝的候选。"""
+        if subscribe.custom_words:
+            positions = set(range(len(self._ordered)))
+        else:
+            target_identity = self.media_identity(subscribe)
+            positions = set(self._unknown)
+            positions.update(self._reconcilable)
+            if target_identity:
+                positions.update(self._by_identity.get(target_identity, set()))
+                positions.update(
+                    position
+                    for position, explicit_identity in self._explicit_identity.items()
+                    if explicit_identity == target_identity
+                )
+
+        routed: CandidateGroups = {}
+        for position, (domain, context) in enumerate(self._ordered):
+            if position not in positions:
+                continue
+            if not subscribe.custom_words and (
+                not self.media_type_matches(context, subscribe)
+                or not self.season_matches(context, subscribe)
+            ):
+                continue
+            routed.setdefault(domain, []).append(context)
+        return routed
+
+    @classmethod
+    def strict_matches(cls, context: Context, subscribe: SubscriptionSnapshot) -> bool:
+        """判断候选自身明确身份、类型和季是否严格命中订阅。"""
+        if not cls.media_type_matches(context, subscribe):
+            return False
+        if not cls.season_matches(context, subscribe):
+            return False
+        subscribe_identity = cls.media_identity(subscribe)
+        return bool(subscribe_identity and subscribe_identity in cls.context_identities(context))
+
+    @classmethod
+    def title_matches(cls, context: Context, subscribe: SubscriptionSnapshot) -> bool:
+        """仅允许身份缺失候选按标题进入低置信诊断兜底。"""
+        if cls.context_identities(context):
+            return False
+        if not cls.media_type_matches(context, subscribe):
+            return False
+        if not cls.season_matches(context, subscribe):
+            return False
+        subscribe_title = cls.normalize_title(subscribe.name)
+        if not subscribe_title:
+            return False
+        meta_info = getattr(context, "meta_info", None)
+        torrent_info = getattr(context, "torrent_info", None)
+        candidate_titles = (
+            getattr(torrent_info, "title", None),
+            getattr(meta_info, "title", None),
+            getattr(meta_info, "name", None),
+        )
+        return any(
+            subscribe_title in candidate_title
+            for candidate_title in (cls.normalize_title(title) for title in candidate_titles)
+            if candidate_title
+        )
+
+    @staticmethod
+    def mark_title_candidate(context: Context, subscribe: SubscriptionSnapshot) -> None:
+        """把标题兜底副本标记为目标回填，避免伪装成候选自身识别结果。"""
+        context.match_source = "title"
+        context.candidate_recognized = False
+        context.media_info_is_target = True
+        context.media_info = MediaInfo(
+            type=subscribe.type,
+            title=subscribe.name,
+            media_source=subscribe.media_source,
+            media_id=subscribe.media_id,
+            season=subscribe.season,
+        )
+
+    @classmethod
+    def media_type_matches(cls, context: Context, subscribe: SubscriptionSnapshot) -> bool:
+        """类型已知且冲突时拒绝，缺失类型保持保守候选。"""
+        subscribe_type = cls.normalize_media_type(subscribe.type)
+        media_info = getattr(context, "media_info", None)
+        meta_info = getattr(context, "meta_info", None)
+        context_types = {
+            cls.normalize_media_type(value)
+            for value in (
+                getattr(media_info, "type", None),
+                getattr(meta_info, "type", None),
+            )
+        }
+        context_types.discard(None)
+        return not subscribe_type or not context_types or all(
+            context_type == subscribe_type for context_type in context_types
+        )
+
+    @classmethod
+    def season_matches(cls, context: Context, subscribe: SubscriptionSnapshot) -> bool:
+        """仅在资源季信息明确排除目标季时拒绝。"""
+        target_season = cls.normalize_int(subscribe.season)
+        if target_season is None:
+            return True
+        meta_info = getattr(context, "meta_info", None)
+        explicit_meta_seasons = cls.meta_seasons(meta_info)
+        if explicit_meta_seasons:
+            return target_season in explicit_meta_seasons
+        media_info = getattr(context, "media_info", None)
+        media_season = cls.normalize_int(getattr(media_info, "season", None))
+        return media_season is None or target_season == media_season
+
+    @classmethod
+    def meta_seasons(cls, meta_info) -> set[int]:
+        """提取标题解析出的显式季范围。"""
+        meta_fields = vars(meta_info) if meta_info else {}
+        if "season_list" in meta_fields:
+            season_list = {
+                season
+                for season in (
+                    cls.normalize_int(item)
+                    for item in (meta_fields.get("season_list") or [])
+                )
+                if season is not None
+            }
+            if season_list:
+                return season_list
+        begin_season = cls.normalize_int(getattr(meta_info, "begin_season", None))
+        end_season = cls.normalize_int(getattr(meta_info, "end_season", None))
+        if begin_season is not None and end_season is not None:
+            start, end = sorted((begin_season, end_season))
+            return set(range(start, end + 1))
+        if begin_season is not None:
+            return {begin_season}
+        if end_season is not None:
+            return {end_season}
+        return set()
+
+    @classmethod
+    def context_identities(cls, context: Context) -> set[tuple[str, str]]:
+        """提取候选媒体信息与标题标签中的通用媒体身份。"""
+        identities = {
+            cls.media_identity(getattr(context, "media_info", None)),
+            cls.media_identity(getattr(context, "meta_info", None)),
+        }
+        return {identity for identity in identities if identity}
+
+    @staticmethod
+    def media_identity(media) -> Optional[tuple[str, str]]:
+        """把动态媒体对象的身份归一为可索引键。"""
+        source, media_id = resolve_media_identity(media=media)
+        if not source or not media_id:
+            return None
+        return str(source), media_id
+
+    @staticmethod
+    def normalize_int(value) -> Optional[int]:
+        """将季号等动态字段转为整数。"""
+        if value is None:
+            return None
+        try:
+            return int(value)
+        except (TypeError, ValueError):
+            return None
+
+    @staticmethod
+    def normalize_media_type(value) -> Optional[str]:
+        """统一媒体类型枚举与字符串形态。"""
+        if isinstance(value, MediaType):
+            value = value.value
+        if value == MediaType.UNKNOWN.value:
+            return None
+        return value
+
+    @staticmethod
+    def normalize_title(value) -> str:
+        """归一标题用于低置信标题匹配。"""
+        return (text_tools.normalize_upper(value or "") or "").strip()

app/chain/subscribe/contract.py

@@ -88,6 +88,7 @@ if TYPE_CHECKING:
         get_subscribed_sites: Callable[..., Any]
         has_music_subscribe: Callable[..., Any]
         match: Callable[..., Any]
+        match_batch: Callable[..., Any]
         media_exists: Callable[..., Any]
         media_files: Callable[..., Any]
         obtain_images: Callable[..., Any]

app/chain/subscribe/match.py

@@ -5,6 +5,7 @@ from datetime import datetime
 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.torrent.download import TorrentHelper
 from app.chain.media import MediaChain
@@ -102,6 +103,17 @@ class SubscribeMatchOwner(_SubscribeOwnerBase):
             progress_callback=progress_callback,
         )
 
+    def match_batch(
+        self,
+        batch: CandidateBatch,
+        progress_callback: Optional[Callable[..., None]] = None,
+    ) -> None:
+        """消费包含完整缓存与本轮增量边界的订阅候选批次。"""
+        return self._execute_match(
+            torrents=batch.candidates,
+            progress_callback=progress_callback,
+        )
+
     def _execute_match(
         self,
         torrents: Dict[str, List[Context]],
@@ -129,6 +141,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase):
                 return
 
             processed_torrents = self._prepare_match_torrents(torrents)
+            candidate_index = CandidateIndex(processed_torrents)
 
             # 所有订阅
             subscribes = self.subscription_repository.list(self.get_states_for_search("R"))
@@ -207,7 +220,8 @@ class SubscribeMatchOwner(_SubscribeOwnerBase):
                     torrenthelper = TorrentHelper()
                     systemconfig = get_configured_system_config()
                     wordsmatcher = WordsMatcher()
-                    for domain, contexts in processed_torrents.items():
+                    routed_torrents = candidate_index.route_for_match(subscribe)
+                    for domain, contexts in routed_torrents.items():
                         if runtime_stop_state.is_system_stopped:
                             break
                         if domains and domain not in domains:

app/chain/subscribe/refresh.py

@@ -77,14 +77,14 @@ class SubscribeRefreshOwner(_SubscribeOwnerBase):
                     data=data,
                 )
 
-        torrents = TorrentsChain().refresh(
+        candidate_batch = TorrentsChain().refresh_batch(
             sites=sites,
             progress_callback=_update_refresh_progress if progress_callback else None,
             # 存在音乐订阅时额外抓取站点音乐专用入口，音乐不一定在默认种子首页
             include_music=self.has_music_subscribe(),
         )
-        self.match(
-            torrents,
+        self.match_batch(
+            candidate_batch,
             progress_callback=_update_match_progress if progress_callback else None,
         )
         if progress_callback:

app/chain/torrents.py

@@ -1,11 +1,12 @@
-import copy
 import re
 import traceback
+from datetime import datetime, timezone
 from typing import Callable, Dict, List, Optional, Union
 
 from app.application.configuration import get_configured_system_config
 from app.application.rss import RssHelper
 from app.application.site.sites import SitesHelper  # pylint: disable=import-error,no-name-in-module
+from app.application.subscription.candidates import CandidateBatch, CandidateIndex
 from app.application.torrent.download import TorrentHelper
 from app.chain.base import ChainBase
 from app.chain.media import MediaChain
@@ -13,7 +14,6 @@ from app.domain import site as site_rules
 from app.domain.context import Context, MediaInfo, MusicInfo, TorrentInfo
 from app.domain.meta.metamusic import MetaMusic
 from app.domain.metainfo import MetaInfo
-from app.foundation import text as text_tools
 from app.runtime.log import logger
 from app.runtime.stop import runtime_stop_state
 from app.schemas.media import resolve_media_identity
@@ -162,195 +162,89 @@ class TorrentsChain(ChainBase):
         主程序只提供缓存读取与轻量候选筛选，不在这里判断站点证据能否扩展
         订阅目标或放行完成；标题兜底候选会显式标记为低置信来源。
         """
-        results: List[Context] = []
-        for contexts in (self.get_torrents(stype=stype) or {}).values():
-            for context in contexts or []:
-                if not context:
-                    continue
-                copied = copy.deepcopy(context)
-                if self._context_matches_subscribe(copied, subscribe):
-                    results.append(copied)
-                    continue
-                if allow_title_match and self._context_title_matches_subscribe(copied, subscribe):
-                    self._mark_title_match_candidate(copied, subscribe)
-                    results.append(copied)
-        return results
+        candidates = {
+            domain: [context for context in contexts or [] if context]
+            for domain, contexts in (self.get_torrents(stype=stype) or {}).items()
+        }
+        return CandidateIndex(candidates).select_cache_candidates(
+            subscribe,
+            allow_title_match=allow_title_match,
+        )
 
     @classmethod
     def _context_matches_subscribe(cls, context: Context, subscribe) -> bool:
         """
         严格身份匹配：候选自身识别出的媒体 ID 命中订阅，且季信息不排除订阅季。
         """
-        if not cls._context_media_type_matches(context, subscribe):
-            return False
-        if not cls._context_season_matches_subscribe(context, subscribe):
-            return False
-
-        subscribe_identity = resolve_media_identity(media=subscribe)
-        context_identities = cls._context_media_identities(context)
-
-        return bool(all(subscribe_identity) and subscribe_identity in context_identities)
+        return CandidateIndex.strict_matches(context, subscribe)
 
     @classmethod
     def _context_title_matches_subscribe(cls, context: Context, subscribe) -> bool:
         """
         标题兜底只服务诊断：仅允许身份缺失候选按标题命中，显式冲突 ID 不兜底。
         """
-        if cls._context_has_media_identity(context):
-            return False
-        if not cls._context_media_type_matches(context, subscribe):
-            return False
-        if not cls._context_season_matches_subscribe(context, subscribe):
-            return False
-
-        subscribe_title = cls._normalize_title(getattr(subscribe, "name", None))
-        if not subscribe_title:
-            return False
-
-        meta_info = getattr(context, "meta_info", None)
-        torrent_info = getattr(context, "torrent_info", None)
-        candidate_titles = [
-            getattr(torrent_info, "title", None),
-            getattr(meta_info, "title", None),
-            getattr(meta_info, "name", None),
-        ]
-        return any(
-            subscribe_title in candidate_title
-            for candidate_title in (cls._normalize_title(title) for title in candidate_titles)
-            if candidate_title
-        )
+        return CandidateIndex.title_matches(context, subscribe)
 
     @staticmethod
     def _mark_title_match_candidate(context: Context, subscribe) -> None:
         """
         标记标题兜底候选，避免下游把目标媒体回填误认为候选自身识别结果。
         """
-        context.match_source = "title"
-        context.candidate_recognized = False
-        context.media_info_is_target = True
-        context.media_info = MediaInfo(
-            type=getattr(subscribe, "type", None),
-            title=getattr(subscribe, "name", None),
-            media_source=getattr(subscribe, "media_source", None),
-            media_id=getattr(subscribe, "media_id", None),
-            season=getattr(subscribe, "season", None),
-        )
+        CandidateIndex.mark_title_candidate(context, subscribe)
 
     @classmethod
     def _context_media_type_matches(cls, context: Context, subscribe) -> bool:
         """
         类型已知且冲突时拒绝；缺失类型不作为缓存候选过滤条件。
         """
-        subscribe_type = cls._normalize_media_type(getattr(subscribe, "type", None))
-        media_info = getattr(context, "media_info", None)
-        meta_info = getattr(context, "meta_info", None)
-        context_types = {
-            cls._normalize_media_type(value)
-            for value in (
-                getattr(media_info, "type", None),
-                getattr(meta_info, "type", None),
-            )
-        }
-        context_types.discard(None)
-        return not subscribe_type or not context_types or all(
-            context_type == subscribe_type for context_type in context_types
-        )
+        return CandidateIndex.media_type_matches(context, subscribe)
 
     @classmethod
     def _context_season_matches_subscribe(cls, context: Context, subscribe) -> bool:
         """
         资源季信息只要明确排除订阅季就拒绝；跨季覆盖目标季留给插件诊断。
         """
-        target_season = cls._normalize_int(getattr(subscribe, "season", None))
-        if target_season is None:
-            return True
-
-        meta_info = getattr(context, "meta_info", None)
-        explicit_meta_seasons = cls._context_meta_seasons(meta_info)
-        if explicit_meta_seasons:
-            return target_season in explicit_meta_seasons
-
-        media_info = getattr(context, "media_info", None)
-        media_season = cls._normalize_int(getattr(media_info, "season", None))
-        return media_season is None or target_season == media_season
+        return CandidateIndex.season_matches(context, subscribe)
 
     @classmethod
     def _context_meta_seasons(cls, meta_info) -> set[int]:
         """
         提取标题解析出的显式季范围；多季包以该范围为准。
         """
-        meta_fields = vars(meta_info) if meta_info else {}
-        if "season_list" in meta_fields:
-            season_list = {
-                season
-                for season in (
-                    cls._normalize_int(item)
-                    for item in (meta_fields.get("season_list") or [])
-                )
-                if season is not None
-            }
-            if season_list:
-                return season_list
-        begin_season = cls._normalize_int(getattr(meta_info, "begin_season", None))
-        end_season = cls._normalize_int(getattr(meta_info, "end_season", None))
-        if begin_season is not None and end_season is not None:
-            start, end = sorted((begin_season, end_season))
-            return set(range(start, end + 1))
-        if begin_season is not None:
-            return {begin_season}
-        if end_season is not None:
-            return {end_season}
-        return set()
+        return CandidateIndex.meta_seasons(meta_info)
 
     @staticmethod
     def _context_has_media_identity(context: Context) -> bool:
         """
         判断候选是否已经带有明确媒体 ID。
         """
-        return bool(TorrentsChain._context_media_identities(context))
+        return bool(CandidateIndex.context_identities(context))
 
     @staticmethod
     def _context_media_identities(context: Context) -> set[tuple[str, str]]:
         """提取候选媒体信息与标题标签中的通用媒体身份。"""
-        identities = {
-            resolve_media_identity(media=getattr(context, "media_info", None)),
-            resolve_media_identity(media=getattr(context, "meta_info", None)),
-        }
-        return {
-            (str(source), media_id)
-            for source, media_id in identities
-            if source and media_id
-        }
+        return CandidateIndex.context_identities(context)
 
     @staticmethod
     def _normalize_int(value) -> Optional[int]:
         """
         将季号等动态字段转为 int，无法解析时视为缺失。
         """
-        if value is None:
-            return None
-        try:
-            return int(value)
-        except (TypeError, ValueError):
-            return None
+        return CandidateIndex.normalize_int(value)
 
     @staticmethod
     def _normalize_media_type(value) -> Optional[str]:
         """
         统一 MediaType 枚举与字符串形态。
         """
-        if isinstance(value, MediaType):
-            value = value.value
-        if value == MediaType.UNKNOWN.value:
-            return None
-        return value
+        return CandidateIndex.normalize_media_type(value)
 
     @staticmethod
     def _normalize_title(value) -> str:
         """
         归一标题用于低置信标题兜底匹配。
         """
-        return (text_tools.normalize_upper(value or "") or "").strip()
+        return CandidateIndex.normalize_title(value)
 
     def clear_torrents(self):
         """
@@ -527,6 +421,8 @@ class TorrentsChain(ChainBase):
             include_music: bool,
             torrents_cache: Dict[str, List[Context]],
             music_cache: Dict[str, List[Context]],
+            fresh_torrents: Dict[str, List[Context]],
+            fresh_music: Dict[str, List[Context]],
     ) -> str:
         """抓取并写入单个站点的影视、音乐资源缓存。"""
         domain = site_rules.extract_domain(indexer.get("domain"))
@@ -586,7 +482,9 @@ class TorrentsChain(ChainBase):
                 continue
             context = self._build_refresh_context(torrent, stype)
             target_cache = music_cache if torrent.category == MediaType.MUSIC.value else torrents_cache
+            target_fresh = fresh_music if torrent.category == MediaType.MUSIC.value else fresh_torrents
             target_cache.setdefault(domain, []).append(context)
+            target_fresh.setdefault(domain, []).append(context)
             if len(target_cache[domain]) > self.runtime_config.torrent_cache_size:
                 target_cache[domain] = target_cache[domain][-self.runtime_config.torrent_cache_size:]
         return domain
@@ -639,13 +537,30 @@ class TorrentsChain(ChainBase):
             progress_callback: Optional[Callable[..., None]] = None,
             include_music: bool = False,
     ) -> Dict[str, List[Context]]:
+        """兼容旧调用返回完整候选字典，批次语义由 ``refresh_batch`` 提供。"""
+        return self.refresh_batch(
+            stype=stype,
+            sites=sites,
+            progress_callback=progress_callback,
+            include_music=include_music,
+        ).candidates
+
+    def refresh_batch(
+            self,
+            stype: Optional[str] = None,
+            sites: List[int] = None,
+            progress_callback: Optional[Callable[..., None]] = None,
+            include_music: bool = False,
+    ) -> CandidateBatch:
         """
-        刷新站点最新资源，识别并缓存起来
+        刷新站点最新资源并返回完整缓存与本轮新增候选。
+
         :param stype: 强制指定缓存类型，spider:爬虫缓存，rss:rss缓存
         :param sites: 强制指定站点ID列表，为空则读取设置的订阅站点
         :param progress_callback: 资源刷新进度更新回调
         :param include_music: 是否额外抓取站点的音乐专用浏览入口，服务音乐订阅
         """
+        started_at = datetime.now(timezone.utc)
 
         # 刷新类型
         if not stype:
@@ -664,6 +579,8 @@ class TorrentsChain(ChainBase):
             music_cache = self.load_cache(self._music_rss_file) or {}
         self._ensure_context_compatibility(torrents_cache, stype=stype)
         self._ensure_context_compatibility(music_cache, stype=stype)
+        fresh_torrents: Dict[str, List[Context]] = {}
+        fresh_music: Dict[str, List[Context]] = {}
 
         # 缓存过滤掉无效种子（影视与音乐缓存分别处理）
         for _cache in (torrents_cache, music_cache):
@@ -704,6 +621,8 @@ class TorrentsChain(ChainBase):
                 include_music=include_music,
                 torrents_cache=torrents_cache,
                 music_cache=music_cache,
+                fresh_torrents=fresh_torrents,
+                fresh_music=fresh_music,
             ))
 
         # 保存缓存到本地，影视与音乐分别存储
@@ -719,6 +638,10 @@ class TorrentsChain(ChainBase):
             torrents_cache = {k: v for k, v in torrents_cache.items() if k in domains}
         if sites and music_cache:
             music_cache = {k: v for k, v in music_cache.items() if k in domains}
+        if sites and fresh_torrents:
+            fresh_torrents = {k: v for k, v in fresh_torrents.items() if k in domains}
+        if sites and fresh_music:
+            fresh_music = {k: v for k, v in fresh_music.items() if k in domains}
 
         if progress_callback:
             progress_callback(
@@ -727,12 +650,37 @@ class TorrentsChain(ChainBase):
                 data={"total": total_indexers, "finished": total_indexers},
             )
 
-        # 订阅匹配需要完整候选，音乐独立缓存在返回值中按站点合并
-        for _domain, _contexts in music_cache.items():
-            if _contexts:
-                torrents_cache.setdefault(_domain, []).extend(_contexts)
+        self._retain_cached_fresh(fresh_torrents, torrents_cache)
+        self._retain_cached_fresh(fresh_music, music_cache)
+        candidates = self._merge_torrent_caches(
+            {domain: list(contexts) for domain, contexts in torrents_cache.items()},
+            music_cache,
+        )
+        fresh_candidates = self._merge_torrent_caches(
+            {domain: list(contexts) for domain, contexts in fresh_torrents.items()},
+            fresh_music,
+        )
+        return CandidateBatch.create(
+            source=stype,
+            candidates=candidates,
+            fresh_candidates=fresh_candidates,
+            sites=domains,
+            started_at=started_at,
+        )
 
-        return torrents_cache
+    @staticmethod
+    def _retain_cached_fresh(
+        fresh_candidates: Dict[str, List[Context]],
+        cached_candidates: Dict[str, List[Context]],
+    ) -> None:
+        """移除因缓存容量裁剪而未进入最终完整缓存的本轮候选。"""
+        for domain, contexts in list(fresh_candidates.items()):
+            cached_ids = {id(context) for context in cached_candidates.get(domain) or []}
+            retained = [context for context in contexts if id(context) in cached_ids]
+            if retained:
+                fresh_candidates[domain] = retained
+            else:
+                fresh_candidates.pop(domain, None)
 
     @staticmethod
     def _ensure_context_compatibility(torrents_cache: Dict[str, List[Context]], stype: Optional[str] = None):

docs/refactor/moviepilot-subscription-governance-roadmap.md

@@ -1,7 +1,7 @@
 # MoviePilot 订阅执行治理
 
 > 状态：`active（2026-09-01 已由 MoviePilot v3 接管）`
-> 当前叶：`SUB-GOV-001B`
+> 当前叶：`SUB-GOV-001C`
 > 迁移来源：`V3-RDY-009A`、`V3-RDY-009A1`、`V3-RDY-009A1B`、
 > `V3-RDY-009A1C`、`V3-RDY-009A1D`
 > 适用范围：V3 订阅刷新、匹配、兜底搜索、下载提交与用户状态闭环
@@ -228,8 +228,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 | `in_progress` |
-| `SUB-GOV-001C` | 将资源匹配与完成对账拆开；只有受候选影响的订阅进入匹配，完成对账独立保持新鲜语义 | 001B | `pending` |
+| `SUB-GOV-001B` | 区分完整缓存与本轮 delta，建立无损候选索引；先证明命中集合与基线完全一致 | 001A | `completed（2026-09-01）` |
+| `SUB-GOV-001C` | 将资源匹配与完成对账拆开；只有受候选影响的订阅进入匹配，完成对账独立保持新鲜语义 | 001B | `in_progress` |
 | `SUB-GOV-001D` | 引入单轮新鲜事实租约，评估稳定元数据复用、轻量季集查询和单媒体服务器事实合并 | 001C | `pending` |
 | `SUB-GOV-001E` | 若 001B–D 后仍有可重复等待热点，再通过隔离压测评估 2/4 worker 的有界准备；提交保持串行 | 001D | `conditional` |
 | `SUB-GOV-002A` | 将 24 小时兜底搜索移出 Match/提交长锁，建立批次、订阅 single-flight、可取消等待和恢复游标 | 001C | `pending` |
@@ -239,7 +239,7 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收
 | `SUB-GOV-003B` | 后端提供订阅及批次业务状态，前端展示排队、匹配、搜索、提交、完成、失败和取消 | 002A, 003A | `pending` |
 | `SUB-GOV-004` | 多条订阅记录指向同一媒体时的跨记录季集去重和产品规则 | 003A | `pending（最低优先级）` |
 
-当前只激活 `SUB-GOV-001B`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B
+当前只激活 `SUB-GOV-001C`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B
 只有在前两条链路的身份和终态稳定后实施。每个叶子完成验收并更新本表后，才激活下一个满足依赖的叶子。
 
 ### 6.1 SUB-GOV-001A 验收证据
@@ -253,6 +253,17 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收
 - 验证：上述重放、现有订阅 Chain 与候选筛选共 `118 passed`，固定 JSON、文档链接和
   `git diff --check` 通过。
 
+### 6.2 SUB-GOV-001B 验收证据
+
+- `CandidateBatch` 同时表达完整缓存、本轮 `fresh_candidates`、预留 `retry_candidates`、来源、站点和批次时间；
+- `TorrentsChain.refresh()` 保持原字典 ABI，新增 `refresh_batch()` 供订阅刷新消费，RSS/Spider 请求模型未改变；
+- `CandidateIndex` 在一次构建后按媒体身份路由，并保持候选原顺序；未知身份、主识别失败、标题兜底、
+  自定义识别词和无显式标题 ID 的可复核冲突均进入保守集合；
+- 严格缓存候选 API 委托新索引并继续返回深拷贝，插件可观察行为保持不变；
+- 专项矩阵验证完整缓存与本轮 delta 分离，并验证索引只排除类型、季或显式身份上 canonical 必然拒绝的候选；
+- 验证：订阅、候选、音乐缓存和架构测试共 `228 passed`，错误级 Pylint 为 0，
+  `scripts/architecture/baseline.py --check-host` 与 `git diff --check` 通过。
+
 ## 7. 上线前验证与验收
 
 ### 7.1 场景

tests/fixtures/architecture/dependency-baseline.json

@@ -1089,8 +1089,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 7690,
-  "edge_sha256": "8bfd75d59fe7bf736dd4deb09d943f7e1668916159d64b3188774ad1915193ef",
+  "edge_count": 7701,
+  "edge_sha256": "9427670889cf46d007599fffd52b4b373f46445f939f209942901497765d233f",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -3406,6 +3406,16 @@
     "app.application.storage -> app.schemas",
     "app.application.storage -> app.schemas.system",
     "app.application.storage -> app.schemas.types",
+    "app.application.subscription.candidates -> app.application",
+    "app.application.subscription.candidates -> app.application.subscription",
+    "app.application.subscription.candidates -> app.application.subscription.contract",
+    "app.application.subscription.candidates -> app.domain",
+    "app.application.subscription.candidates -> app.domain.context",
+    "app.application.subscription.candidates -> app.foundation",
+    "app.application.subscription.candidates -> app.foundation.text",
+    "app.application.subscription.candidates -> app.schemas",
+    "app.application.subscription.candidates -> app.schemas.media",
+    "app.application.subscription.candidates -> app.schemas.types",
     "app.application.subscription.complete -> app.application",
     "app.application.subscription.complete -> app.application.outbox",
     "app.application.subscription.complete -> app.application.subscription",
@@ -4446,6 +4456,7 @@
     "app.chain.subscribe.match -> app.application",
     "app.chain.subscribe.match -> app.application.configuration",
     "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.torrent",
     "app.chain.subscribe.match -> app.application.torrent.download",
@@ -4609,6 +4620,8 @@
     "app.chain.torrents -> app.application.configuration",
     "app.chain.torrents -> app.application.rss",
     "app.chain.torrents -> app.application.site",
+    "app.chain.torrents -> app.application.subscription",
+    "app.chain.torrents -> app.application.subscription.candidates",
     "app.chain.torrents -> app.application.torrent",
     "app.chain.torrents -> app.application.torrent.download",
     "app.chain.torrents -> app.chain",
@@ -4620,8 +4633,6 @@
     "app.chain.torrents -> app.domain.meta.metamusic",
     "app.chain.torrents -> app.domain.metainfo",
     "app.chain.torrents -> app.domain.site",
-    "app.chain.torrents -> app.foundation",
-    "app.chain.torrents -> app.foundation.text",
     "app.chain.torrents -> app.runtime",
     "app.chain.torrents -> app.runtime.log",
     "app.chain.torrents -> app.runtime.stop",
@@ -8783,7 +8794,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 919,
+  "module_count": 920,
   "modules": [
     "app",
     "app.adapters",
@@ -9070,6 +9081,7 @@
     "app.application.site.query",
     "app.application.storage",
     "app.application.subscription",
+    "app.application.subscription.candidates",
     "app.application.subscription.complete",
     "app.application.subscription.contract",
     "app.application.subscription.delete",

tests/test_subscription_candidate_batch.py

@@ -0,0 +1,181 @@
+"""订阅候选批次与无损索引合同测试。"""
+
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+from app.application.subscription.candidates import CandidateBatch, CandidateIndex
+from app.application.subscription.contract import SubscriptionSnapshot
+from app.chain.torrents import TorrentsChain
+from app.domain.context import Context, MediaInfo, TorrentInfo
+from app.domain.metainfo import MetaInfo
+from app.schemas.types import MediaSource, MediaType
+
+
+def _context(
+    title: str,
+    media_id: str = None,
+    *,
+    meta_media_id: str = None,
+    media_type: MediaType = MediaType.TV,
+    season: int = 1,
+    enclosure: str = None,
+) -> Context:
+    """构造可配置身份、类型和季的候选上下文。"""
+    meta = MetaInfo(title=title)
+    meta.type = media_type
+    meta.begin_season = season
+    if meta_media_id:
+        meta.media_source = MediaSource.TMDB
+        meta.media_id = meta_media_id
+    return Context(
+        meta_info=meta,
+        media_info=MediaInfo(
+            media_source=MediaSource.TMDB if media_id else None,
+            media_id=media_id,
+            type=media_type,
+            title=title,
+            season=season,
+        ),
+        torrent_info=TorrentInfo(
+            title=title,
+            description="",
+            enclosure=enclosure or f"https://example.com/{title}",
+            site=1,
+            site_name="Test",
+            category=media_type.value,
+        ),
+        resource_source="rss",
+        match_source=MediaSource.TMDB.value if media_id else "unknown",
+        candidate_recognized=bool(media_id),
+    )
+
+
+def _subscribe(**overrides) -> SubscriptionSnapshot:
+    """构造候选路由使用的电视剧订阅。"""
+    values = {
+        "id": 1,
+        "name": "目标剧集",
+        "type": MediaType.TV.value,
+        "media_source": MediaSource.TMDB,
+        "media_id": "100",
+        "season": 1,
+        "state": "R",
+        "sites": [],
+        "best_version": 0,
+    }
+    values.update(overrides)
+    return SubscriptionSnapshot(**values)
+
+
+def test_refresh_batch_distinguishes_complete_cache_from_fresh_delta():
+    """刷新批次必须保留完整缓存，同时只把本轮新增资源放入 fresh 集合。"""
+    chain = TorrentsChain()
+    existing = _context("既有剧集 S01E01", media_id="100")
+    duplicate = TorrentInfo(
+        title=existing.torrent_info.title,
+        description=existing.torrent_info.description,
+        enclosure="https://example.com/duplicate",
+        site=1,
+        site_name="Test",
+        category=MediaType.TV.value,
+    )
+    fresh = TorrentInfo(
+        title="新增剧集 S01E02",
+        description="",
+        enclosure="https://example.com/fresh",
+        site=1,
+        site_name="Test",
+        category=MediaType.TV.value,
+    )
+    sites_helper = Mock()
+    sites_helper.get_indexers.return_value = [
+        {"id": 1, "name": "Test", "domain": "https://example.com"}
+    ]
+
+    def _load_cache(filename):
+        """仅影视 RSS 缓存预置一条历史候选。"""
+        if filename == TorrentsChain._rss_file:
+            return {"example.com": [existing]}
+        return {}
+
+    with (
+        patch.object(chain, "load_cache", side_effect=_load_cache),
+        patch.object(chain, "rss", return_value=[duplicate, fresh]),
+        patch.object(chain, "save_cache"),
+        patch("app.chain.torrents.SitesHelper", return_value=sites_helper),
+        patch(
+            "app.chain.torrents.MediaChain",
+            return_value=SimpleNamespace(
+                recognize_by_meta=lambda *_args, **_kwargs: MediaInfo(
+                    media_source=MediaSource.TMDB,
+                    media_id="100",
+                    type=MediaType.TV,
+                )
+            ),
+        ),
+    ):
+        batch = chain.refresh_batch(stype="rss", sites=[1])
+
+    assert isinstance(batch, CandidateBatch)
+    assert batch.source == "rss"
+    assert batch.finished_at is not None
+    assert [item.torrent_info.title for item in batch.candidates["example.com"]] == [
+        existing.torrent_info.title,
+        fresh.title,
+    ]
+    assert [item.torrent_info.title for item in batch.fresh_candidates["example.com"]] == [fresh.title]
+    assert CandidateBatch.count(batch.candidates) == 2
+    assert CandidateBatch.count(batch.fresh_candidates) == 1
+
+
+def test_candidate_index_routes_all_canonical_fallback_classes_without_loss():
+    """索引必须保留未知身份、识别失败和可复核冲突，只排除确定冲突。"""
+    exact = _context("目标剧集 S01E01", media_id="100")
+    inferred_conflict = _context("目标剧集 S01E02", media_id="200")
+    failed_with_meta_id = _context(
+        "目标剧集 S01E03",
+        media_id=None,
+        meta_media_id="300",
+    )
+    explicit_conflict = _context(
+        "其他剧集 S01E01",
+        media_id="400",
+        meta_media_id="400",
+    )
+    season_conflict = _context("目标剧集 S02E01", media_id="100", season=2)
+    type_conflict = _context(
+        "目标电影 2026",
+        media_id="100",
+        media_type=MediaType.MOVIE,
+        season=None,
+    )
+    candidates = {
+        "example.com": [
+            exact,
+            inferred_conflict,
+            failed_with_meta_id,
+            explicit_conflict,
+            season_conflict,
+            type_conflict,
+        ]
+    }
+
+    routed = CandidateIndex(candidates).route_for_match(_subscribe())
+
+    assert routed["example.com"] == [exact, inferred_conflict, failed_with_meta_id]
+
+
+def test_candidate_index_custom_words_preserve_complete_candidate_set():
+    """自定义识别词可能改变身份、类型和季，索引不得提前排除任何候选。"""
+    candidates = {
+        "example.com": [
+            _context("其他剧集 S02E01", media_id="400", meta_media_id="400", season=2),
+            _context("其他电影 2026", media_id="500", media_type=MediaType.MOVIE, season=None),
+        ]
+    }
+
+    routed = CandidateIndex(candidates).route_for_match(
+        _subscribe(custom_words="被替换词 => 目标剧集")
+    )
+
+    assert routed == candidates