import copy
import json
import random
import threading
import time
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Callable, Dict, List, Optional, Union, Tuple

from app.schemas.mediaserver import NotExistMediaInfo as _SchemaNotExistMediaInfo
from app.schemas.message import Message as _SchemaMessage
from app.schemas.subscribe import SubscrbieInfo as _SchemaSubscrbieInfo
from app.schemas.subscribe import SubscribeDownloadFileInfo as _SchemaSubscribeDownloadFileInfo
from app.schemas.subscribe import SubscribeEpisodeInfo as _SchemaSubscribeEpisodeInfo
from app.schemas.subscribe import SubscribeLibraryFileInfo as _SchemaSubscribeLibraryFileInfo
from app.schemas.workflow import Subscribe as _SchemaSubscribe
from app.chain import ChainBase
from app.chain._interaction import InteractionChainMixin
from app.chain._music import MusicSubscribeMixin
from app.chain.download import DownloadChain
from app.chain.media import MediaChain
from app.chain.mediaserver import MediaServerChain
from app.chain.search import SearchChain
from app.chain.tmdb import TmdbChain
from app.chain.torrents import TorrentsChain
from app.runtime.config import global_vars
from app.domain.context import (
    Context,
    MediaInfo,
    TorrentInfo,
)
from app.runtime.events import eventmanager, Event
from app.domain.meta.metabase import MetaBase
from app.domain.meta.metamusic import MetaMusic
from app.domain.meta.words import WordsMatcher
from app.domain.metainfo import MetaInfo
from app.application.chain.data import (
    DownloadHistoryPortProxy as DownloadHistoryOper,
    SitePortProxy as SiteOper,
    SubscribePortProxy as SubscribeOper,
)
from app.application.configuration import (
    get_chain_runtime_config_snapshot,
    get_configured_system_config,
)
from app.application.messaging.subscribe import SubscribeInteractionHandler
from app.application.messaging.message import MessageTemplateHelper
from app.application.mediaserver import MediaServerHelper
from app.application.subscription.write import add_subscribe, async_add_subscribe
from app.application.subscription.complete import get_subscription_completion_scope
from app.application.subscription.delete import (
    SubscribeDeletionActor,
    get_sync_delete_subscribe_scope,
)
from app.application.subscription.contract import (
    build_subscribe_meta as _build_subscribe_meta,
    subscribe_media_key,
    subscribe_media_keys,
)
from app.application.subscription.query import SubscriptionQueryService
from app.adapters.external.server import MoviePilotServerHelper
from app.application.torrent import TorrentHelper
from app.runtime.log import logger
from app.schemas.event import SubscribeEpisodesRefreshEventData
from app.schemas.event import SubscribeCompletionCheckEventData
from app.schemas.types import MUSIC_ENTITY_ALBUM, MediaSource, MediaType, SystemConfigKey, NotificationChannel, MessageType, EventType, ChainEventType, \
    ContentType
from app.schemas.media import normalize_media_source, resolve_media_identity

if hasattr(_SchemaSubscribe, "model_fields"):
    Subscribe = _SchemaSubscribe
else:
    class Subscribe(_SchemaSubscribe):
        """隔离测试或旧插件缺少完整订阅模型时使用的轻量快照。"""

        def __init__(self, **kwargs):
            """初始化兼容快照并补齐链路必需的运行字段。"""
            super().__init__(**kwargs)
            for field, default in {
                "best_version_full": 0,
                "current_priority": None,
                "episode_priority": None,
                "total_episode": 0,
                "start_episode": 0,
                "lack_episode": 0,
                "note": [],
            }.items():
                if not hasattr(self, field):
                    setattr(self, field, default)

        def to_dict(self) -> dict:
            """返回快照字段，兼容整理链的响应转换。"""
            return dict(self.__dict__)

# 旧测试与第三方扩展可能替换该名字；它现在只是应用配置端口的本地别名，
# 不再指向数据库 Oper。
SystemConfigOper = get_configured_system_config
_DEFAULT_SYSTEM_CONFIG_PROVIDER = get_configured_system_config


@dataclass(frozen=True, slots=True)
class _SubscribePostCommitContext:
    """订阅提交后副作用所需的不可变业务快照。"""

    title: str
    year: str
    metainfo: MetaBase
    mediainfo: MediaInfo
    media_source: Optional[MediaSource]
    media_id: Optional[str]
    season: Optional[int]
    channel: Optional[NotificationChannel]
    source: Optional[str]
    userid: Optional[str]
    username: Optional[str]
    message: bool
    notification: Optional[dict] = None


@dataclass(slots=True)
class _SubscribeCreateContext:
    """订阅新增各阶段共享的显式状态，避免同步与异步入口各自维护散落变量。"""

    title: str
    year: str
    mtype: Optional[MediaType]
    episode_group: Optional[str]
    season: Optional[int]
    channel: Optional[NotificationChannel]
    source: Optional[str]
    userid: Optional[str]
    username: Optional[str]
    message: bool
    exist_ok: bool
    options: Dict[str, Any]
    explicit_identity: bool
    media_source: Optional[MediaSource]
    media_id: Optional[str]
    requested_music_type: Optional[str]
    metainfo: MetaBase
    mediainfo: Optional[MediaInfo] = None


def _system_config():
    """返回配置端口，并兼容旧测试对本地别名的替换。"""
    if SystemConfigOper is not _DEFAULT_SYSTEM_CONFIG_PROVIDER:
        return SystemConfigOper()
    return get_configured_system_config()


def build_subscribe_meta(subscribe: Subscribe) -> MetaBase:
    """兼容旧导入路径，转发订阅媒体元数据构造。"""
    return _build_subscribe_meta(subscribe)


def _media_recognize_kwargs(mediainfo: MediaInfo) -> dict:
    """从统一媒体信息构造规范识别身份参数。"""
    media_source, media_id = resolve_media_identity(media=mediainfo)
    return {
        "media_source": media_source,
        "media_id": media_id,
    }


def _subscribe_recognize_kwargs(subscribe: Subscribe) -> dict:
    """从订阅记录构造规范识别身份参数。"""
    media_source, media_id = resolve_media_identity(media=subscribe)
    return {
        "media_source": media_source,
        "media_id": media_id,
    }


def _subscribe_media_key(subscribe: Subscribe) -> Union[str, int, None]:
    """兼容旧导入路径，返回订阅缺失集使用的稳定媒体键。"""
    return subscribe_media_key(subscribe)


def _subscribe_media_keys(subscribe: Subscribe) -> List[Union[str, int]]:
    """兼容旧导入路径，返回规范媒体键与旧纯 ID 键。"""
    return subscribe_media_keys(subscribe)


class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
    """
    订阅管理处理链。

    订阅链路同时服务电影、普通电视剧、分集洗版和全集洗版。普通电视剧订阅与
    分集洗版共享按集事实：note 表示目标集已经存在或已经下载，
    episode_priority 表示每集已知下载质量；二者可以互相切换。current_priority
    表示当前洗版模式的资源准入基线：分集洗版按目标范围内的最低按集优先级派生，
    全集洗版由下载层确认完整覆盖后按整包优先级维护。

    实现上保持三个入口分离：下载事实入口写 note / episode_priority，并在确认
    全集覆盖时写全集 current_priority；progress 刷新入口计算 lack_episode，且仅
    分集洗版从按集事实派生 current_priority；完成入口只根据当前模式的最终事实
    和完成策略收敛订阅状态。电影没有按集事实，电影洗版的 current_priority 由
    电影下载优先级 writer 单独维护。
    """

    # 交互处理器类注入，供 InteractionChainMixin 的 parse_callback 委托
    _interaction_handler_type = SubscribeInteractionHandler

    _rlock = threading.RLock()
    # 避免莫名原因导致长时间持有锁
    _LOCK_TIMOUT = 3600 * 2

    @staticmethod
    def __normalize_episode_priority(episode_priority: Optional[dict]) -> Dict[str, int]:
        """
        归一化按集洗版优先级状态。
        """
        if not isinstance(episode_priority, dict):
            return {}

        normalized = {}
        for episode, priority in episode_priority.items():
            if episode is None or priority is None:
                continue
            try:
                normalized[str(int(episode))] = int(priority)
            except (TypeError, ValueError):
                continue
        return normalized

    @classmethod
    def __get_episode_priority(cls, subscribe: Subscribe,
                               total_episode: Optional[int] = None) -> Dict[str, int]:
        """
        获取订阅按集洗版优先级状态。
        """
        episode_priority = cls.__normalize_episode_priority(subscribe.episode_priority)
        if episode_priority:
            return episode_priority

        if (
                subscribe.best_version
                and not cls.__is_full_best_version_enabled(subscribe)
                and subscribe.type == MediaType.TV.value
                and subscribe.current_priority is not None
        ):
            target_episodes = cls.__get_best_version_target_episodes(subscribe, total_episode=total_episode)
            return {
                str(episode): int(subscribe.current_priority)
                for episode in target_episodes
            }
        return {}

    @classmethod
    def get_episode_priority(cls, subscribe: Subscribe) -> Dict[str, int]:
        """
        对外暴露按集洗版优先级状态。
        """
        return cls.__get_episode_priority(subscribe)

    @classmethod
    def __get_best_version_target_episodes(cls, subscribe: Subscribe,
                                           total_episode: Optional[int] = None) -> List[int]:
        """
        获取洗版订阅目标剧集范围。
        """
        if subscribe.type != MediaType.TV.value:
            return []

        start_episode = subscribe.start_episode or 1
        total_episode = total_episode or subscribe.total_episode or 0
        if total_episode < start_episode:
            return []
        return list(range(start_episode, total_episode + 1))

    @classmethod
    def __get_downloaded_best_version_episodes(cls, subscribe: Subscribe,
                                               total_episode: Optional[int] = None) -> List[int]:
        """
        获取洗版订阅目标范围内已下载到任意版本的剧集。

        分集洗版的完成态要求 priority==100，但订阅目标满足查询有时只需要确认
        目标集是否已下载过任意版本，因此这里按 note 与 episode_priority>0 统计。
        """
        if subscribe.type != MediaType.TV.value:
            return []

        start_episode = subscribe.start_episode or 1
        total_episode = total_episode or subscribe.total_episode or 0
        if total_episode < start_episode:
            return []
        target_episodes = set(range(start_episode, total_episode + 1))
        downloaded = 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, total_episode=total_episode).items():
            if not str(episode).isdigit():
                continue
            try:
                if float(priority) > 0:
                    episode_number = int(episode)
                    if episode_number in target_episodes:
                        downloaded.add(episode_number)
            except (TypeError, ValueError):
                continue
        return sorted(downloaded)

    @classmethod
    def __get_pending_best_version_episodes_with_priority(
            cls,
            subscribe: Subscribe,
            episode_priority: Optional[dict] = None,
            total_episode: Optional[int] = None,
    ) -> List[int]:
        """
        使用指定按集优先级状态获取当前仍需继续洗版的剧集。
        """
        target_episodes = cls.__get_best_version_target_episodes(subscribe, total_episode=total_episode)
        if not target_episodes:
            return []

        if episode_priority is None:
            normalized = cls.__get_episode_priority(subscribe, total_episode=total_episode)
        else:
            normalized = cls.__normalize_episode_priority(episode_priority)
        return [episode for episode in target_episodes if normalized.get(str(episode)) != 100]

    @classmethod
    def _get_pending_best_version_episodes(cls, subscribe: Subscribe,
                                           total_episode: Optional[int] = None) -> List[int]:
        """
        获取当前仍需继续洗版的剧集。
        """
        return cls.__get_pending_best_version_episodes_with_priority(subscribe, total_episode=total_episode)

    @classmethod
    def compute_lack_episode(
            cls,
            subscribe: Subscribe,
            no_exists: Optional[Dict[Union[int, str], Dict[int, _SchemaNotExistMediaInfo]]] = None,
    ) -> int:
        """
        计算订阅范围内尚未下载到任何版本的集数。

        普通电视剧订阅以媒体库缺失结果为准；调用方没有缺失结果时按空缺失处理，
        避免入口级刷新失败把未知状态写成异常。洗版电视剧订阅按 note 与
        episode_priority>0 判断是否已有任意版本落点，priority<100 仍表示已下载过任意版本。
        """
        if subscribe.type != MediaType.TV.value:
            return 0

        if not subscribe.best_version:
            no_exists = no_exists or {}
            left_seasons = next(
                (
                    no_exists.get(media_key)
                    for media_key in _subscribe_media_keys(subscribe)
                    if no_exists.get(media_key) is not None
                ),
                {},
            )
            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

        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(
            cls,
            subscribe: Subscribe,
            episode_priority: Optional[dict] = None,
    ) -> int:
        """
        获取洗版订阅当前优先级状态。
        """
        if not subscribe.best_version or subscribe.type != MediaType.TV.value:
            return subscribe.current_priority or 0
        if cls.__is_full_best_version_enabled(subscribe):
            return subscribe.current_priority or 0

        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 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 cls.__is_full_best_version_enabled(subscribe)
                    and 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
            if cls.__is_full_best_version_enabled(subscribe):
                subscribe.current_priority = 0
                update_data["current_priority"] = 0

        update_data.update(cls.__prepare_subscribe_progress_fields(subscribe=subscribe, no_exists={}))
        return update_data

    @classmethod
    def __prepare_best_version_total_change_fields(
            cls,
            subscribe: Subscribe,
            total_episode: int,
            old_total_episode: int,
    ) -> Dict[str, Any]:
        """
        准备洗版电视剧总集数变化后需要写库的字段。

        总集数变化会改变目标范围，按集优先级只保留新范围内的目标集，避免范围外
        旧状态继续参与完成集、缺失集和当前优先级计算。
        """
        update_data: Dict[str, Any] = {"total_episode": total_episode}
        target_episodes = set(cls.__get_best_version_target_episodes(
            subscribe,
            total_episode=total_episode,
        ))
        episode_priority = cls.__get_episode_priority(
            subscribe,
            total_episode=old_total_episode,
        )
        filtered_priority = {
            str(episode): priority
            for episode, priority in episode_priority.items()
            if int(episode) in target_episodes
        }
        subscribe.total_episode = total_episode
        subscribe.episode_priority = filtered_priority
        if cls.__is_full_best_version_enabled(subscribe):
            current_priority = 0 if total_episode > old_total_episode else subscribe.current_priority
        else:
            current_priority = 0 if not target_episodes else cls.get_best_version_current_priority(
                subscribe,
                episode_priority=filtered_priority,
            )
        subscribe.current_priority = current_priority
        update_data["episode_priority"] = filtered_priority
        update_data["current_priority"] = current_priority
        update_data.update(cls.__prepare_subscribe_progress_fields(subscribe=subscribe, no_exists={}))
        return update_data

    @classmethod
    def __prepare_total_episode_change_fields(
            cls,
            subscribe: Subscribe,
            total_episode: int,
            old_total_episode: int,
    ) -> Dict[str, Any]:
        """
        准备已有订阅总集数持久化字段，并同步内存对象上的总集数快照。
        """
        if subscribe.best_version and subscribe.type == MediaType.TV.value:
            return cls.__prepare_best_version_total_change_fields(
                subscribe=subscribe,
                total_episode=total_episode,
                old_total_episode=old_total_episode,
            )

        subscribe.total_episode = total_episode
        return {
            "total_episode": total_episode,
            "lack_episode": max(
                (subscribe.lack_episode or 0) + (total_episode - old_total_episode),
                0,
            ),
        }

    @classmethod
    def __is_best_version_complete(cls, subscribe: Subscribe) -> bool:
        """
        判断洗版订阅是否已完成。
        """
        if not subscribe.best_version:
            return False
        if subscribe.type != MediaType.TV.value:
            return subscribe.current_priority == 100
        if cls.__is_full_best_version_enabled(subscribe):
            return subscribe.current_priority == 100

        target_episodes = cls.__get_best_version_target_episodes(subscribe)
        if not target_episodes:
            return subscribe.current_priority == 100

        episode_priority = cls.__get_episode_priority(subscribe)
        return all(episode_priority.get(str(episode)) == 100 for episode in target_episodes)

    @classmethod
    def is_best_version_complete(cls, subscribe: Subscribe) -> bool:
        """
        对外暴露洗版完成判断。
        """
        return cls.__is_best_version_complete(subscribe)

    @classmethod
    def __is_best_version_complete_with_priority(
            cls,
            subscribe: Subscribe,
            episode_priority: Optional[dict] = None,
    ) -> bool:
        """
        使用指定按集优先级状态判断洗版是否已完成。
        """
        if not subscribe.best_version:
            return False
        if subscribe.type != MediaType.TV.value:
            return subscribe.current_priority == 100
        if cls.__is_full_best_version_enabled(subscribe):
            return subscribe.current_priority == 100

        target_episodes = cls.__get_best_version_target_episodes(subscribe)
        if not target_episodes:
            return subscribe.current_priority == 100

        return not cls.__get_pending_best_version_episodes_with_priority(subscribe, episode_priority)

    @staticmethod
    def __get_downloaded_episodes(downloads: Optional[List[Context]]) -> List[int]:
        """
        获取本次下载实际涉及的剧集。
        """
        if not downloads:
            return []

        downloaded_episodes = set()
        for context in downloads:
            selected_episodes = getattr(context, "selected_episodes", None)
            if selected_episodes is None:
                selected_episodes = context.meta_info.episode_list if context.meta_info else []
            for episode in selected_episodes or []:
                try:
                    downloaded_episodes.add(int(episode))
                except (TypeError, ValueError):
                    continue
        return sorted(downloaded_episodes)

    @classmethod
    def __get_best_version_completed_episodes(cls, subscribe: Subscribe) -> List[int]:
        """
        获取已完成洗版的剧集。
        """
        episode_priority = cls.__get_episode_priority(subscribe)
        target_episodes = set(cls.__get_best_version_target_episodes(subscribe))
        return sorted(
            int(episode) for episode, priority in episode_priority.items()
            if str(episode).isdigit() and int(episode) in target_episodes and priority == 100
        )

    @classmethod
    def __get_best_version_interested_episodes(
            cls,
            subscribe: Subscribe,
            context: Context,
            priority: int,
    ) -> List[int]:
        """
        获取当前资源中仍值得继续洗版的剧集。
        """
        if subscribe.type != MediaType.TV.value:
            return []

        target_episodes = set(cls.__get_best_version_target_episodes(subscribe))
        if not target_episodes:
            return []

        selected_episodes = getattr(context, "selected_episodes", None)
        if selected_episodes is None:
            selected_episodes = context.meta_info.episode_list if context.meta_info else []
        if not selected_episodes:
            episode_priority = cls.__get_episode_priority(subscribe)
            return sorted([
                episode for episode in target_episodes
                if episode_priority.get(str(episode)) is None or priority > episode_priority.get(str(episode))
            ])

        episode_priority = cls.__get_episode_priority(subscribe)
        interested = []
        for episode in selected_episodes:
            try:
                episode_num = int(episode)
            except (TypeError, ValueError):
                continue
            if episode_num not in target_episodes:
                continue
            current_priority = episode_priority.get(str(episode_num))
            if current_priority is None or priority > current_priority:
                interested.append(episode_num)
        return sorted(set(interested))

    @classmethod
    def __prepare_best_version_tv_candidate(
            cls,
            subscribe: Subscribe,
            context: Context,
            priority: int,
    ) -> bool:
        """
        校验电视剧洗版候选，并为分集模式设置允许下载的剧集范围。

        全集模式按当前准入基线筛选；分集模式设置能严格提升质量的目标集范围。
        """
        if cls.__is_full_best_version_enabled(subscribe):
            try:
                return int(priority or 0) > int(subscribe.current_priority or 0)
            except (TypeError, ValueError):
                return False

        interested_episodes = cls.__get_best_version_interested_episodes(
            subscribe=subscribe,
            context=context,
            priority=priority,
        )
        if not interested_episodes:
            return False
        context.allowed_episodes = set(interested_episodes)
        return True

    @classmethod
    def __is_full_best_version_enabled(cls, subscribe: Subscribe) -> bool:
        """
        判断当前订阅是否启用了电视剧全集洗版。
        """
        return (
                bool(subscribe.best_version_full)
                and bool(subscribe.best_version)
                and subscribe.type == MediaType.TV.value
        )

    @classmethod
    def __is_full_season_resource(cls, meta: MetaBase, subscribe: Subscribe) -> bool:
        """
        判断候选资源是否覆盖订阅目标全集范围。
        """
        season_list = meta.season_list or [1]
        if len(season_list) != 1:
            return False
        if subscribe.season is not None and season_list[0] != subscribe.season:
            return False

        episodes = meta.episode_list
        if not episodes:
            # 资源未标出单集时按整季包处理，后续下载前仍会解析种子文件确认完整性。
            return True

        target_episodes = set(cls.__get_best_version_target_episodes(subscribe))
        if not target_episodes:
            return False
        return target_episodes.issubset(set(episodes))

    @classmethod
    def __is_full_season_best_version_resource(cls, meta: MetaBase, subscribe: Subscribe) -> bool:
        """
        判断候选资源是否符合全集洗版资源约束。
        """
        if not cls.__is_full_best_version_enabled(subscribe):
            return True

        return cls.__is_full_season_resource(meta=meta, subscribe=subscribe)

    @classmethod
    def __should_prefer_full_pack_for_episode_best_version(
            cls,
            subscribe: Subscribe,
            priority: int,
    ) -> bool:
        """
        判断分集洗版是否应优先下载整包。

        整包优先级必须严格高于每个目标集；否则交回按集路径，只下载能提升质量的集。
        """
        if (
                subscribe.type != MediaType.TV.value
                or cls.__is_full_best_version_enabled(subscribe)
        ):
            return False

        target_episodes = cls.__get_best_version_target_episodes(subscribe)
        if not target_episodes:
            return False

        try:
            resource_priority = int(priority or 0)
        except (TypeError, ValueError):
            resource_priority = 0

        episode_priority = cls.__get_episode_priority(subscribe)
        return all(
            resource_priority > episode_priority.get(str(episode), 0)
            for episode in target_episodes
        )

    @classmethod
    def __build_full_pack_first_no_exists(
            cls,
            subscribe: Subscribe,
            mediakey: Union[int, str],
    ) -> Optional[Dict[Union[int, str], Dict[int, _SchemaNotExistMediaInfo]]]:
        """
        构造分集洗版优先全集时使用的整季缺失范围。
        """
        if (
                not subscribe.best_version
                or cls.__is_full_best_version_enabled(subscribe)
                or subscribe.type != MediaType.TV.value
        ):
            return None

        target_episodes = cls.__get_best_version_target_episodes(subscribe)
        if not target_episodes:
            return None

        return {
            mediakey: {
                subscribe.season: _SchemaNotExistMediaInfo(
                    season=subscribe.season,
                    episodes=[],
                    total_episode=subscribe.total_episode,
                    start_episode=subscribe.start_episode or 1,
                    require_complete_coverage=True,
                )
            }
        }

    def __download_best_version_with_full_pack_first(
            self,
            contexts: List[Context],
            no_exists: Dict[Union[int, str], Dict[int, _SchemaNotExistMediaInfo]],
            subscribe: Subscribe,
            mediakey: Union[int, str],
            username: Optional[str] = None,
            save_path: Optional[str] = None,
            downloader: Optional[str] = None,
            source: Optional[str] = None,
    ) -> Tuple[List[Context], Dict[Union[int, str], Dict[int, _SchemaNotExistMediaInfo]]]:
        """
        TV 分集洗版先尝试覆盖目标范围的全集资源，失败后回退到按集下载。
        """
        full_pack_no_exists = self.__build_full_pack_first_no_exists(subscribe=subscribe, mediakey=mediakey)
        full_season_contexts = [
            context for context in contexts
            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 []
        target_episodes = self.__get_best_version_target_episodes(subscribe)
        target_range = f"{target_episodes[0]}-{target_episodes[-1]}" if target_episodes else "empty"
        episode_priority_gate = self.__get_episode_priority(subscribe)
        full_pack_contexts = []
        for context in full_season_contexts:
            candidate_priority = context.torrent_info.pri_order
            accepted = self.__should_prefer_full_pack_for_episode_best_version(
                subscribe=subscribe,
                priority=candidate_priority,
            )
            logger.info(
                f"{subscribe.name} 整包候选优先级判断：candidate_priority={candidate_priority}，"
                f"episode_priority={episode_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} 全集候选优先级未高于全部目标集，回退到分集洗版")

        if full_pack_contexts:
            logger.info(f"{subscribe.name} 分集洗版优先尝试全集资源，共匹配到 {len(full_pack_contexts)} 个候选")
            downloads, lefts = DownloadChain().batch_download(
                contexts=full_pack_contexts,
                no_exists=full_pack_no_exists,
                username=username,
                save_path=save_path,
                downloader=downloader,
                source=source,
                custom_words=subscribe.custom_words,
            )
            if downloads:
                return downloads, lefts
            logger.info(f"{subscribe.name} 未下载到全集资源，回退到分集洗版")

        return DownloadChain().batch_download(
            contexts=contexts,
            no_exists=no_exists,
            username=username,
            save_path=save_path,
            downloader=downloader,
            source=source,
            custom_words=subscribe.custom_words,
        )

    def __get_default_kwargs(self, mtype: MediaType, **kwargs) -> dict:
        """
        获取订阅默认配置
        :param mtype: 媒体类型
        :param key: 配置键
        :return: 配置值
        """
        defaults = {
            'quality': self.__get_default_subscribe_config(mtype, "quality") if not kwargs.get(
                "quality") else kwargs.get("quality"),
            'resolution': self.__get_default_subscribe_config(mtype, "resolution") if not kwargs.get(
                "resolution") else kwargs.get("resolution"),
            'effect': self.__get_default_subscribe_config(mtype, "effect") if not kwargs.get(
                "effect") else kwargs.get("effect"),
            'audio_quality': self.__get_default_subscribe_config(mtype, "audio_quality") if not kwargs.get(
                "audio_quality") else kwargs.get("audio_quality"),
            'audio_format': self.__get_default_subscribe_config(mtype, "audio_format") if not kwargs.get(
                "audio_format") else kwargs.get("audio_format"),
            'min_bitrate': self.__get_default_subscribe_config(mtype, "min_bitrate") if not kwargs.get(
                "min_bitrate") else kwargs.get("min_bitrate"),
            'min_bit_depth': self.__get_default_subscribe_config(mtype, "min_bit_depth") if not kwargs.get(
                "min_bit_depth") else kwargs.get("min_bit_depth"),
            'min_sample_rate': self.__get_default_subscribe_config(mtype, "min_sample_rate") if not kwargs.get(
                "min_sample_rate") else kwargs.get("min_sample_rate"),
            'include': self.__get_default_subscribe_config(mtype, "include") if not kwargs.get(
                "include") else kwargs.get("include"),
            'exclude': self.__get_default_subscribe_config(mtype, "exclude") if not kwargs.get(
                "exclude") else kwargs.get("exclude"),
            'best_version': self.__get_default_subscribe_config(mtype, "best_version")
            if kwargs.get("best_version") is None else kwargs.get("best_version"),
            'best_version_full': self.__get_default_subscribe_config(mtype, "best_version_full")
            if kwargs.get("best_version_full") is None else kwargs.get("best_version_full"),
            'search_imdbid': self.__get_default_subscribe_config(mtype, "search_imdbid") if not kwargs.get(
                "search_imdbid") else kwargs.get("search_imdbid"),
            'sites': self.__get_default_subscribe_config(mtype, "sites") or None if not kwargs.get(
                "sites") else kwargs.get("sites"),
            'downloader': self.__get_default_subscribe_config(mtype, "downloader") if not kwargs.get(
                "downloader") else kwargs.get("downloader"),
            'save_path': self.__get_default_subscribe_config(mtype, "save_path") if not kwargs.get(
                "save_path") else kwargs.get("save_path"),
            'filter_groups': self.__get_default_subscribe_config(mtype, "filter_groups") if not kwargs.get(
                "filter_groups") else kwargs.get("filter_groups")
        }
        if mtype == MediaType.MUSIC:
            # 音乐允许按音质洗版，但没有电视剧整包洗版和 IMDB 搜索语义。
            defaults.update({
                "best_version_full": 0,
                "search_imdbid": 0,
            })
        return defaults

    @staticmethod
    def __subscribe_added_link(mtype: MediaType) -> str:
        """返回订阅类型对应的前端详情入口。"""
        if mtype == MediaType.TV:
            return get_chain_runtime_config_snapshot().television_subscribe_url
        if mtype == MediaType.MUSIC:
            return get_chain_runtime_config_snapshot().music_subscribe_url
        return get_chain_runtime_config_snapshot().movie_subscribe_url

    @staticmethod
    def __subscribe_report_payload(context: _SubscribePostCommitContext) -> dict:
        """构造保持旧字段和值语义的订阅统计上报。"""
        mediainfo = context.mediainfo
        music_type = getattr(mediainfo, "music_type", None)
        return {
            "name": context.title,
            "year": context.year,
            "type": context.metainfo.type.value,
            "media_source": context.media_source,
            "media_id": context.media_id,
            "music_type": music_type,
            "total_tracks": getattr(mediainfo, "total_tracks", None)
            if music_type == MUSIC_ENTITY_ALBUM else None,
            "season": context.season,
            "poster": mediainfo.get_poster_image(),
            "backdrop": mediainfo.get_backdrop_image(),
            "vote": mediainfo.vote_average,
            "description": mediainfo.overview,
        }

    def __post_subscribe_added(
        self,
        subscribe_id: int,
        context: _SubscribePostCommitContext,
    ) -> None:
        """同步执行提交后消息、事件和统计，异常不再触碰数据库事务。"""
        if context.notification:
            self.post_message(_SchemaMessage.model_validate(context.notification))
        eventmanager.send_event(EventType.SubscribeAdded, {
            "subscribe_id": subscribe_id,
            "idempotency_key": (
                f"subscribe.added:{subscribe_id}:"
                f"{context.media_source}:{context.media_id}:v1"
            ),
            "username": context.username,
            "mediainfo": context.mediainfo.to_dict(),
        })
        if not MoviePilotServerHelper.sub_reg_durable(
            self.__subscribe_report_payload(context)
        ):
            raise RuntimeError("订阅新增统计上报未确认")

    async def __async_post_subscribe_added(
        self,
        subscribe_id: int,
        context: _SubscribePostCommitContext,
    ) -> None:
        """异步执行提交后消息、事件和统计，保持与同步入口相同顺序。"""
        if context.notification:
            await self.async_post_message(
                _SchemaMessage.model_validate(context.notification)
            )
        await eventmanager.async_send_event(EventType.SubscribeAdded, {
            "subscribe_id": subscribe_id,
            "idempotency_key": (
                f"subscribe.added:{subscribe_id}:"
                f"{context.media_source}:{context.media_id}:v1"
            ),
            "username": context.username,
            "mediainfo": context.mediainfo.to_dict(),
        })
        if not await MoviePilotServerHelper.async_sub_reg_durable(
            self.__subscribe_report_payload(context)
        ):
            raise RuntimeError("订阅新增统计上报未确认")

    @staticmethod
    def __build_subscribe_create_context(
        title: str,
        year: str,
        mtype: Optional[MediaType],
        episode_group: Optional[str],
        season: Optional[int],
        channel: Optional[NotificationChannel],
        source: Optional[str],
        userid: Optional[str],
        username: Optional[str],
        message: bool,
        exist_ok: bool,
        media_source: Optional[MediaSource],
        media_id: Optional[str],
        options: Dict[str, Any],
    ) -> Tuple[Optional[_SubscribeCreateContext], Optional[str]]:
        """规范订阅新增输入，并在任何识别 I/O 前拒绝不完整的显式媒体身份。"""
        explicit_identity = media_source is not None or media_id is not None
        media_source, media_id = resolve_media_identity(
            media_source=media_source,
            media_id=media_id,
        )
        if explicit_identity and (not media_source or not media_id):
            return None, "媒体来源和媒体 ID 必须同时提供"

        metainfo = MetaMusic.parse_query(title) if mtype == MediaType.MUSIC else MetaInfo(title)
        if year:
            metainfo.year = year
        if mtype:
            metainfo.type = mtype
        if season is not None:
            metainfo.type = MediaType.TV
            metainfo.begin_season = season
        if mtype == MediaType.MUSIC and media_id:
            metainfo.media_id = str(media_id)

        return _SubscribeCreateContext(
            title=title,
            year=year,
            mtype=mtype,
            episode_group=episode_group,
            season=season,
            channel=channel,
            source=source,
            userid=userid,
            username=username,
            message=bool(message),
            exist_ok=bool(exist_ok),
            options=options,
            explicit_identity=explicit_identity,
            media_source=media_source,
            media_id=media_id,
            requested_music_type=options.get("music_type"),
            metainfo=metainfo,
        ), None

    @staticmethod
    def __normalize_recognized_subscribe_media(context: _SubscribeCreateContext) -> None:
        """保留非 TMDB 影视标题清洗与从标题补季号的历史行为。"""
        mediainfo = context.mediainfo
        if (
            context.mtype == MediaType.MUSIC
            or not mediainfo
            or mediainfo.media_source == MediaSource.TMDB
        ):
            return
        meta = MetaInfo(mediainfo.title)
        mediainfo.title = meta.name
        if context.season is None:
            context.season = meta.begin_season

    def __recognize_subscribe_media(self, context: _SubscribeCreateContext) -> Optional[str]:
        """同步识别订阅目标；显式身份失败时禁止按标题换成另一个媒体。"""
        if context.media_source and context.media_id:
            context.mediainfo = MediaChain().recognize_media(
                meta=context.metainfo,
                mtype=context.mtype,
                media_source=context.media_source,
                media_id=context.media_id,
                music_type=context.requested_music_type,
                episode_group=context.episode_group,
                cache=False,
            )
        self.__normalize_recognized_subscribe_media(context)
        if not context.mediainfo and not context.explicit_identity:
            context.mediainfo = MediaChain().recognize_by_meta(
                context.metainfo,
                media_source=context.media_source,
                episode_group=context.episode_group,
                obtain_images=False,
                music_type=context.requested_music_type,
            )
            if (
                context.mtype == MediaType.MUSIC
                and context.mediainfo
                and not context.mediainfo.media_source
            ):
                context.mediainfo = None
        return self.__validate_recognized_subscribe_media(context)

    async def __async_recognize_subscribe_media(
        self,
        context: _SubscribeCreateContext,
    ) -> Optional[str]:
        """异步识别订阅目标，并与同步入口共享相同的规范化和校验规则。"""
        if context.media_source and context.media_id:
            context.mediainfo = await MediaChain().async_recognize_media(
                meta=context.metainfo,
                mtype=context.mtype,
                media_source=context.media_source,
                media_id=context.media_id,
                music_type=context.requested_music_type,
                episode_group=context.episode_group,
                cache=False,
            )
        self.__normalize_recognized_subscribe_media(context)
        if not context.mediainfo and not context.explicit_identity:
            context.mediainfo = await MediaChain().async_recognize_by_meta(
                context.metainfo,
                media_source=context.media_source,
                episode_group=context.episode_group,
                obtain_images=False,
                music_type=context.requested_music_type,
            )
            if (
                context.mtype == MediaType.MUSIC
                and context.mediainfo
                and not context.mediainfo.media_source
            ):
                context.mediainfo = None
        return self.__validate_recognized_subscribe_media(context)

    def __validate_recognized_subscribe_media(
        self,
        context: _SubscribeCreateContext,
    ) -> Optional[str]:
        """校验识别结果以及音乐订阅实体，返回兼容旧入口的错误文案。"""
        if not context.mediainfo:
            logger.warning(
                f"未识别到媒体信息，标题：{context.title}，媒体来源：{context.media_source}，"
                f"媒体 ID：{context.media_id}"
            )
            return "未识别到媒体信息"
        if context.mtype != MediaType.MUSIC:
            return None
        music_error = self._validate_music_subscribe_target(
            context.mediainfo,
            requested_music_type=context.requested_music_type,
        )
        if music_error:
            logger.warning(f"音乐订阅目标校验失败：{context.title} - {music_error}")
        return music_error

    def __prepare_subscribe_episodes(self, context: _SubscribeCreateContext) -> Optional[str]:
        """同步补齐电视剧季集信息，并保持外部集数刷新只能扩展创建目标。"""
        mediainfo = context.mediainfo
        if mediainfo.type != MediaType.TV:
            context.season = None
            return None
        if context.season is None:
            context.season = 1
        if not context.options.get("total_episode"):
            if not mediainfo.seasons or context.episode_group:
                mediainfo = MediaChain().recognize_media(
                    mtype=mediainfo.type,
                    **_media_recognize_kwargs(mediainfo),
                    episode_group=context.episode_group,
                    cache=False,
                )
                context.mediainfo = mediainfo
                error = self.__validate_subscribe_seasons(context)
                if error:
                    return error
            current_total_episode = len(mediainfo.seasons.get(context.season) or [])
            total_episode = self.__apply_episodes_refresh(
                current_total_episode,
                season=context.season,
                mediainfo=mediainfo,
                media_source=resolve_media_identity(media=mediainfo)[0],
                media_id=resolve_media_identity(media=mediainfo)[1],
                scene="create",
            )
            error = self.__store_subscribe_episode_total(
                context,
                current_total_episode,
                total_episode,
            )
            if error:
                return error
        self.__fill_subscribe_lack_episode(context)
        return None

    async def __async_prepare_subscribe_episodes(
        self,
        context: _SubscribeCreateContext,
    ) -> Optional[str]:
        """异步补齐电视剧季集信息，并复用同步入口的结果校验和字段写入规则。"""
        mediainfo = context.mediainfo
        if mediainfo.type != MediaType.TV:
            context.season = None
            return None
        if context.season is None:
            context.season = 1
        if not context.options.get("total_episode"):
            if not mediainfo.seasons or context.episode_group:
                mediainfo = await MediaChain().async_recognize_media(
                    mtype=mediainfo.type,
                    **_media_recognize_kwargs(mediainfo),
                    episode_group=context.episode_group,
                    cache=False,
                )
                context.mediainfo = mediainfo
                error = self.__validate_subscribe_seasons(context)
                if error:
                    return error
            current_total_episode = len(mediainfo.seasons.get(context.season) or [])
            total_episode = await self.__async_apply_episodes_refresh(
                current_total_episode,
                season=context.season,
                mediainfo=mediainfo,
                media_source=resolve_media_identity(media=mediainfo)[0],
                media_id=resolve_media_identity(media=mediainfo)[1],
                scene="create",
            )
            error = self.__store_subscribe_episode_total(
                context,
                current_total_episode,
                total_episode,
            )
            if error:
                return error
        self.__fill_subscribe_lack_episode(context)
        return None

    @staticmethod
    def __validate_subscribe_seasons(context: _SubscribeCreateContext) -> Optional[str]:
        """校验补充识别结果是否仍包含创建电视剧订阅所需的季集信息。"""
        if not context.mediainfo:
            logger.error("媒体信息识别失败！")
            return "媒体信息识别失败"
        if not context.mediainfo.seasons:
            logger.error(f"媒体信息中没有季集信息，标题：{context.title}")
            return "媒体信息中没有季集信息"
        return None

    @staticmethod
    def __store_subscribe_episode_total(
        context: _SubscribeCreateContext,
        current_total_episode: int,
        total_episode: int,
    ) -> Optional[str]:
        """写入创建场景最终集数，阻止外部刷新把可靠的当前集数向下覆盖。"""
        if current_total_episode and total_episode < current_total_episode:
            total_episode = current_total_episode
        if not total_episode:
            logger.error(f"未获取到总集数，标题：{context.title}")
            return f"未获取到第 {context.season} 季的总集数"
        context.options["total_episode"] = total_episode
        return None

    @staticmethod
    def __fill_subscribe_lack_episode(context: _SubscribeCreateContext) -> None:
        """未显式指定缺失集数时沿用总集数，保持旧创建默认值。"""
        if not context.options.get("lack_episode"):
            context.options["lack_episode"] = context.options.get("total_episode")

    def __finalize_subscribe_create_context(self, context: _SubscribeCreateContext) -> None:
        """同步补图并写入规范媒体身份和订阅默认配置。"""
        if context.mediainfo.type != MediaType.MUSIC:
            self.obtain_images(mediainfo=context.mediainfo)
        self.__apply_subscribe_create_defaults(context)

    async def __async_finalize_subscribe_create_context(
        self,
        context: _SubscribeCreateContext,
    ) -> None:
        """异步补图并写入规范媒体身份和订阅默认配置。"""
        if context.mediainfo.type != MediaType.MUSIC:
            await self.async_obtain_images(mediainfo=context.mediainfo)
        self.__apply_subscribe_create_defaults(context)

    def __apply_subscribe_create_defaults(self, context: _SubscribeCreateContext) -> None:
        """以最终识别结果覆盖身份字段，并补齐当前媒体类型的默认订阅参数。"""
        context.media_source, context.media_id = resolve_media_identity(media=context.mediainfo)
        context.options.update({
            "media_source": context.media_source,
            "media_id": context.media_id,
        })
        context.options.update(
            self.__get_default_kwargs(context.mediainfo.type, **context.options)
        )

    @staticmethod
    def __subscribe_post_commit_context(
        context: _SubscribeCreateContext,
        notification: Optional[dict] = None,
    ) -> _SubscribePostCommitContext:
        """从创建阶段状态冻结提交后副作用需要的最小快照。"""
        return _SubscribePostCommitContext(
            title=context.title,
            year=context.year,
            metainfo=context.metainfo,
            mediainfo=context.mediainfo,
            media_source=context.media_source,
            media_id=context.media_id,
            season=context.season,
            channel=context.channel,
            source=context.source,
            userid=context.userid,
            username=context.username,
            message=context.message,
            notification=notification,
        )

    def __build_subscribe_notification(
        self,
        context: _SubscribeCreateContext,
    ) -> Optional[dict]:
        """在事务提交前冻结已渲染消息，供即时发送与 outbox 恢复共用。"""
        if not context.message:
            return None
        message = _SchemaMessage(
            channel=context.channel,
            source=context.source,
            mtype=MessageType.Subscribe,
            ctype=ContentType.SubscribeAdded,
            image=context.mediainfo.get_message_image(),
            link=self.__subscribe_added_link(context.mediainfo.type),
            userid=context.userid,
            username=context.username,
        )
        rendered = MessageTemplateHelper.render(
            message,
            meta=context.metainfo,
            mediainfo=context.mediainfo,
            username=context.username,
        ) or message
        return rendered.model_dump(mode="json")

    def __persist_subscribe_create(self, context: _SubscribeCreateContext) -> Tuple[Optional[int], str]:
        """同步提交订阅，并在提交成功后按原顺序执行消息、事件和统计。"""
        post_commit_context = self.__subscribe_post_commit_context(
            context,
            self.__build_subscribe_notification(context),
        )

        def _after_commit(subscribe_id: int) -> None:
            """把同步提交后的副作用委托给单一顺序实现。"""
            self.__post_subscribe_added(subscribe_id, post_commit_context)

        sid, err_msg = add_subscribe(
            mediainfo=context.mediainfo,
            season=context.season,
            username=context.username,
            after_commit=_after_commit,
            notification=post_commit_context.notification,
            **context.options,
        )
        if not sid:
            self.__notify_subscribe_create_failure(context, err_msg)
            return None, err_msg
        return sid, err_msg

    async def __async_persist_subscribe_create(
        self,
        context: _SubscribeCreateContext,
    ) -> Tuple[Optional[int], str]:
        """异步提交订阅，并在提交成功后按原顺序执行消息、事件和统计。"""
        post_commit_context = self.__subscribe_post_commit_context(
            context,
            self.__build_subscribe_notification(context),
        )

        async def _after_commit(subscribe_id: int) -> None:
            """把异步提交后的副作用委托给单一顺序实现。"""
            await self.__async_post_subscribe_added(subscribe_id, post_commit_context)

        sid, err_msg = await async_add_subscribe(
            mediainfo=context.mediainfo,
            season=context.season,
            username=context.username,
            after_commit=_after_commit,
            notification=post_commit_context.notification,
            **context.options,
        )
        if not sid:
            await self.__async_notify_subscribe_create_failure(context, err_msg)
            return None, err_msg
        return sid, err_msg

    def __notify_subscribe_create_failure(
        self,
        context: _SubscribeCreateContext,
        err_msg: str,
    ) -> None:
        """同步记录持久化失败，并按旧规则向原用户反馈。"""
        logger.error(f"{context.mediainfo.title_year} {err_msg}")
        if context.exist_ok or not context.message:
            return
        self.post_message(self.__subscribe_create_failure_message(context, err_msg))

    async def __async_notify_subscribe_create_failure(
        self,
        context: _SubscribeCreateContext,
        err_msg: str,
    ) -> None:
        """异步记录持久化失败，并按旧规则向原用户反馈。"""
        logger.error(f"{context.mediainfo.title_year} {err_msg}")
        if context.exist_ok or not context.message:
            return
        await self.async_post_message(self.__subscribe_create_failure_message(context, err_msg))

    @staticmethod
    def __subscribe_create_failure_message(
        context: _SubscribeCreateContext,
        err_msg: str,
    ) -> _SchemaMessage:
        """构造保持旧标题、图片和接收人字段的订阅失败消息。"""
        return _SchemaMessage(
            channel=context.channel,
            source=context.source,
            mtype=MessageType.Subscribe,
            title=(
                f"{context.mediainfo.title_year} {context.metainfo.season} "
                "添加订阅失败！"
            ),
            text=err_msg,
            image=context.mediainfo.get_message_image(),
            userid=context.userid,
        )

    def add(self, title: str, year: str,
            mtype: MediaType = None,
            episode_group: Optional[str] = None,
            season: Optional[int] = None,
            channel: NotificationChannel = None,
            source: Optional[str] = None,
            userid: Optional[str] = None,
            username: Optional[str] = None,
            message: Optional[bool] = True,
            exist_ok: Optional[bool] = False,
            media_source: Optional[MediaSource] = None,
            media_id: Optional[str] = None,
            **kwargs) -> Tuple[Optional[int], str]:
        """
        识别媒体信息并添加订阅
        """
        logger.info(f'开始添加订阅，标题：{title} ...')
        context, error = self.__build_subscribe_create_context(
            title, year, mtype, episode_group, season, channel, source, userid,
            username, message, exist_ok, media_source, media_id, kwargs,
        )
        if error:
            return None, error
        error = self.__recognize_subscribe_media(context)
        if error:
            return None, error
        error = self.__prepare_subscribe_episodes(context)
        if error:
            return None, error
        self.__finalize_subscribe_create_context(context)
        return self.__persist_subscribe_create(context)

    async def async_add(self, title: str, year: str,
                        mtype: MediaType = None,
                        episode_group: Optional[str] = None,
                        season: Optional[int] = None,
                        channel: NotificationChannel = None,
                        source: Optional[str] = None,
                        userid: Optional[str] = None,
                        username: Optional[str] = None,
                        message: Optional[bool] = True,
                        exist_ok: Optional[bool] = False,
                        media_source: Optional[MediaSource] = None,
                        media_id: Optional[str] = None,
                        **kwargs) -> Tuple[Optional[int], str]:
        """
        异步识别媒体信息并添加订阅
        """
        logger.info(f'开始添加订阅，标题：{title} ...')
        context, error = self.__build_subscribe_create_context(
            title, year, mtype, episode_group, season, channel, source, userid,
            username, message, exist_ok, media_source, media_id, kwargs,
        )
        if error:
            return None, error
        error = await self.__async_recognize_subscribe_media(context)
        if error:
            return None, error
        error = await self.__async_prepare_subscribe_episodes(context)
        if error:
            return None, error
        await self.__async_finalize_subscribe_create_context(context)
        return await self.__async_persist_subscribe_create(context)

    @staticmethod
    def _subscription_query() -> SubscriptionQueryService:
        """构造绑定订阅 Oper 的查询应用服务。"""
        return SubscriptionQueryService(SubscribeOper())

    @classmethod
    def exists(cls, mediainfo: MediaInfo, meta: MetaBase = None):
        """
        判断订阅是否已存在
        """
        return cls._subscription_query().exists(mediainfo, meta)

    def _acquire_run_lock(
            self,
            operation: str,
            progress_callback: Optional[Callable[..., None]],
    ) -> bool:
        """获取订阅任务锁，超时时统一记录并结束本轮进度。"""
        if self._rlock.acquire(blocking=True, timeout=self._LOCK_TIMOUT):
            logger.debug(f"{operation} lock acquired at {datetime.now()}")
            return True
        operation_label = {"search": "搜索", "match": "匹配"}[operation]
        progress_text = {
            "search": "订阅搜索锁等待超时，已跳过本轮",
            "match": "订阅匹配锁等待超时，已跳过本轮",
        }[operation]
        logger.error(f"订阅{operation_label}锁等待超时，已中止本轮执行")
        if progress_callback:
            progress_callback(
                value=100,
                text=progress_text,
            )
        return False

    def search(
            self,
            sid: Optional[int] = None,
            state: Optional[str] = 'N',
            manual: Optional[bool] = False,
            progress_callback: Optional[Callable[..., None]] = None,
            sids: Optional[tuple[int, ...]] = None,
    ) -> None:
        """
        执行订阅搜索。

        保持定时任务、API 和插件使用的公开签名，搜索实现委托给内部执行阶段。
        """
        return self._execute_search(
            sid=sid,
            sids=sids,
            state=state,
            manual=manual,
            progress_callback=progress_callback,
        )

    def _execute_search(
            self,
            sid: Optional[int] = None,
            state: Optional[str] = 'N',
            manual: Optional[bool] = False,
            progress_callback: Optional[Callable[..., None]] = None,
            sids: Optional[tuple[int, ...]] = None,
    ) -> None:
        """
        订阅搜索
        :param sid: 订阅ID，有值时只处理该订阅
        :param state: 订阅状态 N:新建, R:订阅中, P:待定, S:暂停
        :param manual: 是否手动搜索
        :param progress_callback: 定时服务进度更新回调
        :param sids: 订阅ID集合，有值时按给定顺序处理
        :return: 更新订阅状态为R或删除订阅
        """
        lock_acquired = False
        try:
            lock_acquired = self._acquire_run_lock("search", progress_callback)
            if not lock_acquired:
                return

            subscribeoper = SubscribeOper()
            if sid:
                subscribe = subscribeoper.get(sid)
                subscribes = [subscribe] if subscribe else []
            elif sids is not None:
                subscribes = [
                    subscribe
                    for current_id in sids
                    if (subscribe := subscribeoper.get(current_id)) is not None
                ]
            else:
                subscribes = subscribeoper.list(self.get_states_for_search(state))
            total_num = len(subscribes)
            processed_subscribes = []
            if progress_callback:
                progress_callback(
                    value=0,
                    text=f"开始订阅搜索，共 {total_num} 个订阅 ...",
                    data={"total": total_num, "finished": 0},
                )

            try:
                # 遍历订阅
                for index, subscribe in enumerate(subscribes, start=1):
                    if global_vars.is_system_stopped:
                        break
                    processed_subscribes.append(subscribe)
                    if progress_callback:
                        progress_callback(
                            value=(index - 1) / total_num * 100 if total_num else 100,
                            text=f"正在搜索订阅（{index}/{total_num}）{subscribe.name} ...",
                            data={
                                "total": total_num,
                                "finished": index - 1,
                                "current": subscribe.id,
                            },
                        )
                    mediakey = _subscribe_media_key(subscribe)
                    custom_word_list = subscribe.custom_words.split("\n") if subscribe.custom_words else None
                    search_attempted = False
                    # 校验当前时间减订阅创建时间是否大于1分钟，否则跳过先，留出编辑订阅的时间
                    if subscribe.date:
                        now = datetime.now()
                        subscribe_time = datetime.strptime(subscribe.date, '%Y-%m-%d %H:%M:%S')
                        if (now - subscribe_time).total_seconds() < 60:
                            logger.debug(f"订阅标题：{subscribe.name} 新增小于1分钟，暂不搜索...")
                            continue
                    # 随机休眠1-5分钟
                    if not sid and sids is None and state in ['R', 'P']:
                        sleep_time = random.randint(60, 300)
                        logger.info(f'订阅搜索随机休眠 {sleep_time} 秒 ...')
                        if progress_callback:
                            progress_callback(
                                text=f"订阅搜索随机休眠 {sleep_time} 秒后继续 ..."
                            )
                        time.sleep(sleep_time)
                    try:
                        search_attempted = True
                        logger.info(f'开始搜索订阅，标题：{subscribe.name} ...')
                        if subscribe.type == MediaType.MUSIC.value:
                            self._search_music_subscribe(subscribe)
                            continue
                        try:
                            meta = build_subscribe_meta(subscribe)
                        except ValueError:
                            logger.error(f'订阅 {subscribe.name} 类型错误：{subscribe.type}')
                            continue
                        # 识别媒体信息
                        mediainfo: MediaInfo = MediaChain().recognize_media(
                            meta=meta,
                            mtype=meta.type,
                            **_subscribe_recognize_kwargs(subscribe),
                            episode_group=subscribe.episode_group,
                            cache=False,
                        )
                        if not mediainfo:
                            logger.warn(
                                f"未识别到媒体信息，标题：{subscribe.name}，"
                                f"媒体来源：{subscribe.media_source}，媒体 ID：{subscribe.media_id}"
                            )
                            continue

                        # 如果媒体已存在或已下载完毕，跳过当前订阅处理
                        exist_flag, no_exists = self.check_and_handle_existing_media(subscribe=subscribe,
                                                                                     meta=meta,
                                                                                     mediainfo=mediainfo,
                                                                                     mediakey=mediakey)
                        if exist_flag:
                            continue

                        # 站点范围
                        sites = self.get_sub_sites(subscribe)

                        # 优先级过滤规则
                        if subscribe.best_version:
                            rule_groups = subscribe.filter_groups \
                                          or _system_config().get(SystemConfigKey.BestVersionFilterRuleGroups) or []
                        else:
                            rule_groups = subscribe.filter_groups \
                                          or _system_config().get(SystemConfigKey.SubscribeFilterRuleGroups) or []

                        # 搜索，同时电视剧会过滤掉不需要的剧集
                        contexts = SearchChain().process(mediainfo=mediainfo,
                                                         keyword=subscribe.keyword,
                                                         no_exists=no_exists,
                                                         sites=sites,
                                                         rule_groups=rule_groups,
                                                         area="imdbid" if subscribe.search_imdbid else "title",
                                                         custom_words=custom_word_list,
                                                         filter_params=self.get_params(subscribe))
                        if not contexts:
                            logger.warn(f'订阅 {subscribe.keyword or subscribe.name} 未搜索到资源')
                            self.finish_subscribe_or_not(subscribe=subscribe, meta=meta,
                                                         mediainfo=mediainfo, lefts=no_exists)
                            continue

                        # 过滤搜索结果
                        matched_contexts = []
                        try:
                            for context in contexts:
                                if global_vars.is_system_stopped:
                                    break
                                torrent_meta = context.meta_info
                                torrent_info = context.torrent_info
                                torrent_mediainfo = context.media_info

                                # 洗版
                                if subscribe.best_version:
                                    if (
                                            torrent_mediainfo.type == MediaType.TV
                                            and not self.__is_full_season_best_version_resource(
                                        meta=torrent_meta, subscribe=subscribe
                                    )
                                    ):
                                        logger.info(
                                            f"{subscribe.name} 正在全集洗版，{torrent_info.title} 不是全集资源"
                                        )
                                        continue
                                    # 洗版时，不符合订阅集数的不要
                                    if (
                                            torrent_mediainfo.type == MediaType.TV
                                            and not self._is_episode_range_covered(
                                        meta=torrent_meta, subscribe=subscribe
                                    )
                                    ):
                                        logger.info(
                                            f"{subscribe.name} 正在洗版，{torrent_info.title} 不符合订阅集数范围"
                                        )
                                        continue
                                    # 全集洗版按整包准入基线过滤；分集洗版按可提升剧集过滤。
                                    if torrent_mediainfo.type == MediaType.TV:
                                        if not self.__prepare_best_version_tv_candidate(
                                                subscribe=subscribe,
                                                context=context,
                                                priority=torrent_info.pri_order,
                                        ):
                                            logger.info(
                                                f'{subscribe.name} 正在洗版，{torrent_info.title} '
                                                f'优先级未达到当前模式的升级条件')
                                            continue
                                    if (
                                            torrent_mediainfo.type != MediaType.TV
                                            and subscribe.current_priority
                                            and torrent_info.pri_order <= subscribe.current_priority
                                    ):
                                        logger.info(
                                            f'{subscribe.name} 正在洗版，{torrent_info.title} 优先级低于或等于已下载优先级')
                                        continue
                                # 更新订阅自定义属性
                                if subscribe.media_category:
                                    torrent_mediainfo.category = subscribe.media_category
                                if subscribe.episode_group:
                                    torrent_mediainfo.episode_group = subscribe.episode_group
                                matched_contexts.append(context)
                        finally:
                            contexts.clear()
                            del contexts

                        if not matched_contexts:
                            logger.warn(f'订阅 {subscribe.name} 没有符合过滤条件的资源')
                            self.finish_subscribe_or_not(subscribe=subscribe, meta=meta,
                                                         mediainfo=mediainfo, lefts=no_exists)
                            continue

                        # 自动下载
                        downloads, lefts = self.__download_best_version_with_full_pack_first(
                            contexts=matched_contexts,
                            no_exists=no_exists,
                            subscribe=subscribe,
                            mediakey=mediakey,
                            username=subscribe.username,
                            save_path=subscribe.save_path,
                            downloader=subscribe.downloader,
                            source=self.get_subscribe_source_keyword(subscribe)
                        )

                        # 同步外部修改，更新订阅信息
                        subscribe = subscribeoper.get(subscribe.id)

                        # 判断是否应完成订阅
                        if subscribe:
                            self.finish_subscribe_or_not(subscribe=subscribe, meta=meta, mediainfo=mediainfo,
                                                         downloads=downloads, lefts=lefts)
                    finally:
                        # 如果状态为N则更新为R
                        if search_attempted and subscribe and subscribe.state == 'N':
                            subscribeoper.update(subscribe.id, {'state': 'R'})
                        if progress_callback:
                            progress_callback(
                                value=index / total_num * 100 if total_num else 100,
                                text=f"订阅搜索（{index}/{total_num}）处理完成",
                                data={"total": total_num, "finished": index},
                            )

                # 手动触发时发送系统消息
                if manual:
                    if subscribes:
                        if sid:
                            self.messagehelper.put(f'{subscribes[0].name} 搜索完成！', title="订阅搜索", role="system")
                        elif sids is not None:
                            for subscribe in processed_subscribes:
                                self.messagehelper.put(
                                    f'{subscribe.name} 搜索完成！',
                                    title="订阅搜索",
                                    role="system",
                                )
                        else:
                            self.messagehelper.put('所有订阅搜索完成！', title="订阅搜索", role="system")
                    else:
                        self.messagehelper.put('没有找到订阅！', title="订阅搜索", role="system")
                if progress_callback:
                    progress_callback(value=100, text="订阅搜索完成")

            finally:
                subscribes.clear()
                del subscribes
        finally:
            if lock_acquired:
                self._rlock.release()
                logger.debug(f"search Lock released at {datetime.now()}")

    @staticmethod
    def __update_movie_download_priority(
            subscribe: Subscribe,
            mediainfo: MediaInfo,
            downloads: Optional[List[Context]],
    ):
        """
        记录电影本轮下载资源优先级，用作后续电影洗版的起始质量状态。
        """
        if not downloads:
            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.MOVIE.value:
            return

        SubscribeOper().update(subscribe.id, {
            "current_priority": priority,
            "last_update": now
        })
        subscribe.current_priority = priority
        subscribe.last_update = now
        if subscribe.best_version and priority != 100:
            # 正在洗版，更新资源优先级
            logger.info(f'{mediainfo.title_year} 正在洗版，更新资源优先级为 {priority}')

    def finish_subscribe_or_not(self, subscribe: Subscribe, meta: MetaBase, mediainfo: MediaInfo,
                                downloads: List[Context] = None,
                                lefts: Dict[Union[int | str], Dict[int, _SchemaNotExistMediaInfo]] = None,
                                force: Optional[bool] = False):
        """
        判断是否应完成订阅
        """
        media_keys = _subscribe_media_keys(subscribe)
        # 是否有剩余集
        no_lefts = not lefts or not any(lefts.get(media_key) for media_key in media_keys)
        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 downloads and meta.type == MediaType.MOVIE:
            self.__update_movie_download_priority(
                subscribe=subscribe,
                mediainfo=mediainfo,
                downloads=downloads,
            )
        # 是否完成订阅
        if not subscribe.best_version:
            # 普通订阅：先按 lefts 写 lack，再判断完成
            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 (
                        meta.type == MediaType.MUSIC
                        and self._is_music_download_complete(subscribe, mediainfo, downloads)
                    )
                    or force):
                self.__finish_subscribe(subscribe=subscribe, meta=meta, mediainfo=mediainfo)
            else:
                logger.info(f'{mediainfo.title_year} 未下载完整，继续订阅 ...')
            return

        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 meta.type == MediaType.MUSIC and not self._is_music_download_complete(
                subscribe,
                mediainfo,
                downloads,
        ):
            logger.info(f'{mediainfo.title_year} 未下载完整，继续洗版 ...')
            return
        if self.__is_best_version_complete(subscribe):
            # 洗版完成
            self.__finish_subscribe(subscribe=subscribe, meta=meta, mediainfo=mediainfo)
        elif not downloads:
            logger.info(f'{mediainfo.title_year} 继续洗版 ...')

    def refresh(self, progress_callback: Optional[Callable[..., None]] = None) -> None:
        """
        订阅刷新

        :param progress_callback: 定时服务进度更新回调
        """
        # 触发刷新站点资源，从缓存中匹配订阅
        sites = self.get_subscribed_sites()
        if sites is None:
            if progress_callback:
                progress_callback(value=100, text="没有订阅需要刷新")
            return
        def _update_refresh_progress(
                value: Optional[float] = None,
                text: Optional[str] = None,
                data: Optional[dict] = None,
        ) -> None:
            """将站点刷新进度映射到订阅刷新的前半阶段。"""
            if progress_callback:
                progress_callback(
                    value=(value or 0) * 0.6,
                    text=text,
                    data=data,
                )

        def _update_match_progress(
                value: Optional[float] = None,
                text: Optional[str] = None,
                data: Optional[dict] = None,
        ) -> None:
            """将订阅匹配进度映射到订阅刷新的后半阶段。"""
            if progress_callback:
                progress_callback(
                    value=60 + (value or 0) * 0.4,
                    text=text,
                    data=data,
                )

        torrents = TorrentsChain().refresh(
            sites=sites,
            progress_callback=_update_refresh_progress if progress_callback else None,
            # 存在音乐订阅时额外抓取站点音乐专用入口，音乐不一定在默认种子首页
            include_music=self.has_music_subscribe(),
        )
        self.match(
            torrents,
            progress_callback=_update_match_progress if progress_callback else None,
        )
        if progress_callback:
            progress_callback(value=100, text="订阅刷新完成")

    @staticmethod
    def get_sub_sites(subscribe: Subscribe) -> List[int]:
        """
        获取订阅中涉及的站点清单
        :param subscribe: 订阅信息对象
        :return: 涉及的站点清单
        """
        # 从系统配置获取默认订阅站点
        default_sites = _system_config().get(SystemConfigKey.RssSites) or []
        # 如果订阅未指定站点，直接返回默认站点
        if not subscribe.sites:
            return default_sites
        # 如果默认订阅站点未设置，直接返回订阅指定站点
        if not default_sites:
            return subscribe.sites or []
        # 尝试解析订阅中的站点数据
        user_sites = subscribe.sites
        # 计算 user_sites 和 default_sites 的交集
        intersection_sites = [site for site in user_sites if site in default_sites]
        # 如果交集为空，返回默认站点
        return intersection_sites if intersection_sites else default_sites

    def get_subscribed_sites(self) -> Optional[List[int]]:
        """
        获取订阅中涉及的所有站点清单（节约资源）
        :return: 返回[]代表所有站点命中，返回None代表没有订阅
        """
        ret_sites = []
        subscribes = SubscribeOper().list()
        if not subscribes:
            # 没有订阅
            return None
        # 刷新订阅选中的Rss站点
        for subscribe in subscribes:
            # 刷新选中的站点
            if subscribe.state in self.get_states_for_search('R'):
                ret_sites.extend(self.get_sub_sites(subscribe))
        # 去重
        if ret_sites:
            ret_sites = list(set(ret_sites))

        return ret_sites

    def has_music_subscribe(self) -> bool:
        """判断是否存在可搜索状态的音乐订阅，用于决定是否额外刷新站点音乐入口。"""
        return self._subscription_query().has_music(
            self.get_states_for_search('R')
        )

    def _prepare_match_torrents(
            self,
            torrents: Dict[str, List[Context]],
    ) -> Dict[str, List[Context]]:
        """预识别待匹配资源，并保留原上下文供后续订阅复用。"""
        processed_torrents: Dict[str, List[Context]] = {}
        for domain, contexts in torrents.items():
            if global_vars.is_system_stopped:
                break
            processed_torrents[domain] = []
            for context in contexts:
                if global_vars.is_system_stopped:
                    break
                if context.torrent_info and getattr(context.torrent_info, "category", None) in (
                        MediaType.MUSIC,
                        MediaType.MUSIC.value,
                ):
                    # 音乐 RSS 使用订阅目标做实体匹配，不应进入影视识别并累计失败次数。
                    processed_torrents[domain].append(context)
                    continue
                if (
                        not context.media_info
                        or not resolve_media_identity(media=context.media_info)[1]
                ) and context.media_recognize_fail_count < 3:
                    logger.debug(
                        f'尝试重新识别种子：{context.torrent_info.title}，当前失败次数：'
                        f'{context.media_recognize_fail_count}/3'
                    )
                    re_mediainfo = MediaChain().recognize_by_meta(
                        context.meta_info,
                        obtain_images=False,
                    )
                    if re_mediainfo:
                        re_mediainfo.clear()
                        context.media_info = re_mediainfo
                        context.match_source = self.__get_media_id_match_source(re_mediainfo)
                        context.candidate_recognized = bool(
                            resolve_media_identity(media=re_mediainfo)[1]
                        )
                        context.media_info_is_target = False
                        context.media_recognize_fail_count = 0
                        logger.debug(f'种子 {context.torrent_info.title} 重新识别成功')
                    else:
                        context.media_recognize_fail_count += 1
                        logger.debug(
                            f'种子 {context.torrent_info.title} 媒体识别失败，失败次数：'
                            f'{context.media_recognize_fail_count}/3'
                        )
                elif context.media_recognize_fail_count >= 3:
                    logger.debug(f'种子 {context.torrent_info.title} 已达到最大识别失败次数(3次)，跳过识别')
                processed_torrents[domain].append(context)
        return processed_torrents

    def match(
            self,
            torrents: Dict[str, List[Context]],
            progress_callback: Optional[Callable[..., None]] = None,
    ) -> None:
        """
        从缓存中匹配订阅，并自动下载。

        该入口保持订阅刷新、定时任务和插件调用的稳定签名，具体匹配流程由内部阶段执行。
        """
        if not torrents:
            logger.warn('没有缓存资源，无法匹配订阅')
            if progress_callback:
                progress_callback(value=100, text="没有缓存资源，跳过订阅匹配")
            return
        if progress_callback:
            progress_callback(value=0, text="正在预处理订阅资源 ...")
        return self._execute_match(
            torrents=torrents,
            progress_callback=progress_callback,
        )

    def _execute_match(
            self,
            torrents: Dict[str, List[Context]],
            progress_callback: Optional[Callable[..., None]] = None,
    ) -> None:
        """
        从缓存中匹配订阅，并自动下载

        :param torrents: 按站点分组的资源上下文
        :param progress_callback: 订阅匹配进度更新回调
        """
        if not torrents:
            logger.warn('没有缓存资源，无法匹配订阅')
            if progress_callback:
                progress_callback(value=100, text="没有缓存资源，跳过订阅匹配")
            return

        if progress_callback:
            progress_callback(value=0, text="正在预处理订阅资源 ...")

        lock_acquired = False
        try:
            lock_acquired = self._acquire_run_lock("match", progress_callback)
            if not lock_acquired:
                return

            processed_torrents = self._prepare_match_torrents(torrents)

            # 所有订阅
            subscribes = SubscribeOper().list(self.get_states_for_search('R'))
            total_num = len(subscribes)
            if progress_callback:
                progress_callback(
                    value=20,
                    text=f"资源预处理完成，开始匹配 {total_num} 个订阅 ...",
                    data={"total": total_num, "finished": 0},
                )
            try:
                for index, subscribe in enumerate(subscribes, start=1):
                    if global_vars.is_system_stopped:
                        break
                    if progress_callback:
                        progress_callback(
                            value=20 + (
                                (index - 1) / total_num * 80 if total_num else 80
                            ),
                            text=(
                                f"正在匹配订阅（{index}/{total_num}）"
                                f"{subscribe.name} ..."
                            ),
                            data={
                                "total": total_num,
                                "finished": index - 1,
                                "current": subscribe.id,
                            },
                        )
                    logger.info(f'开始匹配订阅，标题：{subscribe.name} ...')
                    if subscribe.type == MediaType.MUSIC.value:
                        music_contexts = [
                            context
                            for contexts in processed_torrents.values()
                            for context in contexts
                        ]
                        self._match_music_subscribe(subscribe, music_contexts)
                        continue
                    mediakey = _subscribe_media_key(subscribe)
                    try:
                        meta = build_subscribe_meta(subscribe)
                    except ValueError:
                        logger.error(f'订阅 {subscribe.name} 类型错误：{subscribe.type}')
                        continue
                    # 订阅的站点域名列表
                    domains = []
                    if subscribe.sites:
                        domains = SiteOper().get_domains_by_ids(subscribe.sites)
                    # 识别媒体信息
                    mediainfo: MediaInfo = MediaChain().recognize_media(
                        meta=meta,
                        mtype=meta.type,
                        **_subscribe_recognize_kwargs(subscribe),
                        episode_group=subscribe.episode_group,
                        cache=False,
                    )
                    if not mediainfo:
                        logger.warn(
                            f"未识别到媒体信息，标题：{subscribe.name}，"
                            f"媒体来源：{subscribe.media_source}，媒体 ID：{subscribe.media_id}"
                        )
                        continue

                    # 如果媒体已存在或已下载完毕，跳过当前订阅处理
                    exist_flag, no_exists = self.check_and_handle_existing_media(subscribe=subscribe, meta=meta,
                                                                                 mediainfo=mediainfo,
                                                                                 mediakey=mediakey)
                    if exist_flag:
                        continue

                    # 清理多余信息
                    mediainfo.clear()

                    # 订阅识别词
                    if subscribe.custom_words:
                        custom_words_list = subscribe.custom_words.split("\n")
                    else:
                        custom_words_list = None

                    # 遍历预识别后的种子
                    _match_context = []
                    torrenthelper = TorrentHelper()
                    systemconfig = _system_config()
                    wordsmatcher = WordsMatcher()
                    for domain, contexts in processed_torrents.items():
                        if global_vars.is_system_stopped:
                            break
                        if domains and domain not in domains:
                            continue
                        logger.debug(f'开始匹配站点：{domain}，共缓存了 {len(contexts)} 个种子...')
                        for context in contexts:
                            if global_vars.is_system_stopped:
                                break
                            # 提取信息
                            _context = copy.copy(context)
                            torrent_meta = _context.meta_info
                            torrent_mediainfo = _context.media_info
                            torrent_info = _context.torrent_info

                            # 不在订阅站点范围的不处理
                            sub_sites = self.get_sub_sites(subscribe)
                            if sub_sites and torrent_info.site not in sub_sites:
                                logger.debug(f"{torrent_info.site_name} - {torrent_info.title} 不符合订阅站点要求")
                                continue

                            # 有自定义识别词时，需要判断是否需要重新识别
                            if custom_words_list:
                                # 使用org_string，应用一次后理论上不能再次应用
                                _, apply_words = wordsmatcher.prepare(torrent_meta.org_string,
                                                                      custom_words=custom_words_list)
                                if apply_words:
                                    logger.info(
                                        f'{torrent_info.site_name} - {torrent_info.title} 因订阅存在自定义识别词，重新识别元数据...')
                                    # 重新识别元数据
                                    torrent_meta = MetaInfo(title=torrent_info.title, subtitle=torrent_info.description,
                                                            custom_words=custom_words_list)
                                    # 更新元数据缓存
                                    _context.meta_info = torrent_meta
                                    # 重新识别媒体信息
                                    torrent_mediainfo = MediaChain().recognize_by_meta(
                                        torrent_meta,
                                        episode_group=subscribe.episode_group,
                                        obtain_images=False,
                                    )
                                    if torrent_mediainfo:
                                        # 清理多余信息
                                        torrent_mediainfo.clear()
                                        # 更新种子缓存
                                        _context.media_info = torrent_mediainfo
                                        _context.match_source = self.__get_media_id_match_source(torrent_mediainfo)
                                        _context.candidate_recognized = bool(
                                            resolve_media_identity(media=torrent_mediainfo)[1]
                                        )
                                        _context.media_info_is_target = False

                            # 如果仍然没有识别到媒体信息，尝试标题匹配
                            if not torrent_mediainfo or not resolve_media_identity(
                                    media=torrent_mediainfo
                            )[1]:
                                logger.debug(
                                    f'{torrent_info.site_name} - {torrent_info.title} 重新识别失败，尝试通过标题匹配...')
                                if not self.__is_title_match_allowed(
                                        mediainfo=mediainfo,
                                        torrent_meta=torrent_meta,
                                        torrent_info=torrent_info,
                                ):
                                    continue
                                # 匹配成功
                                logger.info(
                                    f'{mediainfo.title_year} 通过标题匹配到可选资源：{torrent_info.site_name} - {torrent_info.title}')
                                torrent_mediainfo = mediainfo
                                # 更新种子缓存
                                _context.media_info = mediainfo
                                _context.match_source = "title"
                                _context.candidate_recognized = False
                                _context.media_info_is_target = True

                            # 直接比对媒体信息
                            if torrent_mediainfo and resolve_media_identity(
                                    media=torrent_mediainfo
                            )[1]:
                                if torrent_mediainfo.type != mediainfo.type:
                                    continue
                                torrent_mediainfo = self.__reconcile_candidate_media(
                                    target_mediainfo=mediainfo,
                                    candidate_mediainfo=torrent_mediainfo,
                                    torrent_meta=torrent_meta,
                                    torrent_info=torrent_info,
                                    context=_context,
                                )
                                if not torrent_mediainfo:
                                    continue
                                match_source = _context.match_source
                                if match_source == "title":
                                    # 标题兜底使用的是订阅目标 media_info，不能标记为候选自身识别结果。
                                    _context.candidate_recognized = False
                                    _context.media_info_is_target = True
                                    match_label = "标题复核"
                                elif match_source == "unknown":
                                    _context.match_source = self.__get_media_id_match_source(torrent_mediainfo)
                                    _context.candidate_recognized = True
                                    _context.media_info_is_target = False
                                    match_label = "媒体ID"
                                else:
                                    _context.candidate_recognized = True
                                    _context.media_info_is_target = False
                                    match_label = "媒体ID"
                                logger.info(
                                    f'{mediainfo.title_year} 通过{match_label}匹配到可选资源：'
                                    f'{torrent_info.site_name} - {torrent_info.title}'
                                )
                            else:
                                continue

                            # 如果是电视剧
                            if torrent_mediainfo.type == MediaType.TV:
                                # 有多季的不要
                                if len(torrent_meta.season_list) > 1:
                                    logger.debug(f'{torrent_info.title} 有多季，不处理')
                                    continue
                                # 比对季
                                if torrent_meta.begin_season is not None:
                                    if meta.begin_season != torrent_meta.begin_season:
                                        logger.debug(f'{torrent_info.title} 季不匹配')
                                        continue
                                elif meta.begin_season != 1:
                                    logger.debug(f'{torrent_info.title} 季不匹配')
                                    continue
                                # 非洗版
                                if not subscribe.best_version:
                                    # 不是缺失的剧集不要
                                    if no_exists and no_exists.get(mediakey):
                                        # 缺失集
                                        no_exists_info = no_exists.get(mediakey).get(subscribe.season)
                                        if no_exists_info:
                                            # 是否有交集
                                            if no_exists_info.episodes and \
                                                    torrent_meta.episode_list and \
                                                    not set(no_exists_info.episodes).intersection(
                                                        set(torrent_meta.episode_list)
                                                    ):
                                                logger.debug(
                                                    f'{torrent_info.title} 对应剧集 {torrent_meta.episode_list} 未包含缺失的剧集'
                                                )
                                                continue
                                else:
                                    if not self.__is_full_season_best_version_resource(
                                            meta=torrent_meta,
                                            subscribe=subscribe,
                                    ):
                                        logger.debug(
                                            f"{subscribe.name} 正在全集洗版，{torrent_info.title} 不是全集资源"
                                        )
                                        continue
                                    # 洗版时，不符合订阅集数的不要
                                    if (
                                            meta.type == MediaType.TV
                                            and not self._is_episode_range_covered(
                                        meta=torrent_meta,
                                        subscribe=subscribe,
                                    )
                                    ):
                                        logger.debug(
                                            f"{subscribe.name} 正在洗版，{torrent_info.title} 不符合订阅集数范围"
                                        )
                                        continue

                            # 匹配订阅附加参数
                            if not torrenthelper.filter_torrent(torrent_info=torrent_info,
                                                                filter_params=self.get_params(subscribe)):
                                continue

                            # 优先级过滤规则
                            if subscribe.best_version:
                                rule_groups = subscribe.filter_groups \
                                              or systemconfig.get(SystemConfigKey.BestVersionFilterRuleGroups)
                            else:
                                rule_groups = subscribe.filter_groups \
                                              or systemconfig.get(SystemConfigKey.SubscribeFilterRuleGroups)
                            result: List[TorrentInfo] = self.filter_torrents(
                                rule_groups=rule_groups,
                                torrent_list=[torrent_info],
                                mediainfo=torrent_mediainfo)
                            if result is not None and not result:
                                # 不符合过滤规则
                                logger.debug(f"{torrent_info.title} 不匹配过滤规则")
                                continue

                            # 洗版时，优先级小于已下载优先级的不要
                            if subscribe.best_version:
                                if meta.type == MediaType.TV:
                                    if not self.__prepare_best_version_tv_candidate(
                                            subscribe=subscribe,
                                            context=_context,
                                            priority=torrent_info.pri_order,
                                    ):
                                        logger.info(
                                            f'{subscribe.name} 正在洗版，{torrent_info.title} '
                                            f'优先级未达到当前模式的升级条件')
                                        continue
                                if (
                                        meta.type != MediaType.TV
                                        and subscribe.current_priority
                                        and torrent_info.pri_order <= subscribe.current_priority
                                ):
                                    logger.info(
                                        f'{subscribe.name} 正在洗版，{torrent_info.title} 优先级低于或等于已下载优先级')
                                    continue

                            # 匹配成功
                            logger.info(f'{mediainfo.title_year} 匹配成功：{torrent_info.title}')
                            # 自定义属性
                            if subscribe.media_category:
                                torrent_mediainfo.category = subscribe.media_category
                            if subscribe.episode_group:
                                torrent_mediainfo.episode_group = subscribe.episode_group
                            _match_context.append(_context)

                    if not _match_context:
                        # 未匹配到资源
                        logger.info(f'{mediainfo.title_year} 未匹配到符合条件的资源')
                        self.finish_subscribe_or_not(subscribe=subscribe, meta=meta,
                                                     mediainfo=mediainfo, lefts=no_exists)
                        continue

                    # 开始批量择优下载
                    logger.info(f'{mediainfo.title_year} 匹配完成，共匹配到{len(_match_context)}个资源')
                    downloads, lefts = self.__download_best_version_with_full_pack_first(
                        contexts=_match_context,
                        no_exists=no_exists,
                        subscribe=subscribe,
                        mediakey=mediakey,
                        username=subscribe.username,
                        save_path=subscribe.save_path,
                        downloader=subscribe.downloader,
                        source=self.get_subscribe_source_keyword(subscribe)
                    )

                    # 同步外部修改，更新订阅信息
                    subscribe = SubscribeOper().get(subscribe.id)

                    # 判断是否要完成订阅
                    if subscribe:
                        self.finish_subscribe_or_not(subscribe=subscribe, meta=meta, mediainfo=mediainfo,
                                                     downloads=downloads, lefts=lefts)
            finally:
                processed_torrents.clear()
                del processed_torrents
                subscribes.clear()
                del subscribes
                if progress_callback:
                    progress_callback(
                        value=100,
                        text="订阅资源匹配完成",
                        data={"total": total_num, "finished": total_num},
                    )
        finally:
            if lock_acquired:
                self._rlock.release()
                logger.debug(f"match Lock released at {datetime.now()}")

    def check(self, progress_callback: Optional[Callable[..., None]] = None) -> None:
        """
        定时检查订阅，更新订阅信息

        :param progress_callback: 定时服务进度更新回调
        """
        # 查询所有订阅
        subscribeoper = SubscribeOper()
        subscribes = subscribeoper.list()
        total_num = len(subscribes)
        if progress_callback:
            progress_callback(
                value=0,
                text=f"开始更新订阅元数据，共 {total_num} 个订阅 ...",
                data={"total": total_num, "finished": 0},
            )
        # 遍历订阅
        for index, subscribe in enumerate(subscribes, start=1):
            if global_vars.is_system_stopped:
                break
            logger.info(f'开始更新订阅元数据：{subscribe.name} ...')
            if progress_callback:
                progress_callback(
                    value=(index - 1) / total_num * 100 if total_num else 100,
                    text=f"正在更新订阅元数据（{index}/{total_num}）{subscribe.name} ...",
                    data={
                        "total": total_num,
                        "finished": index - 1,
                        "current": subscribe.id,
                    },
                )
            try:
                meta = build_subscribe_meta(subscribe)
            except ValueError:
                logger.error(f'订阅 {subscribe.name} 类型错误：{subscribe.type}')
                continue
            # 识别媒体信息
            if meta.type == MediaType.MUSIC:
                mediainfo = self._recognize_music_subscribe(subscribe)
            else:
                mediainfo: MediaInfo = MediaChain().recognize_media(
                    meta=meta,
                    mtype=meta.type,
                    **_subscribe_recognize_kwargs(subscribe),
                    episode_group=subscribe.episode_group,
                    cache=False,
                )
            if not mediainfo:
                logger.warn(
                    f"未识别到媒体信息，标题：{subscribe.name}，"
                    f"媒体来源：{subscribe.media_source}，媒体 ID：{subscribe.media_id}"
                )
                continue
            # 对于电视剧，获取当前季的总集数
            episodes = (mediainfo.seasons.get(subscribe.season) or []) \
                if meta.type == MediaType.TV else []
            progress_update = {}
            if subscribe.type == MediaType.TV.value and not subscribe.manual_total_episode and len(episodes):
                current_total_episode = len(episodes)
                # 外部事件只能向上覆盖主程序本次识别到的 TMDB 当前季总集数，已有订阅按最终 total 跟随持久化。
                total_episode = self.__apply_episodes_refresh(
                    current_total_episode, season=subscribe.season, mediainfo=mediainfo,
                    media_source=subscribe.media_source,
                    media_id=subscribe.media_id,
                    subscribe_id=subscribe.id, scene="refresh")
                old_total_episode = subscribe.total_episode or 0
                if total_episode and total_episode < old_total_episode:
                    total_episode = self.__resolve_total_episode_decrease(
                        subscribe=subscribe,
                        candidate_total=total_episode,
                        meta=meta,
                        mediainfo=mediainfo,
                        mediakey=_subscribe_media_key(subscribe),
                    )
                if total_episode and total_episode != old_total_episode:
                    progress_update = self.__prepare_total_episode_change_fields(
                        subscribe=subscribe,
                        total_episode=total_episode,
                        old_total_episode=old_total_episode,
                    )
                else:
                    total_episode = subscribe.total_episode
                    progress_update = {"lack_episode": subscribe.lack_episode}
                    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}，'
                    f'缺失集数为{progress_update.get("lack_episode", subscribe.lack_episode)} ...')
            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={})
            # 更新TMDB信息
            update_data = {
                "name": mediainfo.title,
                "year": str(mediainfo.year) if mediainfo.year is not None else None,
                "vote": mediainfo.vote_average,
                "poster": mediainfo.get_poster_image(),
                "backdrop": mediainfo.get_backdrop_image(),
                "description": mediainfo.overview,
                "media_source": resolve_media_identity(media=mediainfo)[0],
                "media_id": resolve_media_identity(media=mediainfo)[1],
                "total_episode": total_episode,
            }
            if meta.type == MediaType.MUSIC:
                music_type = getattr(mediainfo, "music_type", None)
                update_data.update({
                    "music_type": music_type,
                    "total_tracks": getattr(mediainfo, "total_tracks", None)
                    if music_type == MUSIC_ENTITY_ALBUM else None,
                })
            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} 订阅元数据更新完成')
            if progress_callback:
                progress_callback(
                    value=index / total_num * 100 if total_num else 100,
                    text=f"订阅元数据（{index}/{total_num}）更新完成",
                    data={"total": total_num, "finished": index},
                )
        if progress_callback:
            progress_callback(value=100, text="订阅元数据更新完成")

    def get_subscribe_by_source(self, source: str) -> Optional[Subscribe]:
        """
        从来源获取订阅
        """
        return self._subscription_query().get_by_source(
            self.parse_subscribe_source_keyword(source)
        )

    @staticmethod
    def follow(progress_callback: Optional[Callable[..., None]] = None) -> None:
        """
        刷新follow的用户分享，并自动添加订阅

        :param progress_callback: 定时服务进度更新回调
        """
        follow_users: List[str] = _system_config().get(SystemConfigKey.FollowSubscribers)
        if not follow_users:
            if progress_callback:
                progress_callback(value=100, text="未配置 Follow 订阅用户，跳过刷新")
            return
        logger.info(f'开始刷新follow用户分享订阅 ...')
        success_count = 0
        subscribeoper = SubscribeOper()
        share_subscribes = MoviePilotServerHelper.get_subscribe_shares() or []
        total_num = len(share_subscribes)
        if progress_callback:
            progress_callback(
                value=0,
                text=f"开始刷新 Follow 订阅分享，共 {total_num} 条 ...",
                data={"total": total_num, "finished": 0},
            )
        for index, share_sub in enumerate(share_subscribes, start=1):
            if global_vars.is_system_stopped:
                break
            if progress_callback:
                progress_callback(
                    value=(index - 1) / total_num * 100 if total_num else 100,
                    text=f"正在处理 Follow 订阅分享（{index}/{total_num}）...",
                    data={"total": total_num, "finished": index - 1},
                )
            uid = share_sub.get("share_uid")
            if uid and uid in follow_users:
                # 订阅已存在则跳过
                if subscribeoper.exists(media_source=share_sub.get("media_source"),
                                        media_id=share_sub.get("media_id"),
                                        music_type=share_sub.get("music_type"),
                                        season=share_sub.get("season"),
                                        episode_group=share_sub.get("episode_group")):
                    continue
                # 已经订阅过跳过
                if subscribeoper.exist_history(media_source=share_sub.get("media_source"),
                                               media_id=share_sub.get("media_id"),
                                               music_type=share_sub.get("music_type"),
                                               season=share_sub.get("season"),
                                               episode_group=share_sub.get("episode_group")):
                    continue
                # 去除无效属性
                for key in list(share_sub.keys()):
                    if not hasattr(_SchemaSubscribe(), key):
                        share_sub.pop(key)
                # 类型转换
                subscribe_in = _SchemaSubscribe(**share_sub)
                mtype = MediaType(subscribe_in.type)
                # 非 TMDB 标题可能携带季号，入库前统一拆分。
                if (
                    mtype != MediaType.MUSIC
                    and normalize_media_source(subscribe_in.media_source)
                    not in (None, MediaSource.TMDB)
                ):
                    meta = MetaInfo(subscribe_in.name)
                    subscribe_in.name = meta.name
                    if subscribe_in.season is None:
                        subscribe_in.season = meta.begin_season
                # 标题转换
                if subscribe_in.name:
                    title = subscribe_in.name
                else:
                    title = None
                sid, message = SubscribeChain().add(mtype=mtype,
                                                    title=title,
                                                    year=subscribe_in.year,
                                                    season=subscribe_in.season,
                                                    episode_group=subscribe_in.episode_group,
                                                    media_source=subscribe_in.media_source,
                                                    media_id=subscribe_in.media_id,
                                                    music_type=subscribe_in.music_type,
                                                    total_tracks=subscribe_in.total_tracks,
                                                    username="订阅分享",
                                                    best_version=subscribe_in.best_version,
                                                    save_path=subscribe_in.save_path,
                                                    search_imdbid=subscribe_in.search_imdbid,
                                                    custom_words=subscribe_in.custom_words,
                                                    media_category=subscribe_in.media_category,
                                                    filter_groups=subscribe_in.filter_groups,
                                                    exist_ok=True)
                if sid:
                    success_count += 1
                    logger.info(f'follow用户分享订阅 {title} 添加成功')
                else:
                    logger.error(f'follow用户分享订阅 {title} 添加失败：{message}')
        logger.info(f'follow用户分享订阅刷新完成，共添加 {success_count} 个订阅')
        if progress_callback:
            progress_callback(
                value=100,
                text=f"Follow 订阅分享刷新完成，新增 {success_count} 个订阅",
                data={
                    "total": total_num,
                    "finished": total_num,
                    "added": success_count,
                },
            )

    async def cache_calendar(
            self,
            progress_callback: Optional[Callable[..., None]] = None,
    ) -> None:
        """
        预缓存订阅日历，实际上就是查询一遍所有订阅的媒体信息
        前端请示是异常的，所以需要使用异步缓存方法

        :param progress_callback: 定时服务进度更新回调
        """
        logger.info(f'开始预缓存订阅日历 ...')
        subscribes = await SubscribeOper().async_list()
        total_num = len(subscribes)
        if progress_callback:
            progress_callback(
                value=0,
                text=f"开始预缓存订阅日历，共 {total_num} 个订阅 ...",
                data={"total": total_num, "finished": 0},
            )
        for index, subscribe in enumerate(subscribes, start=1):
            if global_vars.is_system_stopped:
                break
            if progress_callback:
                progress_callback(
                    value=(index - 1) / total_num * 100 if total_num else 100,
                    text=f"正在预缓存订阅日历（{index}/{total_num}）{subscribe.name} ...",
                    data={
                        "total": total_num,
                        "finished": index - 1,
                        "current": subscribe.id,
                    },
                )
            try:
                mtype = MediaType(subscribe.type)
            except ValueError:
                logger.error(f'订阅 {subscribe.name} 类型错误：{subscribe.type}')
                continue
            # 先按订阅的主媒体身份预热对应数据源，再对 TMDB 额外预热分集接口。
            if mtype == MediaType.MUSIC:
                mediainfo = await self._async_recognize_music_subscribe(subscribe)
            else:
                mediainfo: MediaInfo = await MediaChain().async_recognize_media(
                    mtype=mtype,
                    **_subscribe_recognize_kwargs(subscribe),
                    episode_group=subscribe.episode_group,
                    cache=False,
                )
            if not mediainfo:
                logger.warn(
                    f'未识别到媒体信息，标题：{subscribe.name}，'
                    f'媒体源：{subscribe.media_source}，媒体ID：{subscribe.media_id}')
                continue
            if (
                mtype == MediaType.TV
                and mediainfo.media_source == MediaSource.TMDB
                and mediainfo.tmdb_id
            ):
                episodes = await TmdbChain().async_tmdb_episodes(tmdbid=mediainfo.tmdb_id,
                                                                 season=subscribe.season,
                                                                 episode_group=subscribe.episode_group)
                if not episodes:
                    logger.warn(
                        f'未识别到季集信息，标题：{subscribe.name}，tmdbid：{mediainfo.tmdb_id}，季：{subscribe.season}')
                    continue
            if progress_callback:
                progress_callback(
                    value=index / total_num * 100 if total_num else 100,
                    text=f"订阅日历（{index}/{total_num}）预缓存完成",
                    data={"total": total_num, "finished": index},
                )
        logger.info(f'订阅日历预缓存完成')
        if progress_callback:
            progress_callback(value=100, text="订阅日历预缓存完成")

    @staticmethod
    def __update_subscribe_note(subscribe: Subscribe, downloads: Optional[List[Context]]):
        """
        更新已下载信息到note字段
        """
        # 查询现有Note
        if not downloads:
            return
        note = []
        if subscribe.note:
            note = subscribe.note or []
        for context in downloads:
            meta = context.meta_info
            mediainfo = context.media_info
            subscribe_source, subscribe_media_id = resolve_media_identity(media=subscribe)
            media_source, media_id = resolve_media_identity(media=mediainfo)
            if (
                subscribe_source != media_source
                or not subscribe_media_id
                or not media_id
                or subscribe_media_id != media_id
            ):
                continue
            items = []
            if mediainfo.type == MediaType.TV:
                # 电视剧有集数，使用 episode_list
                items = meta.episode_list
            elif mediainfo.type == MediaType.MOVIE:
                # 电影只有一个条目，设置为 [1]
                items = [1]
            elif mediainfo.type == MediaType.MUSIC:
                # 专辑只能记录已由下载层确认的整专资源；单曲任一成功任务即可完成。
                if getattr(subscribe, "music_type", None) != MUSIC_ENTITY_ALBUM \
                        or context.confirmed_full_coverage:
                    items = [1]
            if not items:
                continue
            # 合并已下载的集数或电影项（去重）
            note = list(set(note).union(set(items)))
        # 更新订阅
        if note:
            SubscribeOper().update(subscribe.id, {
                "note": note
            })

    @staticmethod
    def __get_downloaded(subscribe: Subscribe) -> List[int]:
        """
        获取已下载过的集数或电影。

        洗版分支只返回 priority==100 的完成集；priority<100 的集仍要继续搜索更高
        优先级版本，不能并入返回值（会让下游把 pending 减空、订阅卡死）。
        note 由非洗版分支消费，用于洗版关闭后的迁移读取。
        """
        if subscribe.best_version:
            if subscribe.type == MediaType.TV.value:
                completed = SubscribeChain.__get_best_version_completed_episodes(subscribe)
                if completed:
                    logger.info(f'订阅 {subscribe.name} 第{subscribe.season}季 已完成洗版剧集：{completed}')
                return completed
            return []
        note = subscribe.note or []
        if not note:
            return []
        # 针对 TV 类型，返回已下载的集数
        if subscribe.type == MediaType.TV.value:
            logger.info(f'订阅 {subscribe.name} 第{subscribe.season}季 已下载集数：{note}')
            return note
        # 针对 Movie/Music 类型，直接返回已下载的单项内容
        if subscribe.type in (MediaType.MOVIE.value, MediaType.MUSIC.value):
            logger.info(f'订阅 {subscribe.name} 已下载内容：{note}')
            return note
        return []

    @classmethod
    def __prepare_subscribe_progress_fields(
            cls,
            subscribe: Subscribe,
            no_exists: Optional[Dict[Union[int, str], Dict[int, _SchemaNotExistMediaInfo]]] = None,
            touch_last_update: Optional[bool] = False,
    ) -> Dict[str, Any]:
        """
        准备订阅进度持久化字段。

        该方法只返回待写字段，不主动写库。普通电视剧的 no_exists 为空时表示当前缺失结果为空；
        洗版电视剧按 note 与 episode_priority 计算未下载过任何版本的目标集数量。
        """
        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 and not cls.__is_full_best_version_enabled(subscribe):
                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 __apply_subscribe_update(subscribe: Subscribe, update_data: Dict[str, Any]) -> None:
        """
        写入订阅字段并同步当前内存对象，保证后续事件和判断读取最终快照。
        """
        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, _SchemaNotExistMediaInfo]]] = 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 = MediaChain().recognize_media(
                meta=meta,
                mtype=meta.type,
                **_subscribe_recognize_kwargs(subscribe),
                episode_group=subscribe.episode_group,
                cache=False,
            )
            if not mediainfo:
                return {"scene": scene, "updated": False, "fields": [], "reason": "recognize_failed"}
            mediakey = _subscribe_media_key(subscribe)
            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:
                episode_number = int(episode)
            except (TypeError, ValueError):
                ignored.append({"episode": episode, "reason": "invalid"})
                continue
            if episode_number not in target_episodes:
                ignored.append({"episode": episode, "reason": "out_of_range"})
                continue
            if episode_number in note_set:
                ignored.append({"episode": episode, "reason": "duplicate"})
                priority_episodes.add(episode_number)
                continue
            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
            subscribe_identity = resolve_media_identity(media=subscribe)
            media_identity = resolve_media_identity(media=media)
            if subscribe_identity != media_identity:
                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 download.meta_info and download.meta_info.episode_list:
                episodes = download.meta_info.episode_list
            elif download.confirmed_full_coverage:
                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 = download.torrent_info.pri_order
            if (
                    self.__is_full_best_version_enabled(subscribe)
                    and download.confirmed_full_coverage
                    and isinstance(priority, int)
                    and not isinstance(priority, bool)
                    and priority > (subscribe.current_priority or 0)
            ):
                subscribe.current_priority = priority
                update_data["current_priority"] = priority
            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):
        """
        完成订阅
        """
        # 如果订阅状态为待定（P），说明订阅信息尚未完全更新，无法完成订阅
        if subscribe.state == "P":
            return
        # 发送订阅完成判定事件，在写入 DB 前，允许外部据完结策略否决本次自动完成
        completion_event = eventmanager.send_event(
            ChainEventType.SubscribeCompletionCheck,
            SubscribeCompletionCheckEventData(subscribe=subscribe, mediainfo=mediainfo, meta=meta))
        if completion_event and completion_event.event_data:
            completion_data: SubscribeCompletionCheckEventData = completion_event.event_data
            if completion_data.cancel:
                logger.info(f'{mediainfo.title_year} 完成被 [{completion_data.source}] 否决：{completion_data.reason}')
                return
        # 完成订阅
        msgstr = "订阅" if not subscribe.best_version else "洗版"
        logger.info(f'{mediainfo.title_year} 完成{msgstr}')

        # 完成命令在同一 Session/UoW 中写历史、删除订阅并暂存可恢复副作用。
        if mediainfo.type == MediaType.TV:
            link = self.runtime_config.television_subscribe_url
        elif mediainfo.type == MediaType.MUSIC:
            link = self.runtime_config.music_subscribe_url
        else:
            link = self.runtime_config.movie_subscribe_url

        def notify() -> None:
            """提交成功后发送完成通知，保持历史消息 ABI。"""
            self.post_message(_completion_message)

        _completion_message = _SchemaMessage(
            mtype=MessageType.Subscribe,
            ctype=ContentType.SubscribeComplete,
            image=mediainfo.get_message_image(),
            link=link,
            username=subscribe.username,
        )
        _completion_message = MessageTemplateHelper.render(
            _completion_message,
            meta=meta,
            mediainfo=mediainfo,
            msgstr=msgstr,
            username=subscribe.username,
        ) or _completion_message

        with get_subscription_completion_scope() as command:
            command.execute(
                subscribe_id=subscribe.id,
                subscribe_info=subscribe.to_dict(),
                mediainfo=mediainfo.to_dict(),
                notify=notify,
                report=MoviePilotServerHelper.sub_done_durable,
                notification=_completion_message.model_dump(mode="json"),
            )

    def _interaction_handler(self) -> "SubscribeInteractionHandler":
        """构造 /subscribes 交互处理器，业务动作由本链提供。"""
        return SubscribeInteractionHandler(
            messenger=self,
            actions=self,
            repository=SubscribeOper(),
            delete_subscription=self._delete_subscription,
        )

    @staticmethod
    def _delete_subscription(subscribe_id: int) -> bool:
        """通过统一同步命令删除订阅，保留消息入口原有的全局管理权限。"""
        with get_sync_delete_subscribe_scope() as command:
            return command.execute(
                subscribe_id,
                SubscribeDeletionActor(username="", is_superuser=True),
            )

    def remote_delete(self, arg_str: str, channel: NotificationChannel,
                      userid: Union[str, int] = None, source: Optional[str] = None):
        """
        删除订阅
        """
        if not arg_str:
            self.post_message(_SchemaMessage(
                channel=channel,
                source=source,
                title="请输入正确的命令格式：/subscribe_delete [id]，"
                      "[id]为订阅编号",
                userid=userid,
                save_history=False))
            return
        arg_strs = str(arg_str).split()
        for arg_str in arg_strs:
            arg_str = arg_str.strip()
            if not arg_str.isdigit():
                continue
            subscribe_id = int(arg_str)
            if not self._delete_subscription(subscribe_id):
                self.post_message(_SchemaMessage(
                    channel=channel, source=source,
                    title=f"订阅编号 {subscribe_id} 不存在！",
                    userid=userid,
                    save_history=False))
                return
        # 重新发送消息
        self.remote_list(channel=channel, userid=userid, source=source)

    @staticmethod
    def __get_subscribe_no_exits(subscribe_name: str,
                                 no_exists: Dict[Union[int, str], Dict[int, _SchemaNotExistMediaInfo]],
                                 mediakey: Union[str, int],
                                 begin_season: int,
                                 total_episode: Optional[int],
                                 start_episode: Optional[int],
                                 downloaded_episodes: List[int] = None
                                 ) -> Tuple[bool, Dict[Union[int, str], Dict[int, _SchemaNotExistMediaInfo]]]:
        """
        根据订阅开始集数和总集数，结合TMDB信息计算当前订阅的缺失集数
        :param subscribe_name: 订阅名称
        :param no_exists: 缺失季集列表
        :param mediakey: TMDB ID或豆瓣ID
        :param begin_season: 开始季
        :param total_episode: 订阅设定总集数
        :param start_episode: 订阅设定开始集数
        :param downloaded_episodes: 已下载集数
        """
        # 使用订阅的总集数和开始集数替换no_exists
        if not no_exists or not no_exists.get(mediakey):
            return False, no_exists
        no_exists_item = no_exists.get(mediakey)
        if total_episode or start_episode:
            logger.info(f'订阅 {subscribe_name} 设定的开始集数：{start_episode}、总集数：{total_episode}')
            # 该季原缺失信息
            no_exist_season = no_exists_item.get(begin_season)
            if no_exist_season:
                # 原集列表
                episode_list = no_exist_season.episodes
                # 原总集数
                total = no_exist_season.total_episode
                # 原开始集数
                start = no_exist_season.start_episode

                # 更新剧集列表、开始集数、总集数
                if not episode_list:
                    # 整季缺失
                    start_episode = start_episode or start
                    total_episode = total_episode or total
                    original_start = start if start is not None else 1
                    # 空集列表会被下载链解释为整季下载；当订阅开始集裁掉季初范围时，需要转成显式集数。
                    if start_episode and total_episode and start_episode > original_start:
                        episodes = list(range(start_episode, total_episode + 1))
                        if not episodes:
                            return True, {}
                    else:
                        episodes = []
                else:
                    # 部分缺失
                    if not start_episode \
                            and not total_episode:
                        # 无需调整
                        return False, no_exists
                    if not start_episode:
                        # 没有自定义开始集
                        start_episode = start
                    if not total_episode:
                        # 没有自定义总集数
                        total_episode = total
                    # 新的集列表
                    new_episodes = list(range(max(start_episode, start), total_episode + 1))
                    # 与原集列表取交集
                    episodes = list(set(episode_list).intersection(set(new_episodes)))
                    # 交集为空时，说明订阅的剧集均已入库
                    if not episodes:
                        return True, {}
                # 更新集合
                no_exists[mediakey][begin_season] = _SchemaNotExistMediaInfo(
                    season=begin_season,
                    episodes=episodes,
                    total_episode=total_episode,
                    start_episode=start_episode,
                    require_complete_coverage=no_exist_season.require_complete_coverage
                )
        # 根据订阅已下载集数更新缺失集数
        if downloaded_episodes:
            logger.info(f'订阅 {subscribe_name} 已下载集数：{downloaded_episodes}')
            # 该季原缺失信息
            no_exist_season = no_exists_item.get(begin_season)
            if no_exist_season:
                # 原集列表
                episode_list = no_exist_season.episodes
                # 原总集数
                total = no_exist_season.total_episode
                # 原开始集数
                start = no_exist_season.start_episode
                # 整季缺失
                if not episode_list:
                    episode_list = list(range(start, total + 1))
                # 更新剧集列表
                episodes = list(set(episode_list).difference(set(downloaded_episodes)))
                # 如果存在已下载剧集，则差集为空时，说明所有均已存在
                if not episodes:
                    return True, {}
                # 更新集合
                no_exists[mediakey][begin_season] = _SchemaNotExistMediaInfo(
                    season=begin_season,
                    episodes=episodes,
                    total_episode=total,
                    start_episode=start,
                    require_complete_coverage=no_exist_season.require_complete_coverage
                )
            else:
                # 开始集数
                start = start_episode or 1
                # 更新剧集列表
                episodes = list(set(range(start, total_episode + 1)).difference(set(downloaded_episodes)))
                # 如果存在已下载剧集，则差集为空时，说明所有均已存在
                if not episodes:
                    return True, {}
                no_exists[mediakey][begin_season] = _SchemaNotExistMediaInfo(
                    season=begin_season,
                    episodes=episodes,
                    total_episode=total_episode,
                    start_episode=start,
                    require_complete_coverage=False,
                )
        logger.info(f'订阅 {subscribe_name} 缺失剧集数更新为：{no_exists}')
        return False, no_exists

    @eventmanager.register(EventType.SiteDeleted)
    def remove_site(self, event: Event):
        """
        从订阅中移除与站点相关的设置
        """
        if not event:
            return
        event_data = event.event_data or {}
        site_id = event_data.get("site_id")
        if not site_id:
            return
        subscribeoper = SubscribeOper()
        if site_id == "*":
            # 站点被重置
            _system_config().set(SystemConfigKey.RssSites, [])
            for subscribe in subscribeoper.list():
                if not subscribe.sites:
                    continue
                subscribeoper.update(subscribe.id, {
                    "sites": []
                })
            return
        # 从选中的rss站点中移除
        selected_sites = _system_config().get(SystemConfigKey.RssSites) or []
        if site_id in selected_sites:
            selected_sites.remove(site_id)
            _system_config().set(SystemConfigKey.RssSites, selected_sites)
        # 查询所有订阅
        for subscribe in subscribeoper.list():
            if not subscribe.sites:
                continue
            sites = subscribe.sites or []
            if site_id not in sites:
                continue
            sites.remove(site_id)
            subscribeoper.update(subscribe.id, {
                "sites": sites
            })

    @staticmethod
    def __get_default_subscribe_config(mtype: MediaType, default_config_key: str) -> Optional[str]:
        """
        获取默认订阅配置
        """
        default_subscribe_key = None
        if mtype == MediaType.TV:
            default_subscribe_key = SystemConfigKey.DefaultTvSubscribeConfig.value
        if mtype == MediaType.MOVIE:
            default_subscribe_key = SystemConfigKey.DefaultMovieSubscribeConfig.value
        if mtype == MediaType.MUSIC:
            default_subscribe_key = SystemConfigKey.DefaultMusicSubscribeConfig.value

        if not default_subscribe_key:
            return None

        # 默认订阅规则属于持久化用户配置，不再从部署 Settings 猜测同名属性。
        value = _system_config().get(default_subscribe_key)

        if not value:
            return None
        return value.get(default_config_key) or None

    @staticmethod
    def get_params(subscribe: Subscribe):
        """
        获取订阅默认参数
        """
        # 默认过滤规则
        default_rule = _system_config().get(SystemConfigKey.SubscribeDefaultParams) or {}
        return {
            key: value for key, value in {
                "include": subscribe.include or default_rule.get("include"),
                "exclude": subscribe.exclude or default_rule.get("exclude"),
                "quality": subscribe.quality or default_rule.get("quality"),
                "resolution": subscribe.resolution or default_rule.get("resolution"),
                "effect": subscribe.effect or default_rule.get("effect"),
                "audio_quality": getattr(subscribe, "audio_quality", None),
                "audio_format": getattr(subscribe, "audio_format", None),
                "min_bitrate": getattr(subscribe, "min_bitrate", None),
                "min_bit_depth": getattr(subscribe, "min_bit_depth", None),
                "min_sample_rate": getattr(subscribe, "min_sample_rate", None),
                "tv_size": default_rule.get("tv_size"),
                "movie_size": default_rule.get("movie_size"),
                "min_seeders": default_rule.get("min_seeders"),
                "min_seeders_time": default_rule.get("min_seeders_time"),
            }.items() if value is not None}

    def subscribe_files_info(self, subscribe: Subscribe) -> Optional[_SchemaSubscrbieInfo]:
        """
        订阅相关的下载和文件信息
        """
        if not subscribe:
            return None

        # 返回订阅数据
        subscribe_info = _SchemaSubscrbieInfo()

        # 所有集的数据
        episodes: Dict[int, _SchemaSubscribeEpisodeInfo] = {}
        if (
            subscribe.media_source == MediaSource.TMDB.value
            and subscribe.media_id
            and str(subscribe.media_id).isdigit()
            and subscribe.type == MediaType.TV.value
        ):
            # 查询TMDB中的集信息
            tmdb_episodes = TmdbChain().tmdb_episodes(
                tmdbid=int(subscribe.media_id),
                season=subscribe.season,
                episode_group=subscribe.episode_group
            )
            if tmdb_episodes:
                for episode in tmdb_episodes:
                    info = _SchemaSubscribeEpisodeInfo()
                    info.title = episode.name
                    info.description = episode.overview
                    info.backdrop = self.runtime_config.tmdb_image_url(episode.still_path, "w500")
                    episodes[episode.episode_number] = info
        elif subscribe.type == MediaType.TV.value:
            # 根据开始结束集计算集信息
            for i in range(subscribe.start_episode or 1, subscribe.total_episode + 1):
                info = _SchemaSubscribeEpisodeInfo()
                info.title = f'第 {i} 集'
                episodes[i] = info
        else:
            # 电影
            info = _SchemaSubscribeEpisodeInfo()
            info.title = subscribe.name
            episodes[0] = info

        # 所有下载记录
        downloadhis = DownloadHistoryOper()
        download_his = downloadhis.get_by_media_identity(
            media_source=subscribe.media_source,
            media_id=subscribe.media_id,
            music_type=getattr(subscribe, "music_type", None),
        )
        if download_his:
            for his in download_his:
                # 查询下载文件
                files = downloadhis.get_files_by_hash(his.download_hash, state=1)
                if files:
                    for file in files:
                        # 识别文件名
                        file_meta = MetaInfo(file.filepath)
                        # 下载文件信息
                        file_info = _SchemaSubscribeDownloadFileInfo(
                            torrent_title=his.torrent_name,
                            site_name=his.torrent_site,
                            downloader=file.downloader,
                            hash=his.download_hash,
                            file_path=file.fullpath,
                        )
                        if subscribe.type == MediaType.TV.value:
                            season_number = file_meta.begin_season
                            if season_number is not None and season_number != subscribe.season:
                                continue
                            episode_number = file_meta.begin_episode
                            if episode_number and episodes.get(episode_number):
                                episodes[episode_number].download.append(file_info)
                        else:
                            episodes[0].download.append(file_info)

        try:
            meta = build_subscribe_meta(subscribe)
        except ValueError:
            logger.error(f'订阅 {subscribe.name} 类型错误：{subscribe.type}')
            return subscribe_info
        # 识别媒体信息
        mediainfo: MediaInfo = MediaChain().recognize_media(
            meta=meta,
            mtype=meta.type,
            **_subscribe_recognize_kwargs(subscribe),
            episode_group=subscribe.episode_group,
            cache=False,
        )
        if not mediainfo:
            logger.warn(
                f"未识别到媒体信息，标题：{subscribe.name}，"
                f"媒体来源：{subscribe.media_source}，媒体 ID：{subscribe.media_id}"
            )
            return subscribe_info

        # 所有媒体库文件记录
        library_fileitems = self.media_files(mediainfo)
        if library_fileitems:
            for fileitem in library_fileitems:
                # 识别文件名
                file_meta = MetaInfo(fileitem.path)
                # 媒体库文件信息
                file_info = _SchemaSubscribeLibraryFileInfo(
                    storage=fileitem.storage,
                    file_path=fileitem.path,
                )
                if subscribe.type == MediaType.TV.value:
                    season_number = file_meta.begin_season
                    if season_number is not None and season_number != subscribe.season:
                        continue
                    episode_number = file_meta.begin_episode
                    if episode_number and episodes.get(episode_number):
                        episodes[episode_number].library.append(file_info)
                else:
                    episodes[0].library.append(file_info)

        self._append_subscribe_media_servers(subscribe, mediainfo, episodes)

        # 更新订阅信息
        subscribe_info.subscribe = Subscribe(**subscribe.to_dict())
        subscribe_info.episodes = episodes
        return subscribe_info

    def _append_subscribe_media_servers(
        self,
        subscribe: Subscribe,
        mediainfo: MediaInfo,
        episodes: Dict[int, _SchemaSubscribeEpisodeInfo],
    ) -> None:
        """合并媒体服务器条目，跳过已经由本地媒体库记录覆盖的服务。"""
        mediaserver_chain = MediaServerChain()
        server_names = list(MediaServerHelper().get_services().keys())

        def has_server_entry(
            library_list: List[_SchemaSubscribeLibraryFileInfo],
            server_name: Optional[str],
            server_type: Optional[str],
        ) -> bool:
            """判断媒体库列表是否已经包含目标服务条目。"""
            for info in library_list or []:
                if info.server and server_name and info.server == server_name:
                    return True
                if (
                    info.server_type
                    and server_type
                    and info.server_type == server_type
                    and info.server == server_name
                    and (
                        not info.file_path
                        or str(info.file_path).startswith(("http://", "https://"))
                    )
                ):
                    return True
            return False

        for server_name in server_names:
            exists_media = self.media_exists(mediainfo=mediainfo, server=server_name)
            if not exists_media or not (exists_media.server or exists_media.server_type):
                continue
            resolved_server = exists_media.server or server_name
            server_storage = exists_media.server_type or resolved_server
            server_itemid = (
                str(exists_media.itemid) if exists_media.itemid is not None else None
            )
            series_detail_url = None
            if resolved_server and exists_media.itemid is not None:
                series_detail_url = mediaserver_chain.get_play_url(
                    server=resolved_server,
                    item_id=exists_media.itemid,
                )

            if subscribe.type != MediaType.TV.value:
                episode_info = episodes.get(0)
                if episode_info and not has_server_entry(
                    episode_info.library, resolved_server, exists_media.server_type
                ):
                    episode_info.library.append(
                        _SchemaSubscribeLibraryFileInfo(
                            storage=server_storage,
                            file_path=series_detail_url,
                            server=resolved_server,
                            server_type=exists_media.server_type,
                            itemid=server_itemid,
                        )
                    )
                continue

            season_number = subscribe.season if subscribe.season is not None else 1
            exist_episodes = (exists_media.seasons or {}).get(season_number) or []
            episode_item_ids: Dict[int, str] = {}
            if resolved_server and exists_media.itemid is not None:
                episode_item_ids = mediaserver_chain.get_season_episode_ids(
                    server=resolved_server,
                    item_id=exists_media.itemid,
                    season=season_number,
                )
            for episode_number in exist_episodes:
                episode_info = episodes.get(episode_number)
                if not episode_info or has_server_entry(
                    episode_info.library, resolved_server, exists_media.server_type
                ):
                    continue
                episode_itemid = episode_item_ids.get(episode_number) or server_itemid
                detail_url = series_detail_url
                if resolved_server and episode_item_ids.get(episode_number):
                    detail_url = mediaserver_chain.get_play_url(
                        server=resolved_server,
                        item_id=episode_itemid,
                    ) or series_detail_url
                episode_info.library.append(
                    _SchemaSubscribeLibraryFileInfo(
                        storage=server_storage,
                        file_path=detail_url,
                        server=resolved_server,
                        server_type=exists_media.server_type,
                        itemid=(
                            str(episode_itemid) if episode_itemid is not None else None
                        ),
                    )
                )

    def check_and_handle_existing_media(self, subscribe: Subscribe, meta: MetaBase,
                                        mediainfo: MediaInfo, mediakey: Union[str, int]):
        """
        检查媒体是否已经存在，并根据情况执行相应的操作
        1. 查询缺失的媒体信息
        2. 判断是否已经下载完毕
        3. 根据媒体类型（电视剧或电影）执行不同的处理

        :param subscribe: 订阅信息对象
        :param meta: 媒体元数据
        :param mediainfo: 媒体信息
        :param mediakey: 媒体标识符
        :return:
            - exist_flag (bool): 布尔值，表示媒体是否已经完全下载或已存在
            - no_exists (dict): 缺失的媒体信息，包含缺失的集数或其他相关信息
        """
        self.__refresh_total_episode_before_completion(
            subscribe=subscribe,
            mediainfo=mediainfo,
            meta=meta,
            mediakey=mediakey,
        )

        exist_flag, no_exists = self.resolve_subscribe_missing(
            subscribe=subscribe,
            meta=meta,
            mediainfo=mediainfo,
            mediakey=mediakey,
        )

        # 如果已下载完毕，执行订阅完成操作
        if exist_flag:
            logger.info(f'{mediainfo.title_year} 已全部下载')
            self.finish_subscribe_or_not(subscribe=subscribe, meta=meta, mediainfo=mediainfo, force=True)
            return True, no_exists

        # 返回结果，表示媒体未完全下载或存在
        return False, no_exists

    def resolve_subscribe_missing(self, subscribe: Subscribe, meta: MetaBase,
                                  mediainfo: MediaInfo,
                                  mediakey: Optional[Union[str, int]] = None,
                                  best_version_accept_downloaded: bool = False):
        """
        按主程序订阅口径查询当前目标是否仍有缺失，不推进订阅状态。

        该方法只组合媒体库缺集、订阅范围、下载历史和洗版优先级，用于外部策略在
        完成前复用主程序"还要不要搜索/下载"的判断口径。它不得完成订阅、写入
        lack_episode、发送事件或修改数据库。

        best_version_accept_downloaded 仅用于分集洗版的外部完成守卫：为 True 时，
        priority>0 的目标集视为已满足；默认 False 保持主程序洗版完成需 priority==100
        的搜索/完成口径。
        """
        mediakey = mediakey or _subscribe_media_key(subscribe)
        effective_total_episode = self.__resolve_effective_total_episode(subscribe, mediainfo)

        if not subscribe.best_version:
            totals = {}
            if subscribe.season is not None and effective_total_episode:
                totals = {
                    subscribe.season: effective_total_episode
                }
            exist_flag, no_exists = DownloadChain().get_no_exists_info(
                meta=meta,
                mediainfo=mediainfo,
                totals=totals
            )
        elif meta.type != MediaType.TV and self.__is_best_version_complete(subscribe):
            return True, {}
        else:
            exist_flag = False
            if meta.type == MediaType.TV:
                if self.__is_full_best_version_enabled(subscribe):
                    pending_episodes = []
                elif best_version_accept_downloaded:
                    downloaded = set(self.__get_downloaded_best_version_episodes(
                        subscribe, total_episode=effective_total_episode
                    ))
                    start_episode = subscribe.start_episode or 1
                    pending_episodes = [
                        episode for episode in range(start_episode, effective_total_episode + 1)
                        if episode not in downloaded
                    ]
                    if not pending_episodes:
                        return True, {}
                else:
                    pending_episodes = self._get_pending_best_version_episodes(
                        subscribe, total_episode=effective_total_episode
                    )
                    if not pending_episodes:
                        return True, {}
                no_exists = {
                    mediakey: {
                        subscribe.season: _SchemaNotExistMediaInfo(
                            season=subscribe.season,
                            episodes=pending_episodes,
                            total_episode=effective_total_episode,
                            start_episode=subscribe.start_episode or 1,
                            require_complete_coverage=self.__is_full_best_version_enabled(subscribe))
                    }
                }
            else:
                no_exists = {}

        if exist_flag:
            return True, no_exists

        downloaded = self.__get_downloaded(subscribe)
        if self.__is_full_best_version_enabled(subscribe):
            downloaded = []
        if meta.type == MediaType.TV:
            return self.__get_subscribe_no_exits(
                subscribe_name=f'{subscribe.name} {meta.season}',
                no_exists=no_exists,
                mediakey=mediakey,
                begin_season=meta.begin_season,
                total_episode=effective_total_episode,
                start_episode=subscribe.start_episode,
                downloaded_episodes=downloaded
            )
        if meta.type in (MediaType.MOVIE, MediaType.MUSIC):
            return bool(downloaded), no_exists
        return False, no_exists

    def __resolve_total_episode_decrease(
            self,
            subscribe: Subscribe,
            candidate_total: int,
            meta: MetaBase,
            mediainfo: MediaInfo,
            mediakey: Optional[Union[str, int]] = None,
    ) -> int:
        """以旧目标范围内已确认存在的最高集号限制总集数回落。"""
        old_total = subscribe.total_episode or 0
        if candidate_total >= old_total or not old_total:
            return candidate_total
        if subscribe.type != MediaType.TV.value or self.__is_full_best_version_enabled(subscribe):
            return candidate_total

        target_key = mediakey or _subscribe_media_key(subscribe)
        target_season = subscribe.season
        target_start = subscribe.start_episode or 1
        snapshot = copy.copy(subscribe)
        snapshot.total_episode = old_total
        try:
            satisfied, no_exists = self.resolve_subscribe_missing(
                subscribe=snapshot,
                meta=meta,
                mediainfo=mediainfo,
                mediakey=target_key,
                best_version_accept_downloaded=bool(subscribe.best_version),
            )
        except Exception as err:
            logger.warning(f"订阅 {subscribe.name} 已存在分集事实查询失败，按元数据总集数继续：{err}")
            return candidate_total

        if satisfied:
            return old_total
        if not isinstance(no_exists, dict):
            return candidate_total
        seasons = next(
            (
                no_exists.get(media_key)
                for media_key in [target_key, *_subscribe_media_keys(subscribe)]
                if no_exists.get(media_key) is not None
            ),
            None,
        )
        if not isinstance(seasons, dict):
            return candidate_total
        missing_info = seasons.get(target_season)
        if not missing_info:
            return candidate_total
        try:
            scope_matches = missing_info.season == target_season \
                and missing_info.start_episode == target_start \
                and missing_info.total_episode == old_total
            episodes = missing_info.episodes
        except AttributeError:
            return candidate_total
        if not scope_matches:
            return candidate_total
        if not isinstance(episodes, list) or not episodes:
            return candidate_total
        if any(isinstance(episode, bool) or not isinstance(episode, int)
               or episode < target_start or episode > old_total for episode in episodes):
            return candidate_total

        confirmed = set(range(target_start, old_total + 1)).difference(episodes)
        return max(candidate_total, max(confirmed) if confirmed else 0)

    @staticmethod
    def __resolve_effective_total_episode(subscribe: Subscribe, mediainfo: MediaInfo) -> int:
        """
        只读计算完成前有效总集数，不触发事件、不写回订阅。

        主流程会通过 ``__refresh_total_episode_before_completion`` 持久化增长后的总集数；
        该查询接口只需要同样避免旧 total 造成误判，因此仅使用当前 mediainfo 中更大的
        季集数作为临时目标范围。
        """
        current_total = subscribe.total_episode or 0
        if subscribe.type != MediaType.TV.value:
            return current_total
        if subscribe.manual_total_episode:
            return current_total
        if subscribe.season is None:
            return current_total
        media_total = len((mediainfo.seasons or {}).get(subscribe.season) or [])
        if media_total > current_total:
            return media_total
        return current_total

    @staticmethod
    def __apply_episodes_refresh(current_total: int, season: Optional[int], *,
                                 mediainfo: Optional[MediaInfo] = None,
                                 media_source: Optional[MediaSource] = None,
                                 media_id: Optional[str] = None,
                                 subscribe_id: Optional[int] = None,
                                 scene: Optional[str] = None) -> int:
        """
        发送订阅总集数推算事件，允许外部把当前数据源识别到的季总集数向上覆盖。

        用途：插件在"待定集数"等场景经事件注入 total_episode
        无监听者或外部未覆盖时返回入参原值，保证零行为变更。
        :param current_total: 主程序本次识别到的当前季总集数
        :param season: 季号
        :return: 最终采用的总集数
        """
        event_data = SubscribeEpisodesRefreshEventData(
            media_source=media_source, media_id=media_id,
            season=season, mediainfo=mediainfo,
            current_total_episode=current_total, subscribe_id=subscribe_id, scene=scene)
        event = eventmanager.send_event(ChainEventType.SubscribeEpisodesRefresh, event_data)
        if event and event.event_data:
            result: SubscribeEpisodesRefreshEventData = event.event_data
            if result.updated and result.total_episode:
                result.total_episode = max(current_total or 0, result.total_episode)
                return result.total_episode
        return current_total

    @staticmethod
    async def __async_apply_episodes_refresh(current_total: int, season: Optional[int], *,
                                             mediainfo: Optional[MediaInfo] = None,
                                             media_source: Optional[MediaSource] = None,
                                             media_id: Optional[str] = None,
                                             subscribe_id: Optional[int] = None,
                                             scene: Optional[str] = None) -> int:
        """
        __apply_episodes_refresh 的异步版本
        """
        event_data = SubscribeEpisodesRefreshEventData(
            media_source=media_source, media_id=media_id,
            season=season, mediainfo=mediainfo,
            current_total_episode=current_total, subscribe_id=subscribe_id, scene=scene)
        event = await eventmanager.async_send_event(ChainEventType.SubscribeEpisodesRefresh, event_data)
        if event and event.event_data:
            result: SubscribeEpisodesRefreshEventData = event.event_data
            if result.updated and result.total_episode:
                result.total_episode = max(current_total or 0, result.total_episode)
                return result.total_episode
        return current_total

    def __refresh_total_episode_before_completion(
            self,
            subscribe: Subscribe,
            mediainfo: MediaInfo,
            meta: Optional[MetaBase] = None,
            mediakey: Optional[Union[str, int]] = None,
    ) -> None:
        """
        在完成判断前，按最新识别结果兜底修正订阅总集数，防止旧总集数导致误完成。
        """
        if subscribe.type != MediaType.TV.value:
            return
        if subscribe.manual_total_episode:
            return
        if subscribe.season is None:
            return

        current_total_episode = len((mediainfo.seasons or {}).get(subscribe.season) or [])
        # 外部事件只能向上覆盖主程序本次识别到的 TMDB 当前季总集数，已有订阅回落由主程序跟随本次识别结果持久化。
        new_total_episode = self.__apply_episodes_refresh(
            current_total_episode, season=subscribe.season, mediainfo=mediainfo,
            media_source=subscribe.media_source,
            media_id=subscribe.media_id,
            subscribe_id=subscribe.id, scene="precheck")
        old_total_episode = subscribe.total_episode or 0
        if meta is not None and new_total_episode and new_total_episode < old_total_episode:
            new_total_episode = self.__resolve_total_episode_decrease(
                subscribe=subscribe,
                candidate_total=new_total_episode,
                meta=meta,
                mediainfo=mediainfo,
                mediakey=mediakey,
            )
        if not new_total_episode or new_total_episode == old_total_episode:
            return

        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        update_data = self.__prepare_total_episode_change_fields(
            subscribe=subscribe,
            total_episode=new_total_episode,
            old_total_episode=old_total_episode,
        )
        update_data["last_update"] = now
        SubscribeOper().update(subscribe.id, update_data)
        for key, value in update_data.items():
            setattr(subscribe, key, value)
        subscribe.last_update = now
        logger.info(
            f"订阅 {subscribe.name} 第{subscribe.season}季 总集数更新为 {new_total_episode}，"
            f"缺失集数更新为 {subscribe.lack_episode}"
        )

    @classmethod
    def _is_episode_range_covered(cls, meta: MetaBase, subscribe: Subscribe) -> bool:
        """
        判断种子是否覆盖当前仍需洗版的剧集范围。
        """
        episodes = meta.episode_list
        if not episodes:
            # 没有剧集信息，表示该种子为合集
            return True

        pending_episodes = cls._get_pending_best_version_episodes(subscribe)
        if not pending_episodes:
            return True

        return bool(set(episodes).intersection(set(pending_episodes)))

    @staticmethod
    def __get_media_id_match_source(mediainfo: Optional[MediaInfo]) -> str:
        """
        返回候选自身识别命中的明确媒体 ID 类型。
        """
        media_source, media_id = resolve_media_identity(media=mediainfo)
        if media_source and media_id:
            return str(media_source)
        return "unknown"

    @staticmethod
    def __is_title_match_allowed(
            mediainfo: MediaInfo,
            torrent_meta: MetaBase,
            torrent_info: TorrentInfo,
    ) -> bool:
        """判断资源是否可以通过标题兜底匹配订阅目标。"""
        if not TorrentHelper.match_torrent(
                mediainfo=mediainfo,
                torrent_meta=torrent_meta,
                torrent=torrent_info,
        ):
            return False
        if not TorrentHelper.requires_identity_disambiguation(
                mediainfo=mediainfo,
                torrent_meta=torrent_meta,
        ):
            return True
        logger.info(
            f'{torrent_info.site_name} - {torrent_info.title} '
            f'仅通过无年份别名命中且候选媒体身份无法确认，已跳过'
        )
        return False

    @staticmethod
    def __reconcile_candidate_media(
            target_mediainfo: MediaInfo,
            candidate_mediainfo: MediaInfo,
            torrent_meta: MetaBase,
            torrent_info: TorrentInfo,
            context: Context,
    ) -> Optional[MediaInfo]:
        """
        在推断媒体 ID 冲突时，以订阅目标对候选标题做严格复核。

        标题携带明确媒体身份或目标缺少同来源 ID 时保持严格拒绝；只有普通
        标题推断产生冲突且标题、别名、类型和年份复核通过时才回填订阅目标。

        :param target_mediainfo: 订阅目标媒体信息
        :param candidate_mediainfo: RSS 或首页候选的识别结果
        :param torrent_meta: 候选标题解析信息
        :param torrent_info: 候选种子信息
        :param context: 当前订阅使用的候选上下文副本
        :return: 可继续匹配的媒体信息，拒绝时返回 None
        """
        candidate_source, candidate_id = resolve_media_identity(
            media=candidate_mediainfo
        )
        target_source, target_id = resolve_media_identity(media=target_mediainfo)
        conflicts = []
        if candidate_source and candidate_id and (
            candidate_source != target_source or candidate_id != target_id
        ):
            conflicts.append(
                (
                    str(candidate_source),
                    candidate_id,
                    f"{target_source}:{target_id}" if target_source and target_id else None,
                )
            )
        if not conflicts:
            return candidate_mediainfo

        conflict_text = "、".join(
            f"{source} {candidate_id} != {target_id if target_id is not None else '缺失'}"
            for source, candidate_id, target_id in conflicts
        )
        if any(target_id is None for _, _, target_id in conflicts):
            logger.debug(
                f'{torrent_info.site_name} - {torrent_info.title} 候选媒体ID与订阅目标不可比较：'
                f'{conflict_text}'
            )
            return None

        explicit_source, explicit_id = resolve_media_identity(media=torrent_meta)
        if explicit_source and explicit_id:
            logger.debug(
                f'{torrent_info.site_name} - {torrent_info.title} 标题含明确媒体ID '
                f'{explicit_source}:{explicit_id}，保持严格校验：{conflict_text}'
            )
            return None

        if not TorrentHelper.match_torrent(
                mediainfo=target_mediainfo,
                torrent_meta=torrent_meta,
                torrent=torrent_info,
        ):
            logger.debug(
                f'{torrent_info.site_name} - {torrent_info.title} 候选媒体ID冲突且标题复核失败：'
                f'{conflict_text}'
            )
            return None

        evidence_matched, evidence = TorrentHelper.match_same_work_evidence(
            target_mediainfo=target_mediainfo,
            candidate_mediainfo=candidate_mediainfo,
            torrent_meta=torrent_meta,
        )
        if not evidence_matched:
            logger.debug(
                f'{torrent_info.site_name} - {torrent_info.title} 候选媒体ID冲突且缺少同作品证据：'
                f'{evidence}；{conflict_text}'
            )
            return None

        context.media_info = target_mediainfo
        context.match_source = "title"
        context.candidate_recognized = False
        context.media_info_is_target = True
        logger.debug(
            f'{target_mediainfo.title_year} 候选媒体ID冲突（{conflict_text}），'
            f'经标题或别名及{evidence}复核匹配到订阅目标：'
            f'{torrent_info.site_name} - {torrent_info.title}'
        )
        return target_mediainfo

    @staticmethod
    def get_states_for_search(state: str) -> str:
        """
        根据给定的状态返回实际需要搜索的状态列表，支持多个状态用逗号分隔
        :param state: 订阅状态
            N: New（新建，未处理）
            R: Resolved（订阅中）
            P: Pending（待定，信息待进一步更新，允许搜索，不允许完成）
            S: Suspended（暂停，订阅不参与任何动作，暂时停止处理）
        :return: 需要查询的状态列表（多个状态用逗号分隔）
        """
        # 如果状态是 R 或 P，则视为一起搜索，返回 R,P 作为查询条件
        if state in ["R", "P"]:
            return "R,P"
        return state

    @staticmethod
    def get_subscribe_source_keyword(subscribe: Subscribe) -> str:
        """
        构造用于订阅来源的关键字字符串

        :param subscribe: Subscribe 对象
        :return str: 格式化的订阅来源关键字字符串，格式为 "Subscribe|{...}"
        """
        source_keyword = {
            'id': subscribe.id,
            'name': subscribe.name,
            'year': subscribe.year,
            'type': subscribe.type,
            'season': subscribe.season,
            'episode_group': subscribe.episode_group,
            'media_source': subscribe.media_source,
            'media_id': subscribe.media_id,
            'music_type': getattr(subscribe, 'music_type', None),
        }
        return f"Subscribe|{json.dumps(source_keyword, ensure_ascii=False)}"

    @staticmethod
    def parse_subscribe_source_keyword(source_keyword_str: str) -> Optional[dict]:
        """
        解析订阅来源关键字字符串

        :param source_keyword_str: 订阅来源关键字字符串，格式为 "Subscribe|{...}"
        :return Dict: 如果解析失败则返回None
        """
        if not source_keyword_str or not source_keyword_str.startswith("Subscribe|"):
            return None

        try:
            # 分割字符串获取JSON部分
            json_part = source_keyword_str.split("|", 1)[1]
            # 解析JSON字符串
            source_keyword = json.loads(json_part)
            return source_keyword
        except (IndexError, json.JSONDecodeError, TypeError) as e:
            logger.error(f"解析订阅来源关键字失败: {e}")
            return None
