From 2f0c1252da41180856fb7a670f08852fade8f850 Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Mon, 6 Jul 2026 06:54:19 +0800 Subject: [PATCH] feat(subscribe): expose cached site candidates (#6062) --- app/chain/subscribe.py | 146 +++-- app/chain/torrents.py | 233 ++++++++ app/schemas/event.py | 11 +- tests/test_subscribe_chain.py | 559 ++++++++++++++++++++ tests/test_torrents_subscribe_candidates.py | 181 +++++++ 5 files changed, 1080 insertions(+), 50 deletions(-) create mode 100644 tests/test_torrents_subscribe_candidates.py diff --git a/app/chain/subscribe.py b/app/chain/subscribe.py index f108df35..9b66d832 100644 --- a/app/chain/subscribe.py +++ b/app/chain/subscribe.py @@ -317,6 +317,71 @@ class SubscribeChain(ChainBase): update_data.update(cls.__prepare_subscribe_progress_fields(subscribe=subscribe, no_exists={})) return update_data + @classmethod + def __prepare_best_version_total_change_fields( + cls, + subscribe: Subscribe, + total_episode: int, + old_total_episode: int, + ) -> Dict[str, Any]: + """ + 准备洗版电视剧总集数变化后需要写库的字段。 + + 总集数变化会改变目标范围，按集优先级只保留新范围内的目标集，避免范围外 + 旧状态继续参与完成集、缺失集和当前优先级计算。 + """ + update_data: Dict[str, Any] = {"total_episode": total_episode} + target_episodes = set(cls.__get_best_version_target_episodes( + subscribe, + total_episode=total_episode, + )) + episode_priority = cls.__get_episode_priority( + subscribe, + total_episode=old_total_episode, + ) + filtered_priority = { + str(episode): priority + for episode, priority in episode_priority.items() + if int(episode) in target_episodes + } + subscribe.total_episode = total_episode + subscribe.episode_priority = filtered_priority + current_priority = 0 if not target_episodes else cls.get_best_version_current_priority( + subscribe, + episode_priority=filtered_priority, + ) + subscribe.current_priority = current_priority + update_data["episode_priority"] = filtered_priority + update_data["current_priority"] = current_priority + update_data.update(cls.__prepare_subscribe_progress_fields(subscribe=subscribe, no_exists={})) + return update_data + + @classmethod + def __prepare_total_episode_change_fields( + cls, + subscribe: Subscribe, + total_episode: int, + old_total_episode: int, + ) -> Dict[str, Any]: + """ + 准备已有订阅总集数持久化字段，并同步内存对象上的总集数快照。 + """ + if subscribe.best_version and subscribe.type == MediaType.TV.value: + return cls.__prepare_best_version_total_change_fields( + subscribe=subscribe, + total_episode=total_episode, + old_total_episode=old_total_episode, + ) + + subscribe.total_episode = total_episode + return { + "total_episode": total_episode, + "lack_episode": max( + (subscribe.lack_episode or 0) + (total_episode - old_total_episode), + 0, + ), + } + @classmethod def __is_best_version_complete(cls, subscribe: Subscribe) -> bool: """ @@ -387,9 +452,10 @@ class SubscribeChain(ChainBase): 获取已完成洗版的剧集。 """ episode_priority = cls.__get_episode_priority(subscribe) + target_episodes = set(cls.__get_best_version_target_episodes(subscribe)) return sorted( int(episode) for episode, priority in episode_priority.items() - if str(episode).isdigit() and priority == 100 + if str(episode).isdigit() and int(episode) in target_episodes and priority == 100 ) @classmethod @@ -773,11 +839,13 @@ class SubscribeChain(ChainBase): if not mediainfo.seasons: logger.error(f"媒体信息中没有季集信息，标题：{title}，tmdbid：{tmdbid}，doubanid：{doubanid}") return None, "媒体信息中没有季集信息" - total_episode = len(mediainfo.seasons.get(season) or []) - # 允许外部覆盖按 TMDB 算出的总集数（如待定集数） + current_total_episode = len(mediainfo.seasons.get(season) or []) + # 创建场景没有旧订阅事实，仅允许外部补正未知或扩展总集数。 total_episode = self.__apply_episodes_refresh( - total_episode, season=season, mediainfo=mediainfo, + current_total_episode, season=season, mediainfo=mediainfo, tmdbid=mediainfo.tmdb_id, doubanid=mediainfo.douban_id, scene="create") + if current_total_episode and total_episode < current_total_episode: + total_episode = current_total_episode if not total_episode: logger.error(f'未获取到总集数，标题：{title}，tmdbid：{tmdbid}, doubanid：{doubanid}') return None, f"未获取到第 {season} 季的总集数" @@ -958,11 +1026,13 @@ class SubscribeChain(ChainBase): if not mediainfo.seasons: logger.error(f"媒体信息中没有季集信息，标题：{title}，tmdbid：{tmdbid}，doubanid：{doubanid}") return None, "媒体信息中没有季集信息" - total_episode = len(mediainfo.seasons.get(season) or []) - # 允许外部覆盖按 TMDB 算出的总集数（如待定集数） + current_total_episode = len(mediainfo.seasons.get(season) or []) + # 创建场景没有旧订阅事实，仅允许外部补正未知或扩展总集数。 total_episode = await self.__async_apply_episodes_refresh( - total_episode, season=season, mediainfo=mediainfo, + current_total_episode, season=season, mediainfo=mediainfo, tmdbid=mediainfo.tmdb_id, doubanid=mediainfo.douban_id, scene="create") + if current_total_episode and total_episode < current_total_episode: + total_episode = current_total_episode if not total_episode: logger.error(f'未获取到总集数，标题：{title}，tmdbid：{tmdbid}, doubanid：{doubanid}') return None, f"未获取到第 {season} 季的总集数" @@ -1913,28 +1983,20 @@ class SubscribeChain(ChainBase): # 对于电视剧，获取当前季的总集数 episodes = mediainfo.seasons.get(subscribe.season) or [] progress_update = {} - if not subscribe.manual_total_episode and len(episodes): - total_episode = len(episodes) - # 允许外部覆盖按 TMDB 算出的总集数（如待定集数） + if subscribe.type == MediaType.TV.value and not subscribe.manual_total_episode and len(episodes): + current_total_episode = len(episodes) + # 外部事件只能向上覆盖主程序本次识别到的 TMDB 当前季总集数，已有订阅按最终 total 跟随持久化。 total_episode = self.__apply_episodes_refresh( - total_episode, season=subscribe.season, mediainfo=mediainfo, + current_total_episode, season=subscribe.season, mediainfo=mediainfo, tmdbid=subscribe.tmdbid, doubanid=subscribe.doubanid, subscribe_id=subscribe.id, scene="refresh") - if total_episode > (subscribe.total_episode or 0): - if subscribe.best_version and subscribe.type == MediaType.TV.value: - progress_update = self.__prepare_best_version_total_expansion_fields( - subscribe=subscribe, - total_episode=total_episode, - ) - else: - old_total_episode = subscribe.total_episode or 0 - progress_update = { - "total_episode": total_episode, - "lack_episode": max( - (subscribe.lack_episode or 0) + (total_episode - old_total_episode), - 0, - ), - } + old_total_episode = subscribe.total_episode or 0 + if total_episode and total_episode != old_total_episode: + progress_update = self.__prepare_total_episode_change_fields( + subscribe=subscribe, + total_episode=total_episode, + old_total_episode=old_total_episode, + ) else: total_episode = subscribe.total_episode progress_update = {"lack_episode": subscribe.lack_episode} @@ -3724,11 +3786,11 @@ class SubscribeChain(ChainBase): subscribe_id: Optional[int] = None, scene: Optional[str] = None) -> int: """ - 发送订阅总集数推算事件，允许外部据自身策略覆盖按 TMDB 季集数算出的总集数。 + 发送订阅总集数推算事件，允许外部把主程序本次识别到的 TMDB 当前季总集数向上覆盖。 用途：插件在"待定集数"等场景经事件注入 total_episode 无监听者或外部未覆盖时返回入参原值，保证零行为变更。 - :param current_total: 主程序按 TMDB 季集数算出的默认总集数 + :param current_total: 主程序本次识别到的 TMDB 当前季总集数 :param season: 季号 :return: 最终采用的总集数 """ @@ -3739,6 +3801,7 @@ class SubscribeChain(ChainBase): if event and event.event_data: result: SubscribeEpisodesRefreshEventData = event.event_data if result.updated and result.total_episode: + result.total_episode = max(current_total or 0, result.total_episode) return result.total_episode return current_total @@ -3759,6 +3822,7 @@ class SubscribeChain(ChainBase): if event and event.event_data: result: SubscribeEpisodesRefreshEventData = event.event_data if result.updated and result.total_episode: + result.total_episode = max(current_total or 0, result.total_episode) return result.total_episode return current_total @@ -3773,30 +3837,22 @@ class SubscribeChain(ChainBase): if subscribe.season is None: return - new_total_episode = len((mediainfo.seasons or {}).get(subscribe.season) or []) - # 允许外部覆盖按 TMDB 算出的总集数（如待定集数），后续“只增不减”仍作用于覆盖后的结果，避免误减导致提前完成。 + current_total_episode = len((mediainfo.seasons or {}).get(subscribe.season) or []) + # 外部事件只能向上覆盖主程序本次识别到的 TMDB 当前季总集数，已有订阅回落由主程序跟随本次识别结果持久化。 new_total_episode = self.__apply_episodes_refresh( - new_total_episode, season=subscribe.season, mediainfo=mediainfo, + current_total_episode, season=subscribe.season, mediainfo=mediainfo, tmdbid=subscribe.tmdbid, doubanid=subscribe.doubanid, subscribe_id=subscribe.id, scene="precheck") old_total_episode = subscribe.total_episode or 0 - if not new_total_episode or new_total_episode <= old_total_episode: + if not new_total_episode or new_total_episode == old_total_episode: return now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - if subscribe.best_version and subscribe.type == MediaType.TV.value: - update_data = self.__prepare_best_version_total_expansion_fields( - subscribe=subscribe, - total_episode=new_total_episode, - ) - else: - update_data = { - "total_episode": new_total_episode, - "lack_episode": max( - (subscribe.lack_episode or 0) + (new_total_episode - old_total_episode), - 0, - ), - } + update_data = self.__prepare_total_episode_change_fields( + subscribe=subscribe, + total_episode=new_total_episode, + old_total_episode=old_total_episode, + ) update_data["last_update"] = now SubscribeOper().update(subscribe.id, update_data) for key, value in update_data.items(): diff --git a/app/chain/torrents.py b/app/chain/torrents.py index 05065469..a62fb440 100644 --- a/app/chain/torrents.py +++ b/app/chain/torrents.py @@ -1,3 +1,4 @@ +import copy import re import traceback from typing import Callable, Dict, List, Union, Optional @@ -92,6 +93,238 @@ class TorrentsChain(ChainBase): return torrents_cache + def get_subscribe_cache_candidates( + self, + subscribe, + stype: Optional[str] = None, + allow_title_match: bool = False, + ) -> List[Context]: + """ + 按订阅身份读取 RSS/spider 缓存候选，返回不会回写缓存的 Context 副本。 + + 主程序只提供缓存读取与轻量候选筛选，不在这里判断站点证据能否扩展 + 订阅目标或放行完成；标题兜底候选会显式标记为低置信来源。 + """ + 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 + + @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_tmdbid = cls._normalize_id(getattr(subscribe, "tmdbid", None)) + subscribe_doubanid = cls._normalize_id(getattr(subscribe, "doubanid", None)) + context_tmdbids = cls._context_tmdb_ids(context) + context_doubanids = cls._context_douban_ids(context) + + return bool( + subscribe_tmdbid and subscribe_tmdbid in context_tmdbids + or subscribe_doubanid and subscribe_doubanid in context_doubanids + ) + + @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 + ) + + @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), + tmdb_id=getattr(subscribe, "tmdbid", None), + douban_id=getattr(subscribe, "doubanid", None), + season=getattr(subscribe, "season", None), + ) + + @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 + ) + + @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 + + @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() + + @staticmethod + def _context_has_media_identity(context: Context) -> bool: + """ + 判断候选是否已经带有明确媒体 ID。 + """ + return bool(TorrentsChain._context_tmdb_ids(context) or TorrentsChain._context_douban_ids(context)) + + @staticmethod + def _context_tmdb_ids(context: Context) -> set[str]: + """ + 提取候选已有 TMDB ID，兼容 media_info 与标题显式标签。 + """ + media_info = getattr(context, "media_info", None) + meta_info = getattr(context, "meta_info", None) + return { + value for value in ( + TorrentsChain._normalize_id(getattr(media_info, "tmdb_id", None)), + TorrentsChain._normalize_id(getattr(meta_info, "tmdbid", None)), + ) if value + } + + @staticmethod + def _context_douban_ids(context: Context) -> set[str]: + """ + 提取候选已有豆瓣 ID，兼容 media_info 与标题显式标签。 + """ + media_info = getattr(context, "media_info", None) + meta_info = getattr(context, "meta_info", None) + return { + value for value in ( + TorrentsChain._normalize_id(getattr(media_info, "douban_id", None)), + TorrentsChain._normalize_id(getattr(meta_info, "doubanid", None)), + ) if value + } + + @staticmethod + def _normalize_id(value) -> Optional[str]: + """ + 统一比较媒体 ID，避免 int/string 形态差异影响缓存候选筛选。 + """ + if value is None: + return None + value = str(value).strip() + return value or None + + @staticmethod + def _normalize_int(value) -> Optional[int]: + """ + 将季号等动态字段转为 int，无法解析时视为缺失。 + """ + if value is None: + return None + try: + return int(value) + except (TypeError, ValueError): + return None + + @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 + + @staticmethod + def _normalize_title(value) -> str: + """ + 归一标题用于低置信标题兜底匹配。 + """ + return (StringUtils.clear_upper(value or "") or "").strip() + def clear_torrents(self): """ 清理种子缓存数据 diff --git a/app/schemas/event.py b/app/schemas/event.py index 15c35090..aeaa19ed 100644 --- a/app/schemas/event.py +++ b/app/schemas/event.py @@ -570,7 +570,8 @@ class SubscribeEpisodesRefreshEventData(ChainEventData): """ SubscribeEpisodesRefresh 事件的数据模型 - 主程序在推算订阅某季总集数时发出，携带按 TMDB 季集数算出的默认值，外部可据自身策略覆盖total_episode（如待定集数） + 主程序在推算订阅某季总集数时发出，携带主程序本次识别到的 TMDB 当前季总集数； + 外部可据自身策略向上覆盖 total_episode（如待定集数），低于 current_total_episode 的覆盖值会被主程序钳制。 Attributes: # 输入参数 @@ -578,13 +579,13 @@ class SubscribeEpisodesRefreshEventData(ChainEventData): doubanid (Optional[str]): 豆瓣 ID season (Optional[int]): 季号 mediainfo (Any): 媒体信息 - current_total_episode (int): 主程序按 TMDB 季集数算出的默认总集数 + current_total_episode (int): 主程序本次识别到的 TMDB 当前季总集数 subscribe_id (Optional[int]): 订阅 ID；订阅创建场景下尚未入库，为空 scene (Optional[str]): 触发场景，create/refresh/precheck # 输出参数 updated (bool): 外部是否覆盖了总集数，默认 False - total_episode (Optional[int]): 覆盖后的总集数，仅在 updated=True 时生效 + total_episode (Optional[int]): 覆盖后的总集数，仅在 updated=True 时生效；低于 current_total_episode 时由主程序钳制 source (str): 覆盖来源 reason (str): 覆盖原因 """ @@ -594,13 +595,13 @@ class SubscribeEpisodesRefreshEventData(ChainEventData): doubanid: Optional[str] = Field(default=None, description="豆瓣 ID") season: Optional[int] = Field(default=None, description="季号") mediainfo: Any = Field(default=None, description="媒体信息") - current_total_episode: int = Field(default=0, description="按 TMDB 季集数算出的默认总集数") + current_total_episode: int = Field(default=0, description="主程序本次识别到的 TMDB 当前季总集数") subscribe_id: Optional[int] = Field(default=None, description="订阅 ID；创建场景为空") scene: Optional[str] = Field(default=None, description="触发场景：create/refresh/precheck") # 输出参数 updated: bool = Field(default=False, description="外部是否覆盖了总集数") - total_episode: Optional[int] = Field(default=None, description="覆盖后的总集数") + total_episode: Optional[int] = Field(default=None, description="覆盖后的总集数；低于主程序本次识别到的 TMDB 当前季总集数时由主程序钳制") source: str = Field(default="未知来源", description="覆盖来源") reason: str = Field(default="", description="覆盖原因") diff --git a/tests/test_subscribe_chain.py b/tests/test_subscribe_chain.py index 29d0bcd2..1f87f62d 100644 --- a/tests/test_subscribe_chain.py +++ b/tests/test_subscribe_chain.py @@ -1,3 +1,4 @@ +import asyncio import importlib.util import sys import types @@ -206,6 +207,10 @@ def _load_subscribe_chain_class(): class _SubscribeEpisodesRefreshEventData: def __init__(self, **kwargs): + self.updated = kwargs.get("updated", False) + self.total_episode = kwargs.get("total_episode") + self.source = kwargs.get("source", "未知来源") + self.reason = kwargs.get("reason", "") self.__dict__.update(kwargs) class _SubscribeCompletionCheckEventData: @@ -2423,17 +2428,84 @@ class SubscribeProgressEntrypointTest(TestCase): class SubscribeProgressConsolidationTest(TestCase): def _mediainfo(self, total_episode=5): return SimpleNamespace( + type=MediaType.TV, seasons={1: [object() for _ in range(total_episode)]}, title="总集增长剧", + title_year="总集增长剧 (2026)", year="2026", + tmdb_id=31000, + douban_id=None, + bangumi_id=None, vote_average=9.5, overview="overview", imdb_id="tt1234567", tvdb_id=99, get_poster_image=lambda: "poster", get_backdrop_image=lambda: "backdrop", + get_message_image=lambda: "message-image", + to_dict=lambda: {}, ) + @staticmethod + def _event_manager(total_episode=None, *, updated=True): + captured = [] + + def _apply(event_type, event_data): + captured.append((event_type, event_data)) + if hasattr(event_data, "current_total_episode"): + event_data.updated = updated + event_data.total_episode = total_episode + event_data.source = "unit" + event_data.reason = "unit" + return SimpleNamespace(event_data=event_data) + + class _EventManager: + def send_event(self, event_type, event_data): + return _apply(event_type, event_data) + + async def async_send_event(self, event_type, event_data): + return _apply(event_type, event_data) + + return _EventManager(), captured + + def test_apply_episodes_refresh_clamps_external_total_to_current_total(self): + module, SubscribeChain = _load_subscribe_chain_class() + eventmanager, captured = self._event_manager(5) + + with patch.object(module, "eventmanager", eventmanager): + result = SubscribeChain._SubscribeChain__apply_episodes_refresh( + 10, + season=1, + mediainfo=self._mediainfo(total_episode=10), + tmdbid=31030, + doubanid=None, + subscribe_id=31, + scene="precheck", + ) + + self.assertEqual(result, 10) + self.assertEqual(captured[0][1].current_total_episode, 10) + self.assertEqual(captured[0][1].total_episode, 10) + + def test_async_apply_episodes_refresh_clamps_external_total_to_current_total(self): + module, SubscribeChain = _load_subscribe_chain_class() + eventmanager, captured = self._event_manager(5) + + with patch.object(module, "eventmanager", eventmanager): + result = asyncio.run(SubscribeChain._SubscribeChain__async_apply_episodes_refresh( + 10, + season=1, + mediainfo=self._mediainfo(total_episode=10), + tmdbid=31030, + doubanid=None, + subscribe_id=31, + scene="precheck", + )) + + self.assertEqual(result, 10) + self.assertEqual(captured[0][1].current_total_episode, 10) + self.assertEqual(captured[0][1].total_episode, 10) + def test_refresh_total_episode_before_completion_reuses_progress_priority_snapshot(self): module, SubscribeChain = _load_subscribe_chain_class() subscribe = module.Subscribe( @@ -2476,6 +2548,317 @@ class SubscribeProgressConsolidationTest(TestCase): self.assertEqual(updates[-1][1]["lack_episode"], 2) self.assertEqual(updates[-1][1]["current_priority"], 0) + def test_refresh_total_episode_before_completion_follows_recognized_total_decrease(self): + module, SubscribeChain = _load_subscribe_chain_class() + subscribe = module.Subscribe( + id=34, + name="总集回落剧", + type=MediaType.TV.value, + season=1, + total_episode=100, + start_episode=1, + lack_episode=100, + best_version=0, + best_version_full=0, + current_priority=None, + episode_priority={}, + note=[], + tmdbid=31034, + doubanid=None, + manual_total_episode=0, + ) + updates = [] + eventmanager, captured = self._event_manager(updated=False) + + class _SubscribeOper: + def update(self, subscribe_id, payload): + updates.append((subscribe_id, payload)) + + with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + module, + "eventmanager", + eventmanager, + ): + SubscribeChain()._SubscribeChain__refresh_total_episode_before_completion( + subscribe, + self._mediainfo(total_episode=90), + ) + + self.assertEqual(captured[0][1].current_total_episode, 90) + self.assertEqual(subscribe.total_episode, 90) + self.assertEqual(subscribe.lack_episode, 90) + self.assertEqual(updates[-1][1]["total_episode"], 90) + self.assertEqual(updates[-1][1]["lack_episode"], 90) + + def test_refresh_total_episode_before_completion_filters_best_version_priority_on_decrease(self): + module, SubscribeChain = _load_subscribe_chain_class() + subscribe = module.Subscribe( + id=40, + name="洗版回落剧", + type=MediaType.TV.value, + season=1, + total_episode=100, + start_episode=1, + lack_episode=0, + best_version=1, + best_version_full=0, + current_priority=100, + episode_priority={str(episode): 100 for episode in range(1, 101)}, + note=[], + tmdbid=31040, + doubanid=None, + manual_total_episode=0, + ) + updates = [] + eventmanager, _ = self._event_manager(updated=False) + + class _SubscribeOper: + def update(self, subscribe_id, payload): + updates.append((subscribe_id, payload)) + + with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + module, + "eventmanager", + eventmanager, + ): + SubscribeChain()._SubscribeChain__refresh_total_episode_before_completion( + subscribe, + self._mediainfo(total_episode=10), + ) + + expected_priority = {str(episode): 100 for episode in range(1, 11)} + payload = updates[-1][1] + self.assertEqual(subscribe.total_episode, 10) + self.assertEqual(subscribe.episode_priority, expected_priority) + self.assertEqual(payload["episode_priority"], expected_priority) + self.assertEqual(subscribe.lack_episode, 0) + self.assertEqual(subscribe.current_priority, 100) + self.assertEqual( + SubscribeChain._SubscribeChain__get_best_version_completed_episodes(subscribe), + list(range(1, 11)), + ) + + def test_refresh_total_episode_before_completion_resets_legacy_current_priority_when_filtered_empty(self): + module, SubscribeChain = _load_subscribe_chain_class() + subscribe = module.Subscribe( + id=42, + name="洗版空优先级回落剧", + type=MediaType.TV.value, + season=1, + total_episode=100, + start_episode=1, + lack_episode=0, + best_version=1, + best_version_full=0, + current_priority=100, + episode_priority={str(episode): 100 for episode in range(11, 101)}, + note=[], + tmdbid=31042, + doubanid=None, + manual_total_episode=0, + ) + updates = [] + eventmanager, _ = self._event_manager(updated=False) + + class _SubscribeOper: + def update(self, subscribe_id, payload): + updates.append((subscribe_id, payload)) + + with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + module, + "eventmanager", + eventmanager, + ): + SubscribeChain()._SubscribeChain__refresh_total_episode_before_completion( + subscribe, + self._mediainfo(total_episode=10), + ) + + payload = updates[-1][1] + self.assertEqual(subscribe.total_episode, 10) + self.assertEqual(subscribe.episode_priority, {}) + self.assertEqual(subscribe.current_priority, 0) + self.assertEqual(subscribe.lack_episode, 10) + self.assertEqual(payload["episode_priority"], {}) + self.assertEqual(payload["current_priority"], 0) + self.assertEqual(payload["lack_episode"], 10) + self.assertEqual( + SubscribeChain._SubscribeChain__get_best_version_completed_episodes(subscribe), + [], + ) + + def test_refresh_total_episode_before_completion_resets_priority_when_target_range_empty(self): + module, SubscribeChain = _load_subscribe_chain_class() + subscribe = module.Subscribe( + id=44, + name="洗版目标范围为空回落剧", + type=MediaType.TV.value, + season=1, + total_episode=100, + start_episode=11, + lack_episode=0, + best_version=1, + best_version_full=0, + current_priority=100, + episode_priority={str(episode): 100 for episode in range(11, 101)}, + note=[], + tmdbid=31044, + doubanid=None, + manual_total_episode=0, + ) + updates = [] + eventmanager, _ = self._event_manager(updated=False) + + class _SubscribeOper: + def update(self, subscribe_id, payload): + updates.append((subscribe_id, payload)) + + with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + module, + "eventmanager", + eventmanager, + ): + SubscribeChain()._SubscribeChain__refresh_total_episode_before_completion( + subscribe, + self._mediainfo(total_episode=10), + ) + + payload = updates[-1][1] + self.assertEqual(subscribe.total_episode, 10) + self.assertEqual(subscribe.episode_priority, {}) + self.assertEqual(subscribe.current_priority, 0) + self.assertEqual(subscribe.lack_episode, 0) + self.assertEqual(payload["episode_priority"], {}) + self.assertEqual(payload["current_priority"], 0) + self.assertEqual(payload["lack_episode"], 0) + self.assertEqual( + SubscribeChain._SubscribeChain__get_best_version_completed_episodes(subscribe), + [], + ) + + def test_refresh_total_episode_before_completion_clamps_lower_event_total_to_recognized_total(self): + module, SubscribeChain = _load_subscribe_chain_class() + subscribe = module.Subscribe( + id=35, + name="总集事件压低剧", + type=MediaType.TV.value, + season=1, + total_episode=100, + start_episode=1, + lack_episode=100, + best_version=0, + best_version_full=0, + current_priority=None, + episode_priority={}, + note=[], + tmdbid=31035, + doubanid=None, + manual_total_episode=0, + ) + updates = [] + eventmanager, _ = self._event_manager(9) + + class _SubscribeOper: + def update(self, subscribe_id, payload): + updates.append((subscribe_id, payload)) + + with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + module, + "eventmanager", + eventmanager, + ): + SubscribeChain()._SubscribeChain__refresh_total_episode_before_completion( + subscribe, + self._mediainfo(total_episode=10), + ) + + self.assertEqual(subscribe.total_episode, 10) + self.assertEqual(subscribe.lack_episode, 10) + self.assertEqual(updates[-1][1]["total_episode"], 10) + self.assertEqual(updates[-1][1]["lack_episode"], 10) + + def test_refresh_total_episode_before_completion_rejects_manual_total_decrease(self): + module, SubscribeChain = _load_subscribe_chain_class() + subscribe = module.Subscribe( + id=37, + name="手动总集数剧", + type=MediaType.TV.value, + season=1, + total_episode=100, + start_episode=1, + lack_episode=100, + best_version=0, + best_version_full=0, + current_priority=None, + episode_priority={}, + note=[], + tmdbid=31037, + doubanid=None, + manual_total_episode=1, + ) + + class _EventManager: + def send_event(self, *_args, **_kwargs): + raise AssertionError("manual total episode must not ask external refresh") + + class _SubscribeOper: + def update(self, *_args, **_kwargs): + raise AssertionError("manual total episode must not be updated") + + with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + module, + "eventmanager", + _EventManager(), + ): + SubscribeChain()._SubscribeChain__refresh_total_episode_before_completion( + subscribe, + self._mediainfo(total_episode=10), + ) + + self.assertEqual(subscribe.total_episode, 100) + self.assertEqual(subscribe.lack_episode, 100) + + def test_refresh_total_episode_before_completion_rejects_non_tv_decrease(self): + module, SubscribeChain = _load_subscribe_chain_class() + subscribe = module.Subscribe( + id=38, + name="非电视剧", + type=MediaType.MOVIE.value, + season=1, + total_episode=100, + start_episode=1, + lack_episode=100, + best_version=0, + best_version_full=0, + current_priority=None, + episode_priority={}, + note=[], + tmdbid=31038, + doubanid=None, + manual_total_episode=0, + ) + + class _EventManager: + def send_event(self, *_args, **_kwargs): + raise AssertionError("non-tv subscribe must not ask external refresh") + + class _SubscribeOper: + def update(self, *_args, **_kwargs): + raise AssertionError("non-tv subscribe must not be updated") + + with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + module, + "eventmanager", + _EventManager(), + ): + SubscribeChain()._SubscribeChain__refresh_total_episode_before_completion( + subscribe, + self._mediainfo(total_episode=10), + ) + + self.assertEqual(subscribe.total_episode, 100) + self.assertEqual(subscribe.lack_episode, 100) + def test_check_total_growth_reuses_progress_priority_snapshot(self): module, SubscribeChain = _load_subscribe_chain_class() subscribe = module.Subscribe( @@ -2521,6 +2904,182 @@ class SubscribeProgressConsolidationTest(TestCase): self.assertEqual(payload["lack_episode"], 2) self.assertEqual(payload["current_priority"], 0) + def test_check_total_growth_still_uses_larger_event_total(self): + module, SubscribeChain = _load_subscribe_chain_class() + subscribe = module.Subscribe( + id=39, + name="总集事件增长剧", + type=MediaType.TV.value, + season=1, + total_episode=100, + start_episode=1, + lack_episode=100, + best_version=0, + best_version_full=0, + current_priority=None, + episode_priority={}, + note=[], + year="2026", + episode_group=None, + tmdbid=31039, + doubanid=None, + manual_total_episode=0, + ) + updates = [] + eventmanager, captured = self._event_manager(120) + + class _SubscribeOper: + def list(self): + return [subscribe] + + def update(self, subscribe_id, payload): + updates.append((subscribe_id, payload)) + + chain = SubscribeChain() + chain.recognize_media = lambda **kwargs: self._mediainfo(total_episode=10) + + with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + module, + "eventmanager", + eventmanager, + ): + chain.check() + + payload = updates[-1][1] + self.assertEqual(captured[0][1].current_total_episode, 10) + self.assertEqual(payload["total_episode"], 120) + self.assertEqual(payload["lack_episode"], 120) + + def test_check_total_refresh_follows_recognized_total_decrease(self): + module, SubscribeChain = _load_subscribe_chain_class() + subscribe = module.Subscribe( + id=43, + name="总集巡检回落剧", + type=MediaType.TV.value, + season=1, + total_episode=100, + start_episode=1, + lack_episode=100, + best_version=0, + best_version_full=0, + current_priority=None, + episode_priority={}, + note=[], + year="2026", + episode_group=None, + tmdbid=31043, + doubanid=None, + manual_total_episode=0, + ) + updates = [] + eventmanager, captured = self._event_manager(updated=False) + + class _SubscribeOper: + def list(self): + return [subscribe] + + def update(self, subscribe_id, payload): + updates.append((subscribe_id, payload)) + + chain = SubscribeChain() + chain.recognize_media = lambda **kwargs: self._mediainfo(total_episode=90) + + with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + module, + "eventmanager", + eventmanager, + ): + chain.check() + + payload = updates[-1][1] + self.assertEqual(captured[0][1].current_total_episode, 90) + self.assertEqual(payload["total_episode"], 90) + self.assertEqual(payload["lack_episode"], 90) + + def test_check_total_refresh_skips_non_tv_even_when_mediainfo_has_seasons(self): + module, SubscribeChain = _load_subscribe_chain_class() + subscribe = module.Subscribe( + id=45, + name="电影误带季集", + type=MediaType.MOVIE.value, + season=1, + total_episode=100, + start_episode=1, + lack_episode=100, + best_version=0, + best_version_full=0, + current_priority=None, + episode_priority={}, + note=[], + year="2026", + episode_group=None, + tmdbid=31045, + doubanid=None, + manual_total_episode=0, + ) + updates = [] + mediainfo = self._mediainfo(total_episode=10) + mediainfo.type = MediaType.MOVIE + + class _SubscribeOper: + def list(self): + return [subscribe] + + def update(self, subscribe_id, payload): + updates.append((subscribe_id, payload)) + + class _EventManager: + def send_event(self, *_args, **_kwargs): + raise AssertionError("non-tv subscribe must not ask external refresh") + + chain = SubscribeChain() + chain.recognize_media = lambda **kwargs: mediainfo + + with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + module, + "eventmanager", + _EventManager(), + ): + chain.check() + + self.assertEqual(updates[-1][1]["total_episode"], 100) + self.assertEqual(updates[-1][1]["lack_episode"], 100) + + def test_add_create_clamps_event_decrease_to_recognized_total(self): + module, SubscribeChain = _load_subscribe_chain_class() + added = [] + eventmanager, captured = self._event_manager(5) + mediainfo = self._mediainfo(total_episode=10) + chain = SubscribeChain() + chain.recognize_media = lambda **_kwargs: mediainfo + chain.obtain_images = lambda **_kwargs: None + + class _SubscribeOper: + def add(self, **kwargs): + added.append(kwargs) + return 41, None + + with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()), patch.object( + module, + "eventmanager", + eventmanager, + ): + sid, err_msg = chain.add( + title="总集创建剧", + year="2026", + mtype=MediaType.TV, + tmdbid=31041, + season=1, + message=False, + ) + + self.assertEqual(sid, 41) + self.assertIsNone(err_msg) + self.assertEqual(captured[0][1].scene, "create") + self.assertEqual(captured[0][1].current_total_episode, 10) + self.assertEqual(added[-1]["total_episode"], 10) + self.assertEqual(added[-1]["lack_episode"], 10) + def test_completed_episode_uses_schema_function_directly_for_best_version(self): module, SubscribeChain = _load_subscribe_chain_class() values = { diff --git a/tests/test_torrents_subscribe_candidates.py b/tests/test_torrents_subscribe_candidates.py new file mode 100644 index 00000000..9254df09 --- /dev/null +++ b/tests/test_torrents_subscribe_candidates.py @@ -0,0 +1,181 @@ +from types import SimpleNamespace + +from app.chain.torrents import TorrentsChain +from app.core.context import Context, MediaInfo, TorrentInfo +from app.schemas.types import MediaType + + +def _chain() -> TorrentsChain: + """构造不触发外部依赖初始化的种子链实例。""" + return object.__new__(TorrentsChain) + + +def _subscribe(**kwargs): + defaults = { + "tmdbid": 100, + "doubanid": None, + "season": 1, + "name": "测试剧", + "type": MediaType.TV.value, + } + defaults.update(kwargs) + return SimpleNamespace(**defaults) + + +def _ctx( + title: str = "测试剧 S01E05", + *, + tmdb_id: int = 100, + douban_id: str = None, + meta_tmdbid: int = None, + meta_doubanid: str = None, + meta_type=MediaType.TV, + media_season: int = None, + begin_season: int = 1, + end_season: int = None, +) -> Context: + return Context( + meta_info=SimpleNamespace( + title=title, + name="测试剧", + type=meta_type, + tmdbid=meta_tmdbid, + doubanid=meta_doubanid, + begin_season=begin_season, + end_season=end_season, + begin_episode=5, + episode_list=[5], + ), + media_info=MediaInfo( + type=MediaType.TV, + tmdb_id=tmdb_id, + douban_id=douban_id, + season=media_season, + ), + torrent_info=TorrentInfo(title=title), + resource_source="rss", + match_source="tmdbid" if tmdb_id else "unknown", + candidate_recognized=bool(tmdb_id or douban_id), + media_info_is_target=False, + ) + + +def test_cache_candidates_return_deep_copies(monkeypatch): + source = _ctx() + monkeypatch.setattr(TorrentsChain, "get_torrents", lambda self, stype=None: {"site": [source]}) + + result = _chain().get_subscribe_cache_candidates(_subscribe(), stype="rss") + result[0].meta_info.title = "changed" + result[0].media_info.tmdb_id = 999 + + assert result[0] is not source + assert result[0].meta_info is not source.meta_info + assert result[0].media_info is not source.media_info + assert source.meta_info.title == "测试剧 S01E05" + assert source.media_info.tmdb_id == 100 + + +def test_cache_candidates_reject_season_conflict(monkeypatch): + source = _ctx(media_season=2, begin_season=2) + monkeypatch.setattr(TorrentsChain, "get_torrents", lambda self, stype=None: {"site": [source]}) + + assert _chain().get_subscribe_cache_candidates(_subscribe(), stype="rss") == [] + + +def test_cache_candidates_keep_multi_season_candidate_covering_target(monkeypatch): + source = _ctx(media_season=None, begin_season=1, end_season=2) + monkeypatch.setattr(TorrentsChain, "get_torrents", lambda self, stype=None: {"site": [source]}) + + result = _chain().get_subscribe_cache_candidates(_subscribe(), stype="rss") + + assert len(result) == 1 + assert result[0].match_source == "tmdbid" + + +def test_cache_candidates_keep_multi_season_candidate_when_media_season_is_range_start(monkeypatch): + source = _ctx(media_season=1, begin_season=1, end_season=2) + monkeypatch.setattr(TorrentsChain, "get_torrents", lambda self, stype=None: {"site": [source]}) + + result = _chain().get_subscribe_cache_candidates(_subscribe(season=2), stype="rss") + + assert len(result) == 1 + assert result[0].match_source == "tmdbid" + + +def test_cache_candidates_ignore_default_meta_season_list_when_no_explicit_meta_season(monkeypatch): + class _MetaWithDefaultSeasonList: + title = "测试剧 E05" + name = "测试剧" + type = MediaType.TV + begin_season = None + end_season = None + + @property + def season_list(self): + return [1] + + source = _ctx(media_season=2, begin_season=None) + source.meta_info = _MetaWithDefaultSeasonList() + monkeypatch.setattr(TorrentsChain, "get_torrents", lambda self, stype=None: {"site": [source]}) + + result = _chain().get_subscribe_cache_candidates(_subscribe(season=2), stype="rss") + + assert len(result) == 1 + assert result[0].match_source == "tmdbid" + + +def test_title_fallback_requires_explicit_flag(monkeypatch): + source = _ctx(tmdb_id=None) + monkeypatch.setattr(TorrentsChain, "get_torrents", lambda self, stype=None: {"site": [source]}) + + assert _chain().get_subscribe_cache_candidates(_subscribe(), stype="rss") == [] + + +def test_title_fallback_is_diagnostic_only_and_uses_target_media(monkeypatch): + source = _ctx(tmdb_id=None) + monkeypatch.setattr(TorrentsChain, "get_torrents", lambda self, stype=None: {"site": [source]}) + + result = _chain().get_subscribe_cache_candidates(_subscribe(doubanid="200"), stype="rss", allow_title_match=True) + + assert len(result) == 1 + assert result[0].match_source == "title" + assert result[0].candidate_recognized is False + assert result[0].media_info_is_target is True + assert result[0].media_info.tmdb_id == 100 + assert result[0].media_info.douban_id == "200" + assert source.media_info.tmdb_id is None + + +def test_title_fallback_rejects_meta_type_conflict(monkeypatch): + source = _ctx(tmdb_id=None, meta_type=MediaType.MOVIE) + source.media_info.type = None + monkeypatch.setattr(TorrentsChain, "get_torrents", lambda self, stype=None: {"site": [source]}) + + assert _chain().get_subscribe_cache_candidates( + _subscribe(), + stype="rss", + allow_title_match=True, + ) == [] + + +def test_title_fallback_rejects_explicit_conflicting_identity(monkeypatch): + source = _ctx(tmdb_id=999) + source.match_source = "tmdbid" + monkeypatch.setattr(TorrentsChain, "get_torrents", lambda self, stype=None: {"site": [source]}) + + assert _chain().get_subscribe_cache_candidates( + _subscribe(), + stype="rss", + allow_title_match=True, + ) == [] + + +def test_title_fallback_rejects_meta_explicit_conflicting_identity(monkeypatch): + source = _ctx(tmdb_id=None, meta_tmdbid=999) + monkeypatch.setattr(TorrentsChain, "get_torrents", lambda self, stype=None: {"site": [source]}) + + assert _chain().get_subscribe_cache_candidates( + _subscribe(), + stype="rss", + allow_title_match=True, + ) == []