refactor(modules): unify music catalog decisions

app/modules/douban/__init__.py

@@ -1,12 +1,9 @@
 import re
-from typing import Any, List, Optional, Tuple, Union
+from dataclasses import dataclass
+from typing import Any, List, Optional, Tuple, Union, cast
 
 import cn2an
 
-from app.modules._base.media_auxiliary import MediaAuxiliaryProviderMixin
-from app.runtime.settings import get_runtime_setting
-from app.schemas.context import MediaPerson as _SchemaMediaPerson
-
 from app.adapters.network.http import RequestUtils
 from app.domain.context import (
     MediaInfo,
@@ -19,12 +16,15 @@ from app.domain.meta.metamusic import MetaMusic
 from app.domain.metainfo import MetaInfo
 from app.foundation.text import convert as zhconv_convert
 from app.modules import _ModuleBase
+from app.modules._base.media_auxiliary import MediaAuxiliaryProviderMixin
 from app.modules.douban.apiv2 import DoubanApi
 from app.modules.douban.scraper import DoubanScraper
 from app.runtime.execution import retry
 from app.runtime.log import logger
 from app.runtime.rate import rate_limit_exponential
+from app.runtime.settings import get_runtime_setting
 from app.schemas.context import MediaPerson
+from app.schemas.context import MediaPerson as _SchemaMediaPerson
 from app.schemas.exception import APIRateLimitException
 from app.schemas.types import (
     MUSIC_ENTITY_ALBUM,
@@ -37,6 +37,60 @@ from app.schemas.types import (
 )
 
 
+@dataclass(frozen=True, slots=True)
+class _DoubanMusicRecognitionPlan:
+    """描述豆瓣音乐识别的显式身份或候选搜索范围。"""
+
+    meta: Optional[MetaMusic]
+    media_id: Optional[str]
+    album_id: Optional[str]
+    track_id: Optional[str]
+    music_type: Optional[str]
+    keyword: Optional[str]
+
+    def require_meta(self) -> MetaMusic:
+        """返回候选搜索计划必有的音乐元数据。"""
+        if self.meta is None:
+            raise RuntimeError("豆瓣音乐候选识别计划缺少音乐元数据")
+        return self.meta
+
+    def require_album_id(self) -> str:
+        """返回详情识别计划必有的专辑 ID。"""
+        if self.album_id is None:
+            raise RuntimeError("豆瓣音乐详情识别计划缺少专辑 ID")
+        return self.album_id
+
+    def require_keyword(self) -> str:
+        """返回候选识别计划必有的搜索词。"""
+        if self.keyword is None:
+            raise RuntimeError("豆瓣音乐候选识别计划缺少搜索词")
+        return self.keyword
+
+
+@dataclass(frozen=True, slots=True)
+class _DoubanVideoRecognitionPlan:
+    """描述豆瓣影视识别的原生 ID 或有序标题候选。"""
+
+    meta: Optional[MetaBase]
+    media_type: Optional[MediaType]
+    douban_id: Optional[str]
+    search_names: tuple[str, ...]
+
+    def require_meta(self) -> MetaBase:
+        """返回标题识别计划必有的影视元数据。"""
+        if self.meta is None:
+            raise RuntimeError("豆瓣影视标题识别计划缺少元数据")
+        return self.meta
+
+
+@dataclass(frozen=True, slots=True)
+class _DoubanSearchPlan:
+    """保存豆瓣搜索来源准入结果和有效查询文本。"""
+
+    enabled: bool
+    query: Optional[str]
+
+
 class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
     """提供豆瓣影视与豆瓣音乐元数据识别能力。"""
 
@@ -219,40 +273,33 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
             music_type: Optional[str] = None,
     ) -> Optional[MusicInfo]:
         """执行豆瓣音乐详情识别或按专辑名称匹配。"""
-        if media_source != self._music_source:
+        plan = self._music_recognition_plan(
+            meta, media_source, media_id, music_type
+        )
+        if not plan:
             return None
-        resolved_media_id = media_id or (meta.media_id if meta else None)
-        if resolved_media_id:
-            detail_kwargs = (
-                {"music_type": music_type} if music_type is not None else {}
+        if plan.media_id:
+            info = self.doubanapi.music_detail(
+                subject_id=plan.require_album_id()
             )
-            return self.recognize_music(
-                media_source, str(resolved_media_id), **detail_kwargs
-            )
-        if not meta:
-            return None
-        candidates = self.search_music(meta=meta, limit=20, media_source=media_source) or []
-        expected_title = meta.album or meta.title
+            return self._project_music_detail(plan, info)
+        meta = plan.require_meta()
+        result = self.doubanapi.music_search(
+            keyword=plan.require_keyword(), count=20
+        )
+        candidates = self._matching_music_candidates(
+            meta, self._build_music_search_results(result)
+        )
         for candidate in candidates:
-            if not self._same_music_text(expected_title, candidate.title):
-                continue
-            if meta.artists and candidate.artists and not any(
-                self._same_music_text(expected, actual)
-                for expected in meta.artists
-                for actual in candidate.artists
-            ):
-                continue
-            if music_type == MUSIC_ENTITY_ALBUM:
-                return candidate
+            direct_result = self._direct_music_candidate(plan, candidate)
+            if direct_result is not None:
+                return direct_result
             if meta.album and meta.title:
-                album = self.music_album(media_source, candidate.media_id)
+                info = self.doubanapi.music_detail(subject_id=str(candidate.media_id))
+                album = self._douban_music_to_album(info) if info else None
                 matched_track = self._select_douban_music_track(meta, album)
                 if matched_track:
                     return matched_track
-                continue
-            if music_type == MUSIC_ENTITY_RECORDING:
-                continue
-            return candidate
         return None
 
     async def _async_recognize_music_media(
@@ -263,48 +310,27 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
             music_type: Optional[str] = None,
     ) -> Optional[MusicInfo]:
         """异步执行豆瓣音乐详情识别或按专辑名称匹配。"""
-        if media_source != self._music_source:
+        plan = self._music_recognition_plan(
+            meta, media_source, media_id, music_type
+        )
+        if not plan:
             return None
-        resolved_media_id = media_id or (meta.media_id if meta else None)
-        if resolved_media_id:
-            album_id, separator, track_id = str(resolved_media_id).partition(":")
-            if music_type == MUSIC_ENTITY_RECORDING and not separator:
-                return None
-            if music_type == MUSIC_ENTITY_ALBUM and separator:
-                return None
-            info = await self.doubanapi.async_music_detail(subject_id=album_id)
-            album = self._douban_music_to_album(info) if info else None
-            if not album:
-                return None
-            if separator and track_id:
-                return next(
-                    (
-                        track for track in album.tracks
-                        if track.media_id == resolved_media_id
-                        or str(track.track_number or "") == track_id
-                    ),
-                    None,
-                )
-            return album.to_music_info()
-        if not meta:
-            return None
-        keyword = meta.album or meta.title
-        if not keyword:
-            return None
-        result = await self.doubanapi.async_music_search(keyword=keyword, count=20)
-        candidates = self._build_music_search_results(result)
-        expected_title = meta.album or meta.title
+        if plan.media_id:
+            info = await self.doubanapi.async_music_detail(
+                subject_id=plan.require_album_id()
+            )
+            return self._project_music_detail(plan, info)
+        meta = plan.require_meta()
+        result = await self.doubanapi.async_music_search(
+            keyword=plan.require_keyword(), count=20
+        )
+        candidates = self._matching_music_candidates(
+            meta, self._build_music_search_results(result)
+        )
         for candidate in candidates:
-            if not self._same_music_text(expected_title, candidate.title):
-                continue
-            if meta.artists and candidate.artists and not any(
-                self._same_music_text(expected, actual)
-                for expected in meta.artists
-                for actual in candidate.artists
-            ):
-                continue
-            if music_type == MUSIC_ENTITY_ALBUM:
-                return candidate
+            direct_result = self._direct_music_candidate(plan, candidate)
+            if direct_result is not None:
+                return direct_result
             if meta.album and meta.title:
                 info = await self.doubanapi.async_music_detail(
                     subject_id=str(candidate.media_id)
@@ -313,12 +339,107 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
                 matched_track = self._select_douban_music_track(meta, album)
                 if matched_track:
                     return matched_track
-                continue
-            if music_type == MUSIC_ENTITY_RECORDING:
-                continue
-            return candidate
         return None
 
+    @classmethod
+    def _music_recognition_plan(
+            cls,
+            meta: Optional[MetaMusic],
+            media_source: Optional[MediaSource],
+            media_id: Optional[str],
+            music_type: Optional[str],
+    ) -> Optional[_DoubanMusicRecognitionPlan]:
+        """统一完成豆瓣音乐来源、实体类型、ID 和搜索词准入。"""
+        if media_source != cls._music_source:
+            return None
+        resolved_media_id = media_id or (meta.media_id if meta else None)
+        if resolved_media_id:
+            normalized_id = str(resolved_media_id)
+            album_id, separator, track_id = normalized_id.partition(":")
+            if music_type == MUSIC_ENTITY_RECORDING and not separator:
+                return None
+            if music_type == MUSIC_ENTITY_ALBUM and separator:
+                return None
+            return _DoubanMusicRecognitionPlan(
+                meta=meta,
+                media_id=normalized_id,
+                album_id=album_id,
+                track_id=track_id if separator else None,
+                music_type=music_type,
+                keyword=None,
+            )
+        keyword = (meta.album or meta.title) if meta else None
+        if not meta or not keyword:
+            return None
+        return _DoubanMusicRecognitionPlan(
+            meta=meta,
+            media_id=None,
+            album_id=None,
+            track_id=None,
+            music_type=music_type,
+            keyword=keyword,
+        )
+
+    @classmethod
+    def _project_music_detail(
+            cls,
+            plan: _DoubanMusicRecognitionPlan,
+            info: Optional[dict[str, Any]],
+    ) -> Optional[MusicInfo]:
+        """把豆瓣音乐详情按计划投影为专辑或指定曲目。"""
+        album = cls._douban_music_to_album(info) if info else None
+        if not album:
+            return None
+        if plan.track_id:
+            return next(
+                (
+                    track
+                    for track in album.tracks
+                    if track.media_id == plan.media_id
+                    or str(track.track_number or "") == plan.track_id
+                ),
+                None,
+            )
+        return album.to_music_info()
+
+    @classmethod
+    def _matching_music_candidates(
+            cls,
+            meta: MetaMusic,
+            candidates: List[MusicInfo],
+    ) -> List[MusicInfo]:
+        """按专辑标题和可用艺术家线索过滤豆瓣音乐候选。"""
+        expected_title = meta.album or meta.title
+        return [
+            candidate
+            for candidate in candidates
+            if cls._same_music_text(expected_title, candidate.title)
+            and (
+                not meta.artists
+                or not candidate.artists
+                or any(
+                    cls._same_music_text(expected, actual)
+                    for expected in meta.artists
+                    for actual in candidate.artists
+                )
+            )
+        ]
+
+    @staticmethod
+    def _direct_music_candidate(
+            plan: _DoubanMusicRecognitionPlan,
+            candidate: MusicInfo,
+    ) -> Optional[MusicInfo]:
+        """决定候选可直接返回，还是必须继续读取专辑曲目。"""
+        meta = plan.require_meta()
+        if plan.music_type == MUSIC_ENTITY_ALBUM:
+            return candidate
+        if meta.album and meta.title:
+            return None
+        if plan.music_type == MUSIC_ENTITY_RECORDING:
+            return None
+        return candidate
+
     @classmethod
     def _select_douban_music_track(
             cls,
@@ -427,7 +548,10 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         title = cls._douban_music_text(info.get("title") or info.get("name"))
         if not media_id or not title:
             return None
-        attrs = info.get("attrs") if isinstance(info.get("attrs"), dict) else {}
+        attrs = cast(
+            dict[str, Any],
+            info.get("attrs") if isinstance(info.get("attrs"), dict) else {},
+        )
         artists = cls._douban_music_artists(info)
         release_date = cls._douban_music_date(info)
         tags = [
@@ -435,7 +559,10 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
             for item in (info.get("tags") or [])
         ]
         genres = [str(item) for item in info.get("genres") or [] if item]
-        rating = info.get("rating") if isinstance(info.get("rating"), dict) else {}
+        rating = cast(
+            dict[str, Any],
+            info.get("rating") if isinstance(info.get("rating"), dict) else {},
+        )
         album = MusicAlbumInfo(
             media_source=cls._music_source,
             media_id=media_id,
@@ -470,7 +597,10 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
             album: MusicAlbumInfo,
     ) -> List[MusicInfo]:
         """从豆瓣新旧响应结构中提取专辑曲目。"""
-        attrs = info.get("attrs") if isinstance(info.get("attrs"), dict) else {}
+        attrs = cast(
+            dict[str, Any],
+            info.get("attrs") if isinstance(info.get("attrs"), dict) else {},
+        )
         # Frodo 当前音乐详情使用 songs；tracks/attrs.tracks 兼容旧接口响应。
         tracks = info.get("songs") or info.get("tracks") or attrs.get("tracks") or []
         if isinstance(tracks, str):
@@ -538,7 +668,10 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
     @classmethod
     def _douban_music_artists(cls, info: dict[str, Any]) -> List[str]:
         """从豆瓣新旧响应结构中提取艺术家名称。"""
-        attrs = info.get("attrs") if isinstance(info.get("attrs"), dict) else {}
+        attrs = cast(
+            dict[str, Any],
+            info.get("attrs") if isinstance(info.get("attrs"), dict) else {},
+        )
         values = (
             info.get("artists")
             or info.get("artist_names")
@@ -600,7 +733,10 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
     @classmethod
     def _douban_music_date(cls, info: dict[str, Any]) -> Optional[str]:
         """从豆瓣新旧响应结构中提取首个发行日期。"""
-        attrs = info.get("attrs") if isinstance(info.get("attrs"), dict) else {}
+        attrs = cast(
+            dict[str, Any],
+            info.get("attrs") if isinstance(info.get("attrs"), dict) else {},
+        )
         return cls._douban_music_first(info.get("pubdate") or attrs.get("pubdate"))
 
     @staticmethod
@@ -700,61 +836,46 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         :param match_doubaninfo_func: 匹配豆瓣信息的函数
         :return: 识别的媒体信息，包括剧集信息
         """
-        if not doubanid and not meta:
+        plan = self._video_recognition_plan(
+            meta=meta,
+            mtype=mtype,
+            douban_id=doubanid,
+            effective_source=(
+                kwargs.get("media_source")
+                or get_runtime_setting('RECOGNIZE_SOURCE')
+            ),
+        )
+        if not plan:
             return None
-
-        if (
-            meta
-            and not doubanid
-            and (kwargs.get("media_source") or get_runtime_setting('RECOGNIZE_SOURCE')) != "douban"
-        ):
+        if not self._prepare_video_plan(plan):
             return None
-
-        if doubanid:
+        if plan.douban_id:
             info = douban_info_func(
-                doubanid=doubanid,
-                mtype=mtype or (meta.type if meta else None),
+                doubanid=plan.douban_id,
+                mtype=plan.media_type,
             )
-        elif not meta.name:
-            logger.error("识别媒体信息时未提供元数据名称")
-            return None
         else:
-            if mtype:
-                meta.type = mtype
             info = {}
-            for name in self._prepare_search_names(meta):
-                if meta.begin_season is not None:
-                    logger.info(f"正在识别 {name} 第{meta.begin_season}季 ...")
+            plan_meta = plan.require_meta()
+            for name in plan.search_names:
+                if plan_meta.begin_season is not None:
+                    logger.info(f"正在识别 {name} 第{plan_meta.begin_season}季 ...")
                 else:
                     logger.info(f"正在识别 {name} ...")
                 match_info = match_doubaninfo_func(
                     name=name,
-                    mtype=mtype or meta.type,
-                    year=meta.year,
-                    season=meta.begin_season,
+                    mtype=plan.media_type,
+                    year=plan_meta.year,
+                    season=plan_meta.begin_season,
                 )
                 if match_info:
                     info = douban_info_func(
                         doubanid=match_info.get("id"),
-                        mtype=mtype or meta.type,
+                        mtype=plan.media_type,
                     )
                     if info:
                         break
-
-        if info:
-            mediainfo = MediaInfo(douban_info=info)
-            if meta:
-                logger.info(f"{meta.name} 豆瓣识别结果：{mediainfo.type.value} "
-                            f"{mediainfo.title_year} "
-                            f"{mediainfo.douban_id}")
-            else:
-                logger.info(f"{doubanid} 豆瓣识别结果：{mediainfo.type.value} "
-                            f"{mediainfo.title_year}")
-            return mediainfo
-        else:
-            logger.info(f"{meta.name if meta else doubanid} 未匹配到豆瓣媒体信息")
-
-        return None
+        return self._project_video_recognition(plan, info)
 
     async def _async_recognize_media_core(self, meta: MetaBase = None,
                                           mtype: MediaType = None,
@@ -771,61 +892,115 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         :param async_match_doubaninfo_func: 匹配豆瓣信息的异步函数
         :return: 识别的媒体信息，包括剧集信息
         """
-        if not doubanid and not meta:
+        plan = self._video_recognition_plan(
+            meta=meta,
+            mtype=mtype,
+            douban_id=doubanid,
+            effective_source=(
+                kwargs.get("media_source")
+                or get_runtime_setting('RECOGNIZE_SOURCE')
+            ),
+        )
+        if not plan:
             return None
-
-        if (
-            meta
-            and not doubanid
-            and (kwargs.get("media_source") or get_runtime_setting('RECOGNIZE_SOURCE')) != "douban"
-        ):
+        if not self._prepare_video_plan(plan):
             return None
-
-        if doubanid:
+        if plan.douban_id:
             info = await async_douban_info_func(
-                doubanid=doubanid,
-                mtype=mtype or (meta.type if meta else None),
+                doubanid=plan.douban_id,
+                mtype=plan.media_type,
             )
-        elif not meta.name:
-            logger.error("识别媒体信息时未提供元数据名称")
-            return None
         else:
-            if mtype:
-                meta.type = mtype
             info = {}
-            for name in self._prepare_search_names(meta):
-                if meta.begin_season is not None:
-                    logger.info(f"正在识别 {name} 第{meta.begin_season}季 ...")
+            plan_meta = plan.require_meta()
+            for name in plan.search_names:
+                if plan_meta.begin_season is not None:
+                    logger.info(f"正在识别 {name} 第{plan_meta.begin_season}季 ...")
                 else:
                     logger.info(f"正在识别 {name} ...")
                 match_info = await async_match_doubaninfo_func(
                     name=name,
-                    mtype=mtype or meta.type,
-                    year=meta.year,
-                    season=meta.begin_season,
+                    mtype=plan.media_type,
+                    year=plan_meta.year,
+                    season=plan_meta.begin_season,
                 )
                 if match_info:
                     info = await async_douban_info_func(
                         doubanid=match_info.get("id"),
-                        mtype=mtype or meta.type,
+                        mtype=plan.media_type,
                     )
                     if info:
                         break
+        return self._project_video_recognition(plan, info)
 
-        if info:
-            mediainfo = MediaInfo(douban_info=info)
-            if meta:
-                logger.info(f"{meta.name} 豆瓣识别结果：{mediainfo.type.value} "
-                            f"{mediainfo.title_year} "
-                            f"{mediainfo.douban_id}")
-            else:
-                logger.info(f"{doubanid} 豆瓣识别结果：{mediainfo.type.value} "
-                            f"{mediainfo.title_year}")
-            return mediainfo
+    @classmethod
+    def _video_recognition_plan(
+            cls,
+            meta: Optional[MetaBase],
+            mtype: Optional[MediaType],
+            douban_id: Optional[str],
+            effective_source: Optional[Union[MediaSource, str]],
+    ) -> Optional[_DoubanVideoRecognitionPlan]:
+        """统一完成豆瓣影视来源、类型、原生 ID 和搜索名称准入。"""
+        if not douban_id and not meta:
+            return None
+        if meta and not douban_id and effective_source != MediaSource.Douban:
+            return None
+        media_type = mtype or (meta.type if meta else None)
+        if douban_id:
+            return _DoubanVideoRecognitionPlan(
+                meta=meta,
+                media_type=media_type,
+                douban_id=str(douban_id),
+                search_names=(),
+            )
+        plan_meta = meta
+        if plan_meta is None:
+            raise RuntimeError("豆瓣影视标题识别计划缺少元数据")
+        return _DoubanVideoRecognitionPlan(
+            meta=plan_meta,
+            media_type=media_type,
+            douban_id=None,
+            search_names=(
+                tuple(cls._prepare_search_names(plan_meta)) if plan_meta.name else ()
+            ),
+        )
+
+    @staticmethod
+    def _prepare_video_plan(plan: _DoubanVideoRecognitionPlan) -> bool:
+        """应用豆瓣影视识别的旧 ABI 元数据回写并报告缺失标题。"""
+        if plan.douban_id:
+            return True
+        if not plan.search_names:
+            logger.error("识别媒体信息时未提供元数据名称")
+            return False
+        if plan.meta and plan.media_type and plan.meta.type != plan.media_type:
+            # 显式类型历史上会回写解析元数据，保留该可观察行为。
+            plan.meta.type = plan.media_type
+        return True
+
+    @staticmethod
+    def _project_video_recognition(
+            plan: _DoubanVideoRecognitionPlan,
+            info: Optional[dict[str, Any]],
+    ) -> Optional[MediaInfo]:
+        """把豆瓣影视详情统一投影为媒体信息并记录识别结论。"""
+        label = plan.meta.name if plan.meta else plan.douban_id
+        if not info:
+            logger.info(f"{label} 未匹配到豆瓣媒体信息")
+            return None
+        mediainfo = MediaInfo(douban_info=info)
+        if plan.meta:
+            logger.info(
+                f"{plan.meta.name} 豆瓣识别结果：{mediainfo.type.value} "
+                f"{mediainfo.title_year} {mediainfo.douban_id}"
+            )
         else:
-            logger.info(f"{meta.name if meta else doubanid} 未匹配到豆瓣媒体信息")
-
-        return None
+            logger.info(
+                f"{plan.douban_id} 豆瓣识别结果：{mediainfo.type.value} "
+                f"{mediainfo.title_year}"
+            )
+        return mediainfo
 
     def recognize_media(self, meta: MetaBase = None,
                         mtype: MediaType = None,
@@ -1290,15 +1465,7 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         else:
             infos = self.doubanapi.tv_recommend(start=(page - 1) * count, count=count,
                                                 sort=sort, tags=tags)
-        if infos and infos.get("items"):
-            medias = [MediaInfo(douban_info=info) for info in infos.get("items")]
-            return [media for media in medias if media.poster_path
-                    and "movie_large.jpg" not in media.poster_path
-                    and "tv_normal.png" not in media.poster_path
-                    and "movie_large.jpg" not in media.poster_path
-                    and "tv_normal.jpg" not in media.poster_path
-                    and "tv_large.jpg" not in media.poster_path]
-        return []
+        return self._project_discover(infos)
 
     async def async_douban_discover(self, mtype: MediaType, sort: str, tags: str,
                                     page: int = 1, count: int = 30) -> Optional[List[MediaInfo]]:
@@ -1318,15 +1485,27 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         else:
             infos = await self.doubanapi.async_tv_recommend(start=(page - 1) * count, count=count,
                                                             sort=sort, tags=tags)
-        if infos and infos.get("items"):
-            medias = [MediaInfo(douban_info=info) for info in infos.get("items")]
-            return [media for media in medias if media.poster_path
-                    and "movie_large.jpg" not in media.poster_path
-                    and "tv_normal.png" not in media.poster_path
-                    and "movie_large.jpg" not in media.poster_path
-                    and "tv_normal.jpg" not in media.poster_path
-                    and "tv_large.jpg" not in media.poster_path]
-        return []
+        return self._project_discover(infos)
+
+    @staticmethod
+    def _project_discover(infos: Optional[dict[str, Any]]) -> List[MediaInfo]:
+        """过滤豆瓣发现页占位海报并投影为统一媒体列表。"""
+        medias = [
+            MediaInfo(douban_info=info)
+            for info in (infos or {}).get("items") or []
+        ]
+        invalid_posters = (
+            "movie_large.jpg",
+            "tv_normal.png",
+            "tv_normal.jpg",
+            "tv_large.jpg",
+        )
+        return [
+            media
+            for media in medias
+            if media.poster_path
+            and not any(marker in media.poster_path for marker in invalid_posters)
+        ]
 
     def movie_showing(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
         """
@@ -1334,9 +1513,7 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         """
         infos = self.doubanapi.movie_showing(start=(page - 1) * count,
                                              count=count)
-        if infos and infos.get("subject_collection_items"):
-            return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
-        return []
+        return self._project_collection(infos)
 
     async def async_movie_showing(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
         """
@@ -1344,9 +1521,7 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         """
         infos = await self.doubanapi.async_movie_showing(start=(page - 1) * count,
                                                          count=count)
-        if infos and infos.get("subject_collection_items"):
-            return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
-        return []
+        return self._project_collection(infos)
 
     def tv_weekly_chinese(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
         """
@@ -1354,9 +1529,7 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         """
         infos = self.doubanapi.tv_chinese_best_weekly(start=(page - 1) * count,
                                                       count=count)
-        if infos:
-            return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
-        return []
+        return self._project_collection(infos)
 
     async def async_tv_weekly_chinese(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
         """
@@ -1364,9 +1537,7 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         """
         infos = await self.doubanapi.async_tv_chinese_best_weekly(start=(page - 1) * count,
                                                                   count=count)
-        if infos:
-            return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
-        return []
+        return self._project_collection(infos)
 
     def tv_weekly_global(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
         """
@@ -1374,9 +1545,7 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         """
         infos = self.doubanapi.tv_global_best_weekly(start=(page - 1) * count,
                                                      count=count)
-        if infos and infos.get("subject_collection_items"):
-            return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
-        return []
+        return self._project_collection(infos)
 
     async def async_tv_weekly_global(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
         """
@@ -1384,9 +1553,7 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         """
         infos = await self.doubanapi.async_tv_global_best_weekly(start=(page - 1) * count,
                                                                  count=count)
-        if infos and infos.get("subject_collection_items"):
-            return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
-        return []
+        return self._project_collection(infos)
 
     def tv_animation(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
         """
@@ -1394,9 +1561,7 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         """
         infos = self.doubanapi.tv_animation(start=(page - 1) * count,
                                             count=count)
-        if infos and infos.get("subject_collection_items"):
-            return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
-        return []
+        return self._project_collection(infos)
 
     async def async_tv_animation(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
         """
@@ -1404,9 +1569,7 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         """
         infos = await self.doubanapi.async_tv_animation(start=(page - 1) * count,
                                                         count=count)
-        if infos and infos.get("subject_collection_items"):
-            return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
-        return []
+        return self._project_collection(infos)
 
     def movie_hot(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
         """
@@ -1414,9 +1577,7 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         """
         infos = self.doubanapi.movie_hot_gaia(start=(page - 1) * count,
                                               count=count)
-        if infos and infos.get("subject_collection_items"):
-            return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
-        return []
+        return self._project_collection(infos)
 
     async def async_movie_hot(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
         """
@@ -1424,9 +1585,7 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         """
         infos = await self.doubanapi.async_movie_hot_gaia(start=(page - 1) * count,
                                                           count=count)
-        if infos and infos.get("subject_collection_items"):
-            return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
-        return []
+        return self._project_collection(infos)
 
     def tv_hot(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
         """
@@ -1434,9 +1593,7 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         """
         infos = self.doubanapi.tv_hot(start=(page - 1) * count,
                                       count=count)
-        if infos and infos.get("subject_collection_items"):
-            return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
-        return []
+        return self._project_collection(infos)
 
     async def async_tv_hot(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
         """
@@ -1444,9 +1601,17 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         """
         infos = await self.doubanapi.async_tv_hot(start=(page - 1) * count,
                                                   count=count)
-        if infos and infos.get("subject_collection_items"):
-            return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
-        return []
+        return self._project_collection(infos)
+
+    @staticmethod
+    def _project_collection(
+            infos: Optional[dict[str, Any]],
+    ) -> List[MediaInfo]:
+        """把豆瓣榜单或集合响应投影为统一媒体列表。"""
+        return [
+            MediaInfo(douban_info=info)
+            for info in (infos or {}).get("subject_collection_items") or []
+        ]
 
     def search_medias(
         self, meta: MetaBase, media_source: Optional[MediaSourceSelection] = None
@@ -1457,15 +1622,13 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         :param media_source: 请求级搜索数据源
         :return: 媒体信息
         """
-        if not is_media_source_enabled(media_source, MediaSource.Douban):
+        plan = self._search_plan(meta.name if meta else None, media_source)
+        if not plan.enabled:
             return None
-        if not meta.name:
+        if not plan.query:
             return []
-        result = self.doubanapi.search(meta.name)
-        if not result or not result.get("items"):
-            return []
-        # 返回数据
-        return self._build_search_medias_result(meta, result.get("items"))
+        result = self.doubanapi.search(plan.query)
+        return self._project_media_search(meta, result)
 
     async def async_search_medias(
         self, meta: MetaBase, media_source: Optional[MediaSourceSelection] = None
@@ -1476,15 +1639,13 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         :param media_source: 请求级搜索数据源
         :return: 媒体信息
         """
-        if not is_media_source_enabled(media_source, MediaSource.Douban):
+        plan = self._search_plan(meta.name if meta else None, media_source)
+        if not plan.enabled:
             return None
-        if not meta.name:
+        if not plan.query:
             return []
-        result = await self.doubanapi.async_search(meta.name)
-        if not result or not result.get("items"):
-            return []
-        # 返回数据
-        return self._build_search_medias_result(meta, result.get("items"))
+        result = await self.doubanapi.async_search(plan.query)
+        return self._project_media_search(meta, result)
 
     def search_persons(
         self, name: str, media_source: Optional[MediaSourceSelection] = None
@@ -1495,21 +1656,13 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         :param media_source: 请求级搜索数据源
         :return: 人物信息列表
         """
-        if not is_media_source_enabled(media_source, MediaSource.Douban):
+        plan = self._search_plan(name, media_source)
+        if not plan.enabled:
             return None
-        if not name:
+        if not plan.query:
             return []
-        result = self.doubanapi.person_search(keyword=name)
-        if result and result.get('items'):
-            return [MediaPerson(source='douban', **{
-                'id': item.get('target_id'),
-                'name': item.get('target', {}).get('title'),
-                'url': item.get('target', {}).get('url'),
-                'images': item.get('target', {}).get('cover', {}),
-                'avatar': (item.get('target', {}).get('cover_img', {}).get('url')
-                           or '').replace("/l/public/", "/s/public/"),
-            }) for item in result.get('items') if name in item.get('target', {}).get('title')]
-        return []
+        result = self.doubanapi.person_search(keyword=plan.query)
+        return self._project_person_search(plan.query, result)
 
     async def async_search_persons(
         self, name: str, media_source: Optional[MediaSourceSelection] = None
@@ -1520,21 +1673,59 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         :param media_source: 请求级搜索数据源
         :return: 人物信息列表
         """
-        if not is_media_source_enabled(media_source, MediaSource.Douban):
+        plan = self._search_plan(name, media_source)
+        if not plan.enabled:
             return None
-        if not name:
+        if not plan.query:
             return []
-        result = await self.doubanapi.async_person_search(keyword=name)
-        if result and result.get('items'):
-            return [MediaPerson(source='douban', **{
-                'id': item.get('target_id'),
-                'name': item.get('target', {}).get('title'),
-                'url': item.get('target', {}).get('url'),
-                'images': item.get('target', {}).get('cover', {}),
-                'avatar': (item.get('target', {}).get('cover_img', {}).get('url')
-                           or '').replace("/l/public/", "/s/public/"),
-            }) for item in result.get('items') if name in item.get('target', {}).get('title')]
-        return []
+        result = await self.doubanapi.async_person_search(keyword=plan.query)
+        return self._project_person_search(plan.query, result)
+
+    @staticmethod
+    def _search_plan(
+            query: Optional[str],
+            media_source: Optional[MediaSourceSelection],
+    ) -> _DoubanSearchPlan:
+        """统一决定豆瓣搜索是否响应并规范化查询文本。"""
+        enabled = is_media_source_enabled(media_source, MediaSource.Douban)
+        return _DoubanSearchPlan(
+            enabled=enabled,
+            query=str(query).strip() if enabled and query else None,
+        )
+
+    @classmethod
+    def _project_media_search(
+            cls,
+            meta: MetaBase,
+            result: Optional[dict[str, Any]],
+    ) -> List[MediaInfo]:
+        """把豆瓣媒体搜索响应交给统一过滤与季信息投影。"""
+        return cls._build_search_medias_result(
+            meta, (result or {}).get("items")
+        )
+
+    @staticmethod
+    def _project_person_search(
+            name: str,
+            result: Optional[dict[str, Any]],
+    ) -> List[MediaPerson]:
+        """把豆瓣人物搜索响应过滤并投影为统一人物列表。"""
+        persons = []
+        for item in (result or {}).get("items") or []:
+            target = item.get("target") or {}
+            title = target.get("title") or ""
+            if name not in title:
+                continue
+            persons.append(MediaPerson(source="douban", **{
+                "id": item.get("target_id"),
+                "name": title,
+                "url": target.get("url"),
+                "images": target.get("cover", {}),
+                "avatar": (target.get("cover_img", {}).get("url") or "").replace(
+                    "/l/public/", "/s/public/"
+                ),
+            }))
+        return persons
 
     @staticmethod
     def _process_imdbid_result(result: dict, imdbid: str) -> Optional[dict]:
@@ -1664,9 +1855,7 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         """
         infos = self.doubanapi.movie_top250(start=(page - 1) * count,
                                             count=count)
-        if infos and infos.get("subject_collection_items"):
-            return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
-        return []
+        return self._project_collection(infos)
 
     async def async_movie_top250(self, page: int = 1, count: int = 30) -> List[MediaInfo]:
         """
@@ -1674,9 +1863,7 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
         """
         infos = await self.doubanapi.async_movie_top250(start=(page - 1) * count,
                                                         count=count)
-        if infos and infos.get("subject_collection_items"):
-            return [MediaInfo(douban_info=info) for info in infos.get("subject_collection_items")]
-        return []
+        return self._project_collection(infos)
 
     def metadata_nfo(self, mediainfo: MediaInfo, season: int = None, **kwargs) -> Optional[str]:
         """

app/modules/musicbrainz/__init__.py

@@ -2,6 +2,7 @@ import asyncio
 import re
 import threading
 import time
+from dataclasses import dataclass
 from difflib import SequenceMatcher
 from typing import Any, Iterable, Optional, Tuple, Union
 
@@ -32,6 +33,39 @@ from app.schemas.types import (
 )
 
 
+@dataclass(frozen=True, slots=True)
+class _MusicBrainzRecognitionPlan:
+    """描述 MusicBrainz 识别的准入身份、实体范围和缓存策略。"""
+
+    meta: Optional[MetaMusic]
+    media_source: MediaSource
+    media_id: Optional[str]
+    music_type: Optional[str]
+    cache_enabled: bool = True
+
+    @property
+    def search_recording(self) -> bool:
+        """是否允许尝试单曲详情或候选。"""
+        return bool(self.music_type != MUSIC_ENTITY_ALBUM)
+
+    @property
+    def search_album(self) -> bool:
+        """是否允许在单曲未命中后继续尝试专辑。"""
+        return bool(self.music_type != MUSIC_ENTITY_RECORDING)
+
+    def require_meta(self) -> MetaMusic:
+        """返回候选识别计划必有的音乐元数据。"""
+        if self.meta is None:
+            raise RuntimeError("MusicBrainz 候选识别计划缺少音乐元数据")
+        return self.meta
+
+    def require_media_id(self) -> str:
+        """返回详情识别计划必有的原生 ID。"""
+        if self.media_id is None:
+            raise RuntimeError("MusicBrainz 详情识别计划缺少原生 ID")
+        return self.media_id
+
+
 class MusicBrainzModule(_ModuleBase):
     """通过 MusicBrainz 提供音乐元数据搜索和详情识别。"""
 
@@ -185,11 +219,7 @@ class MusicBrainzModule(_ModuleBase):
                 "/recording",
                 params={"query": query, "limit": max(1, min(limit, 100)), "fmt": "json"},
             )
-            results = [
-                info
-                for item in (payload or {}).get("recordings") or []
-                if (info := self._recording_to_info(item))
-            ]
+            results = self._project_recording_search(payload)
             if results:
                 return results
         return []
@@ -209,15 +239,22 @@ class MusicBrainzModule(_ModuleBase):
                     "fmt": "json",
                 },
             )
-            results = [
-                info
-                for item in (payload or {}).get("recordings") or []
-                if (info := self._recording_to_info(item))
-            ]
+            results = self._project_recording_search(payload)
             if results:
                 return results
         return []
 
+    @classmethod
+    def _project_recording_search(
+            cls, payload: Optional[dict[str, Any]]
+    ) -> list[MusicInfo]:
+        """把 Recording 搜索响应投影为统一音乐候选。"""
+        return [
+            info
+            for item in (payload or {}).get("recordings") or []
+            if (info := cls._recording_to_info(item))
+        ]
+
     @classmethod
     def _recording_queries(cls, meta: MetaMusic) -> list[str]:
         """构造 Recording 检索式阶梯，由严到宽逐级放宽避免零命中。
@@ -344,11 +381,7 @@ class MusicBrainzModule(_ModuleBase):
                     "fmt": "json",
                 },
             )
-            results = [
-                album.to_music_info()
-                for item in (payload or {}).get("release-groups") or []
-                if (album := self._release_group_to_album(item))
-            ]
+            results = self._project_album_search(payload)
             if results:
                 return results
         return []
@@ -368,15 +401,22 @@ class MusicBrainzModule(_ModuleBase):
                     "fmt": "json",
                 },
             )
-            results = [
-                album.to_music_info()
-                for item in (payload or {}).get("release-groups") or []
-                if (album := self._release_group_to_album(item))
-            ]
+            results = self._project_album_search(payload)
             if results:
                 return results
         return []
 
+    @classmethod
+    def _project_album_search(
+            cls, payload: Optional[dict[str, Any]]
+    ) -> list[MusicInfo]:
+        """把 Release Group 搜索响应投影为统一音乐候选。"""
+        return [
+            album.to_music_info()
+            for item in (payload or {}).get("release-groups") or []
+            if (album := cls._release_group_to_album(item))
+        ]
+
     @classmethod
     def _album_queries(cls, meta: MetaMusic) -> list[str]:
         """构造专辑检索式阶梯：专辑名+艺术家 → 仅专辑名 → 去括号/卷号变体。"""
@@ -456,8 +496,7 @@ class MusicBrainzModule(_ModuleBase):
         """
         if not tracks:
             return None
-        best_album: Optional[MusicAlbumInfo] = None
-        best_score = 0.0
+        details: list[dict[str, Any]] = []
         for release in self._search_release_candidates(meta, tracks, limit=limit):
             release_id = release.get("id")
             if not release_id:
@@ -468,15 +507,8 @@ class MusicBrainzModule(_ModuleBase):
             )
             if not detail:
                 continue
-            summary = self._release_track_summary(detail)
-            score = self._score_release(meta, tracks, detail, summary)
-            if score > best_score:
-                best_score = score
-                best_album = self._release_to_album(detail)
-        # 得分低于阈值时宁可不匹配，避免把曲目写到错误的专辑上
-        if best_score < self._album_match_threshold:
-            return None
-        return best_album
+            details.append(detail)
+        return self._select_release_match(meta, tracks, details)
 
     async def async_match_music_album(
             self,
@@ -487,8 +519,7 @@ class MusicBrainzModule(_ModuleBase):
         """异步按目录线索和曲目特征匹配 MusicBrainz 发行版本。"""
         if not tracks:
             return None
-        best_album: Optional[MusicAlbumInfo] = None
-        best_score = 0.0
+        details: list[dict[str, Any]] = []
         releases = await self._async_search_release_candidates(
             meta,
             tracks,
@@ -504,14 +535,27 @@ class MusicBrainzModule(_ModuleBase):
             )
             if not detail:
                 continue
-            summary = self._release_track_summary(detail)
-            score = self._score_release(meta, tracks, detail, summary)
+            details.append(detail)
+        return self._select_release_match(meta, tracks, details)
+
+    @classmethod
+    def _select_release_match(
+            cls,
+            meta: MetaMusic,
+            tracks: list[MetaMusic],
+            details: Iterable[dict[str, Any]],
+    ) -> Optional[MusicAlbumInfo]:
+        """对已获取的发行详情统一打分并投影最佳专辑。"""
+        best_album: Optional[MusicAlbumInfo] = None
+        best_score = 0.0
+        for detail in details:
+            summary = cls._release_track_summary(detail)
+            score = cls._score_release(meta, tracks, detail, summary)
             if score > best_score:
                 best_score = score
-                best_album = self._release_to_album(detail)
-        if best_score < self._album_match_threshold:
-            return None
-        return best_album
+                best_album = cls._release_to_album(detail)
+        # 得分低于阈值时宁可不匹配，避免把曲目写到错误的专辑上
+        return best_album if best_score >= cls._album_match_threshold else None
 
     _album_match_threshold = 60.0
 
@@ -529,11 +573,7 @@ class MusicBrainzModule(_ModuleBase):
                 "/release",
                 params={"query": query, "limit": max(1, min(limit, 25)), "fmt": "json"},
             )
-            for item in (payload or {}).get("releases") or []:
-                release_id = item.get("id")
-                if release_id and release_id not in seen:
-                    seen.add(release_id)
-                    releases.append(item)
+            self._merge_release_candidates(releases, seen, payload)
             if len(releases) >= limit:
                 break
         return releases[:limit]
@@ -556,15 +596,24 @@ class MusicBrainzModule(_ModuleBase):
                     "fmt": "json",
                 },
             )
-            for item in (payload or {}).get("releases") or []:
-                release_id = item.get("id")
-                if release_id and release_id not in seen:
-                    seen.add(release_id)
-                    releases.append(item)
+            self._merge_release_candidates(releases, seen, payload)
             if len(releases) >= limit:
                 break
         return releases[:limit]
 
+    @staticmethod
+    def _merge_release_candidates(
+            releases: list[dict[str, Any]],
+            seen: set[str],
+            payload: Optional[dict[str, Any]],
+    ) -> None:
+        """按首次命中顺序合并并去重 Release 搜索响应。"""
+        for item in (payload or {}).get("releases") or []:
+            release_id = item.get("id")
+            if release_id and release_id not in seen:
+                seen.add(release_id)
+                releases.append(item)
+
     @classmethod
     def _release_queries(cls, meta: MetaMusic, tracks: list[MetaMusic]) -> list[str]:
         """构造专辑搜索表达式：优先专辑名+歌手，无专辑线索时用曲名兜底。"""
@@ -768,67 +817,31 @@ class MusicBrainzModule(_ModuleBase):
             **kwargs,
     ) -> Optional[MusicInfo]:
         """跟随统一媒体识别分发，仅在音乐类型请求下返回 MusicBrainz 识别结果。"""
-        music_type = kwargs.get("music_type")
-        # 显式选择其它音乐源时必须让出识别管线，且不能复用 MusicBrainz 缓存。
-        if media_source and media_source != self._source:
+        plan = self._recognition_plan(
+            meta=meta,
+            mtype=mtype,
+            media_source=media_source,
+            media_id=media_id,
+            music_type=kwargs.get("music_type"),
+            cache_enabled=bool(kwargs.get("cache", True)),
+        )
+        if not plan:
             return None
-        # 非音乐请求交给影视识别模块，不占用识别管线
-        if not isinstance(meta, MetaMusic) and mtype != MediaType.MUSIC and media_source != self._source:
-            return None
-        # 无 MetaMusic 元数据时仅响应本数据源的详情识别请求
-        if not isinstance(meta, MetaMusic):
-            if media_source == self._source and media_id:
-                detail_kwargs = (
-                    {"music_type": music_type} if music_type is not None else {}
-                )
-                return self.recognize_music(
-                    media_source, str(media_id), **detail_kwargs
-                )
-            return None
-        # 显式身份只允许按该 ID 和实体类型读取，失败后不能按标题替换成其它目标。
-        resolved_source = media_source or meta.media_source
-        resolved_media_id = media_id or meta.media_id
-        if resolved_source and resolved_media_id:
+        if plan.media_id:
             detail_kwargs = (
-                {"music_type": music_type} if music_type is not None else {}
+                {"music_type": plan.music_type}
+                if plan.music_type is not None
+                else {}
             )
             info = self.recognize_music(
-                resolved_source,
-                str(resolved_media_id),
+                plan.media_source,
+                plan.require_media_id(),
                 **detail_kwargs,
             )
-            if info:
-                self._update_recognize_cache(meta, info)
+            if info and plan.meta:
+                self._update_recognize_cache(plan.meta, info)
             return info
-        # 专辑名称识别不复用 Recording 缓存，避免同名实体互相覆盖。
-        if music_type == MUSIC_ENTITY_ALBUM:
-            albums = self._search_albums(meta, limit=10)
-            return self._select_album_candidate(meta, albums)
-        # 识别缓存命中直接响应，避免重复搜索占用 MusicBrainz 限流配额
-        cache_enabled = bool(kwargs.get("cache", True))
-        if cache_enabled and self.cache:
-            cached_info = self.cache.get(meta)
-            if cached_info:
-                if cached_info.media_id:
-                    logger.info(f"{meta.title} 使用音乐识别缓存：{cached_info.title}")
-                else:
-                    logger.info(f"{meta.title} 使用音乐识别缓存：无法识别")
-                cached_info.recognize_cache_hit = True
-                return cached_info
-        # 无身份时按标题搜索并挑选可信候选，检索不到时返回元数据兑底
-        # 文件识别只能从 Recording 中挑选，专辑或艺术家同名结果不能成为音轨身份。
-        candidates = self._search_recordings(meta, limit=10)
-        matched = self._select_candidate(meta, candidates, media_source=resolved_source or self._source)
-        # 整专/单曲发行类资源在 Recording 检索无果时，回退按专辑实体识别；
-        # 专辑挑选要求标题与艺术家同时命中，无艺术家线索时回退检索必然无果，
-        # 直接跳过避免浪费限流配额（批量识别场景可减少约半数请求）
-        if not matched and meta.artists and music_type != MUSIC_ENTITY_RECORDING:
-            albums = self._search_albums(meta, limit=10)
-            matched = self._select_album_candidate(meta, albums)
-        result = matched or self._info_from_meta(meta)
-        # 无远端身份的兑底结果同样入缓存，避免批量识别时反复搜索同一文件
-        self._update_recognize_cache(meta, result)
-        return result
+        return self._recognize_from_candidates_sync(plan)
 
     def _update_recognize_cache(self, meta: MetaMusic, info: Optional[MusicInfo]) -> None:
         """识别完成后把结果写入本地识别缓存，未挂载缓存时静默跳过。"""
@@ -867,56 +880,171 @@ class MusicBrainzModule(_ModuleBase):
             **kwargs,
     ) -> Optional[MusicInfo]:
         """异步识别 MusicBrainz 音乐详情或按元数据匹配单曲。"""
-        music_type = kwargs.get("music_type")
-        if media_source and media_source != self._source:
+        plan = self._recognition_plan(
+            meta=meta,
+            mtype=mtype,
+            media_source=media_source,
+            media_id=media_id,
+            music_type=kwargs.get("music_type"),
+            cache_enabled=bool(kwargs.get("cache", True)),
+        )
+        if not plan:
             return None
-        if not isinstance(meta, MetaMusic) and mtype != MediaType.MUSIC and media_source != self._source:
+        if plan.media_id:
+            detail_kwargs = (
+                {"music_type": plan.music_type}
+                if plan.music_type is not None
+                else {}
+            )
+            info = await self.async_recognize_music(
+                plan.media_source,
+                plan.require_media_id(),
+                **detail_kwargs,
+            )
+            if info and plan.meta:
+                self._update_recognize_cache(plan.meta, info)
+            return info
+        return await self._recognize_from_candidates_async(plan)
+
+    @classmethod
+    def _recognition_plan(
+            cls,
+            meta: Optional[MetaBase],
+            mtype: Optional[MediaType],
+            media_source: Optional[MediaSource],
+            media_id: Optional[str],
+            music_type: Optional[str],
+            cache_enabled: bool,
+    ) -> Optional[_MusicBrainzRecognitionPlan]:
+        """统一完成来源、音乐类型、显式身份和缓存准入。"""
+        if media_source and media_source != cls._source:
             return None
         if not isinstance(meta, MetaMusic):
-            if media_source == self._source and media_id:
-                return await self.async_recognize_music(
-                    media_source,
-                    str(media_id),
-                    music_type=music_type,
-                )
-            return None
+            if mtype != MediaType.MUSIC and media_source != cls._source:
+                return None
+            if media_source != cls._source or not media_id:
+                return None
+            return _MusicBrainzRecognitionPlan(
+                meta=None,
+                media_source=cls._source,
+                media_id=str(media_id),
+                music_type=music_type,
+                cache_enabled=False,
+            )
         resolved_source = media_source or meta.media_source
         resolved_media_id = media_id or meta.media_id
-        if resolved_source and resolved_media_id:
-            info = await self.async_recognize_music(
-                resolved_source,
-                str(resolved_media_id),
-                music_type=music_type,
-            )
-            if info:
-                self._update_recognize_cache(meta, info)
-            return info
-        if music_type == MUSIC_ENTITY_ALBUM:
-            albums = await self._async_search_albums(meta, limit=10)
-            return self._select_album_candidate(meta, albums)
-        cache_enabled = bool(kwargs.get("cache", True))
-        if cache_enabled and self.cache:
-            cached_info = self.cache.get(meta)
-            if cached_info:
-                if cached_info.media_id:
-                    logger.info(f"{meta.title} 使用音乐识别缓存：{cached_info.title}")
-                else:
-                    logger.info(f"{meta.title} 使用音乐识别缓存：无法识别")
-                cached_info.recognize_cache_hit = True
-                return cached_info
-        candidates = await self._async_search_recordings(meta, limit=10)
-        matched = self._select_candidate(
-            meta,
-            candidates,
-            media_source=resolved_source or self._source,
+        return _MusicBrainzRecognitionPlan(
+            meta=meta,
+            media_source=resolved_source or cls._source,
+            media_id=(
+                str(resolved_media_id)
+                if resolved_source and resolved_media_id
+                else None
+            ),
+            music_type=music_type,
+            cache_enabled=cache_enabled,
         )
-        if not matched and meta.artists and music_type != MUSIC_ENTITY_RECORDING:
-            albums = await self._async_search_albums(meta, limit=10)
-            matched = self._select_album_candidate(meta, albums)
+
+    def _cached_recognition(
+            self, plan: _MusicBrainzRecognitionPlan
+    ) -> Optional[MusicInfo]:
+        """读取并标记一次候选识别缓存命中。"""
+        meta = plan.require_meta()
+        if not plan.cache_enabled or not self.cache:
+            return None
+        cached_info = self.cache.get(meta)
+        if not cached_info:
+            return None
+        if cached_info.media_id:
+            logger.info(f"{meta.title} 使用音乐识别缓存：{cached_info.title}")
+        else:
+            logger.info(f"{meta.title} 使用音乐识别缓存：无法识别")
+        cached_info.recognize_cache_hit = True
+        return cached_info
+
+    @classmethod
+    def _select_recognition_candidate(
+            cls,
+            plan: _MusicBrainzRecognitionPlan,
+            recordings: Iterable[MusicInfo],
+            albums: Iterable[MusicInfo] = (),
+    ) -> Optional[MusicInfo]:
+        """按同一来源和实体规则选择 Recording 或专辑候选。"""
+        meta = plan.require_meta()
+        if plan.music_type == MUSIC_ENTITY_ALBUM:
+            return cls._select_album_candidate(meta, albums)
+        matched = cls._select_candidate(
+            meta, recordings, media_source=plan.media_source
+        )
+        if matched or not plan.search_album or not meta.artists:
+            return matched
+        return cls._select_album_candidate(meta, albums)
+
+    def _finalize_recognition(
+            self,
+            plan: _MusicBrainzRecognitionPlan,
+            matched: Optional[MusicInfo],
+    ) -> MusicInfo:
+        """统一生成候选识别兜底并写入本地缓存。"""
+        meta = plan.require_meta()
         result = matched or self._info_from_meta(meta)
         self._update_recognize_cache(meta, result)
         return result
 
+    def _recognize_from_candidates_sync(
+            self, plan: _MusicBrainzRecognitionPlan
+    ) -> Optional[MusicInfo]:
+        """同步获取候选，所有准入与选择由共享决策方法完成。"""
+        meta = plan.require_meta()
+        if plan.music_type != MUSIC_ENTITY_ALBUM:
+            cached_info = self._cached_recognition(plan)
+            if cached_info:
+                return cached_info
+        recordings = (
+            self._search_recordings(meta, limit=10)
+            if plan.search_recording else []
+        )
+        preliminary = self._select_recognition_candidate(plan, recordings)
+        albums = (
+            self._search_albums(meta, limit=10)
+            if (
+                plan.music_type == MUSIC_ENTITY_ALBUM
+                or (not preliminary and plan.search_album and meta.artists)
+            )
+            else []
+        )
+        matched = self._select_recognition_candidate(plan, recordings, albums)
+        if plan.music_type == MUSIC_ENTITY_ALBUM:
+            return matched
+        return self._finalize_recognition(plan, matched)
+
+    async def _recognize_from_candidates_async(
+            self, plan: _MusicBrainzRecognitionPlan
+    ) -> Optional[MusicInfo]:
+        """异步获取候选，所有准入与选择由共享决策方法完成。"""
+        meta = plan.require_meta()
+        if plan.music_type != MUSIC_ENTITY_ALBUM:
+            cached_info = self._cached_recognition(plan)
+            if cached_info:
+                return cached_info
+        recordings = (
+            await self._async_search_recordings(meta, limit=10)
+            if plan.search_recording else []
+        )
+        preliminary = self._select_recognition_candidate(plan, recordings)
+        albums = (
+            await self._async_search_albums(meta, limit=10)
+            if (
+                plan.music_type == MUSIC_ENTITY_ALBUM
+                or (not preliminary and plan.search_album and meta.artists)
+            )
+            else []
+        )
+        matched = self._select_recognition_candidate(plan, recordings, albums)
+        if plan.music_type == MUSIC_ENTITY_ALBUM:
+            return matched
+        return self._finalize_recognition(plan, matched)
+
     @classmethod
     def _select_candidate(
             cls,
@@ -1170,23 +1298,25 @@ class MusicBrainzModule(_ModuleBase):
             music_type: Optional[str] = None,
     ) -> Optional[MusicInfo]:
         """按 MusicBrainz 标准 ID 和实体类型获取详情；空类型保留旧版探测顺序。"""
-        if media_source != self._source or not media_id:
+        plan = self._detail_plan(media_source, media_id, music_type)
+        if not plan:
             return None
-        if music_type != MUSIC_ENTITY_ALBUM:
+        if plan.search_recording:
             payload = self._request_json(
-                f"/recording/{media_id}",
+                f"/recording/{plan.require_media_id()}",
                 params={
                     "inc": "artists+releases+release-groups+isrcs+genres",
                     "fmt": "json",
                 },
             )
-            if payload:
-                return self._recording_to_info(payload)
-            if music_type == MUSIC_ENTITY_RECORDING:
+            result = self._project_recording_detail(payload)
+            if result:
+                return result
+            if not plan.search_album:
                 return None
         # MusicBrainz 各实体共用 UUID 形式，统一详情入口在 Recording 未命中后继续探测专辑。
-        album = self.music_album(media_source, media_id)
-        return album.to_music_info() if album else None
+        album = self.music_album(self._source, plan.require_media_id())
+        return self._project_album_result(album)
 
     async def async_recognize_music(
             self,
@@ -1195,21 +1325,57 @@ class MusicBrainzModule(_ModuleBase):
             music_type: Optional[str] = None,
     ) -> Optional[MusicInfo]:
         """异步按 MusicBrainz 标准 ID 和实体类型获取详情。"""
-        if media_source != self._source or not media_id:
+        plan = self._detail_plan(media_source, media_id, music_type)
+        if not plan:
             return None
-        if music_type != MUSIC_ENTITY_ALBUM:
+        if plan.search_recording:
             payload = await self._async_request_json(
-                f"/recording/{media_id}",
+                f"/recording/{plan.require_media_id()}",
                 params={
                     "inc": "artists+releases+release-groups+isrcs+genres",
                     "fmt": "json",
                 },
             )
-            if payload:
-                return self._recording_to_info(payload)
-            if music_type == MUSIC_ENTITY_RECORDING:
+            result = self._project_recording_detail(payload)
+            if result:
+                return result
+            if not plan.search_album:
                 return None
-        album = await self._async_music_album(media_source, media_id)
+        album = await self._async_music_album(
+            self._source, plan.require_media_id()
+        )
+        return self._project_album_result(album)
+
+    @classmethod
+    def _detail_plan(
+            cls,
+            media_source: MediaSource,
+            media_id: str,
+            music_type: Optional[str],
+    ) -> Optional[_MusicBrainzRecognitionPlan]:
+        """统一校验详情来源并冻结 Recording 到专辑的探测顺序。"""
+        if media_source != cls._source or not media_id:
+            return None
+        return _MusicBrainzRecognitionPlan(
+            meta=None,
+            media_source=cls._source,
+            media_id=str(media_id),
+            music_type=music_type,
+            cache_enabled=False,
+        )
+
+    @classmethod
+    def _project_recording_detail(
+            cls, payload: Optional[dict[str, Any]]
+    ) -> Optional[MusicInfo]:
+        """把 Recording 详情响应投影为统一音乐信息。"""
+        return cls._recording_to_info(payload) if payload else None
+
+    @staticmethod
+    def _project_album_result(
+            album: Optional[MusicAlbumInfo],
+    ) -> Optional[MusicInfo]:
+        """把专辑详情统一投影到音乐识别返回类型。"""
         return album.to_music_info() if album else None
 
     async def _async_music_album(
@@ -1218,7 +1384,7 @@ class MusicBrainzModule(_ModuleBase):
             media_id: str,
     ) -> Optional[MusicAlbumInfo]:
         """异步按 MusicBrainz Release Group ID 获取专辑详情及曲目。"""
-        if media_source != self._source or not media_id:
+        if not self._detail_plan(media_source, media_id, MUSIC_ENTITY_ALBUM):
             return None
         payload = await self._async_request_json(
             f"/release-group/{media_id}",
@@ -1227,16 +1393,13 @@ class MusicBrainzModule(_ModuleBase):
                 "fmt": "json",
             },
         )
-        if not payload:
-            return None
-        album = self._release_group_to_album(payload)
+        album = self._project_album_detail(payload)
         if not album:
             return None
-        album.releases = self._release_variants(payload.get("releases") or [])
-        album.tracks = await self._async_album_tracks(
-            album,
-            payload.get("releases") or [],
+        tracks_payload = await self._async_album_tracks_payload(
+            payload.get("releases") or []
         )
+        album.tracks = self._project_album_tracks(album, tracks_payload)
         return album
 
     def music_album(
@@ -1245,7 +1408,7 @@ class MusicBrainzModule(_ModuleBase):
             media_id: str,
     ) -> Optional[MusicAlbumInfo]:
         """按 MusicBrainz Release Group ID 获取标准化专辑详情及曲目。"""
-        if media_source != self._source or not media_id:
+        if not self._detail_plan(media_source, media_id, MUSIC_ENTITY_ALBUM):
             return None
         payload = self._request_json(
             f"/release-group/{media_id}",
@@ -1254,13 +1417,23 @@ class MusicBrainzModule(_ModuleBase):
                 "fmt": "json",
             },
         )
-        if not payload:
-            return None
-        album = self._release_group_to_album(payload)
+        album = self._project_album_detail(payload)
         if not album:
             return None
-        album.releases = self._release_variants(payload.get("releases") or [])
-        album.tracks = self._album_tracks(album, payload.get("releases") or [])
+        tracks_payload = self._album_tracks_payload(payload.get("releases") or [])
+        album.tracks = self._project_album_tracks(album, tracks_payload)
+        return album
+
+    @classmethod
+    def _project_album_detail(
+            cls, payload: Optional[dict[str, Any]]
+    ) -> Optional[MusicAlbumInfo]:
+        """把 Release Group 详情投影为尚未装载曲目的专辑。"""
+        if not payload:
+            return None
+        album = cls._release_group_to_album(payload)
+        if album:
+            album.releases = cls._release_variants(payload.get("releases") or [])
         return album
 
     def music_artist(
@@ -1519,41 +1692,40 @@ class MusicBrainzModule(_ModuleBase):
         )
 
     @classmethod
-    def _album_tracks(
-            cls,
-            album: MusicAlbumInfo,
-            releases: list[dict[str, Any]],
-    ) -> list[MusicInfo]:
-        """读取专辑代表性发行版本的曲目，作为专辑内音乐列表。"""
+    def _album_tracks_payload(
+            cls, releases: list[dict[str, Any]]
+    ) -> Optional[dict[str, Any]]:
+        """同步读取专辑代表性发行版本的原始曲目响应。"""
         release = cls._select_track_release(releases)
         if not release.get("id"):
-            return []
+            return None
         payload = cls._request_json(
             f"/release/{release['id']}",
             params={"inc": "recordings+artist-credits", "fmt": "json"},
         )
-        tracks: list[MusicInfo] = []
-        for medium in (payload or {}).get("media") or []:
-            for track in medium.get("tracks") or []:
-                info = cls._track_to_info(album, medium, track)
-                if info:
-                    tracks.append(info)
-        return tracks
+        return payload if isinstance(payload, dict) else None
 
     @classmethod
-    async def _async_album_tracks(
-            cls,
-            album: MusicAlbumInfo,
-            releases: list[dict[str, Any]],
-    ) -> list[MusicInfo]:
-        """异步读取专辑代表性发行版本的曲目。"""
+    async def _async_album_tracks_payload(
+            cls, releases: list[dict[str, Any]]
+    ) -> Optional[dict[str, Any]]:
+        """异步读取专辑代表性发行版本的原始曲目响应。"""
         release = cls._select_track_release(releases)
         if not release.get("id"):
-            return []
+            return None
         payload = await cls._async_request_json(
             f"/release/{release['id']}",
             params={"inc": "recordings+artist-credits", "fmt": "json"},
         )
+        return payload if isinstance(payload, dict) else None
+
+    @classmethod
+    def _project_album_tracks(
+            cls,
+            album: MusicAlbumInfo,
+            payload: Optional[dict[str, Any]],
+    ) -> list[MusicInfo]:
+        """把代表性发行版本响应投影为专辑曲目列表。"""
         tracks: list[MusicInfo] = []
         for medium in (payload or {}).get("media") or []:
             for track in medium.get("tracks") or []:

app/modules/theaudiodb/__init__.py

@@ -1,17 +1,19 @@
+from dataclasses import dataclass
 from typing import Any, Optional, Tuple, Union
 
-from app.runtime.cache import cached
-from app.runtime.settings import get_runtime_setting
-
+from app.adapters.network.http import AsyncRequestUtils, RequestUtils
 from app.domain.context import (
     MusicAlbumInfo,
     MusicArtistInfo,
     MusicInfo,
 )
+from app.domain.media import is_media_source_selected
 from app.domain.meta.metabase import MetaBase
 from app.domain.meta.metamusic import MetaMusic
-from app.runtime.log import logger
 from app.modules import _ModuleBase
+from app.runtime.cache import cached
+from app.runtime.log import logger
+from app.runtime.settings import get_runtime_setting
 from app.schemas.types import (
     MUSIC_ENTITY_ALBUM,
     MUSIC_ENTITY_RECORDING,
@@ -21,8 +23,37 @@ from app.schemas.types import (
     MediaType,
     ModuleType,
 )
-from app.adapters.network.http import AsyncRequestUtils, RequestUtils
-from app.domain.media import is_media_source_selected
+
+
+@dataclass(frozen=True, slots=True)
+class _TheAudioDbRecognitionPlan:
+    """描述 TheAudioDB 识别允许的详情或候选检索步骤。"""
+
+    meta: Optional[MetaMusic]
+    media_id: Optional[str]
+    music_type: Optional[str]
+
+    @property
+    def search_recording(self) -> bool:
+        """是否需要尝试单曲详情或候选。"""
+        return bool(self.music_type != MUSIC_ENTITY_ALBUM)
+
+    @property
+    def search_album(self) -> bool:
+        """是否允许在单曲未命中后继续尝试专辑。"""
+        return bool(self.music_type != MUSIC_ENTITY_RECORDING)
+
+    def require_meta(self) -> MetaMusic:
+        """返回候选识别计划必有的音乐元数据。"""
+        if self.meta is None:
+            raise RuntimeError("TheAudioDB 候选识别计划缺少音乐元数据")
+        return self.meta
+
+    def require_media_id(self) -> str:
+        """返回详情识别计划必有的原生 ID。"""
+        if self.media_id is None:
+            raise RuntimeError("TheAudioDB 详情识别计划缺少原生 ID")
+        return self.media_id
 
 
 class TheAudioDbModule(_ModuleBase):
@@ -101,34 +132,28 @@ class TheAudioDbModule(_ModuleBase):
             **kwargs,
     ) -> Optional[MusicInfo]:
         """仅响应显式 TheAudioDB 音乐请求，并返回带原生 ID 的标准音乐信息。"""
-        music_type = kwargs.get("music_type")
-        if media_source != self._source:
+        plan = self._recognition_plan(
+            meta=meta,
+            mtype=mtype,
+            media_source=media_source,
+            media_id=media_id,
+            music_type=kwargs.get("music_type"),
+        )
+        if not plan:
             return None
-        if not isinstance(meta, MetaMusic):
-            if mtype == MediaType.MUSIC and media_id:
-                detail_kwargs = (
-                    {"music_type": music_type} if music_type is not None else {}
-                )
-                return self.recognize_music(
-                    media_source, str(media_id), **detail_kwargs
-                )
-            return None
-        resolved_media_id = media_id or meta.media_id
-        if resolved_media_id:
-            detail_kwargs = (
-                {"music_type": music_type} if music_type is not None else {}
-            )
+        if plan.media_id:
             return self.recognize_music(
-                media_source, str(resolved_media_id), **detail_kwargs
+                self._source, plan.require_media_id(), music_type=plan.music_type
             )
-        if music_type != MUSIC_ENTITY_ALBUM:
-            matched = self._select_track(meta, self._search_tracks(meta))
+        plan_meta = plan.require_meta()
+        if plan.search_recording:
+            matched = self._select_track(plan_meta, self._search_tracks(plan_meta))
             if matched:
                 return matched
-            if music_type == MUSIC_ENTITY_RECORDING:
+            if not plan.search_album:
                 return None
-        album = self._select_album(meta, self._search_albums(meta))
-        return album.to_music_info() if album else None
+        album = self._select_album(plan_meta, self._search_albums(plan_meta))
+        return self._project_album_result(album)
 
     async def async_recognize_media(
             self,
@@ -139,38 +164,60 @@ class TheAudioDbModule(_ModuleBase):
             **kwargs,
     ) -> Optional[MusicInfo]:
         """异步识别 TheAudioDB 音乐详情或按元数据匹配单曲。"""
-        music_type = kwargs.get("music_type")
-        if media_source != self._source:
+        plan = self._recognition_plan(
+            meta=meta,
+            mtype=mtype,
+            media_source=media_source,
+            media_id=media_id,
+            music_type=kwargs.get("music_type"),
+        )
+        if not plan:
             return None
-        if not isinstance(meta, MetaMusic):
-            if mtype == MediaType.MUSIC and media_id:
-                return await self.async_recognize_music(
-                    media_source,
-                    str(media_id),
-                    music_type=music_type,
-                )
-            return None
-        resolved_media_id = media_id or meta.media_id
-        if resolved_media_id:
+        if plan.media_id:
             return await self.async_recognize_music(
-                media_source,
-                str(resolved_media_id),
-                music_type=music_type,
+                self._source,
+                plan.require_media_id(),
+                music_type=plan.music_type,
             )
-        if music_type != MUSIC_ENTITY_ALBUM:
+        plan_meta = plan.require_meta()
+        if plan.search_recording:
             matched = self._select_track(
-                meta,
-                await self._async_search_tracks(meta),
+                plan_meta,
+                await self._async_search_tracks(plan_meta),
             )
             if matched:
                 return matched
-            if music_type == MUSIC_ENTITY_RECORDING:
+            if not plan.search_album:
                 return None
         album = self._select_album(
-            meta,
-            await self._async_search_albums(meta),
+            plan_meta,
+            await self._async_search_albums(plan_meta),
+        )
+        return self._project_album_result(album)
+
+    @classmethod
+    def _recognition_plan(
+            cls,
+            meta: Optional[MetaBase],
+            mtype: Optional[MediaType],
+            media_source: Optional[MediaSource],
+            media_id: Optional[str],
+            music_type: Optional[str],
+    ) -> Optional[_TheAudioDbRecognitionPlan]:
+        """统一完成来源、音乐类型和原生 ID 准入。"""
+        if media_source != cls._source:
+            return None
+        if not isinstance(meta, MetaMusic):
+            if mtype != MediaType.MUSIC or not media_id:
+                return None
+            return _TheAudioDbRecognitionPlan(
+                meta=None, media_id=str(media_id), music_type=music_type
+            )
+        return _TheAudioDbRecognitionPlan(
+            meta=meta,
+            media_id=str(media_id or meta.media_id) if media_id or meta.media_id else None,
+            music_type=music_type,
         )
-        return album.to_music_info() if album else None
 
     def recognize_music(
             self,
@@ -179,17 +226,20 @@ class TheAudioDbModule(_ModuleBase):
             music_type: Optional[str] = None,
     ) -> Optional[MusicInfo]:
         """按 TheAudioDB 原生 ID 和实体类型获取详情；空类型保留旧版探测顺序。"""
-        if media_source != self._source or not media_id:
+        plan = self._detail_plan(media_source, media_id, music_type)
+        if not plan:
             return None
-        if music_type != MUSIC_ENTITY_ALBUM:
-            payload = self._request_json("track.php", {"h": media_id})
-            track = self._first_entity(payload, "track", "tracks")
-            if track:
-                return self._track_to_info(track)
-            if music_type == MUSIC_ENTITY_RECORDING:
+        if plan.search_recording:
+            payload = self._request_json(
+                "track.php", {"h": plan.require_media_id()}
+            )
+            result = self._project_track_detail(payload)
+            if result:
+                return result
+            if not plan.search_album:
                 return None
-        album = self.music_album(media_source, media_id)
-        return album.to_music_info() if album else None
+        album = self.music_album(self._source, plan.require_media_id())
+        return self._project_album_result(album)
 
     async def async_recognize_music(
             self,
@@ -198,16 +248,50 @@ class TheAudioDbModule(_ModuleBase):
             music_type: Optional[str] = None,
     ) -> Optional[MusicInfo]:
         """异步按 TheAudioDB 原生 ID 和实体类型获取详情。"""
-        if media_source != self._source or not media_id:
+        plan = self._detail_plan(media_source, media_id, music_type)
+        if not plan:
             return None
-        if music_type != MUSIC_ENTITY_ALBUM:
-            payload = await self._async_request_json("track.php", {"h": media_id})
-            track = self._first_entity(payload, "track", "tracks")
-            if track:
-                return self._track_to_info(track)
-            if music_type == MUSIC_ENTITY_RECORDING:
+        if plan.search_recording:
+            payload = await self._async_request_json(
+                "track.php", {"h": plan.require_media_id()}
+            )
+            result = self._project_track_detail(payload)
+            if result:
+                return result
+            if not plan.search_album:
                 return None
-        album = await self._async_music_album(media_source, media_id)
+        album = await self._async_music_album(
+            self._source, plan.require_media_id()
+        )
+        return self._project_album_result(album)
+
+    @classmethod
+    def _detail_plan(
+            cls,
+            media_source: MediaSource,
+            media_id: str,
+            music_type: Optional[str],
+    ) -> Optional[_TheAudioDbRecognitionPlan]:
+        """统一校验详情来源并冻结实体探测顺序。"""
+        if media_source != cls._source or not media_id:
+            return None
+        return _TheAudioDbRecognitionPlan(
+            meta=None, media_id=str(media_id), music_type=music_type
+        )
+
+    @classmethod
+    def _project_track_detail(
+            cls, payload: Optional[dict[str, Any]]
+    ) -> Optional[MusicInfo]:
+        """把单曲详情响应投影为统一音乐信息。"""
+        track = cls._first_entity(payload, "track", "tracks")
+        return cls._track_to_info(track) if track else None
+
+    @staticmethod
+    def _project_album_result(
+            album: Optional[MusicAlbumInfo],
+    ) -> Optional[MusicInfo]:
+        """把专辑候选统一投影到音乐识别返回类型。"""
         return album.to_music_info() if album else None
 
     async def _async_music_album(
@@ -216,20 +300,14 @@ class TheAudioDbModule(_ModuleBase):
             media_id: str,
     ) -> Optional[MusicAlbumInfo]:
         """异步按 TheAudioDB 专辑 ID 获取标准化专辑详情和曲目。"""
-        if media_source != self._source or not media_id:
+        if not self._detail_plan(media_source, media_id, MUSIC_ENTITY_ALBUM):
             return None
         payload = await self._async_request_json("album.php", {"m": media_id})
-        item = self._first_entity(payload, "album", "albums")
-        if not item:
+        album = self._project_album_header(payload)
+        if not album:
             return None
-        album = self._album_to_info(item)
         tracks_payload = await self._async_request_json("track.php", {"m": media_id})
-        album.tracks = [
-            info
-            for track in self._entities(tracks_payload, "track", "tracks")
-            if (info := self._track_to_info(track, album=album))
-        ]
-        return album
+        return self._project_album_tracks(album, tracks_payload)
 
     def music_album(
             self,
@@ -237,18 +315,35 @@ class TheAudioDbModule(_ModuleBase):
             media_id: str,
     ) -> Optional[MusicAlbumInfo]:
         """按 TheAudioDB 专辑 ID 获取标准化专辑详情和曲目。"""
-        if media_source != self._source or not media_id:
+        if not self._detail_plan(media_source, media_id, MUSIC_ENTITY_ALBUM):
             return None
         payload = self._request_json("album.php", {"m": media_id})
-        item = self._first_entity(payload, "album", "albums")
-        if not item:
+        album = self._project_album_header(payload)
+        if not album:
             return None
-        album = self._album_to_info(item)
         tracks_payload = self._request_json("track.php", {"m": media_id})
+        return self._project_album_tracks(album, tracks_payload)
+
+    @classmethod
+    def _project_album_header(
+            cls,
+            payload: Optional[dict[str, Any]],
+    ) -> Optional[MusicAlbumInfo]:
+        """把专辑详情响应投影为尚未装载曲目的专辑。"""
+        item = cls._first_entity(payload, "album", "albums")
+        return cls._album_to_info(item) if item else None
+
+    @classmethod
+    def _project_album_tracks(
+            cls,
+            album: MusicAlbumInfo,
+            tracks_payload: Optional[dict[str, Any]],
+    ) -> MusicAlbumInfo:
+        """把曲目响应附加到已投影的 TheAudioDB 专辑。"""
         album.tracks = [
             info
-            for track in self._entities(tracks_payload, "track", "tracks")
-            if (info := self._track_to_info(track, album=album))
+            for track in cls._entities(tracks_payload, "track", "tracks")
+            if (info := cls._track_to_info(track, album=album))
         ]
         return album
 
@@ -314,56 +409,75 @@ class TheAudioDbModule(_ModuleBase):
 
     def _search_tracks(self, meta: MetaMusic) -> list[MusicInfo]:
         """使用曲名和艺术家搜索 TheAudioDB 单曲。"""
-        title = meta.title
-        artist = meta.artists[0] if meta.artists else meta.album_artist
-        if not title or not artist:
+        params = self._track_search_params(meta)
+        if not params:
             return []
-        params = {"t": title, "s": artist}
         payload = self._request_json("searchtrack.php", params)
-        return [
-            info
-            for item in self._entities(payload, "track", "tracks")
-            if (info := self._track_to_info(item))
-        ]
+        return self._project_track_search(payload)
 
     async def _async_search_tracks(self, meta: MetaMusic) -> list[MusicInfo]:
         """异步使用曲名和艺术家搜索 TheAudioDB 单曲。"""
-        title = meta.title
-        artist = meta.artists[0] if meta.artists else meta.album_artist
-        if not title or not artist:
+        params = self._track_search_params(meta)
+        if not params:
             return []
-        params = {"t": title, "s": artist}
         payload = await self._async_request_json("searchtrack.php", params)
+        return self._project_track_search(payload)
+
+    @staticmethod
+    def _track_search_params(meta: MetaMusic) -> Optional[dict[str, str]]:
+        """从音乐元数据归一化单曲搜索参数。"""
+        artist = meta.artists[0] if meta.artists else meta.album_artist
+        if not meta.title or not artist:
+            return None
+        return {"t": meta.title, "s": artist}
+
+    @classmethod
+    def _project_track_search(
+            cls, payload: Optional[dict[str, Any]]
+    ) -> list[MusicInfo]:
+        """把单曲搜索响应投影为统一候选列表。"""
         return [
             info
-            for item in self._entities(payload, "track", "tracks")
-            if (info := self._track_to_info(item))
+            for item in cls._entities(payload, "track", "tracks")
+            if (info := cls._track_to_info(item))
         ]
 
     def _search_albums(self, meta: MetaMusic) -> list[MusicAlbumInfo]:
         """使用专辑名和艺术家搜索 TheAudioDB 专辑。"""
-        album_name = meta.album or meta.title
-        artist = meta.artists[0] if meta.artists else meta.album_artist
-        if not album_name or not artist:
+        params = self._album_search_params(meta)
+        if not params:
             return []
-        params = {"a": album_name, "s": artist}
         payload = self._request_json("searchalbum.php", params)
-        return [self._album_to_info(item) for item in self._entities(payload, "album", "albums")]
+        return self._project_album_search(payload)
 
     async def _async_search_albums(
             self,
             meta: MetaMusic,
     ) -> list[MusicAlbumInfo]:
         """异步使用专辑名和艺术家搜索 TheAudioDB 专辑。"""
+        params = self._album_search_params(meta)
+        if not params:
+            return []
+        payload = await self._async_request_json("searchalbum.php", params)
+        return self._project_album_search(payload)
+
+    @staticmethod
+    def _album_search_params(meta: MetaMusic) -> Optional[dict[str, str]]:
+        """从音乐元数据归一化专辑搜索参数。"""
         album_name = meta.album or meta.title
         artist = meta.artists[0] if meta.artists else meta.album_artist
         if not album_name or not artist:
-            return []
-        params = {"a": album_name, "s": artist}
-        payload = await self._async_request_json("searchalbum.php", params)
+            return None
+        return {"a": album_name, "s": artist}
+
+    @classmethod
+    def _project_album_search(
+            cls, payload: Optional[dict[str, Any]]
+    ) -> list[MusicAlbumInfo]:
+        """把专辑搜索响应投影为统一候选列表。"""
         return [
-            self._album_to_info(item)
-            for item in self._entities(payload, "album", "albums")
+            cls._album_to_info(item)
+            for item in cls._entities(payload, "album", "albums")
         ]
 
     def _search_artists(self, meta: MetaMusic) -> list[MusicArtistInfo]:

docs/architecture-optimization-checklist.md

@@ -78,8 +78,8 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
 | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement |
 | Python 源码量 | 约 298,700 行 | 60 个文件超过 1,000 行，10 个超过 2,000 行 |
 | 长方法 | 299 个超过 80 行 | 70 个超过 150 行，23 个超过 250 行；大量是私有方法 |
-| 全量 mypy 历史债务 | 10,100 / 592 文件 | 识别链 typed plan/result、TMDB API、消息、Indexer、目录源与 Recommend 双 ABI 共用计划继续清除参数、返回值和泛型注解债；strict frontier 当前覆盖 41 个文件 |
-| Ruff 历史诊断 | 640 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
+| 全量 mypy 历史债务 | 10,092 / 592 文件 | 识别链 typed plan/result、TMDB API、消息、Indexer、目录源与 Recommend 双 ABI 共用计划继续清除参数、返回值和泛型注解债；strict frontier 当前覆盖 41 个文件 |
+| Ruff 历史诊断 | 638 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
 | 覆盖率低水位 | Application 81.00%，Domain 80.67% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
 
 ### 3.3 热点文件

tests/fixtures/architecture/mypy-baseline.json

@@ -1909,7 +1909,7 @@
     "union-attr": 18
   },
   "app/modules/douban/__init__.py": {
-    "arg-type": 2,
+    "arg-type": 1,
     "assignment": 27,
     "empty-body": 1,
     "misc": 2,
@@ -1918,7 +1918,7 @@
     "no-untyped-def": 11,
     "return-value": 2,
     "type-arg": 13,
-    "union-attr": 9
+    "union-attr": 2
   },
   "app/modules/douban/apiv2.py": {
     "misc": 6,

tests/fixtures/architecture/ruff-baseline.json

@@ -385,9 +385,6 @@
     "F541": 2,
     "I001": 1
   },
-  "app/modules/douban/__init__.py": {
-    "I001": 1
-  },
   "app/modules/douban/scraper.py": {
     "I001": 1
   },
@@ -567,9 +564,6 @@
   "app/modules/telegram/telegram.py": {
     "I001": 3
   },
-  "app/modules/theaudiodb/__init__.py": {
-    "I001": 1
-  },
   "app/modules/themoviedb/category.py": {
     "F541": 1,
     "I001": 1

tests/test_catalog_source_sync_async_parity.py

@@ -1,4 +1,4 @@
-"""AniList、Bangumi 与 IMDb Module 同步/异步决策一致性测试。"""
+"""目录来源 Module 同步/异步决策一致性测试。"""
 
 import asyncio
 from copy import deepcopy
@@ -7,13 +7,22 @@ from unittest.mock import AsyncMock, Mock
 
 import pytest
 
-from app.domain.context import MediaInfo
+from app.domain.context import MediaInfo, MusicAlbumInfo, MusicInfo
 from app.domain.meta.metabase import MetaBase
+from app.domain.meta.metamusic import MetaMusic
 from app.modules.anilist import AniListModule
 from app.modules.bangumi import BangumiModule
+from app.modules.douban import DoubanModule
 from app.modules.imdb import ImdbModule
 from app.modules.imdb.api import ImdbAka, ImdbApi, ImdbTitle
-from app.schemas.types import MediaSource, MediaType
+from app.modules.musicbrainz import MusicBrainzModule
+from app.modules.theaudiodb import TheAudioDbModule
+from app.schemas.types import (
+    MUSIC_ENTITY_ALBUM,
+    MUSIC_ENTITY_RECORDING,
+    MediaSource,
+    MediaType,
+)
 
 
 def _meta(
@@ -43,6 +52,22 @@ def _signature(media: Optional[MediaInfo]) -> Optional[tuple[Any, ...]]:
     )
 
 
+def _music_signature(music: Optional[MusicInfo]) -> Optional[tuple[Any, ...]]:
+    """提取音乐识别双 ABI 必须一致的来源、身份和展示字段。"""
+    if music is None:
+        return None
+    return (
+        music.media_source,
+        music.media_id,
+        music.music_type,
+        music.title,
+        tuple(music.artists),
+        music.album,
+        music.album_id,
+        music.track_number,
+    )
+
+
 def _anilist_info() -> dict:
     """构造可覆盖识别与搜索投影的 AniList 详情。"""
     return {
@@ -267,3 +292,192 @@ def test_catalog_search_sync_async_projection_parity(
     assert [_signature(item) for item in sync_results] == [
         _signature(item) for item in async_results
     ]
+
+
+@pytest.mark.parametrize(
+    ("media_source", "expected"),
+    [
+        (MediaSource.Douban, True),
+        (MediaSource.TMDB, False),
+    ],
+)
+def test_douban_video_recognition_sync_async_decision_parity(
+    media_source: MediaSource,
+    expected: bool,
+) -> None:
+    """豆瓣影视来源准入、原生 ID 规范化和详情投影在双 ABI 下应一致。"""
+    info = {
+        "id": "200",
+        "title": "测试电影",
+        "type": "movie",
+        "year": "2024",
+    }
+    module = DoubanModule()
+    module.douban_info = Mock(return_value=deepcopy(info))
+    module.async_douban_info = AsyncMock(return_value=deepcopy(info))
+
+    sync_result = module.recognize_media(
+        mtype=MediaType.MOVIE,
+        media_source=media_source,
+        media_id="200",
+    )
+    async_result = asyncio.run(
+        module.async_recognize_media(
+            mtype=MediaType.MOVIE,
+            media_source=media_source,
+            media_id="200",
+        )
+    )
+
+    assert bool(sync_result) is expected
+    assert _signature(sync_result) == _signature(async_result)
+    if expected:
+        module.douban_info.assert_called_once_with(
+            doubanid="200", mtype=MediaType.MOVIE
+        )
+        module.async_douban_info.assert_awaited_once_with(
+            doubanid="200", mtype=MediaType.MOVIE
+        )
+    else:
+        module.douban_info.assert_not_called()
+        module.async_douban_info.assert_not_awaited()
+
+
+def test_douban_music_candidate_sync_async_decision_parity() -> None:
+    """豆瓣音乐标题、艺术家和曲目候选选择仅由共享决策完成。"""
+    search_payload = {
+        "items": [{
+            "target_type": "music",
+            "target": {
+                "id": "1401853",
+                "title": "范特西",
+                "card_subtitle": "周杰伦 / 2001",
+            },
+        }]
+    }
+    detail_payload = {
+        "id": "1401853",
+        "title": "范特西",
+        "singer": [{"name": "周杰伦"}],
+        "songs": [{
+            "title": "爱在西元前",
+            "track_number": 1,
+            "artist_names": ["周杰伦"],
+        }],
+    }
+    module = DoubanModule()
+    module.doubanapi = Mock()
+    module.doubanapi.music_search.return_value = deepcopy(search_payload)
+    module.doubanapi.music_detail.return_value = deepcopy(detail_payload)
+    module.doubanapi.async_music_search = AsyncMock(
+        return_value=deepcopy(search_payload)
+    )
+    module.doubanapi.async_music_detail = AsyncMock(
+        return_value=deepcopy(detail_payload)
+    )
+    meta = MetaMusic(
+        title="爱在西元前",
+        album="范特西",
+        artists=["周杰伦"],
+    )
+
+    sync_result = module.recognize_media(
+        meta=meta,
+        mtype=MediaType.MUSIC,
+        media_source=MediaSource.DoubanMusic,
+        music_type=MUSIC_ENTITY_RECORDING,
+    )
+    async_result = asyncio.run(
+        module.async_recognize_media(
+            meta=meta,
+            mtype=MediaType.MUSIC,
+            media_source=MediaSource.DoubanMusic,
+            music_type=MUSIC_ENTITY_RECORDING,
+        )
+    )
+
+    assert _music_signature(sync_result) == _music_signature(async_result)
+    assert sync_result and sync_result.media_id == "1401853:1"
+    module.doubanapi.music_search.assert_called_once_with(
+        keyword="范特西", count=20
+    )
+    module.doubanapi.async_music_search.assert_awaited_once_with(
+        keyword="范特西", count=20
+    )
+
+
+def test_musicbrainz_candidate_sync_async_decision_parity() -> None:
+    """MusicBrainz 候选选择和兜底决策一致，仅检索 I/O 不同。"""
+    wrong = MusicInfo(
+        media_source=MediaSource.MusicBrainz,
+        media_id="wrong",
+        title="晴天",
+        artists=["其他歌手"],
+    )
+    expected = MusicInfo(
+        media_source=MediaSource.MusicBrainz,
+        media_id="recording-1",
+        title="晴天",
+        artists=["周杰伦"],
+        album="叶惠美",
+        album_id="album-1",
+    )
+    module = MusicBrainzModule()
+    module.cache = None
+    module._search_recordings = Mock(return_value=[wrong, expected])
+    module._async_search_recordings = AsyncMock(return_value=[wrong, expected])
+    module._search_albums = Mock(return_value=[])
+    module._async_search_albums = AsyncMock(return_value=[])
+    meta = MetaMusic(title="晴天", artists=["周杰伦"], album="叶惠美")
+
+    sync_result = module.recognize_media(meta=meta, cache=False)
+    async_result = asyncio.run(module.async_recognize_media(meta=meta, cache=False))
+
+    assert _music_signature(sync_result) == _music_signature(async_result)
+    assert sync_result and sync_result.media_id == "recording-1"
+    module._search_recordings.assert_called_once_with(meta, limit=10)
+    module._async_search_recordings.assert_awaited_once_with(meta, limit=10)
+    module._search_albums.assert_not_called()
+    module._async_search_albums.assert_not_awaited()
+
+
+def test_theaudiodb_album_candidate_sync_async_decision_parity() -> None:
+    """TheAudioDB 专辑实体准入、候选选择和结果投影在双 ABI 下应一致。"""
+    album = MusicAlbumInfo(
+        media_source=MediaSource.TheAudioDB,
+        media_id="2109619",
+        title="Parachutes",
+        artists=["Coldplay"],
+    )
+    module = TheAudioDbModule()
+    module._search_tracks = Mock(return_value=[])
+    module._async_search_tracks = AsyncMock(return_value=[])
+    module._search_albums = Mock(return_value=[album])
+    module._async_search_albums = AsyncMock(return_value=[album])
+    meta = MetaMusic(
+        title="Parachutes",
+        album="Parachutes",
+        artists=["Coldplay"],
+    )
+
+    sync_result = module.recognize_media(
+        meta=meta,
+        mtype=MediaType.MUSIC,
+        media_source=MediaSource.TheAudioDB,
+        music_type=MUSIC_ENTITY_ALBUM,
+    )
+    async_result = asyncio.run(
+        module.async_recognize_media(
+            meta=meta,
+            mtype=MediaType.MUSIC,
+            media_source=MediaSource.TheAudioDB,
+            music_type=MUSIC_ENTITY_ALBUM,
+        )
+    )
+
+    assert _music_signature(sync_result) == _music_signature(async_result)
+    assert sync_result and sync_result.media_id == "2109619"
+    module._search_tracks.assert_not_called()
+    module._async_search_tracks.assert_not_awaited()
+    module._search_albums.assert_called_once_with(meta)
+    module._async_search_albums.assert_awaited_once_with(meta)