From 36fb82b7aa84fec69993b7bb505a80260f9ac04a Mon Sep 17 00:00:00 2001
From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com>
Date: Mon, 29 Jun 2026 06:43:51 +0800
Subject: [PATCH 1/2] feat(subscribe): record episode download facts (#6015)

---
 app/chain/download.py         |   13 +-
 app/chain/subscribe.py        |  686 ++++++++++++++-------
 app/core/context.py           |    3 +
 app/schemas/context.py        |    2 +
 app/schemas/subscribe.py      |   54 +-
 docs/subscribe-lifecycle.md   |   85 +++
 tests/test_download_chain.py  |   73 +++
 tests/test_subscribe_chain.py | 1053 ++++++++++++++++++++++++++++++---
 8 files changed, 1671 insertions(+), 298 deletions(-)
 create mode 100644 docs/subscribe-lifecycle.md

diff --git a/app/chain/download.py b/app/chain/download.py
index c0c6a651..fe697401 100644
--- a/app/chain/download.py
+++ b/app/chain/download.py
@@ -941,6 +941,7 @@ class DownloadChain(ChainBase):
                             continue
                         # 种子季是需要季或者子集
                         if set(torrent_season).issubset(set(need_season)):
+                            complete_coverage_matched = False
                             if len(torrent_season) == 1:
                                 # 只有一季的可能是命名错误，需要打开种子鉴别，只有实际集数大于等于总集数才下载
                                 logger.info(f"开始下载种子 {torrent.title} ...")
@@ -965,7 +966,13 @@ class DownloadChain(ChainBase):
                                 required_episodes = __get_required_episodes(need_mid, torrent_season[0]) \
                                     if __requires_complete_coverage(need_tv_info) else set()
                                 need_total = __get_season_episodes(need_mid, torrent_season[0])
-                                if required_episodes and not required_episodes.issubset(torrent_episodes_set):
+                                complete_coverage_matched = bool(required_episodes) \
+                                    and required_episodes.issubset(torrent_episodes_set)
+                                if complete_coverage_matched:
+                                    logger.info(
+                                        f"{meta.org_string} 解析文件集数已完整覆盖目标范围："
+                                        f"{StringUtils.format_ep(sorted(required_episodes))}")
+                                if required_episodes and not complete_coverage_matched:
                                     missing_episodes = sorted(required_episodes.difference(torrent_episodes_set))
                                     logger.info(
                                         f"{meta.org_string} 解析文件集数未覆盖目标范围，"
@@ -998,6 +1005,8 @@ class DownloadChain(ChainBase):
 
                             if download_id:
                                 # 下载成功
+                                if complete_coverage_matched:
+                                    context.confirmed_full_coverage = True
                                 logger.info(f"{torrent.title} 添加下载成功")
                                 downloaded_list.append(context)
                                 # 更新仍需季集
@@ -1079,6 +1088,8 @@ class DownloadChain(ChainBase):
                                                                    downloader=downloader)
                                 if download_id:
                                     # 下载成功
+                                    if __requires_complete_coverage(tv):
+                                        context.confirmed_full_coverage = True
                                     logger.info(f"{meta.title} 添加下载成功")
                                     downloaded_list.append(context)
                                     # 更新仍需集数
diff --git a/app/chain/subscribe.py b/app/chain/subscribe.py
index ffb222a8..4d2c04e8 100644
--- a/app/chain/subscribe.py
+++ b/app/chain/subscribe.py
@@ -58,7 +58,18 @@ def build_subscribe_meta(subscribe: Subscribe) -> MetaBase:
 
 class SubscribeChain(ChainBase):
     """
-    订阅管理处理链
+    订阅管理处理链。
+
+    订阅链路同时服务电影、普通电视剧、分集洗版和全集洗版。普通电视剧订阅与
+    分集洗版共享按集事实：note 表示目标集已经存在或已经下载，
+    episode_priority 表示每集已知下载质量；二者可以互相切换。全集洗版关注
+    完整目标范围的整体质量，只有下载层确认整包完整覆盖目标范围后，才把资源
+    写成目标范围内的按集事实。
+
+    实现上保持三个入口分离：下载事实入口只写 note / episode_priority；
+    progress 刷新入口只把当前事实计算为 lack_episode 和电视剧洗版
+    current_priority；完成入口只根据最终事实和完成策略收敛订阅状态。电影没有
+    按集事实，电影洗版的 current_priority 由电影下载优先级 writer 单独维护。
     """
 
     _rlock = threading.RLock()
@@ -150,7 +161,7 @@ class SubscribeChain(ChainBase):
                 continue
             if episode_number in target_episodes:
                 downloaded.add(episode_number)
-        for episode, priority in (subscribe.episode_priority or {}).items():
+        for episode, priority in cls.__get_episode_priority(subscribe, total_episode=total_episode).items():
             if not str(episode).isdigit():
                 continue
             try:
@@ -191,42 +202,60 @@ class SubscribeChain(ChainBase):
         return cls.__get_pending_best_version_episodes_with_priority(subscribe, total_episode=total_episode)
 
     @classmethod
-    def compute_completed_episode(cls, subscribe: Subscribe) -> Optional[int]:
+    def compute_lack_episode(
+            cls,
+            subscribe: Subscribe,
+            no_exists: Optional[Dict[Union[int, str], Dict[int, schemas.NotExistMediaInfo]]] = None,
+    ) -> int:
         """
-        计算订阅"已完成"集数派生值，仅用于响应填充，不入库。
+        计算订阅范围内尚未下载到任何版本的集数。
 
-        语义：
-        - 普通订阅 (best_version=0)：``max(total_episode - lack_episode, 0)``，即媒体库已入库集数。
-        - 洗版订阅 (best_version=1，含分集与全集洗版)：
-          ``(start_episode - 1) + (episode_priority 中 priority==100 且 ep ∈ [start, total] 的命中数)``。
-          start_episode 之前的集不在订阅范围内，视为"逻辑上已完成"，与主文案分母 total_episode 对齐。
-
-        - 入参：完整 Subscribe ORM/Schema 对象，需至少包含 best_version、type、start_episode、
-          total_episode、lack_episode、episode_priority 字段。
-        - 返回：完成集数；电影或缺少 total_episode 时返回 None。
+        普通电视剧订阅以媒体库缺失结果为准；调用方没有缺失结果时按空缺失处理，
+        避免入口级刷新失败把未知状态写成异常。洗版电视剧订阅按 note 与
+        episode_priority>0 判断是否已有任意版本落点，priority<100 仍表示已下载过任意版本。
         """
-        total_episode = subscribe.total_episode or 0
-        if subscribe.type != MediaType.TV.value or not total_episode:
-            return None
-
-        start_episode = subscribe.start_episode or 1
+        if subscribe.type != MediaType.TV.value:
+            return 0
 
         if not subscribe.best_version:
-            lack = subscribe.lack_episode or 0
-            return max(total_episode - lack, 0)
+            no_exists = no_exists or {}
+            mediakey = subscribe.tmdbid or subscribe.doubanid
+            left_seasons = no_exists.get(mediakey) or {}
+            for season_info in left_seasons.values():
+                if season_info.season != subscribe.season:
+                    continue
+                left_episodes = season_info.episodes
+                if not left_episodes:
+                    return season_info.total_episode or 0
+                return len(left_episodes)
+            return 0
 
-        # 洗版口径：start 之前的集视为已完成 + 范围内 priority==100 命中。
-        # ``start_episode > total_episode`` 是异常配置，需把"起始集前"偏移截断到 total，
-        # 避免 completed 越过分母 total_episode。
-        episode_priority = subscribe.episode_priority or {}
-        priority_completed = sum(
-            1
-            for ep_key, priority in episode_priority.items()
-            if str(ep_key).isdigit()
-            and start_episode <= int(ep_key) <= total_episode
-            and priority == 100
-        )
-        return min(max(start_episode - 1, 0), total_episode) + priority_completed
+        total_episode = subscribe.total_episode or 0
+        if not total_episode:
+            return 0
+        start_episode = subscribe.start_episode or 1
+        if total_episode < start_episode:
+            return 0
+
+        target_episodes = set(range(start_episode, total_episode + 1))
+        downloaded: set = set()
+        for episode in subscribe.note or []:
+            try:
+                episode_number = int(episode)
+            except (TypeError, ValueError):
+                continue
+            if episode_number in target_episodes:
+                downloaded.add(episode_number)
+        for episode, priority in cls.__get_episode_priority(subscribe).items():
+            try:
+                if float(priority) <= 0:
+                    continue
+                episode_number = int(episode)
+            except (TypeError, ValueError):
+                continue
+            if episode_number in target_episodes:
+                downloaded.add(episode_number)
+        return len(target_episodes - downloaded)
 
     @classmethod
     def get_best_version_current_priority(
@@ -240,19 +269,54 @@ class SubscribeChain(ChainBase):
         if not subscribe.best_version or subscribe.type != MediaType.TV.value:
             return subscribe.current_priority or 0
 
-        pending_episodes = cls.__get_pending_best_version_episodes_with_priority(subscribe, episode_priority)
-        if not pending_episodes:
-            return 100
+        target_episodes = cls.__get_best_version_target_episodes(subscribe)
+        if not target_episodes:
+            return subscribe.current_priority or 0
 
         if episode_priority is None:
             normalized = cls.__get_episode_priority(subscribe)
         else:
             normalized = cls.__normalize_episode_priority(episode_priority)
-        return max(
-            (normalized.get(str(episode), 0) for episode in pending_episodes),
+        return min(
+            (normalized.get(str(episode), 0) for episode in target_episodes),
             default=0,
         )
 
+    @classmethod
+    def __prepare_best_version_total_expansion_fields(
+            cls,
+            subscribe: Subscribe,
+            total_episode: int,
+    ) -> Dict[str, Any]:
+        """
+        准备洗版电视剧总集数扩展后需要写库的字段。
+
+        该方法会同步传入对象上的 total_episode / episode_priority，方便同一链路后续
+        按最终快照继续计算进度；实际数据库写入由调用方统一执行。
+        """
+        update_data: Dict[str, Any] = {"total_episode": total_episode}
+        old_total_episode = subscribe.total_episode or 0
+        subscribe.total_episode = total_episode
+
+        if subscribe.best_version and subscribe.type == MediaType.TV.value:
+            episode_priority = cls.__get_episode_priority(
+                subscribe,
+                total_episode=old_total_episode,
+            )
+            if not episode_priority and subscribe.current_priority is not None:
+                episode_priority = {
+                    str(episode): int(subscribe.current_priority)
+                    for episode in cls.__get_best_version_target_episodes(
+                        subscribe,
+                        total_episode=old_total_episode,
+                    )
+                }
+            subscribe.episode_priority = episode_priority
+            update_data["episode_priority"] = episode_priority
+
+        update_data.update(cls.__prepare_subscribe_progress_fields(subscribe=subscribe, no_exists={}))
+        return update_data
+
     @classmethod
     def __is_best_version_complete(cls, subscribe: Subscribe) -> bool:
         """
@@ -414,7 +478,7 @@ class SubscribeChain(ChainBase):
     @classmethod
     def __is_full_season_priority_higher_than_all_targets(cls, subscribe: Subscribe, priority: int) -> bool:
         """
-        判断整季资源优先级是否高于订阅目标范围内所有分集。
+        判断整季资源优先级是否高于订阅目标范围的整体优先级门槛。
         """
         if subscribe.type != MediaType.TV.value:
             return False
@@ -428,12 +492,13 @@ class SubscribeChain(ChainBase):
         except (TypeError, ValueError):
             resource_priority = 0
 
-        episode_priority = cls.__get_episode_priority(subscribe)
-        for episode in target_episodes:
-            current_priority = episode_priority.get(str(episode), 0)
-            if resource_priority <= current_priority:
-                return False
-        return True
+        try:
+            current_priority = int(subscribe.current_priority) if subscribe.current_priority is not None \
+                else cls.get_best_version_current_priority(subscribe)
+        except (TypeError, ValueError):
+            current_priority = 0
+
+        return resource_priority > current_priority
 
     @classmethod
     def __build_full_pack_first_no_exists(
@@ -462,6 +527,7 @@ class SubscribeChain(ChainBase):
                     episodes=[],
                     total_episode=subscribe.total_episode,
                     start_episode=subscribe.start_episode or 1,
+                    require_complete_coverage=True,
                 )
             }
         }
@@ -486,16 +552,30 @@ class SubscribeChain(ChainBase):
             if context.media_info.type == MediaType.TV
                and self.__is_full_season_resource(meta=context.meta_info, subscribe=subscribe)
         ] if full_pack_no_exists else []
-        full_pack_contexts = [
-            context for context in full_season_contexts
-            if self.__is_full_season_priority_higher_than_all_targets(
+        target_episodes = self.__get_best_version_target_episodes(subscribe)
+        target_range = f"{target_episodes[0]}-{target_episodes[-1]}" if target_episodes else "empty"
+        try:
+            current_priority_gate = int(subscribe.current_priority) if subscribe.current_priority is not None \
+                else self.get_best_version_current_priority(subscribe)
+        except (TypeError, ValueError):
+            current_priority_gate = 0
+        full_pack_contexts = []
+        for context in full_season_contexts:
+            candidate_priority = getattr(context.torrent_info, "pri_order", 0)
+            accepted = self.__is_full_season_priority_higher_than_all_targets(
                 subscribe=subscribe,
-                priority=context.torrent_info.pri_order,
+                priority=candidate_priority,
             )
-        ]
+            logger.info(
+                f"{subscribe.name} 整包候选优先级判断：candidate_priority={candidate_priority}，"
+                f"current_priority={current_priority_gate}，target_range={target_range}，"
+                f"decision={'accept' if accepted else 'reject'}"
+            )
+            if accepted:
+                full_pack_contexts.append(context)
 
         if full_season_contexts and not full_pack_contexts:
-            logger.info(f"{subscribe.name} 全集候选优先级未高于所有目标集，回退到分集洗版")
+            logger.info(f"{subscribe.name} 全集候选优先级未高于 current_priority 门槛，回退到分集洗版")
 
         if full_pack_contexts:
             logger.info(f"{subscribe.name} 分集洗版优先尝试全集资源，共匹配到 {len(full_pack_contexts)} 个候选")
@@ -1187,61 +1267,25 @@ class SubscribeChain(ChainBase):
                 self._rlock.release()
                 logger.debug(f"search Lock released at {datetime.now()}")
 
-    def update_subscribe_priority(self, subscribe: Subscribe, meta: MetaBase,
-                                  mediainfo: MediaInfo, downloads: Optional[List[Context]]):
+    @staticmethod
+    def __update_movie_best_version_download_priority(
+            subscribe: Subscribe,
+            mediainfo: MediaInfo,
+            downloads: Optional[List[Context]],
+    ):
         """
-        更新订阅已下载资源的优先级
+        记录电影洗版本轮下载资源优先级。
         """
         if not downloads:
             return
         if not subscribe.best_version:
             return
-        # 当前下载资源的优先级
         priority = max([item.torrent_info.pri_order for item in downloads])
         now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
 
-        if subscribe.type == MediaType.TV.value:
-            episode_priority = self.__get_episode_priority(subscribe)
-            updated = False
-            for download in downloads:
-                download_priority = download.torrent_info.pri_order
-                downloaded_episodes = self.__get_downloaded_episodes([download])
-                if not downloaded_episodes and self.__is_full_season_resource(download.meta_info, subscribe):
-                    # 整包下载时资源标题常不携带集数，视为覆盖当前订阅的全部目标集。
-                    downloaded_episodes = self.__get_best_version_target_episodes(subscribe)
-                if not downloaded_episodes:
-                    continue
-                for episode in downloaded_episodes:
-                    episode_key = str(episode)
-                    old_priority = episode_priority.get(episode_key)
-                    if old_priority is None or download_priority > old_priority:
-                        episode_priority[episode_key] = download_priority
-                        updated = True
-
-            if not updated and not episode_priority:
-                return
-
-            current_priority = self.get_best_version_current_priority(subscribe, episode_priority)
-            # lack_episode 由 finish_subscribe_or_not -> __update_lack_episodes 按媒体库实况维护，本处不写
-            update_data: Dict[str, Any] = {
-                "episode_priority": episode_priority,
-                "last_update": now,
-                "current_priority": current_priority,
-            }
-
-            SubscribeOper().update(subscribe.id, update_data)
-            subscribe.episode_priority = episode_priority
-            subscribe.current_priority = current_priority
-            subscribe.last_update = now
-
-            completed_episodes = self.__get_best_version_completed_episodes(subscribe)
-            if not self.__is_best_version_complete(subscribe):
-                logger.info(
-                    f'{mediainfo.title_year} 正在洗版，更新剧集优先级为 {priority}，已完成剧集：{completed_episodes}'
-                )
+        if subscribe.type != MediaType.MOVIE.value:
             return
 
-        # 订阅存在待定策略，不管是否已完成，均需更新订阅信息
         SubscribeOper().update(subscribe.id, {
             "current_priority": priority,
             "last_update": now
@@ -1262,17 +1306,20 @@ class SubscribeChain(ChainBase):
         mediakey = subscribe.tmdbid or subscribe.doubanid
         # 是否有剩余集
         no_lefts = not lefts or not lefts.get(mediakey)
-        # 不论是否洗版，只要本轮有下载产生就要把集数追加进 subscribe.note，
-        # 保证"已下载过哪些集"这条事实在所有订阅模式下都有可靠落点；洗版分支
-        # 之前只写 episode_priority，导致用户切回普通订阅时丢失下载历史，并让
-        # __get_downloaded 在洗版下无法从 note 拿到 priority 未达 100 但实际下过的集。
-        if downloads:
+        if downloads and meta.type == MediaType.TV:
+            self.__record_subscribe_download_facts(subscribe=subscribe, mediainfo=mediainfo, downloads=downloads)
+        elif downloads:
             self.__update_subscribe_note(subscribe=subscribe, downloads=downloads)
         # 是否完成订阅
         if not subscribe.best_version:
             # 普通订阅：先按 lefts 写 lack，再判断完成
-            self.__update_lack_episodes(lefts=lefts, subscribe=subscribe, mediainfo=mediainfo,
-                                        update_date=bool(downloads))
+            if meta.type == MediaType.TV:
+                self.__refresh_subscribe_progress_with_no_exists(
+                    no_exists=lefts,
+                    subscribe=subscribe,
+                    touch_last_update=bool(downloads),
+                    scene="download",
+                )
             if ((no_lefts and meta.type == MediaType.TV)
                     or (downloads and meta.type == MediaType.MOVIE)
                     or force):
@@ -1281,13 +1328,20 @@ class SubscribeChain(ChainBase):
                 logger.info(f'{mediainfo.title_year} 未下载完整，继续订阅 ...')
             return
 
-        # 洗版订阅：本轮若有下载先更新 episode_priority / current_priority，让 __update_lack_episodes
-        # 读取到包含本轮新下载的集；否则 lack 会慢一个搜索周期才反映新下载。
-        if downloads:
-            self.update_subscribe_priority(subscribe=subscribe, meta=meta,
-                                           mediainfo=mediainfo, downloads=downloads)
-        self.__update_lack_episodes(lefts=lefts, subscribe=subscribe, mediainfo=mediainfo,
-                                    update_date=bool(downloads))
+        if downloads and meta.type == MediaType.MOVIE:
+            # 电影没有按集质量事实，只能用 current_priority 表达洗版下载质量。
+            self.__update_movie_best_version_download_priority(
+                subscribe=subscribe,
+                mediainfo=mediainfo,
+                downloads=downloads,
+            )
+        if meta.type == MediaType.TV:
+            self.__refresh_subscribe_progress_with_no_exists(
+                no_exists=lefts,
+                subscribe=subscribe,
+                touch_last_update=bool(downloads),
+                scene="download",
+            )
         if self.__is_best_version_complete(subscribe):
             # 洗版完成
             self.__finish_subscribe(subscribe=subscribe, meta=meta, mediainfo=mediainfo)
@@ -1726,7 +1780,7 @@ class SubscribeChain(ChainBase):
                 continue
             # 对于电视剧，获取当前季的总集数
             episodes = mediainfo.seasons.get(subscribe.season) or []
-            current_priority = None
+            progress_update = {}
             if not subscribe.manual_total_episode and len(episodes):
                 total_episode = len(episodes)
                 # 允许外部覆盖按 TMDB 算出的总集数（如待定集数）
@@ -1734,23 +1788,34 @@ class SubscribeChain(ChainBase):
                     total_episode, season=subscribe.season, mediainfo=mediainfo,
                     tmdbid=subscribe.tmdbid, doubanid=subscribe.doubanid,
                     subscribe_id=subscribe.id, scene="refresh")
-                lack_episode = max((subscribe.lack_episode or 0) + (total_episode - (subscribe.total_episode or 0)), 0)
-                if subscribe.best_version and subscribe.type == MediaType.TV.value:
-                    # 为新增集补齐 episode_priority 初始项（priority=0）
-                    old_total_episode = subscribe.total_episode or 0
-                    episode_priority = self.__get_episode_priority(subscribe)
-                    for episode in range(old_total_episode + 1, total_episode + 1):
-                        episode_priority.setdefault(str(episode), 0)
-                    subscribe.total_episode = total_episode
-                    subscribe.episode_priority = episode_priority
-                    current_priority = self.get_best_version_current_priority(subscribe, episode_priority)
+                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,
+                            ),
+                        }
+                else:
+                    total_episode = subscribe.total_episode
+                    progress_update = {"lack_episode": subscribe.lack_episode}
+                    if subscribe.best_version and subscribe.type == MediaType.TV.value:
+                        progress_update = self.__prepare_subscribe_progress_fields(subscribe=subscribe, no_exists={})
                 logger.info(
-                    f'订阅 {subscribe.name} 总集数变化，更新总集数为{total_episode}，缺失集数为{lack_episode} ...')
+                    f'订阅 {subscribe.name} 总集数变化，更新总集数为{total_episode}，'
+                    f'缺失集数为{progress_update.get("lack_episode", subscribe.lack_episode)} ...')
             else:
                 total_episode = subscribe.total_episode
-                lack_episode = subscribe.lack_episode
+                progress_update = {"lack_episode": subscribe.lack_episode}
                 if subscribe.best_version and subscribe.type == MediaType.TV.value:
-                    current_priority = self.get_best_version_current_priority(subscribe)
+                    progress_update = self.__prepare_subscribe_progress_fields(subscribe=subscribe, no_exists={})
             # 更新TMDB信息
             update_data = {
                 "name": mediainfo.title,
@@ -1762,15 +1827,10 @@ class SubscribeChain(ChainBase):
                 "imdbid": mediainfo.imdb_id,
                 "tvdbid": mediainfo.tvdb_id,
                 "total_episode": total_episode,
-                "lack_episode": lack_episode
             }
-            if subscribe.best_version and subscribe.type == MediaType.TV.value:
-                update_data["current_priority"] = current_priority
-                if not subscribe.manual_total_episode and len(episodes):
-                    update_data["episode_priority"] = subscribe.episode_priority
-                subscribe.current_priority = current_priority
-            subscribe.total_episode = total_episode
-            subscribe.lack_episode = lack_episode
+            update_data.update(progress_update)
+            for key, value in progress_update.items():
+                setattr(subscribe, key, value)
             subscribeoper.update(subscribe.id, update_data)
             logger.info(f'{subscribe.name} 订阅元数据更新完成')
 
@@ -1955,80 +2015,297 @@ class SubscribeChain(ChainBase):
             return note
         return []
 
-    @staticmethod
-    def __update_lack_episodes(lefts: Dict[Union[int, str], Dict[int, schemas.NotExistMediaInfo]],
-                               subscribe: Subscribe,
-                               mediainfo: MediaInfo,
-                               update_date: Optional[bool] = False):
+    @classmethod
+    def __prepare_subscribe_progress_fields(
+            cls,
+            subscribe: Subscribe,
+            no_exists: Optional[Dict[Union[int, str], Dict[int, schemas.NotExistMediaInfo]]] = None,
+            touch_last_update: Optional[bool] = False,
+    ) -> Dict[str, Any]:
         """
-        写入订阅 lack_episode，可选同时刷新 last_update。
+        准备订阅进度持久化字段。
 
-        lack 统一语义为"订阅范围内尚未下载到任何版本的集数"。
-        - 普通订阅：lack 从 ``lefts`` 提取（lefts 已在 ``__get_subscribe_no_exits`` 里扣过 note）
-        - 洗版订阅：lack = ``[start, total]`` 范围内既不在 note 也不在 episode_priority(>0) 命中的集数。
-          洗版的 lefts 由 ``check_and_handle_existing_media`` 按 priority<100 构造，承担"搜索目标"职责，
-          与"未下载"维度并不同义——若复用会把"已下载但待升级"的集错算成 lack。
+        该方法只返回待写字段，不主动写库。普通电视剧的 no_exists 为空时表示当前缺失结果为空；
+        洗版电视剧按 note 与 episode_priority 计算未下载过任何版本的目标集数量。
         """
-        update_data = {}
-        if update_date:
-            update_data["last_update"] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+        update_data: Dict[str, Any] = {}
         if subscribe.type == MediaType.TV.value:
+            if no_exists is None and not subscribe.best_version:
+                no_exists = {}
+            update_data["lack_episode"] = cls.compute_lack_episode(subscribe, no_exists=no_exists)
             if subscribe.best_version:
-                lack_episode = SubscribeChain.__compute_best_version_lack_episode(subscribe)
-                logger.info(f"{mediainfo.title_year} 季 {subscribe.season} 剩余未下载剧集数为{lack_episode} ...")
-            elif not lefts:
-                # lefts 为空：媒体库实缺为 0
-                lack_episode = 0
-                logger.info(f'{mediainfo.title_year} 没有缺失集数，直接更新为 0 ...')
-            else:
-                mediakey = subscribe.tmdbid or subscribe.doubanid
-                left_seasons = lefts.get(mediakey)
-                lack_episode = 0
-                if left_seasons:
-                    for season_info in left_seasons.values():
-                        season = season_info.season
-                        if season == subscribe.season:
-                            left_episodes = season_info.episodes
-                            if not left_episodes:
-                                lack_episode = season_info.total_episode
-                            else:
-                                lack_episode = len(left_episodes)
-                            logger.info(f"{mediainfo.title_year} 季 {season} 更新缺失集数为{lack_episode} ...")
-                            break
-            update_data["lack_episode"] = lack_episode
-        # 更新数据库
-        if update_data:
-            SubscribeOper().update(subscribe.id, update_data)
+                update_data["current_priority"] = cls.get_best_version_current_priority(subscribe)
+        if update_data and touch_last_update:
+            update_data["last_update"] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+        return update_data
 
     @staticmethod
-    def __compute_best_version_lack_episode(subscribe: Subscribe) -> int:
+    def __apply_subscribe_update(subscribe: Subscribe, update_data: Dict[str, Any]) -> None:
         """
-        计算洗版订阅"未下载集数"：在 ``[start, total]`` 范围内排除已在 ``note`` 或
-        ``episode_priority`` (>0) 中记账的集。priority<100 但 >0 的集视为"已下载、待升级"，
-        不计入 lack——与 UI 上"已下载 = total - lack"展示口径一致。
+        写入订阅字段并同步当前内存对象，保证后续事件和判断读取最终快照。
         """
-        total_episode = subscribe.total_episode or 0
-        if not total_episode:
-            return 0
-        start_episode = subscribe.start_episode or 1
-        if total_episode < start_episode:
-            return 0
-        target_episodes = set(range(start_episode, total_episode + 1))
-        downloaded: set = set()
-        for ep in (subscribe.note or []):
+        if not update_data:
+            return
+        SubscribeOper().update(subscribe.id, update_data)
+        for key, value in update_data.items():
+            setattr(subscribe, key, value)
+
+    def __refresh_subscribe_progress_with_no_exists(
+            self,
+            subscribe: Subscribe,
+            no_exists: Optional[Dict[Union[int, str], Dict[int, schemas.NotExistMediaInfo]]] = None,
+            touch_last_update: Optional[bool] = False,
+            scene: str = "download",
+    ) -> Dict[str, Any]:
+        """
+        使用已解析的缺失信息刷新订阅进度，避免下载链路重复查询媒体库。
+        """
+        old_lack_episode = subscribe.lack_episode
+        old_current_priority = subscribe.current_priority
+        update_data = self.__prepare_subscribe_progress_fields(
+            subscribe=subscribe,
+            no_exists=no_exists,
+            touch_last_update=touch_last_update,
+        )
+        if not update_data:
+            return {"scene": scene, "updated": False, "fields": [], "reason": "unsupported_subscribe_type"}
+
+        self.__apply_subscribe_update(subscribe, update_data)
+        logger.info(
+            f"订阅 {subscribe.id} 进度刷新：scene={scene}，fields={list(update_data)}，"
+            f"lack_episode {old_lack_episode}->{subscribe.lack_episode}，"
+            f"current_priority {old_current_priority}->{subscribe.current_priority}，reason=updated"
+        )
+        return {
+            "scene": scene,
+            "updated": True,
+            "fields": list(update_data),
+            "lack_episode": update_data.get("lack_episode", subscribe.lack_episode),
+            "current_priority": update_data.get("current_priority", subscribe.current_priority),
+            "reason": "updated",
+        }
+
+    def refresh_subscribe_progress(self, subscribe: Subscribe, *, scene: str = "update") -> Dict[str, Any]:
+        """
+        按主程序口径重新计算并持久化订阅进度。
+        """
+        if subscribe.type != MediaType.TV.value:
+            return {"scene": scene, "updated": False, "fields": [], "reason": "unsupported_subscribe_type"}
+
+        no_exists = None
+        mediainfo = None
+        if not subscribe.best_version:
+            meta = build_subscribe_meta(subscribe)
+            mediainfo = self.recognize_media(
+                meta=meta,
+                mtype=meta.type,
+                tmdbid=subscribe.tmdbid,
+                doubanid=subscribe.doubanid,
+                bangumiid=getattr(subscribe, "bangumiid", None),
+                episode_group=getattr(subscribe, "episode_group", None),
+                cache=False,
+            )
+            if not mediainfo:
+                return {"scene": scene, "updated": False, "fields": [], "reason": "recognize_failed"}
+            mediakey = subscribe.tmdbid or subscribe.doubanid
+            exist_flag, no_exists = self.resolve_subscribe_missing(
+                subscribe=subscribe,
+                meta=meta,
+                mediainfo=mediainfo,
+                mediakey=mediakey,
+            )
+            if not exist_flag and not no_exists:
+                return {"scene": scene, "updated": False, "fields": [], "reason": "resolve_missing_failed"}
+
+        return self.__refresh_subscribe_progress_with_no_exists(
+            subscribe=subscribe,
+            no_exists=no_exists,
+            touch_last_update=False,
+            scene=scene,
+        )
+
+    def backfill_existing_episodes(
+            self,
+            subscribe: Subscribe,
+            episodes: List[Union[int, str]],
+            priority: Optional[int] = None,
+            scene: str = "backfill",
+    ) -> Dict[str, Any]:
+        """
+        将媒体库既有剧集补写为订阅下载事实，并按需刷新进度字段。
+        """
+        accepted = []
+        ignored = []
+        priority_updated = []
+        priority_ignored = []
+        target_episodes = set(self.__get_best_version_target_episodes(subscribe))
+        note = sorted({int(episode) for episode in subscribe.note or [] if str(episode).isdigit()})
+        note_set = set(note)
+        priority_episodes = set()
+
+        for episode in episodes or []:
             try:
-                downloaded.add(int(ep))
+                episode_number = int(episode)
             except (TypeError, ValueError):
+                ignored.append({"episode": episode, "reason": "invalid"})
                 continue
-        for ep_str, priority in (subscribe.episode_priority or {}).items():
-            if not str(ep_str).isdigit():
+            if episode_number not in target_episodes:
+                ignored.append({"episode": episode, "reason": "out_of_range"})
                 continue
-            try:
-                if float(priority) > 0:
-                    downloaded.add(int(ep_str))
-            except (TypeError, ValueError):
+            if episode_number in note_set:
+                ignored.append({"episode": episode, "reason": "duplicate"})
+                priority_episodes.add(episode_number)
                 continue
-        return len(target_episodes - downloaded)
+            accepted.append(episode_number)
+            note_set.add(episode_number)
+            priority_episodes.add(episode_number)
+
+        summary: Dict[str, Any] = {
+            "scene": scene,
+            "accepted": accepted,
+            "ignored": ignored,
+            "priority_updated": priority_updated,
+            "priority_ignored": priority_ignored,
+            "fields": [],
+        }
+        update_data: Dict[str, Any] = {}
+        if accepted:
+            note = sorted(note_set)
+            subscribe.note = note
+            update_data["note"] = note
+
+        priority_valid = isinstance(priority, int) and not isinstance(priority, bool) and 1 <= priority <= 100
+        if priority is not None and not priority_valid:
+            summary["ignored_priority"] = priority
+        if priority_valid:
+            episode_priority = self.__get_episode_priority(subscribe)
+            for episode_number in sorted(priority_episodes):
+                episode_key = str(episode_number)
+                old_priority = episode_priority.get(episode_key)
+                if old_priority is None or priority > old_priority:
+                    episode_priority[episode_key] = priority
+                    priority_updated.append(episode_number)
+                else:
+                    priority_ignored.append({
+                        "episode": episode_number,
+                        "reason": "duplicate" if old_priority == priority else "not_higher_priority",
+                    })
+            if priority_updated:
+                subscribe.episode_priority = episode_priority
+                update_data["episode_priority"] = episode_priority
+
+        should_refresh_progress = subscribe.type == MediaType.TV.value and (accepted or priority_updated)
+        progress_summary = None
+        if should_refresh_progress and subscribe.best_version:
+            update_data.update(self.__prepare_subscribe_progress_fields(
+                subscribe=subscribe,
+                touch_last_update=True,
+            ))
+
+        if update_data:
+            self.__apply_subscribe_update(subscribe, update_data)
+            summary["fields"] = list(update_data)
+        if should_refresh_progress and not subscribe.best_version:
+            progress_summary = self.refresh_subscribe_progress(subscribe, scene=scene)
+        if progress_summary is not None:
+            summary["progress"] = progress_summary
+        summary["updated"] = bool(update_data)
+        if progress_summary:
+            summary["updated"] = summary["updated"] or bool(progress_summary.get("updated"))
+        return summary
+
+    def __record_subscribe_download_facts(
+            self,
+            subscribe: Subscribe,
+            *,
+            mediainfo: MediaInfo,
+            downloads: Optional[List[Context]],
+    ) -> Dict[str, Any]:
+        """
+        记录主程序本轮下载产生的订阅事实，并返回本轮覆盖摘要。
+        """
+        if not downloads:
+            return {"episodes": [], "fields": [], "updated": False}
+
+        covered_episodes = set()
+        written_priorities: Dict[str, int] = {}
+        used_full_coverage_fallback = False
+        episode_priority = self.__get_episode_priority(subscribe)
+        note_set = {
+            int(episode)
+            for episode in subscribe.note or []
+            if str(episode).isdigit()
+        }
+        update_data: Dict[str, Any] = {}
+
+        for download in downloads:
+            media = download.media_info
+            if subscribe.tmdbid and getattr(media, "tmdb_id", None) and media.tmdb_id != subscribe.tmdbid:
+                continue
+            if subscribe.doubanid and getattr(media, "douban_id", None) and media.douban_id != subscribe.doubanid:
+                continue
+
+            if subscribe.type == MediaType.MOVIE.value and media.type == MediaType.MOVIE:
+                note_set.add(1)
+                covered_episodes.add(1)
+                continue
+
+            if subscribe.type != MediaType.TV.value or media.type != MediaType.TV:
+                continue
+
+            selected_episodes = getattr(download, "selected_episodes", None)
+            if selected_episodes:
+                episodes = selected_episodes
+            elif getattr(download, "meta_info", None) and download.meta_info.episode_list:
+                episodes = download.meta_info.episode_list
+            elif getattr(download, "confirmed_full_coverage", False):
+                episodes = self.__get_best_version_target_episodes(subscribe)
+                used_full_coverage_fallback = True
+            else:
+                episodes = []
+
+            valid_episodes = []
+            for episode in episodes:
+                try:
+                    episode_number = int(episode)
+                except (TypeError, ValueError):
+                    continue
+                if episode_number not in self.__get_best_version_target_episodes(subscribe):
+                    continue
+                valid_episodes.append(episode_number)
+            if not valid_episodes:
+                continue
+
+            priority = getattr(download.torrent_info, "pri_order", None)
+            for episode_number in valid_episodes:
+                note_set.add(episode_number)
+                covered_episodes.add(episode_number)
+                episode_key = str(episode_number)
+                old_priority = episode_priority.get(episode_key)
+                if isinstance(priority, int) and not isinstance(priority, bool) \
+                        and (old_priority is None or priority > old_priority):
+                    episode_priority[episode_key] = priority
+                    written_priorities[episode_key] = priority
+
+        if covered_episodes:
+            update_data["note"] = sorted(note_set)
+            if subscribe.type == MediaType.TV.value:
+                update_data["episode_priority"] = episode_priority
+            update_data["last_update"] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+
+        if update_data:
+            self.__apply_subscribe_update(subscribe, update_data)
+            logger.info(
+                f"{mediainfo.title_year} 订阅 {subscribe.id} 第 {subscribe.season} 季记录下载事实："
+                f"mode=best_version:{subscribe.best_version},full:{subscribe.best_version_full}，"
+                f"covered_episodes={sorted(covered_episodes)}，episode_priority={written_priorities}，"
+                f"confirmed_full_coverage_fallback={used_full_coverage_fallback}"
+            )
+        return {
+            "episodes": sorted(covered_episodes),
+            "fields": list(update_data),
+            "updated": bool(update_data),
+        }
 
     def __finish_subscribe(self, subscribe: Subscribe, mediainfo: MediaInfo, meta: MetaBase):
         """
@@ -3307,19 +3584,28 @@ class SubscribeChain(ChainBase):
         if not new_total_episode or new_total_episode <= old_total_episode:
             return
 
-        old_lack_episode = subscribe.lack_episode or 0
-        new_lack_episode = old_lack_episode + (new_total_episode - old_total_episode)
         now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-        SubscribeOper().update(subscribe.id, {
-            "total_episode": new_total_episode,
-            "lack_episode": new_lack_episode,
-            "last_update": now
-        })
-        subscribe.total_episode = new_total_episode
-        subscribe.lack_episode = new_lack_episode
+        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["last_update"] = now
+        SubscribeOper().update(subscribe.id, update_data)
+        for key, value in update_data.items():
+            setattr(subscribe, key, value)
         subscribe.last_update = now
         logger.info(
-            f"订阅 {subscribe.name} 第{subscribe.season}季 总集数更新为 {new_total_episode}，缺失集数更新为 {new_lack_episode}"
+            f"订阅 {subscribe.name} 第{subscribe.season}季 总集数更新为 {new_total_episode}，"
+            f"缺失集数更新为 {subscribe.lack_episode}"
         )
 
     @classmethod
diff --git a/app/core/context.py b/app/core/context.py
index 44e9a8dd..827c6b10 100644
--- a/app/core/context.py
+++ b/app/core/context.py
@@ -919,6 +919,8 @@ class Context:
     media_info_is_target: bool = False
     # 调用方对本候选允许下载的剧集集合，None 表示不限制，空集合表示拒绝交付任何集。
     allowed_episodes: Optional[Set[int]] = None
+    # 下载层确认候选资源覆盖完整目标范围，供订阅事实写入判断整包资源。
+    confirmed_full_coverage: bool = False
 
     def to_dict(self):
         """
@@ -935,4 +937,5 @@ class Context:
             "media_info_is_target": self.media_info_is_target,
             # 保留 None / 空集 / 非空集 三态语义，避免下游误把"显式拒绝"当成"不限制"。
             "allowed_episodes": sorted(self.allowed_episodes) if self.allowed_episodes is not None else None,
+            "confirmed_full_coverage": self.confirmed_full_coverage,
         }
diff --git a/app/schemas/context.py b/app/schemas/context.py
index 3ea6976b..ccc5dc99 100644
--- a/app/schemas/context.py
+++ b/app/schemas/context.py
@@ -314,6 +314,8 @@ class Context(BaseModel):
     candidate_recognized: Optional[bool] = False
     # 当前 media_info 是否为目标媒体回填
     media_info_is_target: Optional[bool] = False
+    # 下载层确认候选资源覆盖完整目标范围，供订阅事实写入判断整包资源
+    confirmed_full_coverage: Optional[bool] = False
 
 
 class MediaSeason(BaseModel):
diff --git a/app/schemas/subscribe.py b/app/schemas/subscribe.py
index d1cf1bb5..a16338ed 100644
--- a/app/schemas/subscribe.py
+++ b/app/schemas/subscribe.py
@@ -5,6 +5,39 @@ from pydantic import BaseModel, Field, ConfigDict, model_validator
 from app.schemas.types import MediaType
 
 
+def compute_subscribe_completed_episode(subscribe: Any) -> Optional[int]:
+    """
+    计算订阅"已完成"集数派生值，仅用于响应填充，不入库。
+
+    普通电视剧按 ``total_episode - lack_episode`` 计算；洗版电视剧按订阅目标范围内
+    priority==100 的分集数量，加上起始集前的逻辑完成集数计算。
+    """
+    total_episode = getattr(subscribe, "total_episode", None) or 0
+    if getattr(subscribe, "type", None) != MediaType.TV.value or not total_episode:
+        return None
+
+    start_episode = getattr(subscribe, "start_episode", None) or 1
+    if not getattr(subscribe, "best_version", None):
+        lack = getattr(subscribe, "lack_episode", None) or 0
+        return max(total_episode - lack, 0)
+
+    episode_priority = getattr(subscribe, "episode_priority", None) or {}
+    if not episode_priority and getattr(subscribe, "current_priority", None) is not None:
+        # 兼容只有整体优先级的洗版快照，响应派生值需与链路侧按集口径保持一致。
+        episode_priority = {
+            str(episode): int(getattr(subscribe, "current_priority"))
+            for episode in range(start_episode, total_episode + 1)
+        }
+    priority_completed = sum(
+        1
+        for ep_key, priority in episode_priority.items()
+        if str(ep_key).isdigit()
+        and start_episode <= int(ep_key) <= total_episode
+        and priority == 100
+    )
+    return min(max(start_episode - 1, 0), total_episode) + priority_completed
+
+
 class Subscribe(BaseModel):
     id: Optional[int] = None
     # 订阅名称
@@ -95,26 +128,7 @@ class Subscribe(BaseModel):
         if self.completed_episode is not None:
             # 调用方显式提供过的值不覆盖
             return self
-        total_episode = self.total_episode or 0
-        if self.type != MediaType.TV.value or not total_episode:
-            return self
-        start_episode = self.start_episode or 1
-        if not self.best_version:
-            lack = self.lack_episode or 0
-            self.completed_episode = max(total_episode - lack, 0)
-            return self
-        # 洗版口径：起始集前视为逻辑完成 + [start, total] 范围内 priority==100 命中。
-        # ``start_episode > total_episode`` 属于异常配置，需把 "起始集前" 偏移截断到 total，
-        # 防止 completed_episode 越过分母 total_episode。
-        episode_priority = self.episode_priority or {}
-        priority_completed = sum(
-            1
-            for ep_key, priority in episode_priority.items()
-            if str(ep_key).isdigit()
-            and start_episode <= int(ep_key) <= total_episode
-            and priority == 100
-        )
-        self.completed_episode = min(max(start_episode - 1, 0), total_episode) + priority_completed
+        self.completed_episode = compute_subscribe_completed_episode(self)
         return self
 
 
diff --git a/docs/subscribe-lifecycle.md b/docs/subscribe-lifecycle.md
new file mode 100644
index 00000000..ce406729
--- /dev/null
+++ b/docs/subscribe-lifecycle.md
@@ -0,0 +1,85 @@
+# 订阅生命周期与订阅模式
+
+本文说明主程序当前订阅领域的业务语义、字段所有权和使用方式，供维护订阅搜索、下载、进度刷新、重置和插件协同时参考。
+
+## 核心模型
+
+订阅是一条持续运行的媒体目标。它把用户想要的内容、搜索下载策略、已下载事实和完成进度保存在同一个订阅记录中。
+
+订阅链路围绕四类状态工作：
+
+| 状态类别 | 主要字段 | 含义 | 写入入口 |
+| --- | --- | --- | --- |
+| 目标范围 | `type`、`season`、`start_episode`、`total_episode` | 订阅需要覆盖的媒体范围。电影视为单个目标；电视剧按季和集范围处理。 | 新增订阅、订阅编辑、剧集刷新 |
+| 下载事实 | `note`、`episode_priority` | 主程序或可信回填入口确认已经存在或已经下载的目标。`note` 表示存在事实，`episode_priority` 表示电视剧每集已知下载质量。 | 下载事实入口、backfill 入口、reset 入口 |
+| 进度摘要 | `lack_episode`、`current_priority`、`completed_episode` | 面向搜索、完成判定和展示的派生进度。`completed_episode` 由响应层按当前事实计算。 | progress 刷新入口、响应构造 |
+| 生命周期状态 | `state`、`last_update` | 订阅是否继续搜索，以及最近一次系统写入时间。 | 搜索下载流程、完成订阅、reset、状态更新 |
+
+字段所有权应保持集中：下载事实入口写事实，progress 刷新入口写进度摘要，完成入口写完成状态，普通更新入口不应绕过这些语义直接改派生字段。
+
+## 生命周期
+
+| 阶段 | 触发来源 | 主程序行为 | 关键输出 |
+| --- | --- | --- | --- |
+| 创建订阅 | 用户、API、Agent、插件 | 记录媒体目标、订阅模式、季集范围、保存路径和下载策略。 | 初始订阅记录 |
+| 搜索匹配 | 定时任务、手动搜索、RSS/站点刷新 | 按订阅目标和模式过滤候选资源，计算需要下载或升级的目标。 | 匹配候选、待下载上下文 |
+| 下载选择 | 下载链路 | 将候选资源交给下载器，下载层可返回明确集数或完整覆盖确认。 | 下载上下文、剩余缺口 |
+| 事实记录 | 下载完成或可信 backfill | 写入 `note`，电视剧同步写入或提升 `episode_priority`。 | 已下载事实快照 |
+| 进度刷新 | 下载流程、backfill、剧集范围变化、显式刷新 | 普通电视剧按媒体库缺失结果刷新 `lack_episode`；洗版电视剧按目标范围和按集事实刷新 `lack_episode` 与 `current_priority`。 | 进度字段 |
+| 完成判定 | 下载流程末端、强制完成 | 根据媒体类型和订阅模式判断是否完成，完成前允许完成检查事件否决。 | 订阅历史、完成状态 |
+| 重置/回填 | 用户操作、插件、维护工具 | reset 清空事实和进度；backfill 写入调用方确认的外部存在事实。 | 新事实快照和进度摘要 |
+
+## 订阅模式
+
+MoviePilot 当前以媒体类型和洗版方式组合出常用订阅模式。普通电视剧订阅和分集洗版都是“按集目标”订阅：前者每集下载一次即可，后者允许同一集继续升级质量。全集洗版是整体资源质量目标，整包候选需要完整覆盖目标范围。
+
+| 模式 | 适用媒体 | 目标口径 | 下载事实 | 进度口径 | 完成口径 | 典型用途 |
+| --- | --- | --- | --- | --- | --- | --- |
+| 电影普通订阅 | 电影 | 单个电影目标 | `note=[1]` 表示电影已下载。 | 不维护电视剧缺集进度。 | 有下载或强制完成即可完成。 | 追新电影、补电影库 |
+| 电影洗版 | 电影 | 单个电影目标的整体质量 | `note=[1]` 表示电影已下载。 | `current_priority` 表示当前电影资源质量。 | `current_priority == 100` 或完成策略满足时完成。 | 用更高质量版本替换已有电影 |
+| 普通电视剧订阅 | 电视剧 | 季内目标集范围 | `note` 记录已存在或已下载集；`episode_priority` 记录每集已知下载质量。 | `lack_episode` 来自媒体库缺失结果。 | 目标范围无缺集时完成。 | 追剧、补缺集 |
+| 分集洗版 | 电视剧 | 季内目标集范围 | `note` 记录存在事实；`episode_priority` 记录每集质量。 | `lack_episode` 统计从未下载过任何版本的集；`current_priority` 是目标范围最低已知质量。 | 目标范围内每集达到顶级质量时完成洗版。 | 对单集逐步升级质量 |
+| 全集洗版 | 电视剧 | 完整目标集范围的整体质量 | 整包下载确认完整覆盖后，按目标范围写 `note` 和 `episode_priority`。 | `current_priority` 作为整包候选整体质量门槛，并由按集事实刷新。 | 完整目标范围达到顶级质量时完成洗版。 | 用完整季包替换已有剧集 |
+
+## 按集订阅与全集洗版
+
+普通电视剧订阅和分集洗版共享同一个按集事实源：
+
+- `note` 表示某一集已经存在或已经下载。
+- `episode_priority` 表示某一集已知的下载质量，缺失 key 表示没有质量事实。
+- `lack_episode` 只表达“还缺多少集没有任何版本”，不表达“还有多少集需要升级质量”。
+- `current_priority` 对电视剧洗版是目标范围内最低已知质量摘要，缺失质量事实按 `0` 参与计算。
+
+因此普通电视剧订阅和分集洗版可以互相转换。普通订阅下载产生的按集质量事实可被分集洗版继续使用；分集洗版产生的 `note` 也可被普通订阅继续作为已下载事实。
+
+全集洗版关注完整目标范围的整体资源质量。整包候选必须完整覆盖 `[start_episode, total_episode]`，不是只覆盖当前缺口集。下载层确认完整覆盖后，主程序才可在资源缺少显式集数时按目标范围补写事实。
+
+## 字段语义
+
+| 字段 | 语义 | 关键约束 |
+| --- | --- | --- |
+| `note` | 已存在或已下载的电影项/电视剧集。 | 电影使用 `[1]`；电视剧使用集数列表。 |
+| `episode_priority` | 电视剧按集质量事实，键为集数，值为下载优先级。 | 只升不降；普通订阅和洗版订阅都可以写入。 |
+| `lack_episode` | 电视剧目标范围内仍缺少任意版本的集数。 | 普通订阅来自媒体库缺失结果；洗版订阅按 `note` 和 `episode_priority > 0` 计算。 |
+| `current_priority` | 洗版整体质量摘要。 | 电影洗版直接维护；电视剧洗版按目标范围内最低已知质量刷新。 |
+| `completed_episode` | 展示用完成集数。 | 响应层按当前事实计算，不作为主要事实源。 |
+| `state` | 订阅运行状态。 | 完成、暂停、重置等入口按业务状态维护。 |
+
+## 用户视角
+
+| 想要做什么 | 推荐模式 | 行为说明 | 注意事项 |
+| --- | --- | --- | --- |
+| 新剧更新后自动下载 | 普通电视剧订阅 | 每集下载到任意版本即可，后续缺集继续追。 | 下载过的集会作为事实保留，后续可切到分集洗版。 |
+| 已有剧集想逐集升级质量 | 分集洗版 | 每集按质量继续升级，达到顶级质量后不再下载该集。 | 低质量已下载集不计入缺集，但仍会作为待升级目标。 |
+| 想用完整季包替换整季 | 全集洗版 | 优先寻找完整覆盖目标范围的整包资源，并按整体质量门槛判断是否下载。 | 整包必须覆盖目标集范围，不只是覆盖当前缺口。 |
+| 新电影自动下载 | 电影普通订阅 | 下载到电影后完成订阅。 | 电影没有分集事实。 |
+| 电影已有版本想升级 | 电影洗版 | 用 `current_priority` 表达当前电影质量，下载更高质量版本。 | 完成策略按电影整体质量判断。 |
+| 手动入库或插件扫描到已有集 | backfill | 可信调用方写入已存在集；可选择是否提供 quality priority。 | `priority=None` 只表达存在事实；提供有效 priority 才表达质量事实。 |
+
+## 维护边界
+
+- 用户/API/Agent 的普通订阅编辑只负责目标和配置变更，不直接维护下载事实和进度摘要。
+- 主程序下载链路产生的电视剧下载必须同时维护 `note` 和 `episode_priority`。
+- backfill 入口只接收调用方确认的外部存在事实，不主动扫描媒体库。
+- progress 刷新入口负责把当前事实转换为 `lack_episode` 和 TV 洗版 `current_priority`。
+- 插件可以调用主程序公开入口补事实或刷新进度，不应复制主程序缺集、洗版完成或当前优先级计算规则。
diff --git a/tests/test_download_chain.py b/tests/test_download_chain.py
index 661ecb92..305a2c96 100644
--- a/tests/test_download_chain.py
+++ b/tests/test_download_chain.py
@@ -330,6 +330,7 @@ def _build_tv_context(episode_list=None):
         ),
         torrent_info=SimpleNamespace(title="Test Show S01 2160p", site_name="TestSite"),
         allowed_episodes=None,
+        confirmed_full_coverage=False,
     )
 
 
@@ -362,6 +363,40 @@ def test_batch_download_rejects_complete_coverage_when_files_do_not_cover_target
 
     assert downloads == []
     assert lefts == no_exists
+    assert context.confirmed_full_coverage is False
+    chain.download_single.assert_not_called()
+
+
+def test_batch_download_rejects_complete_coverage_when_only_missing_episodes_match(monkeypatch):
+    """
+    完整覆盖要求目标范围全集，不能只覆盖当前缺口集。
+    """
+    _FakeBatchTorrentHelper.episodes = [4, 5]
+    monkeypatch.setattr(download_module, "TorrentHelper", _FakeBatchTorrentHelper)
+    monkeypatch.setattr(download_module.eventmanager, "send_event", lambda *args, **kwargs: None)
+
+    chain = DownloadChain.__new__(DownloadChain)
+    chain.download_torrent = MagicMock(return_value=(b"torrent-content", "", ["demo.mkv"]))
+    chain.download_single = MagicMock(return_value="hash")
+
+    context = _build_tv_context()
+    no_exists = {
+        1: {
+            1: NotExistMediaInfo(
+                season=1,
+                episodes=[4, 5],
+                total_episode=5,
+                start_episode=1,
+                require_complete_coverage=True,
+            )
+        }
+    }
+
+    downloads, lefts = chain.batch_download(contexts=[context], no_exists=no_exists)
+
+    assert downloads == []
+    assert lefts == no_exists
+    assert context.confirmed_full_coverage is False
     chain.download_single.assert_not_called()
 
 
@@ -462,9 +497,43 @@ def test_batch_download_accepts_complete_coverage_when_files_cover_target_range(
 
     assert downloads == [context]
     assert lefts == {}
+    assert context.confirmed_full_coverage is True
     chain.download_single.assert_called_once()
 
 
+def test_batch_download_rejects_complete_coverage_when_files_have_same_count_but_wrong_range(monkeypatch):
+    """
+    完整覆盖按目标集号集合判断，不能让同数量的偏移局部包通过。
+    """
+    _FakeBatchTorrentHelper.episodes = list(range(1, 45))
+    monkeypatch.setattr(download_module, "TorrentHelper", _FakeBatchTorrentHelper)
+    monkeypatch.setattr(download_module.eventmanager, "send_event", lambda *args, **kwargs: None)
+
+    chain = DownloadChain.__new__(DownloadChain)
+    chain.download_torrent = MagicMock(return_value=(b"torrent-content", "", ["demo.mkv"]))
+    chain.download_single = MagicMock(return_value="hash")
+
+    context = _build_tv_context()
+    no_exists = {
+        1: {
+            1: NotExistMediaInfo(
+                season=1,
+                episodes=[],
+                total_episode=143,
+                start_episode=100,
+                require_complete_coverage=True,
+            )
+        }
+    }
+
+    downloads, lefts = chain.batch_download(contexts=[context], no_exists=no_exists)
+
+    assert downloads == []
+    assert lefts == no_exists
+    assert context.confirmed_full_coverage is False
+    chain.download_single.assert_not_called()
+
+
 def test_batch_download_accepts_complete_coverage_when_title_episodes_cover_target(monkeypatch):
     """
     显式标出完整范围的候选也可满足完整覆盖任务。
@@ -494,6 +563,7 @@ def test_batch_download_accepts_complete_coverage_when_title_episodes_cover_targ
 
     assert downloads == [context]
     assert lefts == {}
+    assert context.confirmed_full_coverage is True
     chain.download_torrent.assert_not_called()
     chain.download_single.assert_called_once()
 
@@ -527,6 +597,7 @@ def test_batch_download_rejects_complete_coverage_when_title_episodes_are_partia
 
     assert downloads == []
     assert lefts == no_exists
+    assert context.confirmed_full_coverage is False
     chain.download_torrent.assert_not_called()
     chain.download_single.assert_not_called()
 
@@ -561,6 +632,7 @@ def test_batch_download_complete_coverage_ignores_allowed_episode_narrowing(monk
 
     assert downloads == []
     assert lefts == no_exists
+    assert context.confirmed_full_coverage is False
     chain.download_torrent.assert_not_called()
     chain.download_single.assert_not_called()
 
@@ -593,4 +665,5 @@ def test_batch_download_keeps_count_check_without_complete_coverage(monkeypatch)
 
     assert downloads == [context]
     assert lefts == {}
+    assert context.confirmed_full_coverage is False
     chain.download_single.assert_called_once()
diff --git a/tests/test_subscribe_chain.py b/tests/test_subscribe_chain.py
index 1e50bf1d..7ef9f75c 100644
--- a/tests/test_subscribe_chain.py
+++ b/tests/test_subscribe_chain.py
@@ -6,6 +6,7 @@ from types import SimpleNamespace
 from unittest import TestCase
 from unittest.mock import patch
 
+from app import schemas
 from app.schemas.types import MediaType
 from app.testing import stub_modules
 
@@ -36,6 +37,9 @@ def _load_subscribe_chain_class():
         async def async_post_message(self, *args, **kwargs):
             return None
 
+        def recognize_media(self, *args, **kwargs):
+            return None
+
     chain_module.ChainBase = _ChainBase
 
     interaction_module = ensure_module("app.helper.interaction", types.ModuleType("app.helper.interaction"))
@@ -103,7 +107,30 @@ def _load_subscribe_chain_class():
     meta_module.MetaBase = SimpleNamespace
 
     metainfo_module = ensure_module("app.core.metainfo", types.ModuleType("app.core.metainfo"))
-    metainfo_module.MetaInfo = lambda *args, **kwargs: SimpleNamespace(episode_list=[])
+
+    class _MetaInfo(SimpleNamespace):
+        """提供订阅刷新测试需要的 MetaInfo 核心字段。"""
+
+        def __init__(self, title="", *args, **kwargs):
+            super().__init__(name=title, episode_list=[])
+
+        @property
+        def season_seq(self):
+            if getattr(self, "begin_season", None) is not None:
+                return str(self.begin_season)
+            if getattr(self, "type", None) == MediaType.TV:
+                return "1"
+            return ""
+
+        @property
+        def season(self):
+            if getattr(self, "begin_season", None) is not None:
+                return f"S{str(self.begin_season).rjust(2, '0')}"
+            if getattr(self, "type", None) == MediaType.TV:
+                return "S01"
+            return ""
+
+    metainfo_module.MetaInfo = _MetaInfo
 
     words_module = ensure_module("app.core.meta.words", types.ModuleType("app.core.meta.words"))
 
@@ -414,7 +441,7 @@ class SubscribeChainTest(TestCase):
             """标记测试已经进入标题匹配函数体。"""
 
         class _PlainTorrentHelper:
-            """模拟未声明 staticmethod 的历史 TorrentHelper 形态。"""
+            """模拟需要按类调用的 TorrentHelper 形态。"""
 
             def match_torrent(mediainfo, torrent_meta, torrent):
                 """标记类级调用已经正确进入匹配逻辑。"""
@@ -588,13 +615,18 @@ class SubscribeChainTest(TestCase):
     def test_best_version_progress_helpers_return_remaining_priority(self):
         subscribe = self._build_subscribe(
             total_episode=5,
-            episode_priority={"1": 100, "2": 80, "3": 90, "4": 100, "5": 70},
+            episode_priority={"1": 100, "2": 80, "4": 100, "5": 70},
             current_priority=100,
         )
 
-        self.assertEqual(SubscribeChain.get_best_version_current_priority(subscribe), 90)
+        self.assertEqual(SubscribeChain.get_best_version_current_priority(subscribe), 0)
         self.assertFalse(SubscribeChain.is_best_version_complete(subscribe))
 
+    def test_best_version_current_priority_uses_legacy_fallback_when_episode_priority_empty(self):
+        subscribe = self._build_subscribe(total_episode=3, current_priority=80, episode_priority=None)
+
+        self.assertEqual(SubscribeChain.get_best_version_current_priority(subscribe), 80)
+
     def test_best_version_progress_helpers_mark_complete_when_all_target_episodes_done(self):
         subscribe = self._build_subscribe(
             total_episode=3,
@@ -966,6 +998,34 @@ class SubscribeChainTest(TestCase):
         self.assertTrue(satisfied)
         self.assertEqual(no_exists, {})
 
+    def test_resolve_subscribe_missing_accepts_downloaded_legacy_current_priority_targets(self):
+        """外部完成守卫读取按集事实时，应保留 current_priority 整体快照兼容。"""
+        subscribe = self._build_subscribe(
+            best_version=1,
+            best_version_full=0,
+            total_episode=3,
+            current_priority=80,
+            episode_priority=None,
+            note=[],
+        )
+        meta = SimpleNamespace(type=MediaType.TV, begin_season=1, season=1)
+        mediainfo = SimpleNamespace(
+            type=MediaType.TV,
+            seasons={1: [1, 2, 3]},
+            title_year="Test Show (2026)",
+        )
+
+        satisfied, no_exists = SubscribeChain().resolve_subscribe_missing(
+            subscribe=subscribe,
+            meta=meta,
+            mediainfo=mediainfo,
+            mediakey=1,
+            best_version_accept_downloaded=True,
+        )
+
+        self.assertTrue(satisfied)
+        self.assertEqual(no_exists, {})
+
     def test_resolve_subscribe_missing_default_best_version_requires_top_priority(self):
         """主程序洗版完成口径默认仍要求目标分集达到最高优先级。"""
         subscribe = self._build_subscribe(
@@ -1130,6 +1190,7 @@ class SubscribeChainTest(TestCase):
         self.assertEqual(result["media-key"][1].episodes, [])
         self.assertEqual(result["media-key"][1].start_episode, 44)
         self.assertEqual(result["media-key"][1].total_episode, 48)
+        self.assertTrue(result["media-key"][1].require_complete_coverage)
 
     def test_is_episode_range_covered_matches_pending_episodes(self):
         subscribe = self._build_subscribe(
@@ -1335,8 +1396,8 @@ class SubscribeChainTest(TestCase):
         self.assertEqual(calls[0]["contexts"], [episode_context, full_pack_context])
         self.assertIs(calls[0]["no_exists"], no_exists)
 
-    def test_episode_best_version_skips_full_pack_first_when_pack_priority_below_one_episode(self):
-        """验证全集低于任一目标分集优先级时不会整包优先。"""
+    def test_episode_best_version_tries_full_pack_when_priority_exceeds_current_priority(self):
+        """整包候选按 current_priority 整体门槛判断，单集较高优先级不阻断整包优先。"""
         subscribe = self._build_subscribe(
             best_version_full=0,
             total_episode=3,
@@ -1362,7 +1423,7 @@ class SubscribeChainTest(TestCase):
         calls = []
 
         class _FakeDownloadChain:
-            """记录回退下载调用，验证低优先级全集不进入整包优先分支。"""
+            """记录整包优先和回退调用，验证整体门槛口径。"""
 
             def batch_download(self, **kwargs):
                 calls.append(kwargs)
@@ -1378,12 +1439,13 @@ class SubscribeChainTest(TestCase):
 
         self.assertEqual(downloads, [])
         self.assertIs(lefts, no_exists)
-        self.assertEqual(len(calls), 1)
+        self.assertEqual(len(calls), 2)
         self.assertEqual(calls[0]["contexts"], [full_pack_context])
-        self.assertIs(calls[0]["no_exists"], no_exists)
+        self.assertEqual(calls[0]["no_exists"]["media-key"][1].episodes, [])
+        self.assertIs(calls[1]["no_exists"], no_exists)
 
     def test_full_pack_priority_check_uses_current_priority_fallback(self):
-        """验证旧订阅没有分集状态时使用 current_priority 兜底判断。"""
+        """没有按集优先级状态时使用 current_priority 兜底判断。"""
         subscribe = self._build_subscribe(total_episode=3, current_priority=80, episode_priority=None)
 
         self.assertFalse(
@@ -1392,14 +1454,53 @@ class SubscribeChainTest(TestCase):
                 priority=80,
             )
         )
+
+    def test_full_best_version_priority_check_uses_current_priority_directly(self):
+        """全集洗版优先判断使用 current_priority，避免部分 episode_priority 破坏整体优先级语义。"""
+        subscribe = self._build_subscribe(
+            best_version_full=1,
+            total_episode=3,
+            current_priority=80,
+            episode_priority={"1": 100},
+        )
+
+        self.assertFalse(
+            SubscribeChain._SubscribeChain__is_full_season_priority_higher_than_all_targets(
+                subscribe=subscribe,
+                priority=80,
+            )
+        )
         self.assertTrue(
             SubscribeChain._SubscribeChain__is_full_season_priority_higher_than_all_targets(
                 subscribe=subscribe,
                 priority=81,
             )
         )
+        subscribe.current_priority = 0
+        self.assertTrue(
+            SubscribeChain._SubscribeChain__is_full_season_priority_higher_than_all_targets(
+                subscribe=subscribe,
+                priority=10,
+            )
+        )
 
-    def test_update_subscribe_priority_uses_selected_episodes(self):
+    def test_full_pack_priority_check_computes_gate_when_current_priority_is_none(self):
+        """没有持久 current_priority 时，整包门槛按当前按集事实计算。"""
+        subscribe = self._build_subscribe(
+            best_version_full=1,
+            total_episode=3,
+            current_priority=None,
+            episode_priority={"1": 100, "2": 80},
+        )
+
+        self.assertTrue(
+            SubscribeChain._SubscribeChain__is_full_season_priority_higher_than_all_targets(
+                subscribe=subscribe,
+                priority=1,
+            )
+        )
+
+    def test_record_download_facts_uses_selected_episodes(self):
         subscribe = self._build_subscribe(
             total_episode=4,
             episode_priority={"1": 100, "2": 80, "3": 70, "4": 60},
@@ -1414,16 +1515,12 @@ class SubscribeChainTest(TestCase):
         chain = SubscribeChain()
         mediainfo = SimpleNamespace(title_year="Test Show (2026)")
 
-        with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper") as subscribe_oper_cls, patch.object(
-            SubscribeChain,
-            "_SubscribeChain__finish_subscribe",
-        ) as finish_mock:
+        with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper") as subscribe_oper_cls:
             subscribe_oper = subscribe_oper_cls.return_value
             subscribe_oper.update.return_value = None
 
-            chain.update_subscribe_priority(
+            snapshot = chain._SubscribeChain__record_subscribe_download_facts(
                 subscribe=subscribe,
-                meta=SimpleNamespace(),
                 mediainfo=mediainfo,
                 downloads=[download],
             )
@@ -1431,15 +1528,15 @@ class SubscribeChainTest(TestCase):
         subscribe_oper.update.assert_called_once()
         payload = subscribe_oper.update.call_args.args[1]
         self.assertEqual(payload["episode_priority"], {"1": 100, "2": 80, "3": 90, "4": 60})
-        self.assertEqual(payload["current_priority"], 90)
-        # update_subscribe_priority 不再回写 lack_episode；lack 由下载链路末端的 __update_lack_episodes 维护
+        self.assertEqual(payload["note"], [3])
+        self.assertEqual(snapshot["episodes"], [3])
+        self.assertNotIn("current_priority", payload)
         self.assertNotIn("lack_episode", payload)
         self.assertEqual(subscribe.episode_priority, {"1": 100, "2": 80, "3": 90, "4": 60})
-        self.assertEqual(subscribe.current_priority, 90)
+        self.assertEqual(subscribe.current_priority, 80)
         self.assertEqual(subscribe.lack_episode, 3)
-        finish_mock.assert_not_called()
 
-    def test_update_subscribe_priority_updates_all_target_episodes_without_finishing(self):
+    def test_record_download_facts_updates_all_target_episodes_without_finishing(self):
         subscribe = self._build_subscribe(
             total_episode=3,
             episode_priority={"1": 100, "2": 90, "3": 80},
@@ -1451,35 +1548,30 @@ class SubscribeChainTest(TestCase):
             self._build_download(priority=100, selected_episodes=[3]),
         ]
         chain = SubscribeChain()
-        meta = SimpleNamespace()
         mediainfo = SimpleNamespace(title_year="Test Show (2026)")
 
-        with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper") as subscribe_oper_cls, patch.object(
-            SubscribeChain,
-            "_SubscribeChain__finish_subscribe",
-        ) as finish_mock, patch.object(SUBSCRIBE_CHAIN_MODULE, "logger") as logger_mock:
+        with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper") as subscribe_oper_cls, \
+                patch.object(SUBSCRIBE_CHAIN_MODULE, "logger") as logger_mock:
             subscribe_oper = subscribe_oper_cls.return_value
             subscribe_oper.update.return_value = None
 
-            chain.update_subscribe_priority(
+            chain._SubscribeChain__record_subscribe_download_facts(
                 subscribe=subscribe,
-                meta=meta,
                 mediainfo=mediainfo,
                 downloads=downloads,
             )
 
         payload = subscribe_oper.update.call_args.args[1]
         self.assertEqual(payload["episode_priority"], {"1": 100, "2": 100, "3": 100})
-        self.assertEqual(payload["current_priority"], 100)
-        # 完成判定仍由 finish_subscribe_or_not 统一处理，避免优先级更新和流程尾部重复完成
+        self.assertEqual(payload["note"], [2, 3])
+        self.assertNotIn("current_priority", payload)
         self.assertNotIn("lack_episode", payload)
-        finish_mock.assert_not_called()
         self.assertFalse(
             [call for call in logger_mock.info.call_args_list if "洗版完成" in call.args[0]],
-            "update_subscribe_priority should not emit completion logs before finish_subscribe_or_not finishes",
+            "record_subscribe_download_facts should not emit completion logs before finish_subscribe_or_not finishes",
         )
 
-    def test_full_best_version_updates_all_episodes_when_pack_has_no_episode_metadata(self):
+    def test_download_facts_require_full_coverage_confirmation_when_pack_has_no_episode_metadata(self):
         subscribe = self._build_subscribe(
             best_version_full=1,
             total_episode=3,
@@ -1489,30 +1581,23 @@ class SubscribeChainTest(TestCase):
         )
         download = self._build_download(priority=100, selected_episodes=[], meta_episodes=[])
         chain = SubscribeChain()
-        meta = SimpleNamespace()
         mediainfo = SimpleNamespace(title_year="Test Show (2026)")
 
-        with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper") as subscribe_oper_cls, patch.object(
-            SubscribeChain,
-            "_SubscribeChain__finish_subscribe",
-        ) as finish_mock:
+        with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper") as subscribe_oper_cls:
             subscribe_oper = subscribe_oper_cls.return_value
             subscribe_oper.update.return_value = None
 
-            chain.update_subscribe_priority(
+            snapshot = chain._SubscribeChain__record_subscribe_download_facts(
                 subscribe=subscribe,
-                meta=meta,
                 mediainfo=mediainfo,
                 downloads=[download],
             )
 
-        payload = subscribe_oper.update.call_args.args[1]
-        self.assertEqual(payload["episode_priority"], {"1": 100, "2": 100, "3": 100})
-        self.assertEqual(payload["current_priority"], 100)
-        self.assertNotIn("lack_episode", payload)
-        finish_mock.assert_not_called()
+        self.assertEqual(snapshot["episodes"], [])
+        subscribe_oper.update.assert_not_called()
+        self.assertEqual(subscribe.episode_priority, {"1": 80, "2": 80, "3": 80})
 
-    def test_episode_best_version_updates_all_episodes_when_full_pack_has_no_episode_metadata(self):
+    def test_download_facts_write_all_targets_when_full_coverage_is_confirmed(self):
         subscribe = self._build_subscribe(
             best_version_full=0,
             total_episode=3,
@@ -1521,29 +1606,25 @@ class SubscribeChainTest(TestCase):
             lack_episode=3,
         )
         download = self._build_download(priority=100, selected_episodes=[], meta_episodes=[])
+        download.confirmed_full_coverage = True
         chain = SubscribeChain()
-        meta = SimpleNamespace()
         mediainfo = SimpleNamespace(title_year="Test Show (2026)")
 
-        with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper") as subscribe_oper_cls, patch.object(
-            SubscribeChain,
-            "_SubscribeChain__finish_subscribe",
-        ) as finish_mock:
+        with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper") as subscribe_oper_cls:
             subscribe_oper = subscribe_oper_cls.return_value
             subscribe_oper.update.return_value = None
 
-            chain.update_subscribe_priority(
+            chain._SubscribeChain__record_subscribe_download_facts(
                 subscribe=subscribe,
-                meta=meta,
                 mediainfo=mediainfo,
                 downloads=[download],
             )
 
         payload = subscribe_oper.update.call_args.args[1]
         self.assertEqual(payload["episode_priority"], {"1": 100, "2": 100, "3": 100})
-        self.assertEqual(payload["current_priority"], 100)
+        self.assertEqual(payload["note"], [1, 2, 3])
+        self.assertNotIn("current_priority", payload)
         self.assertNotIn("lack_episode", payload)
-        finish_mock.assert_not_called()
 
     def test_finish_subscribe_or_not_does_not_finish_best_version_twice_after_download_completion(self):
         """洗版订阅本轮下载已触发完成时，流程尾部不应对同一订阅再次完成。"""
@@ -1578,7 +1659,7 @@ class SubscribeChainTest(TestCase):
 
         finish_mock.assert_called_once_with(subscribe=subscribe, meta=meta, mediainfo=mediainfo)
 
-    def test_check_resets_current_priority_when_new_episodes_expand_target_range(self):
+    def test_check_keeps_sparse_priority_when_new_episodes_expand_target_range(self):
         subscribe = self._build_subscribe(
             total_episode=3,
             episode_priority={"1": 100, "2": 100, "3": 100},
@@ -1609,7 +1690,9 @@ class SubscribeChainTest(TestCase):
         self.assertEqual(payload["total_episode"], 5)
         self.assertEqual(payload["lack_episode"], 2)
         self.assertEqual(payload["current_priority"], 0)
-        self.assertEqual(payload["episode_priority"], {"1": 100, "2": 100, "3": 100, "4": 0, "5": 0})
+        self.assertEqual(payload["episode_priority"], {"1": 100, "2": 100, "3": 100})
+        self.assertNotIn("4", payload["episode_priority"])
+        self.assertNotIn("5", payload["episode_priority"])
         self.assertEqual(subscribe.total_episode, 5)
         self.assertEqual(subscribe.lack_episode, 2)
         self.assertEqual(subscribe.current_priority, 0)
@@ -1642,7 +1725,7 @@ class SubscribeChainTest(TestCase):
     def test_best_version_interested_episodes_uses_title_episode_list_for_full_pack(self):
         """整包候选（标题展开的集列表）只把仍可提升优先级的集纳入 interested。
 
-        防回归场景：标题显示"第53-104集"，实际目标范围只有 1..92，episode_priority
+        标题显示"第53-104集"，实际目标范围只有 1..92，episode_priority
         已经把 1..82 升到 100，E83 已经记到 99。同 pri_order=99 的同一资源再来时，
         interested 应只剩 [84..92]，绝不能含 E83，否则后续下载层会再下一次同优先级。
         """
@@ -1703,7 +1786,7 @@ class SubscribeFilterAllowedEpisodesTest(TestCase):
 
         self.assertIsNotNone(context.allowed_episodes)
         self.assertEqual(context.allowed_episodes, set(range(84, 93)))
-        # 关键回归点：E83 已达到 99，不在允许集内；下游交集后即不会再下 E83。
+        # E83 已达到 99，不在允许集内；下游交集后即不会再下 E83。
         self.assertNotIn(83, context.allowed_episodes)
 
     def test_filter_leaves_allowed_episodes_none_when_no_upgrade(self):
@@ -1734,7 +1817,7 @@ class SubscribeFilterAllowedEpisodesTest(TestCase):
         判定。此用例复刻 match() 的过滤序列，验证浅拷贝后的 _context 在写入
         allowed_episodes 时不会污染原始 context，且写入结果与 search() 一致。
         若 match() 分支漏写 allowed_episodes，下游 batch_download 将看不到允许集
-        约束，回归到 2c458317 之前的同优先级重复下载状态。
+        约束，导致同优先级资源重复下载。
         """
         import copy
 
@@ -1770,9 +1853,8 @@ class SubscribeFilterAllowedEpisodesTest(TestCase):
 class SubscribeNoteTrackingTest(TestCase):
     """覆盖洗版与非洗版下 subscribe.note 的下载历史追踪。
 
-    回归目标：finish_subscribe_or_not 必须在所有订阅模式下都把本轮下载的集数追加进
-    subscribe.note；__get_downloaded 在洗版分支必须把 note 与 episode_priority==100
-    的完成集合并返回，避免迁移或低优先级下载场景下已下集被误判为"未下载"。
+    finish_subscribe_or_not 有下载事实时必须追加 note；__get_downloaded 在洗版
+    分支只返回 priority==100 的完成集，普通订阅分支继续读取 note。
     """
 
     def _build_subscribe(self, **overrides):
@@ -1793,12 +1875,7 @@ class SubscribeNoteTrackingTest(TestCase):
         )
 
     def test_finish_subscribe_writes_note_for_best_version_downloads(self):
-        """洗版分支若产生 downloads，subscribe.note 必须被追加，不再被 best_version 标志拦截。
-
-        旧逻辑只在非洗版分支调用 __update_subscribe_note，导致 best_version=1 时
-        下载历史只落在 episode_priority；用户切回普通订阅或排障对账时缺失"下过哪些集"
-        的事实源。这条用例验证修复后两个分支都会写 note。
-        """
+        """洗版分支若产生 downloads，subscribe.note 必须被追加。"""
         subscribe = self._build_subscribe(
             best_version=1,
             total_episode=92,
@@ -1819,7 +1896,7 @@ class SubscribeNoteTrackingTest(TestCase):
 
         with patch.object(SUBSCRIBE_CHAIN_MODULE, "SubscribeOper", _SubscribeOper), patch.object(
             SubscribeChain,
-            "update_subscribe_priority",
+            "_SubscribeChain__update_movie_best_version_download_priority",
         ), patch.object(
             SubscribeChain,
             "_SubscribeChain__finish_subscribe",
@@ -1877,7 +1954,7 @@ class SubscribeNoteTrackingTest(TestCase):
         )
 
     def test_get_downloaded_best_version_returns_only_completed_episodes(self):
-        """关键回归：洗版分支不得把 note 合并进 __get_downloaded 返回值。
+        """洗版分支不得把 note 合并进 __get_downloaded 返回值。
 
         否则 check_and_handle_existing_media → __get_subscribe_no_exits 会把
         priority<100 但已下载的集从 pending no_exists 中减掉，配合 force=True 但
@@ -1899,17 +1976,839 @@ class SubscribeNoteTrackingTest(TestCase):
         self.assertNotIn(3, downloaded)
 
     def test_get_downloaded_non_best_version_reads_note_after_wash_migration(self):
-        """迁移场景：洗版期间 finish_subscribe_or_not 把下载集写入 note；
-        用户随后把 best_version 关掉，订阅切回普通模式时 __get_downloaded
-        从非洗版分支读取 note，旧洗版集仍能作为"已下载"被识别，避免重新匹配。
-        """
+        """订阅切回普通模式时 __get_downloaded 从非洗版分支读取 note。"""
         subscribe = self._build_subscribe(
             best_version=0,
             total_episode=5,
-            episode_priority={"1": 100, "2": 99},  # 旧洗版残留，普通分支不读
+            episode_priority={"1": 100, "2": 99},  # 普通分支不读取按集洗版优先级。
             note=[1, 2, 3],
         )
 
         downloaded = SubscribeChain._SubscribeChain__get_downloaded(subscribe)
 
         self.assertEqual(downloaded, [1, 2, 3])
+
+
+class SubscribeProgressEntrypointTest(TestCase):
+    def setUp(self):
+        self.module, self.SubscribeChain = _load_subscribe_chain_class()
+
+    def _build_subscribe(self, **overrides):
+        values = {
+            "id": 1,
+            "name": "测试剧",
+            "type": MediaType.TV.value,
+            "season": 1,
+            "start_episode": 1,
+            "total_episode": 5,
+            "lack_episode": 5,
+            "note": [],
+            "best_version": 1,
+            "best_version_full": 0,
+            "current_priority": None,
+            "episode_priority": {},
+            "last_update": None,
+            "tmdbid": 10001,
+            "doubanid": None,
+            "year": "2026",
+            "manual_total_episode": 0,
+        }
+        values.update(overrides)
+        return self.module.Subscribe(**values)
+
+    def test_compute_lack_episode_counts_best_version_note_and_positive_priority(self):
+        subscribe = self._build_subscribe(
+            note=[1, "bad"],
+            episode_priority={"2": 80, "3": 0, "4": 100},
+        )
+
+        lack = self.SubscribeChain.compute_lack_episode(subscribe)
+
+        self.assertEqual(lack, 2)
+
+    def test_compute_lack_episode_normal_tv_no_exists_boundaries(self):
+        subscribe = self._build_subscribe(best_version=0, note=[1])
+        missing_all = {
+            10001: {
+                1: self.module.schemas.NotExistMediaInfo(
+                    season=1, episodes=[], total_episode=5, start_episode=1
+                )
+            }
+        }
+        missing_some = {
+            10001: {
+                1: self.module.schemas.NotExistMediaInfo(
+                    season=1, episodes=[2, 4], total_episode=5, start_episode=1
+                )
+            }
+        }
+
+        self.assertEqual(self.SubscribeChain.compute_lack_episode(subscribe, no_exists={}), 0)
+        self.assertEqual(self.SubscribeChain.compute_lack_episode(subscribe, no_exists={"other": {}}), 0)
+        self.assertEqual(self.SubscribeChain.compute_lack_episode(subscribe, no_exists=missing_all), 5)
+        self.assertEqual(self.SubscribeChain.compute_lack_episode(subscribe, no_exists=missing_some), 2)
+
+    def test_compute_lack_episode_defaults_empty_no_exists_for_normal_tv(self):
+        subscribe = self._build_subscribe(best_version=0, note=[1])
+
+        self.assertEqual(self.SubscribeChain.compute_lack_episode(subscribe), 0)
+
+    def test_note_only_backfill_does_not_satisfy_best_version_quality_target(self):
+        subscribe = self._build_subscribe(
+            total_episode=3,
+            note=[1],
+            episode_priority={},
+            lack_episode=2,
+        )
+
+        self.assertEqual(self.SubscribeChain.compute_lack_episode(subscribe), 2)
+        self.assertEqual(self.SubscribeChain._get_pending_best_version_episodes(subscribe), [1, 2, 3])
+
+    def test_backfill_existing_episodes_writes_note_only_without_priority(self):
+        subscribe = self._build_subscribe(note=[1], episode_priority={"2": 80}, lack_episode=4)
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append((subscribe_id, payload))
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()):
+            summary = self.SubscribeChain().backfill_existing_episodes(
+                subscribe,
+                [1, 2, 3, 9, "bad"],
+                priority=None,
+                scene="unit",
+            )
+
+        self.assertEqual(summary["accepted"], [2, 3])
+        self.assertEqual(summary["ignored"], [
+            {"episode": 1, "reason": "duplicate"},
+            {"episode": 9, "reason": "out_of_range"},
+            {"episode": "bad", "reason": "invalid"},
+        ])
+        self.assertEqual(subscribe.note, [1, 2, 3])
+        self.assertEqual(subscribe.episode_priority, {"2": 80})
+        self.assertEqual(subscribe.lack_episode, 2)
+        self.assertEqual(updates[-1][1]["lack_episode"], 2)
+
+    def test_backfill_existing_episodes_writes_priority_only_upwards(self):
+        subscribe = self._build_subscribe(note=[], episode_priority={"1": 90, "2": 100}, lack_episode=5)
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append(payload)
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()):
+            summary = self.SubscribeChain().backfill_existing_episodes(
+                subscribe,
+                [1, 2, 3],
+                priority=100,
+                scene="unit",
+            )
+
+        self.assertEqual(summary["accepted"], [1, 2, 3])
+        self.assertEqual(subscribe.note, [1, 2, 3])
+        self.assertEqual(subscribe.episode_priority, {"1": 100, "2": 100, "3": 100})
+        self.assertEqual(subscribe.current_priority, 0)
+        self.assertEqual(updates[-1]["current_priority"], 0)
+
+    def test_backfill_existing_episodes_ignores_invalid_priority_and_does_not_downgrade(self):
+        subscribe = self._build_subscribe(note=[], episode_priority={"1": 90}, lack_episode=5)
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append(payload)
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()):
+            invalid = self.SubscribeChain().backfill_existing_episodes(
+                subscribe,
+                [1, 2],
+                priority=101,
+                scene="unit",
+            )
+            lower = self.SubscribeChain().backfill_existing_episodes(
+                subscribe,
+                [1, 2],
+                priority=80,
+                scene="unit",
+            )
+            boolean_priority = self.SubscribeChain().backfill_existing_episodes(
+                subscribe,
+                [3],
+                priority=True,
+                scene="unit",
+            )
+
+        self.assertEqual(invalid["accepted"], [1, 2])
+        self.assertEqual(invalid["ignored_priority"], 101)
+        self.assertEqual(lower["accepted"], [])
+        self.assertEqual(lower["ignored"], [
+            {"episode": 1, "reason": "duplicate"},
+            {"episode": 2, "reason": "duplicate"},
+        ])
+        self.assertEqual(lower["priority_ignored"], [
+            {"episode": 1, "reason": "not_higher_priority"},
+        ])
+        self.assertEqual(lower["priority_updated"], [2])
+        self.assertEqual(boolean_priority["accepted"], [3])
+        self.assertEqual(boolean_priority["ignored_priority"], True)
+        self.assertEqual(subscribe.note, [1, 2, 3])
+        self.assertEqual(subscribe.episode_priority, {"1": 90, "2": 80})
+
+    def test_backfill_existing_episodes_accepts_note_without_downgrading_priority(self):
+        subscribe = self._build_subscribe(note=[], episode_priority={"1": 90}, lack_episode=5)
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append(payload)
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()):
+            summary = self.SubscribeChain().backfill_existing_episodes(
+                subscribe,
+                [1],
+                priority=80,
+                scene="unit",
+            )
+
+        self.assertEqual(summary["accepted"], [1])
+        self.assertEqual(summary["priority_updated"], [])
+        self.assertEqual(subscribe.note, [1])
+        self.assertEqual(subscribe.episode_priority, {"1": 90})
+        self.assertNotIn("episode_priority", updates[-1])
+
+    def test_backfill_existing_episodes_updates_priority_for_existing_note(self):
+        subscribe = self._build_subscribe(note=[1], episode_priority={}, lack_episode=4)
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append(payload)
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()):
+            summary = self.SubscribeChain().backfill_existing_episodes(
+                subscribe,
+                [1],
+                priority=100,
+                scene="unit",
+            )
+
+        self.assertEqual(summary["accepted"], [])
+        self.assertEqual(summary["ignored"], [{"episode": 1, "reason": "duplicate"}])
+        self.assertEqual(summary["priority_updated"], [1])
+        self.assertEqual(subscribe.note, [1])
+        self.assertEqual(subscribe.episode_priority, {"1": 100})
+        self.assertEqual(updates[-1]["episode_priority"], {"1": 100})
+
+    def test_backfill_existing_episodes_marks_current_priority_complete_only_when_all_targets_are_top(self):
+        subscribe = self._build_subscribe(note=[], episode_priority={"1": 90}, lack_episode=5)
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append(payload)
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()):
+            summary = self.SubscribeChain().backfill_existing_episodes(
+                subscribe,
+                [1, 2, 3, 4, 5],
+                priority=100,
+                scene="unit",
+            )
+
+        self.assertEqual(summary["accepted"], [1, 2, 3, 4, 5])
+        self.assertEqual(subscribe.current_priority, 100)
+        self.assertEqual(updates[-1]["current_priority"], 100)
+
+    def test_backfill_materializes_legacy_current_priority_before_partial_write(self):
+        subscribe = self._build_subscribe(
+            total_episode=3,
+            current_priority=80,
+            episode_priority=None,
+            note=[],
+            lack_episode=0,
+        )
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append(payload)
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()):
+            self.SubscribeChain().backfill_existing_episodes(
+                subscribe,
+                [3],
+                priority=100,
+                scene="unit",
+            )
+
+        self.assertEqual(subscribe.episode_priority, {"1": 80, "2": 80, "3": 100})
+        self.assertEqual(subscribe.note, [3])
+        self.assertEqual(subscribe.current_priority, 80)
+        self.assertEqual(updates[-1]["episode_priority"], {"1": 80, "2": 80, "3": 100})
+
+    def test_backfill_existing_episodes_refreshes_normal_tv_with_public_progress_entrypoint(self):
+        subscribe = self._build_subscribe(best_version=0, note=[], lack_episode=5)
+        progress_calls = []
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append(payload)
+
+        chain = self.SubscribeChain()
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()), \
+                patch.object(chain, "refresh_subscribe_progress", return_value={
+                    "scene": "unit",
+                    "updated": True,
+                    "fields": ["lack_episode"],
+                    "lack_episode": 4,
+                    "reason": "updated",
+                }) as refresh_progress:
+            summary = chain.backfill_existing_episodes(
+                subscribe,
+                [1],
+                priority=None,
+                scene="unit",
+            )
+            progress_calls.append(refresh_progress.call_args)
+
+        refresh_progress.assert_called_once_with(subscribe, scene="unit")
+        self.assertEqual(summary["accepted"], [1])
+        self.assertEqual(summary["progress"]["fields"], ["lack_episode"])
+        self.assertNotIn("lack_episode", updates[0])
+        self.assertEqual(subscribe.note, [1])
+        self.assertTrue(progress_calls)
+
+    def test_refresh_subscribe_progress_lowers_current_priority_for_partial_historical_episode_priority(self):
+        subscribe = self._build_subscribe(
+            total_episode=3,
+            current_priority=80,
+            episode_priority={"1": 100},
+            lack_episode=0,
+        )
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append(payload)
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()):
+            summary = self.SubscribeChain()._SubscribeChain__refresh_subscribe_progress_with_no_exists(
+                subscribe=subscribe,
+                no_exists={},
+                scene="unit",
+            )
+
+        self.assertTrue(summary["updated"])
+        self.assertEqual(subscribe.current_priority, 0)
+        self.assertEqual(updates[-1]["current_priority"], 0)
+
+    def test_refresh_subscribe_progress_normal_tv_uses_resolve_missing_successfully(self):
+        subscribe = self._build_subscribe(best_version=0, lack_episode=5)
+        mediainfo = SimpleNamespace(
+            type=MediaType.TV,
+            tmdb_id=10001,
+            douban_id=None,
+            title_year="测试剧 (2026)",
+            seasons={1: [1, 2, 3, 4, 5]},
+        )
+        no_exists = {
+            10001: {
+                1: self.module.schemas.NotExistMediaInfo(
+                    season=1, episodes=[2, 4], total_episode=5, start_episode=1
+                )
+            }
+        }
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append(payload)
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()), \
+                patch.object(self.SubscribeChain, "recognize_media", return_value=mediainfo), \
+                patch.object(self.SubscribeChain, "resolve_subscribe_missing", return_value=(False, no_exists)) as resolve_missing:
+            summary = self.SubscribeChain().refresh_subscribe_progress(subscribe, scene="unit")
+
+        resolve_missing.assert_called_once()
+        _, kwargs = resolve_missing.call_args
+        self.assertIs(kwargs["subscribe"], subscribe)
+        self.assertIsNotNone(kwargs["meta"])
+        self.assertEqual(kwargs["meta"].type, MediaType.TV)
+        self.assertEqual(kwargs["meta"].name, subscribe.name)
+        self.assertEqual(kwargs["meta"].season_seq, "1")
+        self.assertIs(kwargs["mediainfo"], mediainfo)
+        self.assertEqual(kwargs["mediakey"], 10001)
+        self.assertTrue(summary["updated"])
+        self.assertEqual(summary["lack_episode"], 2)
+        self.assertEqual(subscribe.lack_episode, 2)
+        self.assertEqual(updates[-1]["lack_episode"], 2)
+
+    def test_refresh_subscribe_progress_normal_tv_resolve_failure_does_not_write_zero(self):
+        subscribe = self._build_subscribe(best_version=0, lack_episode=5)
+        mediainfo = SimpleNamespace(
+            type=MediaType.TV,
+            tmdb_id=10001,
+            douban_id=None,
+            title_year="测试剧 (2026)",
+            seasons={1: [1, 2, 3, 4, 5]},
+        )
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                raise AssertionError("resolve failure must not write progress")
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()), \
+                patch.object(self.SubscribeChain, "recognize_media", return_value=mediainfo), \
+                patch.object(self.SubscribeChain, "resolve_subscribe_missing", return_value=(False, {})):
+            summary = self.SubscribeChain().refresh_subscribe_progress(subscribe, scene="unit")
+
+        self.assertFalse(summary["updated"])
+        self.assertIn("reason", summary)
+        self.assertEqual(subscribe.lack_episode, 5)
+
+    def test_refresh_subscribe_progress_normal_tv_recognition_failure_does_not_write_zero(self):
+        subscribe = self._build_subscribe(best_version=0, lack_episode=5)
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                raise AssertionError("recognition failure must not write progress")
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()), \
+                patch.object(self.SubscribeChain, "recognize_media", return_value=None):
+            summary = self.SubscribeChain().refresh_subscribe_progress(subscribe, scene="unit")
+
+        self.assertFalse(summary["updated"])
+        self.assertIn("reason", summary)
+        self.assertEqual(subscribe.lack_episode, 5)
+
+    def test_refresh_subscribe_progress_rejects_raw_no_exists_for_public_signature(self):
+        subscribe = self._build_subscribe(best_version=0, lack_episode=5)
+
+        with self.assertRaises(TypeError):
+            self.SubscribeChain().refresh_subscribe_progress(subscribe, no_exists={})
+
+    def test_finish_subscribe_progress_writer_keeps_empty_lefts_as_zero_for_normal_tv(self):
+        subscribe = self._build_subscribe(best_version=0, lack_episode=5)
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append(payload)
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()), \
+                patch.object(self.SubscribeChain, "_SubscribeChain__finish_subscribe"):
+            self.SubscribeChain().finish_subscribe_or_not(
+                subscribe=subscribe,
+                meta=SimpleNamespace(type=MediaType.TV),
+                mediainfo=SimpleNamespace(title_year="测试剧 (2026)"),
+                downloads=None,
+                lefts=None,
+            )
+
+        self.assertEqual(subscribe.lack_episode, 0)
+        self.assertEqual(updates[-1]["lack_episode"], 0)
+
+
+class SubscribeProgressConsolidationTest(TestCase):
+    def _mediainfo(self, total_episode=5):
+        return SimpleNamespace(
+            seasons={1: [object() for _ in range(total_episode)]},
+            title="总集增长剧",
+            year="2026",
+            vote_average=9.5,
+            overview="overview",
+            imdb_id="tt1234567",
+            tvdb_id=99,
+            get_poster_image=lambda: "poster",
+            get_backdrop_image=lambda: "backdrop",
+        )
+
+    def test_refresh_total_episode_before_completion_reuses_progress_priority_snapshot(self):
+        module, SubscribeChain = _load_subscribe_chain_class()
+        subscribe = module.Subscribe(
+            id=31,
+            name="总集增长剧",
+            type=MediaType.TV.value,
+            season=1,
+            total_episode=3,
+            start_episode=1,
+            lack_episode=0,
+            best_version=1,
+            best_version_full=0,
+            current_priority=80,
+            episode_priority=None,
+            note=[],
+            tmdbid=31031,
+            doubanid=None,
+            manual_total_episode=0,
+        )
+        mediainfo = self._mediainfo(total_episode=5)
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append((subscribe_id, payload))
+
+        with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()):
+            SubscribeChain()._SubscribeChain__refresh_total_episode_before_completion(
+                subscribe,
+                mediainfo,
+            )
+
+        self.assertEqual(subscribe.total_episode, 5)
+        self.assertEqual(
+            subscribe.episode_priority,
+            {"1": 80, "2": 80, "3": 80},
+        )
+        self.assertEqual(subscribe.lack_episode, 2)
+        self.assertEqual(subscribe.current_priority, 0)
+        self.assertEqual(updates[-1][1]["lack_episode"], 2)
+        self.assertEqual(updates[-1][1]["current_priority"], 0)
+
+    def test_check_total_growth_reuses_progress_priority_snapshot(self):
+        module, SubscribeChain = _load_subscribe_chain_class()
+        subscribe = module.Subscribe(
+            id=33,
+            name="总集增长剧",
+            type=MediaType.TV.value,
+            season=1,
+            total_episode=3,
+            start_episode=1,
+            lack_episode=0,
+            best_version=1,
+            best_version_full=0,
+            current_priority=80,
+            episode_priority=None,
+            note=[],
+            year="2026",
+            episode_group=None,
+            tmdbid=31033,
+            doubanid=None,
+            manual_total_episode=0,
+        )
+        updates = []
+
+        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=5)
+
+        with patch.object(module, "SubscribeOper", return_value=_SubscribeOper()):
+            chain.check()
+
+        payload = updates[-1][1]
+        self.assertEqual(payload["total_episode"], 5)
+        self.assertEqual(
+            payload["episode_priority"],
+            {"1": 80, "2": 80, "3": 80},
+        )
+        self.assertEqual(payload["lack_episode"], 2)
+        self.assertEqual(payload["current_priority"], 0)
+
+    def test_completed_episode_uses_schema_function_directly_for_best_version(self):
+        module, SubscribeChain = _load_subscribe_chain_class()
+        values = {
+            "id": 32,
+            "name": "完成集数剧",
+            "type": MediaType.TV.value,
+            "season": 1,
+            "total_episode": 8,
+            "start_episode": 3,
+            "lack_episode": 2,
+            "best_version": 1,
+            "episode_priority": {"3": 100, "4": 80, "5": 100, "8": 100},
+        }
+        chain_subscribe = module.Subscribe(**values)
+        schema_subscribe = schemas.Subscribe(**values)
+
+        self.assertFalse(hasattr(SubscribeChain, "compute_completed_episode"))
+        self.assertEqual(schema_subscribe.completed_episode, schemas.compute_subscribe_completed_episode(chain_subscribe))
+
+    def test_completed_episode_uses_current_priority_when_episode_priority_empty(self):
+        module, _ = _load_subscribe_chain_class()
+        values = {
+            "id": 33,
+            "name": "完成集数旧快照剧",
+            "type": MediaType.TV.value,
+            "season": 1,
+            "total_episode": 3,
+            "start_episode": 1,
+            "lack_episode": 0,
+            "best_version": 1,
+            "current_priority": 100,
+            "episode_priority": None,
+        }
+
+        chain_subscribe = module.Subscribe(**values)
+        schema_subscribe = schemas.Subscribe(**values)
+
+        self.assertEqual(schema_subscribe.completed_episode, 3)
+        self.assertEqual(schemas.compute_subscribe_completed_episode(chain_subscribe), 3)
+
+
+class SubscribeDownloadFactsTest(TestCase):
+    def setUp(self):
+        self.module, self.SubscribeChain = _load_subscribe_chain_class()
+
+    def _build_subscribe(self, **overrides):
+        values = {
+            "id": 3,
+            "name": "下载事实剧",
+            "type": MediaType.TV.value,
+            "season": 1,
+            "start_episode": 1,
+            "total_episode": 4,
+            "lack_episode": 4,
+            "note": [],
+            "best_version": 0,
+            "best_version_full": 0,
+            "current_priority": None,
+            "episode_priority": {},
+            "tmdbid": 30003,
+            "doubanid": None,
+            "manual_total_episode": 0,
+        }
+        values.update(overrides)
+        return self.module.Subscribe(**values)
+
+    def _download(self, episodes=None, pri_order=80, selected_episodes=None, confirmed_full_coverage=False):
+        return SimpleNamespace(
+            selected_episodes=selected_episodes,
+            confirmed_full_coverage=confirmed_full_coverage,
+            torrent_info=SimpleNamespace(pri_order=pri_order),
+            meta_info=SimpleNamespace(episode_list=episodes or [], season_list=[1]),
+            media_info=SimpleNamespace(type=MediaType.TV, tmdb_id=30003, douban_id=None),
+        )
+
+    def test_normal_tv_download_records_note_and_episode_priority_without_current_priority(self):
+        subscribe = self._build_subscribe(best_version=0)
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append(payload)
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()):
+            snapshot = self.SubscribeChain()._SubscribeChain__record_subscribe_download_facts(
+                subscribe,
+                mediainfo=SimpleNamespace(title_year="下载事实剧 (2026)"),
+                downloads=[self._download(episodes=[1, 2], pri_order=70)],
+            )
+
+        self.assertEqual(snapshot["episodes"], [1, 2])
+        self.assertEqual(subscribe.note, [1, 2])
+        self.assertEqual(subscribe.episode_priority, {"1": 70, "2": 70})
+        self.assertIsNone(subscribe.current_priority)
+        self.assertNotIn("current_priority", updates[-1])
+
+    def test_normal_tv_download_records_full_pack_confirmed_coverage_episode_priority(self):
+        subscribe = self._build_subscribe(best_version=0, best_version_full=0, total_episode=3, episode_priority={})
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append(payload)
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()):
+            snapshot = self.SubscribeChain()._SubscribeChain__record_subscribe_download_facts(
+                subscribe,
+                mediainfo=SimpleNamespace(title_year="下载事实剧 (2026)"),
+                downloads=[
+                    self._download(
+                        episodes=[],
+                        pri_order=80,
+                        selected_episodes=[],
+                        confirmed_full_coverage=True,
+                    )
+                ],
+            )
+
+        self.assertEqual(snapshot["episodes"], [1, 2, 3])
+        self.assertEqual(subscribe.note, [1, 2, 3])
+        self.assertEqual(subscribe.episode_priority, {"1": 80, "2": 80, "3": 80})
+        self.assertNotIn("current_priority", updates[-1])
+
+    def test_full_resource_without_episode_list_does_not_fallback_without_download_confirmation(self):
+        subscribe = self._build_subscribe(best_version=1, best_version_full=1, episode_priority={"1": 60})
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append(payload)
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()):
+            snapshot = self.SubscribeChain()._SubscribeChain__record_subscribe_download_facts(
+                subscribe,
+                mediainfo=SimpleNamespace(title_year="下载事实剧 (2026)"),
+                downloads=[self._download(episodes=[], pri_order=90, selected_episodes=[])],
+            )
+
+        self.assertEqual(snapshot["episodes"], [])
+        self.assertEqual(subscribe.note, [])
+        self.assertEqual(subscribe.episode_priority, {"1": 60})
+        self.assertEqual(updates, [])
+
+    def test_full_resource_without_episode_list_uses_target_range_only_when_confirmed(self):
+        subscribe = self._build_subscribe(best_version=1, best_version_full=1, episode_priority={"1": 60})
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append(payload)
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()):
+            snapshot = self.SubscribeChain()._SubscribeChain__record_subscribe_download_facts(
+                subscribe,
+                mediainfo=SimpleNamespace(title_year="下载事实剧 (2026)"),
+                downloads=[
+                    self._download(
+                        episodes=[],
+                        pri_order=90,
+                        selected_episodes=[],
+                        confirmed_full_coverage=True,
+                    )
+                ],
+            )
+
+        self.assertEqual(snapshot["episodes"], [1, 2, 3, 4])
+        self.assertEqual(subscribe.note, [1, 2, 3, 4])
+        self.assertEqual(subscribe.episode_priority, {"1": 90, "2": 90, "3": 90, "4": 90})
+        self.assertNotIn("current_priority", updates[-1])
+
+    def test_normal_subscription_without_episode_list_does_not_use_target_range_without_download_confirmation(self):
+        subscribe = self._build_subscribe(best_version=0, best_version_full=0)
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append(payload)
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()):
+            snapshot = self.SubscribeChain()._SubscribeChain__record_subscribe_download_facts(
+                subscribe,
+                mediainfo=SimpleNamespace(title_year="下载事实剧 (2026)"),
+                downloads=[
+                    self._download(
+                        episodes=[],
+                        pri_order=90,
+                        selected_episodes=[],
+                        confirmed_full_coverage=False,
+                    )
+                ],
+            )
+
+        self.assertEqual(snapshot["episodes"], [])
+        self.assertEqual(subscribe.note, [])
+        self.assertEqual(subscribe.episode_priority, {})
+        self.assertEqual(updates, [])
+
+    def test_movie_best_version_download_keeps_current_priority_without_episode_priority(self):
+        subscribe = self._build_subscribe(
+            type=MediaType.MOVIE.value,
+            best_version=1,
+            best_version_full=0,
+            current_priority=60,
+            episode_priority={},
+            note=[],
+            tmdbid=30003,
+            total_episode=1,
+            lack_episode=1,
+        )
+        download = self._download(episodes=[], pri_order=90)
+        download.media_info = SimpleNamespace(type=MediaType.MOVIE, tmdb_id=30003, douban_id=None)
+        download.meta_info = SimpleNamespace(episode_list=[], season_list=[])
+        updates = []
+
+        class _SubscribeOper:
+            def update(self, subscribe_id, payload):
+                updates.append(payload)
+
+        with patch.object(self.module, "SubscribeOper", return_value=_SubscribeOper()), \
+                patch.object(self.SubscribeChain, "_SubscribeChain__finish_subscribe"):
+            self.SubscribeChain().finish_subscribe_or_not(
+                subscribe=subscribe,
+                meta=SimpleNamespace(type=MediaType.MOVIE),
+                mediainfo=SimpleNamespace(title_year="下载事实电影 (2026)"),
+                downloads=[download],
+                lefts={},
+            )
+
+        self.assertEqual(subscribe.current_priority, 90)
+        self.assertTrue(subscribe.last_update)
+        self.assertEqual(subscribe.episode_priority, {})
+        self.assertIn({"current_priority": 90, "last_update": subscribe.last_update}, updates)
+
+    def test_movie_best_version_download_does_not_call_tv_progress_writer(self):
+        subscribe = self._build_subscribe(
+            type=MediaType.MOVIE.value,
+            best_version=1,
+            best_version_full=0,
+            current_priority=60,
+            episode_priority={},
+            note=[],
+            tmdbid=30003,
+            total_episode=1,
+            lack_episode=1,
+        )
+        download = self._download(episodes=[], pri_order=90)
+        download.media_info = SimpleNamespace(type=MediaType.MOVIE, tmdb_id=30003, douban_id=None)
+        download.meta_info = SimpleNamespace(episode_list=[], season_list=[])
+        chain = self.SubscribeChain()
+
+        with patch.object(self.module, "SubscribeOper") as subscribe_oper_cls, \
+                patch.object(chain, "_SubscribeChain__refresh_subscribe_progress_with_no_exists") as refresh_mock, \
+                patch.object(chain, "_SubscribeChain__finish_subscribe"):
+            subscribe_oper = subscribe_oper_cls.return_value
+            subscribe_oper.update.return_value = None
+
+            chain.finish_subscribe_or_not(
+                subscribe=subscribe,
+                meta=SimpleNamespace(type=MediaType.MOVIE),
+                mediainfo=SimpleNamespace(title_year="下载事实电影 (2026)"),
+                downloads=[download],
+                lefts={},
+            )
+
+        refresh_mock.assert_not_called()
+
+    def test_movie_normal_download_does_not_call_tv_progress_writer(self):
+        subscribe = self._build_subscribe(
+            type=MediaType.MOVIE.value,
+            best_version=0,
+            best_version_full=0,
+            current_priority=None,
+            episode_priority={},
+            note=[],
+            tmdbid=30003,
+            total_episode=1,
+            lack_episode=1,
+        )
+        download = self._download(episodes=[], pri_order=90)
+        download.media_info = SimpleNamespace(type=MediaType.MOVIE, tmdb_id=30003, douban_id=None)
+        download.meta_info = SimpleNamespace(episode_list=[], season_list=[])
+        chain = self.SubscribeChain()
+
+        with patch.object(self.module, "SubscribeOper") as subscribe_oper_cls, \
+                patch.object(chain, "_SubscribeChain__refresh_subscribe_progress_with_no_exists") as refresh_mock, \
+                patch.object(chain, "_SubscribeChain__finish_subscribe"):
+            subscribe_oper = subscribe_oper_cls.return_value
+            subscribe_oper.update.return_value = None
+
+            chain.finish_subscribe_or_not(
+                subscribe=subscribe,
+                meta=SimpleNamespace(type=MediaType.MOVIE),
+                mediainfo=SimpleNamespace(title_year="下载事实电影 (2026)"),
+                downloads=[download],
+                lefts={},
+            )
+
+        refresh_mock.assert_not_called()

From a4f6e138815cbcb6378e32e344896ab7f5e3d069 Mon Sep 17 00:00:00 2001
From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com>
Date: Mon, 29 Jun 2026 06:44:40 +0800
Subject: [PATCH 2/2] fix(transfer): expose file meta in intercept event
 (#6019)

---
 app/modules/filemanager/transhandler.py |  4 ++
 app/schemas/event.py                    |  2 +
 tests/test_transfer_job_manager.py      | 75 +++++++++++++++++++++++++
 3 files changed, 81 insertions(+)

diff --git a/app/modules/filemanager/transhandler.py b/app/modules/filemanager/transhandler.py
index e207445d..95803d40 100644
--- a/app/modules/filemanager/transhandler.py
+++ b/app/modules/filemanager/transhandler.py
@@ -536,6 +536,7 @@ class TransHandler:
                 # 整理文件
                 new_item, err_msg = self.__transfer_file(
                     fileitem=fileitem,
+                    meta=in_meta,
                     mediainfo=mediainfo,
                     target_storage=target_storage,
                     target_file=new_file,
@@ -962,6 +963,7 @@ class TransHandler:
     def __transfer_file(
         self,
         fileitem: FileItem,
+        meta: Optional[MetaBase],
         mediainfo: MediaInfo,
         source_oper: StorageBase,
         target_oper: StorageBase,
@@ -974,6 +976,7 @@ class TransHandler:
         """
         整理一个文件，同时处理其他相关文件
         :param fileitem: 原文件
+        :param meta: 元数据
         :param mediainfo: 媒体信息
         :param source_oper: 源存储操作对象
         :param target_oper: 目标存储操作对象
@@ -990,6 +993,7 @@ class TransHandler:
         )
         event_data = TransferInterceptEventData(
             fileitem=fileitem,
+            meta=meta,
             mediainfo=mediainfo,
             target_storage=target_storage,
             target_path=target_file,
diff --git a/app/schemas/event.py b/app/schemas/event.py
index 75c9f21e..15c35090 100644
--- a/app/schemas/event.py
+++ b/app/schemas/event.py
@@ -388,6 +388,7 @@ class TransferInterceptEventData(ChainEventData):
     Attributes:
         # 输入参数
         fileitem (FileItem): 源文件
+        meta (Any): 元数据
         target_storage (str): 目标存储
         target_path (Path): 目标路径
         transfer_type (str): 整理方式（copy、move、link、softlink等）
@@ -401,6 +402,7 @@ class TransferInterceptEventData(ChainEventData):
 
     # 输入参数
     fileitem: FileItem = Field(..., description="源文件")
+    meta: Optional[Any] = Field(default=None, description="元数据")
     mediainfo: Any = Field(..., description="媒体信息")
     target_storage: str = Field(..., description="目标存储")
     target_path: Path = Field(..., description="目标路径")
diff --git a/tests/test_transfer_job_manager.py b/tests/test_transfer_job_manager.py
index d92f85d0..7e02f74b 100644
--- a/tests/test_transfer_job_manager.py
+++ b/tests/test_transfer_job_manager.py
@@ -265,6 +265,81 @@ class TransferJobManagerTest(unittest.TestCase):
         self.assertEqual(target_item, transferinfo.target_item)
         self.assertEqual(target_folder, transferinfo.target_diritem)
 
+    def test_single_file_transfer_intercept_event_carries_file_meta(self):
+        """
+        单文件整理拦截事件应携带元数据，便于事件处理器按季集匹配。
+        """
+        handler = TransHandler()
+        source_item = FileItem(
+            storage="alist",
+            path="/downloads/Test.Show.S02E03.mkv",
+            type="file",
+            name="Test.Show.S02E03.mkv",
+            basename="Test.Show.S02E03",
+            extension="mkv",
+            size=1024,
+            modify_time=1715939275.0,
+        )
+        target_path = Path("/library")
+        target_file = Path("/library/Test.Show.S02E03.mkv")
+        target_folder = FileItem(
+            storage="alist",
+            type="dir",
+            path="/library/",
+            name="library",
+            basename="library",
+        )
+        target_item = FileItem(
+            storage="alist",
+            path=target_file.as_posix(),
+            type="file",
+            name=target_file.name,
+            basename=target_file.stem,
+            extension="mkv",
+            size=1024,
+        )
+        source_oper = SimpleNamespace(
+            is_support_transtype=lambda transfer_type: True,
+            move=lambda fileitem, path, name: True,
+        )
+        target_oper = SimpleNamespace(
+            get_folder=lambda path: target_folder,
+            get_item=lambda path: None,
+        )
+        in_meta = MetaVideo("Test.Show.S02E03")
+
+        with patch.object(
+                TransHandler, "get_rename_path", return_value=target_file
+        ), patch(
+                "app.modules.filemanager.transhandler.DirectoryHelper.get_media_root_path",
+                return_value=Path("/library"),
+        ), patch.object(
+                TransHandler,
+                "_TransHandler__transfer_command",
+                return_value=(target_item, ""),
+        ), patch(
+                "app.modules.filemanager.transhandler.eventmanager.send_event",
+                return_value=None,
+        ) as send_event:
+            transferinfo = handler.transfer_media(
+                fileitem=source_item,
+                in_meta=in_meta,
+                mediainfo=make_media_info(),
+                target_storage="alist",
+                target_path=target_path,
+                transfer_type="move",
+                source_oper=source_oper,
+                target_oper=target_oper,
+                need_scrape=True,
+                need_notify=True,
+            )
+
+        self.assertTrue(transferinfo.success)
+        event_data = send_event.call_args.args[1]
+        self.assertIs(in_meta, event_data.meta)
+        self.assertEqual(2, event_data.meta.begin_season)
+        self.assertEqual(3, event_data.meta.begin_episode)
+
     def test_success_callback_uses_transfer_result_target_diritem(self):
         """
         回调发送刮削事件时应直接使用整理结果里的目标目录项。