diff --git a/app/application/subscription/candidates.py b/app/application/subscription/candidates.py new file mode 100644 index 000000000..0cf5e1bef --- /dev/null +++ b/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() diff --git a/app/chain/subscribe/contract.py b/app/chain/subscribe/contract.py index d700bba6a..7814d628b 100644 --- a/app/chain/subscribe/contract.py +++ b/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] diff --git a/app/chain/subscribe/match.py b/app/chain/subscribe/match.py index 0fedc5888..195369531 100644 --- a/app/chain/subscribe/match.py +++ b/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: diff --git a/app/chain/subscribe/refresh.py b/app/chain/subscribe/refresh.py index e7d81e775..4d04723ec 100644 --- a/app/chain/subscribe/refresh.py +++ b/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: diff --git a/app/chain/torrents.py b/app/chain/torrents.py index 7cda327e3..60e23ddca 100644 --- a/app/chain/torrents.py +++ b/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): diff --git a/docs/refactor/moviepilot-subscription-governance-roadmap.md b/docs/refactor/moviepilot-subscription-governance-roadmap.md index d46ab1876..a3e12c1d7 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-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 场景 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 1311b88ec..378760401 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/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", diff --git a/tests/test_subscription_candidate_batch.py b/tests/test_subscription_candidate_batch.py new file mode 100644 index 000000000..3b3dee143 --- /dev/null +++ b/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