feat(subscribe): expose cached site candidates (#6062)

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():

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):
         """
         清理种子缓存数据

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="覆盖原因")
 

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 = {

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,
+    ) == []