From 840153e7c7064e66e015974db4e981ce77cb114b Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 29 Aug 2026 20:26:29 +0800 Subject: [PATCH] refactor(modules): unify music catalog decisions --- app/modules/douban/__init__.py | 697 +++++++++++------- app/modules/musicbrainz/__init__.py | 558 +++++++++----- app/modules/theaudiodb/__init__.py | 330 ++++++--- docs/architecture-optimization-checklist.md | 4 +- .../fixtures/architecture/mypy-baseline.json | 4 +- .../fixtures/architecture/ruff-baseline.json | 6 - .../test_catalog_source_sync_async_parity.py | 220 +++++- 7 files changed, 1250 insertions(+), 569 deletions(-) diff --git a/app/modules/douban/__init__.py b/app/modules/douban/__init__.py index a62db8e66..369810adf 100644 --- a/app/modules/douban/__init__.py +++ b/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]: """ diff --git a/app/modules/musicbrainz/__init__.py b/app/modules/musicbrainz/__init__.py index b8fec4ad8..13ebda5f6 100644 --- a/app/modules/musicbrainz/__init__.py +++ b/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 []: diff --git a/app/modules/theaudiodb/__init__.py b/app/modules/theaudiodb/__init__.py index a297693d0..1811dd29f 100644 --- a/app/modules/theaudiodb/__init__.py +++ b/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]: diff --git a/docs/architecture-optimization-checklist.md b/docs/architecture-optimization-checklist.md index 7474af07e..175c164fb 100644 --- a/docs/architecture-optimization-checklist.md +++ b/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 热点文件 diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index 1979ff79a..b8019bf83 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/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, diff --git a/tests/fixtures/architecture/ruff-baseline.json b/tests/fixtures/architecture/ruff-baseline.json index 3412d860f..f5bb2b84b 100644 --- a/tests/fixtures/architecture/ruff-baseline.json +++ b/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 diff --git a/tests/test_catalog_source_sync_async_parity.py b/tests/test_catalog_source_sync_async_parity.py index f218e00c8..7e33e5733 100644 --- a/tests/test_catalog_source_sync_async_parity.py +++ b/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)