diff --git a/app/chain/media.py b/app/chain/media.py index 66833a494..7081654f2 100644 --- a/app/chain/media.py +++ b/app/chain/media.py @@ -1,10 +1,7 @@ -import asyncio import os import re -from concurrent.futures import ThreadPoolExecutor from copy import deepcopy from dataclasses import dataclass -from difflib import SequenceMatcher from pathlib import Path from tempfile import NamedTemporaryFile, TemporaryDirectory from threading import Lock @@ -37,7 +34,6 @@ from app.schemas.types import ( MUSIC_ENTITY_RECORDING, ChainEventType, EventType, - MediaRecognizeType, MediaType, ScrapingTarget, ScrapingMetadata, @@ -48,12 +44,10 @@ from app.utils.http import RequestUtils from app.utils.media import ( is_music_media_source, normalize_media_source, - resolve_media_identity, ) from app.utils.mixins import ConfigReloadMixin from app.utils.singleton import Singleton from app.utils.string import StringUtils -from app.utils.zhconv import convert as zhconv_convert recognize_lock = Lock() scraping_lock = Lock() @@ -214,15 +208,6 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): flags=re.IGNORECASE, ) _video_primary_source = "themoviedb" - _video_fallback_source_order = ("douban", "bangumi", "anilist") - _video_source_subtypes = { - "themoviedb": MediaRecognizeType.TMDB, - "douban": MediaRecognizeType.Douban, - "bangumi": MediaRecognizeType.Bangumi, - "anilist": MediaRecognizeType.AniList, - } - _video_title_min_similarity = 0.72 - _video_recognize_min_score = 65.0 def __init__(self): super().__init__() @@ -234,7 +219,7 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): module_kwargs: dict, cache: bool, ) -> Optional[MediaInfo]: - """统一同步媒体识别路由，音乐请求只进入音乐数据源。""" + """统一同步媒体识别路由，未指定来源时影视和音乐只使用各自主数据源。""" meta = module_kwargs.get("meta") mtype = module_kwargs.get("mtype") source = module_kwargs.get("source") @@ -259,7 +244,7 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): return music_chain.recognize_best(meta=meta, cache=cache) return None if not source and isinstance(meta, MetaBase): - return self._recognize_video_best(module_kwargs, cache) + module_kwargs = {**module_kwargs, "source": self._video_primary_source} return super()._run_native_media_recognize(module_kwargs, cache) async def _async_run_native_media_recognize( @@ -267,7 +252,7 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): module_kwargs: dict, cache: bool, ) -> Optional[MediaInfo]: - """统一异步媒体识别路由，音乐请求只进入音乐数据源。""" + """统一异步媒体识别路由，未指定来源时影视和音乐只使用各自主数据源。""" meta = module_kwargs.get("meta") mtype = module_kwargs.get("mtype") source = module_kwargs.get("source") @@ -294,327 +279,9 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): return await music_chain.async_recognize_best(meta=meta, cache=cache) return None if not source and isinstance(meta, MetaBase): - return await self._async_recognize_video_best(module_kwargs, cache) + module_kwargs = {**module_kwargs, "source": self._video_primary_source} return await super()._async_run_native_media_recognize(module_kwargs, cache) - def _recognize_video_best( - self, - module_kwargs: dict, - cache: bool, - ) -> Optional[MediaInfo]: - """先验证 TMDB 主结果，仅在未可靠命中时并发比较内置影视副源。""" - meta = module_kwargs.get("meta") - mtype = module_kwargs.get("mtype") - primary_result = self._recognize_video_from_source( - module_kwargs=module_kwargs, - source=self._video_primary_source, - cache=cache, - ) - primary = self._select_best_video_candidate( - meta=meta, - mtype=mtype, - candidates=[(self._video_primary_source, primary_result)], - source_order=(self._video_primary_source,), - ) - if primary: - return primary - - logger.info(f"{meta.name} 未可靠命中 TMDB，开始并发查询影视辅助数据源 ...") - with ThreadPoolExecutor( - max_workers=len(self._video_fallback_source_order), - thread_name_prefix="video-recognize", - ) as executor: - futures = { - source: executor.submit( - self._recognize_video_from_source, - module_kwargs, - source, - cache, - ) - for source in self._video_fallback_source_order - } - candidates = [ - (source, futures[source].result()) - for source in self._video_fallback_source_order - ] - return self._select_best_video_candidate( - meta=meta, - mtype=mtype, - candidates=candidates, - source_order=self._video_fallback_source_order, - ) - - async def _async_recognize_video_best( - self, - module_kwargs: dict, - cache: bool, - ) -> Optional[MediaInfo]: - """异步先验证 TMDB 主结果，仅在未可靠命中时并发比较影视副源。""" - meta = module_kwargs.get("meta") - mtype = module_kwargs.get("mtype") - primary_result = await self._async_recognize_video_from_source( - module_kwargs=module_kwargs, - source=self._video_primary_source, - cache=cache, - ) - primary = self._select_best_video_candidate( - meta=meta, - mtype=mtype, - candidates=[(self._video_primary_source, primary_result)], - source_order=(self._video_primary_source,), - ) - if primary: - return primary - - logger.info(f"{meta.name} 未可靠命中 TMDB，开始并发查询影视辅助数据源 ...") - results = await asyncio.gather(*( - self._async_recognize_video_from_source( - module_kwargs=module_kwargs, - source=source, - cache=cache, - ) - for source in self._video_fallback_source_order - )) - return self._select_best_video_candidate( - meta=meta, - mtype=mtype, - candidates=zip(self._video_fallback_source_order, results), - source_order=self._video_fallback_source_order, - ) - - def _video_recognize_module(self, source: str) -> Optional[Any]: - """按内置影视来源枚举对应的运行中识别模块。""" - subtype = self._video_source_subtypes.get(normalize_media_source(source)) - if not subtype: - return None - return next(self.modulemanager.get_running_subtype_module(subtype), None) - - @staticmethod - def _video_source_kwargs(module_kwargs: dict, source: str) -> dict: - """复制单源识别参数和元数据，避免并发模块互相修改解析状态。""" - source_kwargs = dict(module_kwargs) - if source_kwargs.get("meta"): - source_kwargs["meta"] = deepcopy(source_kwargs["meta"]) - source_kwargs["source"] = source - return source_kwargs - - def _recognize_video_from_source( - self, - module_kwargs: dict, - source: str, - cache: bool, - ) -> Optional[MediaInfo]: - """同步调用指定内置影视源，隔离单个来源的查询异常。""" - module = self._video_recognize_module(source) - if not module: - return None - try: - with fresh(not cache): - return module.recognize_media( - **self._video_source_kwargs(module_kwargs, source) - ) - except Exception as err: - logger.warning(f"{source} 影视自动识别失败：{err}") - return None - - async def _async_recognize_video_from_source( - self, - module_kwargs: dict, - source: str, - cache: bool, - ) -> Optional[MediaInfo]: - """异步调用指定内置影视源，隔离单个来源的查询异常。""" - module = self._video_recognize_module(source) - if not module: - return None - source_kwargs = self._video_source_kwargs(module_kwargs, source) - try: - async with async_fresh(not cache): - async_method = getattr(module, "async_recognize_media", None) - if async_method: - return await async_method(**source_kwargs) - return await run_in_threadpool(module.recognize_media, **source_kwargs) - except Exception as err: - logger.warning(f"{source} 影视自动识别失败：{err}") - return None - - @staticmethod - def _normalize_video_candidate( - result: Any, - source: str, - ) -> Optional[MediaInfo]: - """校验单源影视结果的领域、来源和原生身份。""" - if not isinstance(result, MediaInfo): - return None - normalized_source = normalize_media_source(source) - if normalize_media_source(result.source) != normalized_source: - return None - identity_source, media_id = resolve_media_identity(media=result) - if identity_source != normalized_source or not media_id: - return None - if result.type not in {MediaType.MOVIE, MediaType.TV}: - return None - return result - - @staticmethod - def _normalize_video_match_name(value: Any) -> str: - """统一影视标题的繁简、大小写、空白和标点差异。""" - if not isinstance(value, str) or not value.strip(): - return "" - return StringUtils.clear_upper(zhconv_convert(value, "zh-hans")) - - @classmethod - def _normalized_video_names(cls, values: Iterable[Any]) -> set[str]: - """生成原始标题及去除季集标记后的标准标题集合。""" - names: set[str] = set() - for value in values: - normalized = cls._normalize_video_match_name(value) - if normalized: - names.add(normalized) - if not isinstance(value, str) or not value.strip(): - continue - try: - parsed_name = MetaInfo(value).name - except Exception: - parsed_name = None - normalized_parsed = cls._normalize_video_match_name(parsed_name) - if normalized_parsed: - names.add(normalized_parsed) - return names - - @classmethod - def _video_title_similarity( - cls, - meta: MetaBase, - candidate: MediaInfo, - ) -> float: - """计算解析标题与候选全部标题及别名之间的最大相似度。""" - expected_names = cls._normalized_video_names([ - getattr(meta, "name", None), - getattr(meta, "cn_name", None), - getattr(meta, "en_name", None), - ]) - candidate_names = cls._normalized_video_names([ - candidate.title, - candidate.en_title, - candidate.original_title, - candidate.original_name, - *(candidate.names or []), - ]) - return max( - ( - SequenceMatcher(None, expected, actual).ratio() - for expected in expected_names - for actual in candidate_names - ), - default=0.0, - ) - - @staticmethod - def _video_year(value: Any) -> Optional[int]: - """从年份或日期字段中提取四位年份。""" - match = re.search(r"\d{4}", str(value or "")) - return int(match.group()) if match else None - - @classmethod - def _video_candidate_year( - cls, - meta: MetaBase, - candidate: MediaInfo, - ) -> Optional[int]: - """电视剧优先使用请求季年份，电影和整剧使用作品年份。""" - season = getattr(meta, "begin_season", None) - if candidate.type != MediaType.TV or season is None: - return cls._video_year(candidate.year) - season_years = candidate.season_years or {} - season_year = ( - season_years.get(season) - or season_years.get(str(season)) - ) - if season_year: - return cls._video_year(season_year) - if candidate.season == season or not candidate.seasons: - return cls._video_year(candidate.year) - return None - - @classmethod - def _video_candidate_score( - cls, - meta: MetaBase, - candidate: MediaInfo, - mtype: Optional[MediaType] = None, - ) -> Optional[float]: - """按标题、类型、年份和季信息计算跨影视源可比较的证据分。""" - expected_type = mtype if mtype in {MediaType.MOVIE, MediaType.TV} else meta.type - if expected_type in {MediaType.MOVIE, MediaType.TV} and candidate.type != expected_type: - return None - - title_similarity = cls._video_title_similarity(meta, candidate) - if title_similarity < cls._video_title_min_similarity: - return None - score = title_similarity * 70 - - expected_year = cls._video_year(getattr(meta, "year", None)) - candidate_year = cls._video_candidate_year(meta, candidate) - if expected_year and candidate_year: - year_delta = abs(expected_year - candidate_year) - if year_delta > 1: - return None - score += 20 if year_delta == 0 else 8 - - requested_season = getattr(meta, "begin_season", None) - if requested_season is not None: - if candidate.type != MediaType.TV: - return None - available_seasons = { - int(season) - for season in (candidate.seasons or {}) - if str(season).isdigit() - } - if ( - available_seasons - and requested_season not in available_seasons - and candidate.season != requested_season - ): - return None - if ( - requested_season in available_seasons - or candidate.season == requested_season - ): - score += 10 - return score - - @classmethod - def _select_best_video_candidate( - cls, - meta: MetaBase, - mtype: Optional[MediaType], - candidates: Iterable[tuple[str, Any]], - source_order: Iterable[str], - ) -> Optional[MediaInfo]: - """按统一证据分选择影视候选，同分时使用确定的数据源顺序。""" - order = {source: index for index, source in enumerate(source_order)} - ranked: list[tuple[float, int, MediaInfo]] = [] - for source, result in candidates: - candidate = cls._normalize_video_candidate(result, source) - if not candidate: - continue - score = cls._video_candidate_score(meta, candidate, mtype) - if score is None or score < cls._video_recognize_min_score: - continue - logger.debug( - f"影视自动识别候选：{source} {candidate.title_year}，评分 {score:.1f}" - ) - ranked.append((score, -order.get(source, len(order)), candidate)) - if not ranked: - return None - ranked.sort(key=lambda item: (item[0], item[1]), reverse=True) - score, _, best = ranked[0] - logger.info( - f"影视自动识别采用 {best.source}：{best.title_year}，匹配评分 {score:.1f}" - ) - return best - def on_config_changed(self): self.scraping_policies = ScrapingConfig.from_system_config() @@ -1477,6 +1144,117 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): setattr(info, key, value) return info + @staticmethod + def _clear_music_identity(meta: MetaMusic) -> MetaMusic: + """复制音乐元数据并清除远程身份，供直查失败后按要素重新匹配。""" + clean_meta = MetaMusic.from_dict(meta.to_dict()) + clean_meta.media_source = None + clean_meta.media_id = None + return clean_meta + + @staticmethod + def _is_remote_music_info(info: Optional[MusicInfo]) -> bool: + """判断音乐识别结果是否携带可复用的远程身份。""" + return bool(info and info.source and info.media_id) + + @staticmethod + def _recognize_musicbrainz_recording( + meta: MetaMusic, + recording_id: str, + ) -> Optional[MusicInfo]: + """按已知 MusicBrainz Recording ID 直接读取单曲详情。""" + identity_meta = MetaMusic.from_dict(meta.to_dict()) + identity_meta.media_source = "musicbrainz" + identity_meta.media_id = recording_id + return MusicChain().recognize_from_source( + source="musicbrainz", + meta=identity_meta, + mediaid=recording_id, + music_type=MUSIC_ENTITY_RECORDING, + ) + + @staticmethod + async def _async_recognize_musicbrainz_recording( + meta: MetaMusic, + recording_id: str, + ) -> Optional[MusicInfo]: + """异步按已知 MusicBrainz Recording ID 直接读取单曲详情。""" + identity_meta = MetaMusic.from_dict(meta.to_dict()) + identity_meta.media_source = "musicbrainz" + identity_meta.media_id = recording_id + return await MusicChain().async_recognize_from_source( + source="musicbrainz", + meta=identity_meta, + mediaid=recording_id, + music_type=MUSIC_ENTITY_RECORDING, + ) + + def _recognize_music_meta_tier( + self, + meta: Optional[MetaMusic], + source: Optional[str], + tier_name: str, + ) -> Optional[MusicInfo]: + """识别单个音乐元数据证据层，标签中的 MBID 优先直查。""" + if not meta: + return None + normalized_source = normalize_media_source(source) + search_meta = meta + if meta.media_source == "musicbrainz" and meta.media_id: + if normalized_source in (None, "musicbrainz"): + direct = self._recognize_musicbrainz_recording( + meta=meta, + recording_id=str(meta.media_id), + ) + if self._is_remote_music_info(direct): + logger.info(f"音乐识别命中{tier_name}层 MusicBrainz ID 直查") + return direct + search_meta = self._clear_music_identity(meta) + if not search_meta.title: + return None + result = self.recognize_media( + meta=search_meta, + source=source, + music_type=MUSIC_ENTITY_RECORDING, + ) + if self._is_remote_music_info(result): + logger.info(f"音乐识别命中{tier_name}层：{result.title}") + return result + return None + + async def _async_recognize_music_meta_tier( + self, + meta: Optional[MetaMusic], + source: Optional[str], + tier_name: str, + ) -> Optional[MusicInfo]: + """异步识别单个音乐元数据证据层，标签中的 MBID 优先直查。""" + if not meta: + return None + normalized_source = normalize_media_source(source) + search_meta = meta + if meta.media_source == "musicbrainz" and meta.media_id: + if normalized_source in (None, "musicbrainz"): + direct = await self._async_recognize_musicbrainz_recording( + meta=meta, + recording_id=str(meta.media_id), + ) + if self._is_remote_music_info(direct): + logger.info(f"音乐识别命中{tier_name}层 MusicBrainz ID 直查") + return direct + search_meta = self._clear_music_identity(meta) + if not search_meta.title: + return None + result = await self.async_recognize_media( + meta=search_meta, + source=source, + music_type=MUSIC_ENTITY_RECORDING, + ) + if self._is_remote_music_info(result): + logger.info(f"音乐识别命中{tier_name}层：{result.title}") + return result + return None + @staticmethod def _music_album_dir_fallback(path: Union[str, Path]) -> Optional[MusicInfo]: """单曲识别无远端身份时，查找所在目录专辑匹配中属于当前文件的结果。""" @@ -1490,15 +1268,50 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): return None return matched.get(str(file_path.resolve())) + @staticmethod + async def _async_music_album_dir_fallback( + path: Union[str, Path], + ) -> Optional[MusicInfo]: + """异步查找所在目录专辑匹配中属于当前文件的结果。""" + file_path = Path(path) + if not file_path.exists() or not file_path.is_file(): + return None + try: + matched = await MusicChain().async_recognize_album_directory( + file_path.parent + ) + except Exception as err: + logger.debug(f"专辑目录匹配失败：{file_path.parent} - {err}") + return None + return matched.get(str(file_path.resolve())) + def recognize_music_by_path( self, path: Union[str, Path], source: Optional[str] = None, ) -> Tuple[MetaMusic, MusicInfo]: - """同步根据音频标签和文件名识别音乐，并保留离线最小结果。""" - meta = self.read_path_meta(path) - # 统一识别入口分发到音乐模块，模块负责详情/搜索/匹配/兑底 - info = self.recognize_media(meta=meta, source=source) + """按指纹、文件标签、文件名三级顺序识别本地音乐。""" + meta, tag_meta, filename_meta = MusicChain.read_path_evidence(path) + info = None + normalized_source = normalize_media_source(source) + if normalized_source in (None, "musicbrainz"): + recording_id = MusicChain().identify_by_fingerprint(path) + if recording_id: + info = self._recognize_musicbrainz_recording(meta, recording_id) + if self._is_remote_music_info(info): + logger.info("音乐识别命中 AcoustID 指纹层，已跳过标签和文件名识别") + if not self._is_remote_music_info(info): + info = self._recognize_music_meta_tier( + meta=tag_meta, + source=source, + tier_name="文件标签", + ) + if not self._is_remote_music_info(info): + info = self._recognize_music_meta_tier( + meta=filename_meta, + source=source, + tier_name="文件名", + ) result = self._merge_music_audio_quality( info or self._music_info_from_path_meta(meta), meta ) @@ -1514,17 +1327,40 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): path: Union[str, Path], source: Optional[str] = None, ) -> Tuple[MetaMusic, MusicInfo]: - """根据音频标签和文件名识别音乐，远端不可用时仍返回最小音乐信息。""" - # Mutagen 会同步读取本地文件，异步识别入口需要移出事件循环。 - meta = await run_in_threadpool(self.read_path_meta, path) - # 统一识别入口分发到音乐模块，模块负责详情/搜索/匹配/兑底 - info = await self.async_recognize_media(meta=meta, source=source) + """异步按指纹、文件标签、文件名三级顺序识别本地音乐。""" + meta, tag_meta, filename_meta = await run_in_threadpool( + MusicChain.read_path_evidence, + path, + ) + info = None + normalized_source = normalize_media_source(source) + if normalized_source in (None, "musicbrainz"): + recording_id = await MusicChain().async_identify_by_fingerprint(path) + if recording_id: + info = await self._async_recognize_musicbrainz_recording( + meta, + recording_id, + ) + if self._is_remote_music_info(info): + logger.info("音乐识别命中 AcoustID 指纹层，已跳过标签和文件名识别") + if not self._is_remote_music_info(info): + info = await self._async_recognize_music_meta_tier( + meta=tag_meta, + source=source, + tier_name="文件标签", + ) + if not self._is_remote_music_info(info): + info = await self._async_recognize_music_meta_tier( + meta=filename_meta, + source=source, + tier_name="文件名", + ) result = self._merge_music_audio_quality( info or self._music_info_from_path_meta(meta), meta ) if not result.source and source in (None, "musicbrainz"): # 单曲搜索未命中时，按所在目录做专辑级匹配兑底 - matched = await run_in_threadpool(self._music_album_dir_fallback, path) + matched = await self._async_music_album_dir_fallback(path) if matched: result = self._merge_music_audio_quality(matched, meta) return meta, result diff --git a/app/chain/music.py b/app/chain/music.py index fb5cd3185..0895432bc 100644 --- a/app/chain/music.py +++ b/app/chain/music.py @@ -1,6 +1,4 @@ -import asyncio import re -from difflib import SequenceMatcher from pathlib import Path from typing import Any, Iterable, Optional, Union @@ -20,12 +18,10 @@ from app.helper.audio import AudioMetadataHelper from app.log import logger from app.schemas.types import MUSIC_ENTITY_ALBUM, MUSIC_ENTITY_RECORDING, MediaType from app.utils.media import ( - MUSIC_MEDIA_SOURCE_ORDER, is_music_media_source, normalize_media_source, normalize_music_type, ) -from app.utils.zhconv import convert as zhconv_convert class MusicChain(ChainBase): @@ -39,9 +35,8 @@ class MusicChain(ChainBase): _album_dir_cache_max = 128 # 目录级匹配至少需要两个音频文件，单文件由单曲搜索链路处理 _album_match_min_files = 2 - # 自动识别会比较全部来源；该顺序仅用于同分时的确定性选择。 - _recognize_source_order = MUSIC_MEDIA_SOURCE_ORDER - _recognize_min_score = 45.0 + # 自动识别只使用 MusicBrainz；其它来源仅响应显式来源请求。 + _primary_recognize_source = "musicbrainz" @classmethod def parse_query(cls, query: str) -> MetaMusic: @@ -152,40 +147,28 @@ class MusicChain(ChainBase): meta: MetaMusic, cache: bool = True, ) -> Optional[MusicInfo]: - """依次查询全部内置音乐源，统一评分后返回最可信的自动识别结果。""" - candidates: list[MusicInfo] = [] - offline_fallback: Optional[MusicInfo] = None + """执行自动音乐识别，仅调用 MusicBrainz 主数据源。""" with fresh(not cache): - for source in self._recognize_source_order: - result = self._recognize_from_source(meta, source, cache) - candidate = self._normalize_recognize_result(result, source) - if candidate: - candidates.append(candidate) - elif isinstance(result, MusicInfo) and not result.source: - # MusicBrainz 会返回无远端身份的离线结果，全部来源失败时仍需保留。 - offline_fallback = offline_fallback or result - return self._select_best_recognize_candidate(meta, candidates) or offline_fallback + return self.recognize_from_source( + source=self._primary_recognize_source, + meta=meta, + cache=cache, + music_type=MUSIC_ENTITY_RECORDING, + ) async def async_recognize_best( self, meta: MetaMusic, cache: bool = True, ) -> Optional[MusicInfo]: - """并发查询全部内置音乐源，统一评分后返回最可信的自动识别结果。""" + """异步执行自动音乐识别，仅调用 MusicBrainz 主数据源。""" async with async_fresh(not cache): - results = await asyncio.gather(*( - self._async_recognize_from_source(meta, source, cache) - for source in self._recognize_source_order - )) - candidates: list[MusicInfo] = [] - offline_fallback: Optional[MusicInfo] = None - for source, result in zip(self._recognize_source_order, results): - candidate = self._normalize_recognize_result(result, source) - if candidate: - candidates.append(candidate) - elif isinstance(result, MusicInfo) and not result.source: - offline_fallback = offline_fallback or result - return self._select_best_recognize_candidate(meta, candidates) or offline_fallback + return await self.async_recognize_from_source( + source=self._primary_recognize_source, + meta=meta, + cache=cache, + music_type=MUSIC_ENTITY_RECORDING, + ) def recognize_from_source( self, @@ -448,25 +431,6 @@ class MusicChain(ChainBase): """移除大小写、空白和标点差异，生成站点标题匹配使用的紧凑文本。""" return MetaMusic.compact_text(value) - @staticmethod - def _normalize_recognize_result( - result: Any, - source: str, - ) -> Optional[MusicInfo]: - """标准化单个来源结果，并拒绝插件或模块返回的跨来源身份。""" - if isinstance(result, dict): - try: - result = MusicInfo.from_dict(result) - except (TypeError, ValueError): - return None - if not isinstance(result, MusicInfo): - return None - if not result.source or not result.media_id or result.source != source: - return None - if result.music_type != MUSIC_ENTITY_RECORDING: - return None - return result - @staticmethod def _validate_source_recognize_result( result: Optional[MusicInfo], @@ -540,13 +504,7 @@ class MusicChain(ChainBase): } if music_type is not None: recognize_kwargs["music_type"] = music_type - async_method = getattr(module, "async_recognize_media", None) - if async_method: - return await async_method(**recognize_kwargs) - return await run_in_threadpool( - module.recognize_media, - **recognize_kwargs, - ) + return await module.async_recognize_media(**recognize_kwargs) except Exception as err: logger.warning(f"{source} 音乐自动识别失败：{err}") return None @@ -559,117 +517,6 @@ class MusicChain(ChainBase): return module return None - @classmethod - def _select_best_recognize_candidate( - cls, - meta: MetaMusic, - candidates: Iterable[MusicInfo], - ) -> Optional[MusicInfo]: - """按统一证据评分选择最佳音轨，同分时使用默认来源顺序。""" - source_order = { - source: index for index, source in enumerate(cls._recognize_source_order) - } - ranked: list[tuple[float, int, MusicInfo]] = [] - for candidate in candidates: - score = cls._recognize_candidate_score(meta, candidate) - if score is None or score < cls._recognize_min_score: - continue - logger.debug( - f"音乐自动识别候选：{candidate.source} {candidate.title}，评分 {score:.1f}" - ) - ranked.append(( - score, - -source_order.get(candidate.source or "", len(source_order)), - candidate, - )) - if not ranked: - return None - ranked.sort(key=lambda item: (item[0], item[1]), reverse=True) - best_score, _, best = ranked[0] - logger.info( - f"音乐自动识别采用 {best.source}：{best.title}，匹配评分 {best_score:.1f}" - ) - return best - - @classmethod - def _recognize_candidate_score( - cls, - meta: MetaMusic, - candidate: MusicInfo, - ) -> Optional[float]: - """综合曲名、艺术家、专辑、ISRC、时长、曲序和年份计算匹配分。""" - if candidate.music_type != MUSIC_ENTITY_RECORDING: - return None - isrc_match = bool( - meta.isrc - and candidate.isrc - and cls._match_similarity(meta.isrc, candidate.isrc) == 1.0 - ) - if meta.isrc and candidate.isrc and not isrc_match: - return None - - title_similarity = cls._match_similarity(meta.title, candidate.title) - if not isrc_match and (not meta.title or title_similarity < 0.7): - return None - score = title_similarity * 50 - - expected_artists = cls._unique_texts([ - *(meta.artists or []), - meta.album_artist, - ]) - candidate_artists = cls._unique_texts([ - *(candidate.artists or []), - candidate.album_artist, - ]) - if expected_artists: - artist_similarity = max( - ( - cls._match_similarity(expected, actual) - for expected in expected_artists - for actual in candidate_artists - ), - default=0.0, - ) - if not isrc_match and artist_similarity < 0.6: - return None - score += artist_similarity * 25 - - if meta.album and candidate.album: - score += cls._match_similarity(meta.album, candidate.album) * 12 - if meta.duration and candidate.duration: - duration_delta = abs(meta.duration - candidate.duration) / max( - meta.duration, candidate.duration - ) - if duration_delta <= 0.02: - score += 8 - elif duration_delta <= 0.05: - score += 6 - elif duration_delta <= 0.1: - score += 3 - elif duration_delta > 0.2: - score -= 8 - if meta.track_number and candidate.track_number: - score += 3 if meta.track_number == candidate.track_number else -1 - if meta.year and candidate.year: - year_delta = abs(int(meta.year) - int(candidate.year)) - score += 2 if year_delta == 0 else 1 if year_delta == 1 else 0 - if isrc_match: - score += 50 - return score - - @staticmethod - def _match_similarity(left: Optional[str], right: Optional[str]) -> float: - """忽略繁简、大小写和标点后计算两段音乐文本的相似度。""" - normalized_left = MetaMusic.compact_text( - zhconv_convert(str(left or ""), "zh-hans") - ) - normalized_right = MetaMusic.compact_text( - zhconv_convert(str(right or ""), "zh-hans") - ) - if not normalized_left or not normalized_right: - return 0.0 - return SequenceMatcher(None, normalized_left, normalized_right).ratio() - def recognize_album_directory(self, path: str | Path) -> dict[str, MusicInfo]: """按目录级线索批量识别整目录音频，返回 文件路径 到标准音乐信息的映射。 @@ -695,8 +542,23 @@ class MusicChain(ChainBase): return matched async def async_recognize_album_directory(self, path: str | Path) -> dict[str, MusicInfo]: - """目录级批量识别的异步版本，本地文件读取移出事件循环。""" - return await run_in_threadpool(self.recognize_album_directory, path) + """异步按目录级线索批量识别整目录音频。""" + dir_path = Path(path) + if not dir_path.is_dir(): + return {} + files = await run_in_threadpool(self._directory_audio_files, dir_path) + if len(files) < self._album_match_min_files: + return {} + cache_key = str(dir_path) + signature = self._album_directory_signature(dir_path, files) + cached = self._album_dir_cache.get(cache_key) + if cached and cached[0] == signature: + return cached[1] + matched = await self._async_match_album_directory(dir_path, files) + if len(self._album_dir_cache) >= self._album_dir_cache_max: + self._album_dir_cache.clear() + self._album_dir_cache[cache_key] = (signature, matched) + return matched @classmethod def _directory_audio_files(cls, dir_path: Path) -> list[Path]: @@ -739,13 +601,42 @@ class MusicChain(ChainBase): file_path = Path(path) if file_path.exists() and file_path.is_file(): return AudioMetadataHelper.read(file_path) - meta = MetaMusic( - org_string=file_path.name, - title=file_path.stem, - audio_format=file_path.suffix.lstrip(".").upper() or None, + return AudioMetadataHelper.read_filename(file_path) + + @classmethod + def read_path_evidence( + cls, + path: Union[str, Path], + ) -> tuple[MetaMusic, Optional[MetaMusic], MetaMusic]: + """分别返回合并元数据、纯标签元数据和纯文件名元数据。""" + file_path = Path(path) + filename_meta = AudioMetadataHelper.read_filename(file_path) + tag_meta = None + if file_path.exists() and file_path.is_file(): + tag_meta = AudioMetadataHelper.read_tags(file_path) + if not tag_meta: + return filename_meta, None, filename_meta + merged_meta = MetaMusic.from_dict(tag_meta.to_dict()).apply_path_context(file_path) + return merged_meta, tag_meta, filename_meta + + def identify_by_fingerprint(self, path: Union[str, Path]) -> Optional[str]: + """调用音频指纹模块识别 MusicBrainz Recording ID。""" + result = self.run_module( + "identify_music_by_fingerprint", + path=Path(path), ) - # apply_path_context 会先剥离曲序再解析文件名，不能在构造阶段提前解析一次。 - return meta.apply_path_context(file_path) + return str(result) if result else None + + async def async_identify_by_fingerprint( + self, + path: Union[str, Path], + ) -> Optional[str]: + """异步调用音频指纹模块识别 MusicBrainz Recording ID。""" + result = await self.async_run_module( + "async_identify_music_by_fingerprint", + path=Path(path), + ) + return str(result) if result else None def _match_album_directory( self, @@ -760,8 +651,9 @@ class MusicChain(ChainBase): logger.debug(f"目录缺少专辑识别线索，跳过专辑匹配：{dir_path}") return {} candidates = self.run_module("match_music_album", meta=album_meta, tracks=metas) + candidate_items = candidates if isinstance(candidates, list) else [candidates] album = next( - (item for item in candidates or [] if isinstance(item, MusicAlbumInfo) and item.tracks), + (item for item in candidate_items if isinstance(item, MusicAlbumInfo) and item.tracks), None, ) if not album: @@ -772,6 +664,40 @@ class MusicChain(ChainBase): matched[str(file.resolve())] = info return matched + async def _async_match_album_directory( + self, + dir_path: Path, + files: list[Path], + ) -> dict[str, MusicInfo]: + """异步执行目录级专辑匹配，本地标签读取保持在线程池中。""" + metas = await run_in_threadpool(self._read_album_path_metas, files) + album_meta = self._album_meta_from_context(dir_path, metas) + if not (album_meta.album or album_meta.title or album_meta.artists): + logger.debug(f"目录缺少专辑识别线索，跳过专辑匹配：{dir_path}") + return {} + candidates = await self.async_run_module( + "async_match_music_album", + meta=album_meta, + tracks=metas, + ) + candidate_items = candidates if isinstance(candidates, list) else [candidates] + album = next( + (item for item in candidate_items if isinstance(item, MusicAlbumInfo) and item.tracks), + None, + ) + if not album: + return {} + logger.info(f"目录 {dir_path.name} 匹配到专辑：{album.title_year}（{album.source}）") + matched: dict[str, MusicInfo] = {} + for file, info in self._align_album_tracks(files, metas, album.tracks).items(): + matched[str(file.resolve())] = info + return matched + + @classmethod + def _read_album_path_metas(cls, files: list[Path]) -> list[MetaMusic]: + """批量读取专辑目录中的本地音频元数据。""" + return [cls.read_path_meta(file) for file in files] + @classmethod def _album_meta_from_context(cls, dir_path: Path, metas: list[MetaMusic]) -> MetaMusic: """汇总目录名和文件标签中的专辑线索，作为专辑搜索条件。""" diff --git a/app/core/config.py b/app/core/config.py index b8f776c7b..1ffed00f1 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -231,6 +231,8 @@ class ConfigModel(BaseModel): # ==================== 音乐配置 ==================== # 音乐封面代理地址（用于解决 coverartarchive.org 无法访问导致的封面不显示问题，留空则使用官方地址） MUSIC_COVER_PROXY: str = "" + # AcoustID 应用 API Key，用于查询本地音频的 Chromaprint 指纹 + ACOUSTID_API_KEY: str = "b1auxfOzAg" # TheAudioDB API Key，默认使用官方公开的免费 V1 Key，可通过环境变量覆盖 THEAUDIODB_API_KEY: str = "123" diff --git a/app/helper/audio.py b/app/helper/audio.py index 28503c772..2c0114bf7 100644 --- a/app/helper/audio.py +++ b/app/helper/audio.py @@ -1,5 +1,6 @@ from pathlib import Path from typing import Any, Optional, Union +from uuid import UUID from mutagen import File as MutagenFile from mutagen.flac import FLAC, Picture @@ -9,6 +10,7 @@ from mutagen.mp4 import MP4, MP4Cover from app.core.context import MusicInfo from app.core.meta import MetaMusic from app.log import logger +from app.schemas.types import MUSIC_ENTITY_RECORDING class AudioMetadataHelper: @@ -17,29 +19,36 @@ class AudioMetadataHelper: @classmethod def read(cls, path: Path) -> MetaMusic: """读取本地音频标签，并以完整文件名模式和目录线索补充缺失字段。""" - def filename_fallback() -> MetaMusic: - """构造无标签结果，完整文件名解析只在确有需要时执行。""" - return MetaMusic( - org_string=path.name, - title=path.stem, - audio_format=path.suffix.lstrip(".").upper() or None, - ).apply_path_context(path) + tag_meta = cls.read_tags(path) + if tag_meta: + return tag_meta.apply_path_context(path) + return cls.read_filename(path) + @classmethod + def read_tags(cls, path: Path) -> Optional[MetaMusic]: + """只读取本地音频标签和流参数，不使用文件名或目录补齐。""" try: audio = MutagenFile(path, easy=True) except Exception as err: logger.warning(f"读取音频标签失败：{path} - {err}") - return filename_fallback() + return None if not audio: - return filename_fallback() + return None tags = audio.tags or {} track_number, total_tracks = cls._number_pair(cls._first(tags, "tracknumber")) disc_number, total_discs = cls._number_pair(cls._first(tags, "discnumber")) + musicbrainz_id = cls._normalize_musicbrainz_id( + cls._first_of( + tags, + "musicbrainz_trackid", + "musicbrainz_recordingid", + ) + ) info = getattr(audio, "info", None) return MetaMusic( org_string=path.name, - title=cls._first(tags, "title") or path.stem, + title=cls._first(tags, "title"), artists=cls._values(tags, "artist"), album=cls._first(tags, "album"), album_artist=cls._first(tags, "albumartist"), @@ -55,6 +64,17 @@ class AudioMetadataHelper: bitrate=cls._optional_int(getattr(info, "bitrate", None)), duration=round(info.length) if info and getattr(info, "length", None) else None, isrc=cls._first(tags, "isrc"), + media_source="musicbrainz" if musicbrainz_id else None, + media_id=musicbrainz_id, + ) + + @staticmethod + def read_filename(path: Path) -> MetaMusic: + """只从文件名和目录结构解析音乐元数据。""" + return MetaMusic( + org_string=path.name, + title=path.stem, + audio_format=path.suffix.lstrip(".").upper() or None, ).apply_path_context(path) @classmethod @@ -123,8 +143,26 @@ class AudioMetadataHelper: "tracknumber": track_number, "discnumber": disc_number, "isrc": getattr(music, "isrc", None), + "musicbrainz_trackid": cls._musicbrainz_recording_id(music), } + @staticmethod + def _musicbrainz_recording_id( + music: Union[MetaMusic, MusicInfo], + ) -> Optional[str]: + """仅将 MusicBrainz 单曲身份写入 recording 标签，避免误写专辑 ID。""" + if getattr(music, "media_source", None) == "musicbrainz": + media_id = getattr(music, "media_id", None) + return str(media_id) if media_id else None + if ( + getattr(music, "source", None) == "musicbrainz" + and getattr(music, "music_type", MUSIC_ENTITY_RECORDING) + == MUSIC_ENTITY_RECORDING + ): + media_id = getattr(music, "media_id", None) + return str(media_id) if media_id else None + return None + @staticmethod def _number_text(current: Optional[int], total: Optional[int]) -> Optional[str]: """把曲序或碟号转换为常见的 current/total 标签文本。""" @@ -200,6 +238,22 @@ class AudioMetadataHelper: values = cls._values(tags, key) return values[0] if values else None + @classmethod + def _first_of(cls, tags: Any, *keys: str) -> Optional[str]: + """按顺序返回多个音频标签中的第一个非空值。""" + for key in keys: + if value := cls._first(tags, key): + return value + return None + + @staticmethod + def _normalize_musicbrainz_id(value: Optional[str]) -> Optional[str]: + """校验并规范化音频标签中的 MusicBrainz UUID。""" + try: + return str(UUID(str(value))) + except (AttributeError, TypeError, ValueError): + return None + @staticmethod def _number_pair(value: Optional[str]) -> tuple[Optional[int], Optional[int]]: """解析 track/disc 标签中的当前编号和总数。""" diff --git a/app/modules/acoustid/__init__.py b/app/modules/acoustid/__init__.py new file mode 100644 index 000000000..49d83bcc2 --- /dev/null +++ b/app/modules/acoustid/__init__.py @@ -0,0 +1,389 @@ +import asyncio +import json +import shutil +import subprocess +import threading +import time +from collections import OrderedDict +from pathlib import Path +from typing import Any, Optional, Tuple, Union +from uuid import UUID + +from app.core.config import settings +from app.log import logger +from app.modules import _ModuleBase +from app.schemas.types import ModuleType, OtherModulesType +from app.utils.http import AsyncRequestUtils, RequestUtils + + +class AcoustIdModule(_ModuleBase): + """通过 Chromaprint 本地指纹和 AcoustID API 识别 MusicBrainz Recording ID。""" + + _base_url = "https://api.acoustid.org/v2/lookup" + _minimum_score = 0.9 + _request_interval = 0.34 + _fingerprint_timeout = 60 + _cache_max = 1024 + _request_lock = threading.Lock() + _last_request_at = 0.0 + + def __init__(self) -> None: + """初始化 fpcalc 路径和进程内文件指纹识别缓存。""" + super().__init__() + self._fpcalc_path: Optional[str] = None + self._cache: OrderedDict[tuple[str, int, int], Optional[str]] = OrderedDict() + self._cache_lock = threading.Lock() + + def init_module(self) -> None: + """定位 fpcalc 工具并清空可能过期的文件识别缓存。""" + self._fpcalc_path = shutil.which("fpcalc") + with self._cache_lock: + self._cache.clear() + if not self._fpcalc_path: + logger.warning("AcoustID 已配置，但未找到 fpcalc，音频指纹识别将跳过") + + def init_setting(self) -> Tuple[str, Union[str, bool]]: + """仅在配置 AcoustID 应用 API Key 后启用模块。""" + return "ACOUSTID_API_KEY", True + + def stop(self) -> None: + """停止模块并释放进程内文件识别缓存。""" + with self._cache_lock: + self._cache.clear() + + def test(self) -> Tuple[bool, str]: + """检查 API Key、fpcalc 和 AcoustID API 的基础连通性。""" + if not str(settings.ACOUSTID_API_KEY or "").strip(): + return False, "AcoustID API Key 未配置" + if not self._fpcalc_path: + return False, "未找到 fpcalc，请先安装 Chromaprint" + response = RequestUtils( + ua=settings.USER_AGENT, + proxies=settings.PROXY, + timeout=15, + ).get_res( + url=self._base_url, + params={"client": settings.ACOUSTID_API_KEY, "format": "json"}, + ) + if response is None: + return False, "AcoustID 网络连接失败" + try: + if response.status_code >= 500: + return False, f"AcoustID 服务异常：{response.status_code}" + return True, "" + finally: + response.close() + + @staticmethod + def get_name() -> str: + """返回模块展示名称。""" + return "AcoustID" + + @staticmethod + def get_type() -> ModuleType: + """返回模块所属的其它音乐能力类型。""" + return ModuleType.Other + + @staticmethod + def get_subtype() -> OtherModulesType: + """返回 AcoustID 模块子类型。""" + return OtherModulesType.AcoustId + + @staticmethod + def get_priority() -> int: + """返回音频指纹识别模块执行优先级。""" + return 0 + + def identify_music_by_fingerprint(self, path: Path) -> Optional[str]: + """读取本地音频指纹并返回高置信匹配的 MusicBrainz Recording ID。""" + file_path = Path(path) + if not self._fpcalc_path or not file_path.is_file(): + return None + cache_key = self._file_cache_key(file_path) + if cache_key: + found, cached_id = self._get_cached(cache_key) + if found: + return cached_id + fingerprint = self._generate_fingerprint(file_path) + recording_id = ( + self._lookup_recording_id(*fingerprint) if fingerprint else None + ) + if cache_key: + self._set_cached(cache_key, recording_id) + return recording_id + + async def async_identify_music_by_fingerprint( + self, + path: Path, + ) -> Optional[str]: + """异步读取音频指纹并返回高置信匹配的 MusicBrainz Recording ID。""" + file_path = Path(path) + if not self._fpcalc_path or not file_path.is_file(): + return None + cache_key = self._file_cache_key(file_path) + if cache_key: + found, cached_id = self._get_cached(cache_key) + if found: + return cached_id + fingerprint = await self._async_generate_fingerprint(file_path) + recording_id = ( + await self._async_lookup_recording_id(*fingerprint) + if fingerprint + else None + ) + if cache_key: + self._set_cached(cache_key, recording_id) + return recording_id + + @staticmethod + def _file_cache_key(path: Path) -> Optional[tuple[str, int, int]]: + """按规范路径、文件大小和修改时间构造可自动失效的缓存键。""" + try: + stat = path.stat() + return str(path.resolve()), stat.st_size, stat.st_mtime_ns + except OSError: + return None + + def _get_cached( + self, + cache_key: tuple[str, int, int], + ) -> tuple[bool, Optional[str]]: + """读取并触摸文件识别 LRU 缓存，区分未缓存与已缓存未命中。""" + with self._cache_lock: + if cache_key not in self._cache: + return False, None + value = self._cache.pop(cache_key) + self._cache[cache_key] = value + return True, value + + def _set_cached( + self, + cache_key: tuple[str, int, int], + recording_id: Optional[str], + ) -> None: + """写入文件识别 LRU 缓存并淘汰最早使用的条目。""" + with self._cache_lock: + self._cache.pop(cache_key, None) + self._cache[cache_key] = recording_id + while len(self._cache) > self._cache_max: + self._cache.popitem(last=False) + + def _generate_fingerprint(self, path: Path) -> Optional[tuple[int, str]]: + """调用 fpcalc 生成 AcoustID 查询所需的完整时长和压缩指纹。""" + try: + result = subprocess.run( + [self._fpcalc_path, "-json", str(path)], + capture_output=True, + text=True, + check=False, + timeout=self._fingerprint_timeout, + ) + except (OSError, subprocess.TimeoutExpired) as err: + logger.warning(f"生成音频指纹失败：{path} - {err}") + return None + if result.returncode != 0: + logger.warning( + f"生成音频指纹失败：{path} - fpcalc 退出码 {result.returncode}" + ) + return None + return self._parse_fingerprint_output(path, result.stdout) + + async def _async_generate_fingerprint( + self, + path: Path, + ) -> Optional[tuple[int, str]]: + """异步调用 fpcalc 生成 AcoustID 查询所需的指纹。""" + process = None + try: + process = await asyncio.create_subprocess_exec( + self._fpcalc_path, + "-json", + str(path), + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, _ = await asyncio.wait_for( + process.communicate(), + timeout=self._fingerprint_timeout, + ) + except asyncio.TimeoutError: + if process and process.returncode is None: + process.kill() + await process.communicate() + logger.warning(f"生成音频指纹超时：{path}") + return None + except asyncio.CancelledError: + if process and process.returncode is None: + process.kill() + await process.communicate() + raise + except OSError as err: + logger.warning(f"生成音频指纹失败：{path} - {err}") + return None + if process.returncode != 0: + logger.warning( + f"生成音频指纹失败：{path} - fpcalc 退出码 {process.returncode}" + ) + return None + return self._parse_fingerprint_output( + path, + stdout.decode("utf-8", errors="replace"), + ) + + @staticmethod + def _parse_fingerprint_output( + path: Path, + output: str, + ) -> Optional[tuple[int, str]]: + """解析 fpcalc JSON 输出中的音频时长和压缩指纹。""" + try: + payload = json.loads(output) + duration = round(float(payload.get("duration") or 0)) + fingerprint = str(payload.get("fingerprint") or "").strip() + except (AttributeError, TypeError, ValueError) as err: + logger.warning(f"fpcalc 输出解析失败：{path} - {err}") + return None + if duration <= 0 or not fingerprint: + logger.warning(f"fpcalc 未返回有效音频指纹：{path}") + return None + return duration, fingerprint + + @classmethod + def _reserve_request_delay(cls) -> float: + """为同步和异步 AcoustID 请求统一预留下一个发送时间。""" + with cls._request_lock: + now = time.monotonic() + request_at = max(now, cls._last_request_at + cls._request_interval) + cls._last_request_at = request_at + return max(0.0, request_at - now) + + @classmethod + def _wait_for_rate_limit(cls) -> None: + """同步等待 AcoustID 公共接口的已预留请求时间。""" + if delay := cls._reserve_request_delay(): + time.sleep(delay) + + @classmethod + async def _async_wait_for_rate_limit(cls) -> None: + """异步等待 AcoustID 公共接口的已预留请求时间。""" + if delay := cls._reserve_request_delay(): + await asyncio.sleep(delay) + + def _lookup_recording_id( + self, + duration: int, + fingerprint: str, + ) -> Optional[str]: + """查询 AcoustID 指纹库并提取 MusicBrainz Recording ID。""" + api_key = str(settings.ACOUSTID_API_KEY or "").strip() + if not api_key: + return None + self._wait_for_rate_limit() + response = RequestUtils( + ua=settings.USER_AGENT, + proxies=settings.PROXY, + timeout=30, + ).post_res( + url=self._base_url, + data={ + "client": api_key, + "duration": duration, + "fingerprint": fingerprint, + "meta": "recordingids", + "format": "json", + }, + ) + if response is None: + logger.warning("AcoustID 指纹查询失败：无响应") + return None + try: + if response.status_code != 200: + logger.warning(f"AcoustID 指纹查询失败：HTTP {response.status_code}") + return None + payload = response.json() + return self._select_recording_id(payload) + except (TypeError, ValueError) as err: + logger.warning(f"AcoustID 响应解析失败：{err}") + return None + finally: + response.close() + + async def _async_lookup_recording_id( + self, + duration: int, + fingerprint: str, + ) -> Optional[str]: + """异步查询 AcoustID 指纹库并提取 MusicBrainz Recording ID。""" + api_key = str(settings.ACOUSTID_API_KEY or "").strip() + if not api_key: + return None + await self._async_wait_for_rate_limit() + response = await AsyncRequestUtils( + ua=settings.USER_AGENT, + proxies=settings.PROXY, + timeout=30, + ).post_res( + url=self._base_url, + data={ + "client": api_key, + "duration": duration, + "fingerprint": fingerprint, + "meta": "recordingids", + "format": "json", + }, + ) + if response is None: + logger.warning("AcoustID 指纹查询失败：无响应") + return None + try: + if response.status_code != 200: + logger.warning(f"AcoustID 指纹查询失败：HTTP {response.status_code}") + return None + return self._select_recording_id(response.json()) + except (TypeError, ValueError) as err: + logger.warning(f"AcoustID 响应解析失败：{err}") + return None + finally: + await response.aclose() + + @classmethod + def _select_recording_id(cls, payload: Any) -> Optional[str]: + """按匹配分从 AcoustID 响应中选择首个有效 MusicBrainz Recording ID。""" + if not isinstance(payload, dict) or payload.get("status") != "ok": + return None + results = payload.get("results") or [] + ranked = sorted( + (item for item in results if isinstance(item, dict)), + key=lambda item: cls._score(item.get("score")), + reverse=True, + ) + for item in ranked: + score = cls._score(item.get("score")) + if score < cls._minimum_score: + break + for recording in item.get("recordings") or []: + recording_id = cls._normalize_recording_id( + recording.get("id") if isinstance(recording, dict) else None + ) + if recording_id: + logger.info( + f"AcoustID 指纹命中 MusicBrainz：{recording_id}，匹配度 {score:.3f}" + ) + return recording_id + return None + + @staticmethod + def _score(value: Any) -> float: + """将 AcoustID 匹配分安全转换为零到一之间的浮点数。""" + try: + return max(0.0, min(float(value), 1.0)) + except (TypeError, ValueError): + return 0.0 + + @staticmethod + def _normalize_recording_id(value: Any) -> Optional[str]: + """校验并规范化 MusicBrainz UUID，拒绝异常外部响应进入详情路径。""" + try: + return str(UUID(str(value))) + except (AttributeError, TypeError, ValueError): + return None diff --git a/app/modules/douban/__init__.py b/app/modules/douban/__init__.py index 4485010d1..088a436bf 100644 --- a/app/modules/douban/__init__.py +++ b/app/modules/douban/__init__.py @@ -63,7 +63,7 @@ class DoubanModule(_ModuleBase): @staticmethod def get_music_source() -> str: - """返回多源音乐识别使用的数据源标识。""" + """返回音乐识别使用的数据源标识。""" return DoubanModule._music_source @staticmethod diff --git a/app/modules/musicbrainz/__init__.py b/app/modules/musicbrainz/__init__.py index d955d96d1..36d369124 100644 --- a/app/modules/musicbrainz/__init__.py +++ b/app/modules/musicbrainz/__init__.py @@ -1,10 +1,10 @@ +import asyncio import re import threading import time from difflib import SequenceMatcher from typing import Any, Iterable, Optional, Tuple, Union -from fastapi.concurrency import run_in_threadpool from requests import Session from app.core.cache import cached @@ -26,7 +26,7 @@ from app.schemas.types import ( MediaType, ModuleType, ) -from app.utils.http import RequestUtils +from app.utils.http import AsyncRequestUtils, RequestUtils from app.utils.media import is_media_source_selected from app.utils.zhconv import convert as zhconv_convert @@ -123,7 +123,7 @@ class MusicBrainzModule(_ModuleBase): @staticmethod def get_music_source() -> str: - """返回多源音乐识别使用的数据源标识。""" + """返回音乐识别使用的数据源标识。""" return MusicBrainzModule._source @staticmethod @@ -177,6 +177,30 @@ class MusicBrainzModule(_ModuleBase): return results return [] + async def _async_search_recordings( + self, + meta: MetaMusic, + limit: int, + ) -> list[MusicInfo]: + """异步按音频标签条件搜索 Recording 候选。""" + for query in self._recording_queries(meta): + payload = await self._async_request_json( + "/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)) + ] + if results: + return results + return [] + @classmethod def _recording_queries(cls, meta: MetaMusic) -> list[str]: """构造 Recording 检索式阶梯，由严到宽逐级放宽避免零命中。 @@ -312,6 +336,30 @@ class MusicBrainzModule(_ModuleBase): return results return [] + async def _async_search_albums( + self, + meta: MetaMusic, + limit: int, + ) -> list[MusicInfo]: + """异步按标题和可选艺术家搜索 Release Group 专辑候选。""" + for query in self._album_queries(meta): + payload = await self._async_request_json( + "/release-group", + params={ + "query": query, + "limit": max(1, min(limit, 100)), + "fmt": "json", + }, + ) + results = [ + album.to_music_info() + for item in (payload or {}).get("release-groups") or [] + if (album := self._release_group_to_album(item)) + ] + if results: + return results + return [] + @classmethod def _album_queries(cls, meta: MetaMusic) -> list[str]: """构造专辑检索式阶梯：专辑名+艺术家 → 仅专辑名 → 去括号/卷号变体。""" @@ -413,6 +461,41 @@ class MusicBrainzModule(_ModuleBase): return None return best_album + async def async_match_music_album( + self, + meta: MetaMusic, + tracks: list[MetaMusic], + limit: int = 5, + ) -> Optional[MusicAlbumInfo]: + """异步按目录线索和曲目特征匹配 MusicBrainz 发行版本。""" + if not tracks: + return None + best_album: Optional[MusicAlbumInfo] = None + best_score = 0.0 + releases = await self._async_search_release_candidates( + meta, + tracks, + limit=limit, + ) + for release in releases: + release_id = release.get("id") + if not release_id: + continue + detail = await self._async_request_json( + f"/release/{release_id}", + params={"inc": "recordings+media+artist-credits", "fmt": "json"}, + ) + 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 + _album_match_threshold = 60.0 def _search_release_candidates( @@ -438,6 +521,33 @@ class MusicBrainzModule(_ModuleBase): break return releases[:limit] + async def _async_search_release_candidates( + self, + meta: MetaMusic, + tracks: list[MetaMusic], + limit: int, + ) -> list[dict[str, Any]]: + """异步按专辑名和曲名线索搜索并去重候选发行版本。""" + releases: list[dict[str, Any]] = [] + seen: set[str] = set() + for query in self._release_queries(meta, tracks): + payload = await self._async_request_json( + "/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) + if len(releases) >= limit: + break + return releases[:limit] + @classmethod def _release_queries(cls, meta: MetaMusic, tracks: list[MetaMusic]) -> list[str]: """构造专辑搜索表达式：优先专辑名+歌手，无专辑线索时用曲名兜底。""" @@ -739,15 +849,56 @@ class MusicBrainzModule(_ModuleBase): mediaid: Optional[str] = None, **kwargs, ) -> Optional[MusicInfo]: - """同步分发到音乐识别的异步版本，避免阻塞共享事件循环。""" - return await run_in_threadpool( - self.recognize_media, + """异步识别 MusicBrainz 音乐详情或按元数据匹配单曲。""" + music_type = kwargs.get("music_type") + if source and source != self._source: + return None + if not isinstance(meta, MetaMusic) and mtype != MediaType.MUSIC and source != self._source: + return None + if not isinstance(meta, MetaMusic): + if source == self._source and mediaid: + return await self.async_recognize_music( + source, + str(mediaid), + music_type=music_type, + ) + return None + resolved_source = source or meta.media_source + resolved_media_id = mediaid 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, - mtype=mtype, - source=source, - mediaid=mediaid, - **kwargs, + candidates, + source=resolved_source or self._source, ) + 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) + result = matched or self._info_from_meta(meta) + self._update_recognize_cache(meta, result) + return result @classmethod def _select_candidate(cls, meta: MetaMusic, candidates: Iterable[MusicInfo], source: str) -> Optional[MusicInfo]: @@ -1015,6 +1166,57 @@ class MusicBrainzModule(_ModuleBase): album = self.music_album(source, media_id) return album.to_music_info() if album else None + async def async_recognize_music( + self, + source: str, + media_id: str, + music_type: Optional[str] = None, + ) -> Optional[MusicInfo]: + """异步按 MusicBrainz 标准 ID 和实体类型获取详情。""" + if source != self._source or not media_id: + return None + if music_type != MUSIC_ENTITY_ALBUM: + payload = await self._async_request_json( + f"/recording/{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: + return None + album = await self._async_music_album(source, media_id) + return album.to_music_info() if album else None + + async def _async_music_album( + self, + source: str, + media_id: str, + ) -> Optional[MusicAlbumInfo]: + """异步按 MusicBrainz Release Group ID 获取专辑详情及曲目。""" + if source != self._source or not media_id: + return None + payload = await self._async_request_json( + f"/release-group/{media_id}", + params={ + "inc": "artists+releases+media+genres+tags+ratings", + "fmt": "json", + }, + ) + if not payload: + return None + album = self._release_group_to_album(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 [], + ) + return album + def music_album(self, source: str, media_id: str) -> Optional[MusicAlbumInfo]: """按 MusicBrainz Release Group ID 获取标准化专辑详情及曲目。""" if source != self._source or not media_id: @@ -1308,6 +1510,28 @@ class MusicBrainzModule(_ModuleBase): tracks.append(info) return tracks + @classmethod + async def _async_album_tracks( + cls, + album: MusicAlbumInfo, + releases: list[dict[str, Any]], + ) -> list[MusicInfo]: + """异步读取专辑代表性发行版本的曲目。""" + release = cls._select_track_release(releases) + if not release.get("id"): + return [] + payload = await cls._async_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 + @classmethod def _track_to_info( cls, @@ -1513,14 +1737,25 @@ class MusicBrainzModule(_ModuleBase): return cls._session @classmethod - def _wait_for_rate_limit(cls) -> None: - """串行控制 MusicBrainz 公共接口的最小请求间隔。""" + def _reserve_request_delay(cls) -> float: + """为同步和异步 MusicBrainz 请求统一预留发送时间。""" with cls._request_lock: now = time.monotonic() - remaining = cls._request_interval - (now - cls._last_request_at) - if remaining > 0: - time.sleep(remaining) - cls._last_request_at = time.monotonic() + request_at = max(now, cls._last_request_at + cls._request_interval) + cls._last_request_at = request_at + return max(0.0, request_at - now) + + @classmethod + def _wait_for_rate_limit(cls) -> None: + """同步等待 MusicBrainz 公共接口的已预留请求时间。""" + if delay := cls._reserve_request_delay(): + time.sleep(delay) + + @classmethod + async def _async_wait_for_rate_limit(cls) -> None: + """异步等待 MusicBrainz 公共接口的已预留请求时间。""" + if delay := cls._reserve_request_delay(): + await asyncio.sleep(delay) @classmethod @cached(maxsize=settings.CONF.musicbrainz, ttl=settings.CONF.meta, skip_none=True) @@ -1575,3 +1810,56 @@ class MusicBrainzModule(_ModuleBase): finally: response.close() return None + + @classmethod + @cached( + maxsize=settings.CONF.musicbrainz, + ttl=settings.CONF.meta, + skip_none=True, + shared_key="_request_json", + ) + async def _async_request_json( + cls, + path: str, + params: Optional[dict[str, Any]] = None, + ) -> Optional[dict[str, Any]]: + """异步请求 MusicBrainz JSON 接口并统一处理限流与响应错误。""" + attempts = cls._busy_retries + 1 + for attempt in range(attempts): + await cls._async_wait_for_rate_limit() + response = await AsyncRequestUtils( + headers={ + "User-Agent": f"{settings.USER_AGENT} (https://github.com/jxxghp/MoviePilot)", + "Accept": "application/json", + }, + proxies=settings.PROXY, + timeout=20, + ).get_res(f"{cls._base_url}{path}", params=params) + if response is None: + return None + status_code = response.status_code + try: + if status_code == 404: + logger.debug(f"MusicBrainz 资源不存在：{path}") + return {} + if status_code == 429 or status_code >= 500: + logger.warning( + f"MusicBrainz 服务繁忙：{status_code} {response.text[:200]}" + ) + if attempt < attempts - 1: + await asyncio.sleep(cls._busy_backoff * (2 ** attempt)) + continue + return None + if status_code != 200: + logger.warning( + f"MusicBrainz 请求失败：{status_code} {response.text[:200]}" + ) + return None + payload = response.json() + return payload if isinstance(payload, dict) else None + except (TypeError, ValueError) as err: + logger.warning(f"MusicBrainz 响应解析失败：{err}") + return None + finally: + await response.aclose() + return None diff --git a/app/modules/theaudiodb/__init__.py b/app/modules/theaudiodb/__init__.py index befd4d73d..57bf82748 100644 --- a/app/modules/theaudiodb/__init__.py +++ b/app/modules/theaudiodb/__init__.py @@ -1,7 +1,5 @@ from typing import Any, Optional, Tuple, Union -from fastapi.concurrency import run_in_threadpool - from app.core.cache import cached from app.core.config import settings from app.core.context import ( @@ -19,7 +17,7 @@ from app.schemas.types import ( MediaType, ModuleType, ) -from app.utils.http import RequestUtils +from app.utils.http import AsyncRequestUtils, RequestUtils from app.utils.media import is_media_source_selected @@ -52,7 +50,7 @@ class TheAudioDbModule(_ModuleBase): @staticmethod def get_music_source() -> str: - """返回多源音乐识别使用的数据源标识。""" + """返回音乐识别使用的数据源标识。""" return TheAudioDbModule._source @staticmethod @@ -136,15 +134,39 @@ class TheAudioDbModule(_ModuleBase): mediaid: Optional[str] = None, **kwargs, ) -> Optional[MusicInfo]: - """在线程池执行 TheAudioDB 同步识别，避免阻塞事件循环。""" - return await run_in_threadpool( - self.recognize_media, + """异步识别 TheAudioDB 音乐详情或按元数据匹配单曲。""" + music_type = kwargs.get("music_type") + if source != self._source: + return None + if not isinstance(meta, MetaMusic): + if mtype == MediaType.MUSIC and mediaid: + return await self.async_recognize_music( + source, + str(mediaid), + music_type=music_type, + ) + return None + resolved_media_id = mediaid or meta.media_id + if resolved_media_id: + return await self.async_recognize_music( + source, + str(resolved_media_id), + music_type=music_type, + ) + if music_type != MUSIC_ENTITY_ALBUM: + matched = self._select_track( + meta, + await self._async_search_tracks(meta), + ) + if matched: + return matched + if music_type == MUSIC_ENTITY_RECORDING: + return None + album = self._select_album( meta, - mtype=mtype, - source=source, - mediaid=mediaid, - **kwargs, + await self._async_search_albums(meta), ) + return album.to_music_info() if album else None def recognize_music( self, @@ -165,6 +187,46 @@ class TheAudioDbModule(_ModuleBase): album = self.music_album(source, media_id) return album.to_music_info() if album else None + async def async_recognize_music( + self, + source: str, + media_id: str, + music_type: Optional[str] = None, + ) -> Optional[MusicInfo]: + """异步按 TheAudioDB 原生 ID 和实体类型获取详情。""" + if source != self._source or not media_id: + 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: + return None + album = await self._async_music_album(source, media_id) + return album.to_music_info() if album else None + + async def _async_music_album( + self, + source: str, + media_id: str, + ) -> Optional[MusicAlbumInfo]: + """异步按 TheAudioDB 专辑 ID 获取标准化专辑详情和曲目。""" + if source != self._source or not media_id: + return None + payload = await self._async_request_json("album.php", {"m": media_id}) + item = self._first_entity(payload, "album", "albums") + if not item: + 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 + def music_album(self, source: str, media_id: str) -> Optional[MusicAlbumInfo]: """按 TheAudioDB 专辑 ID 获取标准化专辑详情和曲目。""" if source != self._source or not media_id: @@ -231,6 +293,21 @@ class TheAudioDbModule(_ModuleBase): if (info := self._track_to_info(item)) ] + async def _async_search_tracks(self, meta: MetaMusic) -> list[MusicInfo]: + """异步使用曲名和艺术家搜索 TheAudioDB 单曲。""" + title = meta.title + if not title: + return [] + params = {"t": title} + if meta.artists: + params["s"] = meta.artists[0] + payload = await self._async_request_json("searchtrack.php", params) + return [ + info + for item in self._entities(payload, "track", "tracks") + if (info := self._track_to_info(item)) + ] + def _search_albums(self, meta: MetaMusic) -> list[MusicAlbumInfo]: """使用专辑名和艺术家搜索 TheAudioDB 专辑。""" album_name = meta.album or meta.title @@ -242,6 +319,23 @@ class TheAudioDbModule(_ModuleBase): payload = self._request_json("searchalbum.php", params) return [self._album_to_info(item) for item in self._entities(payload, "album", "albums")] + async def _async_search_albums( + self, + meta: MetaMusic, + ) -> list[MusicAlbumInfo]: + """异步使用专辑名和艺术家搜索 TheAudioDB 专辑。""" + album_name = meta.album or meta.title + if not album_name: + return [] + params = {"a": album_name} + if meta.artists: + params["s"] = meta.artists[0] + payload = await self._async_request_json("searchalbum.php", params) + return [ + self._album_to_info(item) + for item in self._entities(payload, "album", "albums") + ] + def _search_artists(self, meta: MetaMusic) -> list[MusicArtistInfo]: """使用艺术家线索搜索 TheAudioDB 艺术家。""" name = meta.artists[0] if meta.artists else meta.title @@ -456,6 +550,44 @@ class TheAudioDbModule(_ModuleBase): return None return payload if isinstance(payload, dict) else None + @classmethod + @cached( + maxsize=settings.CONF.theaudiodb, + ttl=settings.CONF.meta, + skip_none=True, + shared_key="_request_json", + ) + async def _async_request_json( + cls, + endpoint: str, + params: Optional[dict[str, Any]] = None, + ) -> Optional[dict[str, Any]]: + """异步请求 TheAudioDB V1 JSON 接口并统一处理错误响应。""" + api_key = str(settings.THEAUDIODB_API_KEY or "").strip() + if not api_key: + logger.warning("TheAudioDB API Key 未配置，跳过请求") + return None + response = await AsyncRequestUtils( + ua=settings.USER_AGENT, + proxies=settings.PROXY, + timeout=30, + ).get_res( + url=f"{cls._base_url}/{api_key}/{endpoint}", + params=params or {}, + ) + if response is None: + return None + try: + if response.status_code != 200: + return None + payload = response.json() + except ValueError as err: + logger.error(f"TheAudioDB 响应解析失败：{str(err)}") + return None + finally: + await response.aclose() + return payload if isinstance(payload, dict) else None + @staticmethod def _entities( payload: Optional[dict[str, Any]], diff --git a/app/schemas/types.py b/app/schemas/types.py index 6a68fd173..adebd1519 100644 --- a/app/schemas/types.py +++ b/app/schemas/types.py @@ -498,6 +498,8 @@ class OtherModulesType(Enum): ListenBrainz = "ListenBrainz" # LRCLIB 歌词 Lrclib = "LRCLIB" + # AcoustID 音频指纹 + AcoustId = "AcoustID" class NameValueEnum(Enum): diff --git a/docker/Dockerfile b/docker/Dockerfile index da4b84f82..3d44fb345 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -45,6 +45,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ lsof \ nano \ unar \ + libchromaprint-tools \ libjemalloc2 \ && dpkg-reconfigure --frontend noninteractive tzdata \ && curl https://rclone.org/install.sh | bash \ diff --git a/docs/cli.md b/docs/cli.md index e9dcf68a4..a24516df1 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -441,6 +441,7 @@ moviepilot config get PORT moviepilot config set PORT 3001 moviepilot config set NGINX_PORT 3000 moviepilot config set API_TOKEN your-token-here +moviepilot config set ACOUSTID_API_KEY your-acoustid-client-key ``` 查看所有可配置项： @@ -458,6 +459,7 @@ moviepilot config describe API_TOKEN --show-secrets - `config list` 显示当前配置值 - `config keys` 显示配置项名称、类型和默认值 +- `ACOUSTID_API_KEY` 内置可用默认值，也可在前端“高级设置 - 媒体”或配置命令中覆盖；本地安装需要系统可执行路径中存在 Chromaprint `fpcalc`，官方 Docker 镜像已内置 - `config describe` 显示单个配置项的类型、默认值和当前值 ## Tool 命令 diff --git a/docs/mcp-api.md b/docs/mcp-api.md index 8dca5f567..454099037 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -132,7 +132,7 @@ FastAPI 的 HTTP 异常在 v1、v2 均统一使用 `message`，不再返回顶 #### 媒体识别 / 整理 -媒体识别、搜索和手动整理内置支持 `themoviedb`、`douban`、`bangumi`、`anilist` 四种数据源，也允许插件处理自定义来源。影视自动识别在未指定来源时优先使用 TMDB；TMDB 未可靠命中后才并发查询其它内置来源，并按标题、类型、年份和季信息选择最佳兜底结果。音乐自动识别会并发比较全部内置音乐来源。手动操作可通过请求级 `source` 或 `media_source` + `media_id` 严格指定单一来源，不修改系统默认值，也不会跨来源兜底。 +媒体识别、搜索和手动整理内置支持 `themoviedb`、`douban`、`bangumi`、`anilist` 四种影视数据源，也允许插件处理自定义来源。影视自动识别在未指定来源时只使用 TMDB，未命中时不会继续查询其它影视源。音乐路径识别严格按 AcoustID 音频指纹、文件标签、文件名三级依次执行；指纹或标签直接提供 MusicBrainz Recording ID 时，会直接查询 MusicBrainz 详情，标签和文件名标题识别也只使用 MusicBrainz。其它元数据源仅在手动操作通过请求级 `source` 或 `media_source` + `media_id` 明确指定时使用，不修改系统默认值，也不会跨来源兜底。 涉及媒体身份的请求统一以 `media_source` + `media_id` 表示本次选定的主身份，同时保留 `tmdbid`、`doubanid`、`bangumiid`、`anilistid` 作为跨数据源映射和旧客户端兼容字段。两者并非两套独立数据流：显式通用主身份优先，专用 ID 用于补全映射和兼容回退。 @@ -190,7 +190,7 @@ AniList 榜单、探索、详情、人物和推荐接口优先通过 `anilist-ch 音乐元数据使用 `MusicMeta` / `MusicInfo` 独立模型。`music_type=recording` 表示单曲，`album` 表示包含多首曲目的完整专辑，`artist` 仅用于浏览；稳定身份分别使用对应的 `musicbrainz:`。单曲和专辑可进入搜索、订阅、下载、整理、刮削和已配置音乐媒体服务器的入库检查，艺术家不能作为订阅或下载目标。 -音乐识别结果同时提供 `audio_format`、`audio_lossless`、`audio_quality`、`bit_depth`、`sample_rate`、`bitrate`、`audio_specs` 和 `audio_quality_score`。本地文件识别读取实际音频流参数，站点资源识别从标题和描述提取声明参数；码率、采样率的存储单位分别为 bps 和 Hz。 +音乐识别结果同时提供 `audio_format`、`audio_lossless`、`audio_quality`、`bit_depth`、`sample_rate`、`bitrate`、`audio_specs` 和 `audio_quality_score`。本地文件识别读取实际音频流参数，并使用 Chromaprint 的 `fpcalc` 在本地生成指纹后查询 AcoustID；音频文件本身不会上传。站点资源识别从标题和描述提取声明参数；码率、采样率的存储单位分别为 bps 和 Hz。 | 方法 | 路径 | 说明 | | :--- | :--- | :--- | diff --git a/skills/moviepilot-api/SKILL.md b/skills/moviepilot-api/SKILL.md index ba94b5a47..1899e510d 100644 --- a/skills/moviepilot-api/SKILL.md +++ b/skills/moviepilot-api/SKILL.md @@ -111,7 +111,7 @@ All endpoints are under the base URL `{MP_HOST}`. Path parameters are shown as ` ### Media Search (13 endpoints) -When a video recognition request omits `source`, MoviePilot validates TMDB first and only then queries `douban`, `bangumi`, and `anilist` concurrently to score a fallback. Providing `source` or a source-native ID keeps recognition strict to that source. Music recognition without `source` compares all built-in music sources concurrently. +When recognition omits `source`, MoviePilot uses TMDB exclusively for video and MusicBrainz exclusively for music. A miss does not trigger another metadata source. Providing `source` or a source-native ID keeps recognition strict to that manually selected source. | Method | Path | Description | |--------|------|-------------| diff --git a/tests/test_acoustid_module.py b/tests/test_acoustid_module.py new file mode 100644 index 000000000..52df8d932 --- /dev/null +++ b/tests/test_acoustid_module.py @@ -0,0 +1,186 @@ +import asyncio +import json +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +from app.core.config import ConfigModel +from app.modules.acoustid import AcoustIdModule +from app.utils.http import AsyncRequestUtils, RequestUtils + + +RECORDING_ID = "38035858-f990-4fbb-b3b2-f2f8b958eeba" + + +class FakeResponse: + """提供 AcoustID 模块测试所需的最小 HTTP 响应接口。""" + + def __init__(self, payload: dict, status_code: int = 200) -> None: + """保存响应数据并记录资源是否被关闭。""" + self.payload = payload + self.status_code = status_code + self.closed = False + + def json(self) -> dict: + """返回预设 JSON 响应。""" + return self.payload + + def close(self) -> None: + """记录响应资源已释放。""" + self.closed = True + + +class FakeAsyncResponse(FakeResponse): + """提供异步 AcoustID 查询所需的响应关闭接口。""" + + async def aclose(self) -> None: + """记录异步响应资源已释放。""" + self.closed = True + + +def test_acoustid_api_key_has_built_in_default(): + """系统应内置可用 AcoustID 应用 Key，同时允许运行配置覆盖。""" + assert ConfigModel.model_fields["ACOUSTID_API_KEY"].default == "b1auxfOzAg" + + +def test_identify_music_by_fingerprint_queries_acoustid_and_caches_result( + tmp_path, + monkeypatch, +): + """有效指纹应请求 recordingids，并按文件状态缓存 MusicBrainz ID。""" + audio_path = tmp_path / "track.flac" + audio_path.write_bytes(b"audio") + module = AcoustIdModule() + module._fpcalc_path = "/usr/bin/fpcalc" + fpcalc = Mock(return_value=SimpleNamespace( + returncode=0, + stdout=json.dumps({"duration": 243.4, "fingerprint": "AQADtM..."}), + )) + response = FakeResponse({ + "status": "ok", + "results": [{ + "score": 0.98, + "recordings": [{"id": RECORDING_ID}], + }], + }) + post_res = Mock(return_value=response) + monkeypatch.setattr("app.modules.acoustid.subprocess.run", fpcalc) + monkeypatch.setattr(RequestUtils, "post_res", post_res) + monkeypatch.setattr(module, "_wait_for_rate_limit", lambda: None) + monkeypatch.setattr("app.modules.acoustid.settings.ACOUSTID_API_KEY", "client-key") + + first = module.identify_music_by_fingerprint(audio_path) + second = module.identify_music_by_fingerprint(audio_path) + + assert first == RECORDING_ID + assert second == RECORDING_ID + fpcalc.assert_called_once_with( + ["/usr/bin/fpcalc", "-json", str(audio_path)], + capture_output=True, + text=True, + check=False, + timeout=60, + ) + post_res.assert_called_once() + assert post_res.call_args.kwargs["url"] == "https://api.acoustid.org/v2/lookup" + assert post_res.call_args.kwargs["data"] == { + "client": "client-key", + "duration": 243, + "fingerprint": "AQADtM...", + "meta": "recordingids", + "format": "json", + } + assert response.closed is True + + +def test_select_recording_id_requires_high_score_and_valid_uuid(): + """低置信结果和异常外部 ID 不得进入 MusicBrainz 详情查询。""" + payload = { + "status": "ok", + "results": [ + {"score": 0.99, "recordings": [{"id": "not-a-uuid"}]}, + {"score": 0.91, "recordings": [{"id": RECORDING_ID}]}, + { + "score": 0.89, + "recordings": [{"id": "b10bbbfc-cf9e-42e0-be17-e2c3e1d2600d"}], + }, + ], + } + + assert AcoustIdModule._select_recording_id(payload) == RECORDING_ID + assert AcoustIdModule._select_recording_id({ + "status": "ok", + "results": [{ + "score": 0.89, + "recordings": [{"id": RECORDING_ID}], + }], + }) is None + + +def test_identify_music_by_fingerprint_skips_missing_fpcalc(tmp_path, monkeypatch): + """缺少 fpcalc 时应静默跳过指纹层，不发起 AcoustID 请求。""" + audio_path = tmp_path / "track.mp3" + audio_path.write_bytes(b"audio") + module = AcoustIdModule() + post_res = Mock() + monkeypatch.setattr(RequestUtils, "post_res", post_res) + + assert module.identify_music_by_fingerprint(Path(audio_path)) is None + post_res.assert_not_called() + + +def test_async_identify_music_by_fingerprint_uses_async_process_and_http( + tmp_path, + monkeypatch, +): + """异步接口应异步执行 fpcalc 与 HTTP 查询，并复用同一响应筛选规则。""" + audio_path = tmp_path / "track.flac" + audio_path.write_bytes(b"audio") + module = AcoustIdModule() + module._fpcalc_path = "/usr/bin/fpcalc" + + class FakeProcess: + """模拟已成功执行的异步 fpcalc 子进程。""" + + returncode = 0 + + async def communicate(self): + """返回 fpcalc JSON 标准输出和空错误输出。""" + payload = json.dumps({ + "duration": 243.4, + "fingerprint": "AQADtM...", + }) + return payload.encode(), b"" + + create_process = AsyncMock(return_value=FakeProcess()) + response = FakeAsyncResponse({ + "status": "ok", + "results": [{ + "score": 0.98, + "recordings": [{"id": RECORDING_ID}], + }], + }) + post_res = AsyncMock(return_value=response) + wait_rate_limit = AsyncMock() + monkeypatch.setattr( + "app.modules.acoustid.asyncio.create_subprocess_exec", + create_process, + ) + monkeypatch.setattr(AsyncRequestUtils, "post_res", post_res) + monkeypatch.setattr(module, "_async_wait_for_rate_limit", wait_rate_limit) + monkeypatch.setattr("app.modules.acoustid.settings.ACOUSTID_API_KEY", "client-key") + + result = asyncio.run(module.async_identify_music_by_fingerprint(audio_path)) + + assert result == RECORDING_ID + create_process.assert_awaited_once_with( + "/usr/bin/fpcalc", + "-json", + str(audio_path), + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + wait_rate_limit.assert_awaited_once() + post_res.assert_awaited_once() + assert post_res.await_args.kwargs["data"]["meta"] == "recordingids" + assert response.closed is True diff --git a/tests/test_agent_music_tools.py b/tests/test_agent_music_tools.py index 5ccd396bf..8a3093cc3 100644 --- a/tests/test_agent_music_tools.py +++ b/tests/test_agent_music_tools.py @@ -87,8 +87,8 @@ def _album() -> MusicInfo: ) -def test_recognize_music_title_uses_media_chain_automatic_sources(): - """Agent 音乐标题识别应进入 MediaChain 自动多源流程，不再固定 MusicBrainz。""" +def test_recognize_music_title_uses_media_chain_primary_source(): + """Agent 音乐标题识别应由 MediaChain 自动选择 MusicBrainz 主数据源。""" expected = _recording() recognize = AsyncMock(return_value=expected) tool = RecognizeMediaTool(session_id="session-1", user_id="10001") diff --git a/tests/test_audio_metadata.py b/tests/test_audio_metadata.py index 04b147bf1..43672a0a1 100644 --- a/tests/test_audio_metadata.py +++ b/tests/test_audio_metadata.py @@ -14,6 +14,9 @@ from app.core.meta.metamusic import ( from app.helper.audio import AudioMetadataHelper +RECORDING_ID = "38035858-f990-4fbb-b3b2-f2f8b958eeba" + + def test_read_audio_metadata_maps_easy_tags(monkeypatch): """音频标签和技术参数应映射为 MetaMusic。""" audio = SimpleNamespace( @@ -26,6 +29,7 @@ def test_read_audio_metadata_maps_easy_tags(monkeypatch): "tracknumber": ["8/13"], "discnumber": ["1/1"], "isrc": ["USQX91300105"], + "musicbrainz_trackid": [RECORDING_ID], }, info=SimpleNamespace( length=369.4, @@ -49,6 +53,8 @@ def test_read_audio_metadata_maps_easy_tags(monkeypatch): assert meta.audio_lossless is True assert meta.audio_quality == "lossless" assert meta.audio_specs == "FLAC · 16-bit · 44.1 kHz · 1,411 kbps" + assert meta.media_source == "musicbrainz" + assert meta.media_id == RECORDING_ID def test_parse_declared_hires_audio_quality_from_resource_title(): @@ -144,6 +150,38 @@ def test_read_audio_metadata_falls_back_to_filename(monkeypatch): assert meta.audio_format == "MP3" +def test_read_audio_tags_does_not_fill_from_filename(monkeypatch): + """标签识别层应只使用标签证据，避免把文件名线索提前混入。""" + audio = SimpleNamespace( + tags={"title": ["Tagged Title"]}, + info=SimpleNamespace(length=180), + ) + monkeypatch.setattr("app.helper.audio.MutagenFile", lambda *_args, **_kwargs: audio) + + meta = AudioMetadataHelper.read_tags(Path("/music/Daft Punk - Get Lucky 2013.flac")) + + assert meta.title == "Tagged Title" + assert meta.artists == [] + assert meta.year is None + + +def test_read_audio_tags_ignores_invalid_musicbrainz_id(monkeypatch): + """异常 MusicBrainz 标签不得进入 ID 详情路径。""" + audio = SimpleNamespace( + tags={ + "title": ["Tagged Title"], + "musicbrainz_trackid": ["../../unexpected"], + }, + info=SimpleNamespace(length=180), + ) + monkeypatch.setattr("app.helper.audio.MutagenFile", lambda *_args, **_kwargs: audio) + + meta = AudioMetadataHelper.read_tags(Path("/music/track.flac")) + + assert meta.media_source is None + assert meta.media_id is None + + def test_remote_path_meta_parses_track_prefix_once(tmp_path): """远程或尚未落盘的音频路径应先剥离曲序，不能把 08 误识别成艺术家。""" audio_path = tmp_path / "Daft Punk - Random Access Memories (2013)" / "08 - Get Lucky.flac" @@ -245,6 +283,8 @@ def test_write_audio_metadata_maps_music_info_to_easy_tags(monkeypatch): success = AudioMetadataHelper.write( Path("/music/08 - Get Lucky.flac"), MusicInfo( + source="musicbrainz", + media_id=RECORDING_ID, title="Get Lucky", artists=["Daft Punk", "Pharrell Williams"], album="Random Access Memories", @@ -261,6 +301,7 @@ def test_write_audio_metadata_maps_music_info_to_easy_tags(monkeypatch): assert audio.tags["title"] == ["Get Lucky"] assert audio.tags["artist"] == ["Daft Punk", "Pharrell Williams"] assert audio.tags["tracknumber"] == ["8/13"] + assert audio.tags["musicbrainz_trackid"] == [RECORDING_ID] def test_write_audio_metadata_can_embed_cover_without_rewriting_tags(monkeypatch): diff --git a/tests/test_music_album_match.py b/tests/test_music_album_match.py index 829c342bf..92680a09a 100644 --- a/tests/test_music_album_match.py +++ b/tests/test_music_album_match.py @@ -1,3 +1,6 @@ +import asyncio +from unittest.mock import AsyncMock, Mock + import pytest from app.chain.media import MediaChain @@ -164,6 +167,60 @@ def test_recognize_album_directory_maps_files(tmp_path, music_chain, monkeypatch assert calls["count"] == 1 +def test_async_recognize_album_directory_calls_async_module( + tmp_path, + music_chain, + monkeypatch, +): + """异步目录识别应直接调用模块异步接口，并兼容单个专辑返回值。""" + album_dir = tmp_path / "周杰伦 - 七里香 (2004)" + album_dir.mkdir() + files = [] + for index, (name, _length) in enumerate(ALBUM_TRACKS): + file = album_dir / f"{index + 1:02d}.{name}.wav" + file.write_bytes(b"RIFF") + files.append(file) + + album = MusicAlbumInfo( + source="musicbrainz", + media_id="rg-1", + title="七里香", + artists=["周杰伦"], + tracks=[ + MusicInfo( + source="musicbrainz", + media_id=f"rec-{index + 1}", + title=name, + artists=["周杰伦"], + album="七里香", + track_number=index + 1, + duration=length, + ) + for index, (name, length) in enumerate(ALBUM_TRACKS) + ], + ) + async_run_module = AsyncMock(return_value=album) + run_module = Mock(side_effect=AssertionError("异步目录识别不应调用同步模块接口")) + monkeypatch.setattr(music_chain, "async_run_module", async_run_module) + monkeypatch.setattr(music_chain, "run_module", run_module) + monkeypatch.setattr( + music_chain, + "_read_album_path_metas", + lambda _files: _local_tracks(), + ) + + matched = asyncio.run(music_chain.async_recognize_album_directory(album_dir)) + + assert [matched[str(file.resolve())].media_id for file in files] == [ + "rec-1", + "rec-2", + "rec-3", + ] + async_run_module.assert_awaited_once() + assert async_run_module.await_args.args == ("async_match_music_album",) + run_module.assert_not_called() + + def test_recognize_album_directory_skips_single_file(tmp_path, music_chain, monkeypatch): """单文件目录不走专辑匹配，交给单曲识别链路。""" album_dir = tmp_path / "单曲" @@ -240,3 +297,32 @@ def test_recognize_music_by_path_falls_back_to_album_match(tmp_path, monkeypatch assert info.album == "七里香" # 本地音频参数应保留在识别结果中 assert meta.audio_format == "WAV" + + +def test_async_music_album_fallback_calls_async_directory_match(tmp_path, monkeypatch): + """异步路径回退应等待目录异步识别，不得调用同步目录识别。""" + album_dir = tmp_path / "七里香" + album_dir.mkdir() + file = album_dir / "01.我的地盘.wav" + file.write_bytes(b"RIFF") + matched_info = MusicInfo( + source="musicbrainz", + media_id="rec-1", + title="我的地盘", + ) + async_recognize = AsyncMock( + return_value={str(file.resolve()): matched_info} + ) + sync_recognize = Mock(side_effect=AssertionError("异步回退不应调用同步目录识别")) + monkeypatch.setattr( + MusicChain, + "async_recognize_album_directory", + async_recognize, + ) + monkeypatch.setattr(MusicChain, "recognize_album_directory", sync_recognize) + + result = asyncio.run(MediaChain._async_music_album_dir_fallback(file)) + + assert result is matched_info + async_recognize.assert_awaited_once_with(album_dir) + sync_recognize.assert_not_called() diff --git a/tests/test_music_chain.py b/tests/test_music_chain.py index 746d4764c..a58618608 100644 --- a/tests/test_music_chain.py +++ b/tests/test_music_chain.py @@ -1,10 +1,10 @@ +from pathlib import Path from unittest.mock import AsyncMock, Mock from app.chain.media import MediaChain from app.chain.music import MusicChain from app.core.context import MusicAlbumInfo, MusicArtistInfo, MusicInfo from app.core.meta import MetaMusic -from app.helper.audio import AudioMetadataHelper from app.modules.musicbrainz import MusicBrainzModule from app.schemas.types import MediaType @@ -311,7 +311,17 @@ def test_async_recognize_by_path_reads_local_audio_tags(tmp_path, monkeypatch): ) chain = MediaChain() recognize = AsyncMock(return_value=info) - monkeypatch.setattr(AudioMetadataHelper, "read", lambda path: meta) + filename_meta = MetaMusic(title="02. 眼泪成诗") + monkeypatch.setattr( + MusicChain, + "read_path_evidence", + Mock(return_value=(meta, meta, filename_meta)), + ) + monkeypatch.setattr( + MusicChain, + "async_identify_by_fingerprint", + AsyncMock(return_value=None), + ) monkeypatch.setattr(chain, "async_recognize_media", recognize) import asyncio @@ -322,59 +332,42 @@ def test_async_recognize_by_path_reads_local_audio_tags(tmp_path, monkeypatch): assert recognized_meta is meta assert recognized_info is info - recognize.assert_awaited_once_with(meta=meta, source=None) + recognize.assert_awaited_once_with( + meta=meta, + source=None, + music_type="recording", + ) -def test_recognize_best_compares_all_sources_and_prefers_stronger_evidence(monkeypatch): - """自动识别应查询全部来源，并让专辑、时长和曲序证据更完整的候选胜出。""" +def test_recognize_best_only_queries_musicbrainz(monkeypatch): + """自动音乐识别只应调用 MusicBrainz 主数据源。""" chain = MusicChain() meta = MetaMusic( - title="Yellow", - artists=["Coldplay"], - album="Parachutes", - duration=269, - track_number=5, + title="晴天", + artists=["周杰伦"], + album="叶惠美", + year=2003, ) - candidates = { - "musicbrainz": MusicInfo( - source="musicbrainz", - media_id="mb-1", - title="Yellow", - artists=["Coldplay"], - album="Greatest Hits", - duration=240, - track_number=1, - ), - "theaudiodb": MusicInfo( - source="theaudiodb", - media_id="adb-1", - title="Yellow", - artists=["Coldplay"], - album="Parachutes", - duration=269, - track_number=5, - ), - "doubanmusic": MusicInfo( - source="doubanmusic", - media_id="db-1:5", - title="Yellow", - artists=["Coldplay"], - album="Parachutes", - ), - } - requested = [] - - def fake_recognize_source(_meta, source, _cache): - """按来源返回候选并记录实际查询顺序。""" - requested.append(source) - return candidates[source] - - monkeypatch.setattr(chain, "_recognize_from_source", fake_recognize_source) + expected = MusicInfo( + source="musicbrainz", + media_id="mb-1", + title="晴天", + artists=["周杰伦"], + album="叶惠美", + year=2003, + ) + recognize_source = Mock(return_value=expected) + monkeypatch.setattr(chain, "recognize_from_source", recognize_source) result = chain.recognize_best(meta) - assert requested == ["musicbrainz", "theaudiodb", "doubanmusic"] - assert result is candidates["theaudiodb"] + assert result is expected + recognize_source.assert_called_once_with( + source="musicbrainz", + meta=meta, + cache=True, + music_type="recording", + ) def test_recognize_from_source_selects_only_declared_music_module(monkeypatch): @@ -414,76 +407,92 @@ def test_recognize_from_source_selects_only_declared_music_module(monkeypatch): ) -def test_recognize_best_uses_source_order_only_for_equal_scores(monkeypatch): - """候选证据完全相同时应按 MusicBrainz、TheAudioDB、豆瓣音乐顺序稳定选择。""" +def test_recognize_best_does_not_fallback_after_musicbrainz_miss(monkeypatch): + """MusicBrainz 未命中时自动识别不得继续请求其它音乐来源。""" chain = MusicChain() meta = MetaMusic(title="晴天", artists=["周杰伦"]) - candidates = { - source: MusicInfo( - source=source, - media_id=f"{source}-1", - title="晴天", - artists=["周杰伦"], - ) - for source in ("musicbrainz", "theaudiodb", "doubanmusic") - } - monkeypatch.setattr( - chain, - "_recognize_from_source", - lambda _meta, source, _cache: candidates[source], - ) + recognize_source = Mock(return_value=None) + monkeypatch.setattr(chain, "recognize_from_source", recognize_source) - assert chain.recognize_best(meta) is candidates["musicbrainz"] + assert chain.recognize_best(meta) is None + recognize_source.assert_called_once() + assert recognize_source.call_args.kwargs["source"] == "musicbrainz" -def test_async_recognize_best_queries_sources_concurrently(monkeypatch): - """异步自动识别应并发查询各来源，而不是串行等待三个远端请求。""" +def test_async_recognize_best_only_queries_musicbrainz(monkeypatch): + """异步自动音乐识别也只应调用 MusicBrainz 主数据源。""" import asyncio chain = MusicChain() - active = 0 - max_active = 0 + meta = MetaMusic(title="晴天", artists=["周杰伦"]) + expected = MusicInfo( + source="musicbrainz", + media_id="recording-1", + title="晴天", + artists=["周杰伦"], + ) + recognize_source = AsyncMock(return_value=expected) + monkeypatch.setattr(chain, "async_recognize_from_source", recognize_source) - async def fake_async_recognize_source(_meta, source, _cache): - """记录同时执行的来源数并返回同分候选。""" - nonlocal active, max_active - active += 1 - max_active = max(max_active, active) - await asyncio.sleep(0.01) - active -= 1 - return MusicInfo( - source=source, - media_id=f"{source}-1", - title="晴天", - artists=["周杰伦"], - ) + result = asyncio.run(chain.async_recognize_best(meta)) + assert result is expected + recognize_source.assert_awaited_once_with( + source="musicbrainz", + meta=meta, + cache=True, + music_type="recording", + ) + + +def test_async_recognize_from_source_calls_module_async_method(monkeypatch): + """单源异步识别必须直接等待模块异步入口。""" + chain = MusicChain() + expected = MusicInfo( + source="musicbrainz", + media_id="recording-1", + title="晴天", + ) + module = Mock() + module.async_recognize_media = AsyncMock(return_value=expected) + module.recognize_media = Mock( + side_effect=AssertionError("异步识别不应调用同步模块方法") + ) monkeypatch.setattr( chain, - "_async_recognize_from_source", - fake_async_recognize_source, + "_music_recognize_module", + Mock(return_value=module), ) - result = asyncio.run( - chain.async_recognize_best(MetaMusic(title="晴天", artists=["周杰伦"])) + import asyncio + + result = asyncio.run(chain._async_recognize_from_source( + MetaMusic(title="晴天"), + "musicbrainz", + True, + )) + + assert result is expected + module.async_recognize_media.assert_awaited_once() + module.recognize_media.assert_not_called() + + +def test_async_identify_by_fingerprint_uses_async_module_contract(monkeypatch): + """指纹异步链路应请求模块的异步方法名。""" + chain = MusicChain() + async_run_module = AsyncMock(return_value="recording-1") + monkeypatch.setattr(chain, "async_run_module", async_run_module) + + import asyncio + + result = asyncio.run(chain.async_identify_by_fingerprint("/music/track.flac")) + + assert result == "recording-1" + async_run_module.assert_awaited_once_with( + "async_identify_music_by_fingerprint", + path=Path("/music/track.flac"), ) - assert max_active == 3 - assert result and result.source == "musicbrainz" - - -def test_recognize_candidate_rejects_wrong_artist_even_when_title_matches(): - """已知艺术家时，同名异人的候选不能仅凭曲名命中。""" - meta = MetaMusic(title="晴天", artists=["周杰伦"]) - candidate = MusicInfo( - source="theaudiodb", - media_id="wrong-1", - title="晴天", - artists=["其他歌手"], - ) - - assert MusicChain._recognize_candidate_score(meta, candidate) is None - def test_async_chart_forwards_album_entity(monkeypatch): """热门专辑探索应把实体类型透传给 ListenBrainz 榜单模块。""" diff --git a/tests/test_music_recognize_routing.py b/tests/test_music_recognize_routing.py index 639c5eaed..6ec9514b6 100644 --- a/tests/test_music_recognize_routing.py +++ b/tests/test_music_recognize_routing.py @@ -15,6 +15,7 @@ from app.core.meta import MetaMusic from app.modules.anilist import AniListModule from app.modules.bangumi import BangumiModule from app.modules.musicbrainz import MusicBrainzModule +from app.modules.theaudiodb import TheAudioDbModule from app.modules.themoviedb import TheMovieDbModule from app.schemas.types import MediaType @@ -103,10 +104,6 @@ def test_media_chain_recognize_by_path_routes_musicbrainz_source_to_music_chain( def test_async_recognize_music_by_path_reads_local_audio_tags(tmp_path, monkeypatch): """本地音频识别应使用内嵌标签补全艺术家、专辑并保留音频质量参数。""" - from unittest.mock import AsyncMock - - from app.helper.audio import AudioMetadataHelper - audio_path = tmp_path / "02. 眼泪成诗.m4a" audio_path.write_bytes(b"audio") meta = MetaMusic( @@ -119,7 +116,17 @@ def test_async_recognize_music_by_path_reads_local_audio_tags(tmp_path, monkeypa info = _music_info() chain = MediaChain() recognize = AsyncMock(return_value=info) - monkeypatch.setattr(AudioMetadataHelper, "read", lambda path: meta) + filename_meta = MetaMusic(title="02. 眼泪成诗") + monkeypatch.setattr( + MusicChain, + "read_path_evidence", + Mock(return_value=(meta, meta, filename_meta)), + ) + monkeypatch.setattr( + MusicChain, + "async_identify_by_fingerprint", + AsyncMock(return_value=None), + ) monkeypatch.setattr(chain, "async_recognize_media", recognize) recognized_meta, recognized_info = asyncio.run( @@ -128,7 +135,150 @@ def test_async_recognize_music_by_path_reads_local_audio_tags(tmp_path, monkeypa assert recognized_meta is meta assert recognized_info is info - recognize.assert_awaited_once_with(meta=meta, source=None) + recognize.assert_awaited_once_with( + meta=meta, + source=None, + music_type="recording", + ) + + +def test_recognize_music_by_path_fingerprint_mbid_skips_later_tiers(monkeypatch): + """AcoustID 命中后应按 MBID 直查，且不再执行标签和文件名匹配。""" + recording_id = "38035858-f990-4fbb-b3b2-f2f8b958eeba" + merged = MetaMusic(title="Get Lucky", audio_format="FLAC") + tag_meta = MetaMusic(title="Tagged Title") + filename_meta = MetaMusic(title="Filename Title") + expected = MusicInfo( + source="musicbrainz", + media_id=recording_id, + title="Get Lucky", + ) + chain = MediaChain() + direct = Mock(return_value=expected) + later_tier = Mock() + monkeypatch.setattr( + MusicChain, + "read_path_evidence", + Mock(return_value=(merged, tag_meta, filename_meta)), + ) + monkeypatch.setattr( + MusicChain, + "identify_by_fingerprint", + Mock(return_value=recording_id), + ) + monkeypatch.setattr(chain, "_recognize_musicbrainz_recording", direct) + monkeypatch.setattr(chain, "_recognize_music_meta_tier", later_tier) + + recognized_meta, recognized_info = chain.recognize_music_by_path("track.flac") + + assert recognized_meta is merged + assert recognized_info is expected + direct.assert_called_once_with(merged, recording_id) + later_tier.assert_not_called() + + +def test_recognize_music_by_path_tag_mbid_skips_multi_source_matching(monkeypatch): + """指纹未命中但标签含 MBID 时应直查详情，不进入多来源标题匹配。""" + recording_id = "38035858-f990-4fbb-b3b2-f2f8b958eeba" + tag_meta = MetaMusic( + title="Tagged Title", + media_source="musicbrainz", + media_id=recording_id, + ) + filename_meta = MetaMusic(title="Filename Title") + expected = MusicInfo( + source="musicbrainz", + media_id=recording_id, + title="Tagged Title", + ) + chain = MediaChain() + direct = Mock(return_value=expected) + generic = Mock() + tier = Mock(wraps=chain._recognize_music_meta_tier) + monkeypatch.setattr( + MusicChain, + "read_path_evidence", + Mock(return_value=(tag_meta, tag_meta, filename_meta)), + ) + monkeypatch.setattr(MusicChain, "identify_by_fingerprint", Mock(return_value=None)) + monkeypatch.setattr(chain, "_recognize_musicbrainz_recording", direct) + monkeypatch.setattr(chain, "recognize_media", generic) + monkeypatch.setattr(chain, "_recognize_music_meta_tier", tier) + + _, recognized_info = chain.recognize_music_by_path("track.flac") + + assert recognized_info is expected + direct.assert_called_once_with(meta=tag_meta, recording_id=recording_id) + generic.assert_not_called() + assert tier.call_count == 1 + assert tier.call_args.kwargs["tier_name"] == "文件标签" + + +def test_recognize_music_by_path_falls_back_from_tags_to_filename(monkeypatch): + """标签层未获得远端身份时应继续使用文件名层，且顺序不可反转。""" + tag_meta = MetaMusic(title="Tagged Title") + filename_meta = MetaMusic(title="Filename Title") + expected = MusicInfo( + source="musicbrainz", + media_id="recording-from-filename", + title="Filename Title", + ) + chain = MediaChain() + recognize = Mock(side_effect=[MusicInfo(title="Offline Tag"), expected]) + monkeypatch.setattr( + MusicChain, + "read_path_evidence", + Mock(return_value=(tag_meta, tag_meta, filename_meta)), + ) + monkeypatch.setattr(MusicChain, "identify_by_fingerprint", Mock(return_value=None)) + monkeypatch.setattr(chain, "recognize_media", recognize) + + _, recognized_info = chain.recognize_music_by_path("track.flac") + + assert recognized_info is expected + assert [call.kwargs["meta"] for call in recognize.call_args_list] == [ + tag_meta, + filename_meta, + ] + assert all( + call.kwargs["music_type"] == "recording" + for call in recognize.call_args_list + ) + + +def test_async_recognize_music_by_path_fingerprint_mbid_skips_later_tiers(monkeypatch): + """异步路径也应在 AcoustID 命中后直查 MBID 并停止后续层级。""" + recording_id = "38035858-f990-4fbb-b3b2-f2f8b958eeba" + merged = MetaMusic(title="Get Lucky") + expected = MusicInfo( + source="musicbrainz", + media_id=recording_id, + title="Get Lucky", + ) + chain = MediaChain() + direct = AsyncMock(return_value=expected) + later_tier = AsyncMock() + monkeypatch.setattr( + MusicChain, + "read_path_evidence", + Mock(return_value=(merged, MetaMusic(), MetaMusic())), + ) + monkeypatch.setattr( + MusicChain, + "async_identify_by_fingerprint", + AsyncMock(return_value=recording_id), + ) + monkeypatch.setattr(chain, "_async_recognize_musicbrainz_recording", direct) + monkeypatch.setattr(chain, "_async_recognize_music_meta_tier", later_tier) + + recognized_meta, recognized_info = asyncio.run( + chain.async_recognize_music_by_path("track.flac") + ) + + assert recognized_meta is merged + assert recognized_info is expected + direct.assert_awaited_once_with(merged, recording_id) + later_tier.assert_not_awaited() def test_musicbrainz_module_recognize_media_ignores_non_music(): @@ -355,22 +505,49 @@ def test_musicbrainz_module_recognize_media_by_music_type_and_media_id(monkeypat def test_musicbrainz_module_async_recognize_media(monkeypatch): - """异步模块入口应在线程池中调用同步实现。""" + """异步 MusicBrainz 识别应直接调用异步检索而不进入同步入口。""" module = MusicBrainzModule() expected = _music_info() - sync_mock = Mock(return_value=expected) + async_search = AsyncMock(return_value=[expected]) + sync_mock = Mock(side_effect=AssertionError("异步识别不应调用同步入口")) + monkeypatch.setattr(module, "_async_search_recordings", async_search) monkeypatch.setattr(module, "recognize_media", sync_mock) result = asyncio.run(module.async_recognize_media( meta=MetaMusic(title="晴天"), mtype=MediaType.MUSIC )) - sync_mock.assert_called_once() + async_search.assert_awaited_once() + sync_mock.assert_not_called() + assert result is expected + + +def test_theaudiodb_module_async_recognize_media(monkeypatch): + """异步 TheAudioDB 识别应直接调用异步检索而不进入同步入口。""" + module = TheAudioDbModule() + expected = MusicInfo( + source="theaudiodb", + media_id="track-1", + title="晴天", + ) + async_search = AsyncMock(return_value=[expected]) + sync_mock = Mock(side_effect=AssertionError("异步识别不应调用同步入口")) + monkeypatch.setattr(module, "_async_search_tracks", async_search) + monkeypatch.setattr(module, "recognize_media", sync_mock) + + result = asyncio.run(module.async_recognize_media( + meta=MetaMusic(title="晴天"), + mtype=MediaType.MUSIC, + source="theaudiodb", + )) + + async_search.assert_awaited_once() + sync_mock.assert_not_called() assert result is expected def test_chain_recognize_media_returns_musicinfo_and_reports_share(): - """自动多源识别最终选出的 MusicInfo 应只上报一次共享识别。""" + """MusicBrainz 自动识别返回的 MusicInfo 应只上报一次共享识别。""" expected = _music_info() chain = MediaChain() with patch.object(MusicChain, "recognize_best", return_value=expected), patch( @@ -384,7 +561,7 @@ def test_chain_recognize_media_returns_musicinfo_and_reports_share(): def test_chain_async_recognize_media_returns_musicinfo_and_reports_share(): - """异步多源识别最终选出的 MusicInfo 应只上报一次共享识别。""" + """异步 MusicBrainz 自动识别结果应只上报一次共享识别。""" expected = _music_info() chain = MediaChain() with patch.object( diff --git a/tests/test_video_recognize_fallback.py b/tests/test_video_recognize_fallback.py index fe327f270..b7d6f5e05 100644 --- a/tests/test_video_recognize_fallback.py +++ b/tests/test_video_recognize_fallback.py @@ -1,57 +1,29 @@ -"""影视自动识别的 TMDB 优先和多源兜底测试。""" +"""影视自动识别主数据源路由回归测试。""" import asyncio -from threading import Barrier +from typing import Optional from app.chain import ChainBase from app.chain.media import MediaChain from app.core.context import MediaInfo -from app.core.metainfo import MetaInfo +from app.core.meta import MetaBase from app.schemas.types import MediaType -def _video_meta( - title: str = "流浪地球", - year: str = "2019", - mtype: MediaType = MediaType.MOVIE, -) -> MetaInfo: - """构造包含明确标题、年份和类型的影视解析信息。""" - meta = MetaInfo(title) - meta.year = year - meta.type = mtype +def _video_meta() -> MetaBase: + """构造未指定数据源的电影元数据。""" + meta = MetaBase("流浪地球 2019") + meta.name = "流浪地球" + meta.year = "2019" + meta.type = MediaType.MOVIE return meta -def _video_info( - source: str, - media_id: int | str, - title: str = "流浪地球", - year: str = "2019", - mtype: MediaType = MediaType.MOVIE, - **kwargs, -) -> MediaInfo: - """构造带指定内置来源原生身份的标准影视信息。""" - identity_fields = { - "themoviedb": {"tmdb_id": int(media_id)}, - "douban": {"douban_id": str(media_id)}, - "bangumi": {"bangumi_id": int(media_id)}, - "anilist": {"anilist_id": int(media_id)}, - } - return MediaInfo( - source=source, - type=mtype, - title=title, - year=year, - **identity_fields[source], - **kwargs, - ) - - -def _module_kwargs(meta: MetaInfo, source: str | None = None) -> dict: - """构造原生识别路由需要的最小参数。""" +def _module_kwargs(meta: MetaBase, source: Optional[str] = None) -> dict: + """构造原生识别路由使用的模块参数。""" return { "meta": meta, - "mtype": meta.type, + "mtype": MediaType.MOVIE, "source": source, "mediaid": None, "tmdbid": None, @@ -63,160 +35,94 @@ def _module_kwargs(meta: MetaInfo, source: str | None = None) -> dict: } -def test_video_auto_recognize_stops_after_reliable_tmdb(monkeypatch) -> None: - """TMDB 可靠命中时不得查询任何影视副源。""" +def test_video_auto_recognize_only_uses_tmdb(monkeypatch) -> None: + """未指定影视来源时必须固定委托 TMDB，失败后也不切换其它来源。""" chain = object.__new__(MediaChain) - meta = _video_meta() - tmdb = _video_info("themoviedb", 550, names=["The Wandering Earth"]) calls = [] - def recognize_source(module_kwargs, source, cache): - """记录单源调用并仅返回 TMDB 测试结果。""" - calls.append((module_kwargs, source, cache)) - return tmdb if source == "themoviedb" else None - - monkeypatch.setattr(chain, "_recognize_video_from_source", recognize_source) - - result = chain._run_native_media_recognize(_module_kwargs(meta), cache=True) - - assert result is tmdb - assert [call[1] for call in calls] == ["themoviedb"] - - -def test_video_auto_recognize_concurrently_scores_fallback_sources(monkeypatch) -> None: - """TMDB 低置信时应并发查询全部副源，并返回评分最高的候选。""" - chain = object.__new__(MediaChain) - meta = _video_meta() - fallback_barrier = Barrier(3, timeout=2) - candidates = { - "douban": _video_info("douban", "26266893"), - "bangumi": _video_info( - "bangumi", 302875, mtype=MediaType.TV - ), - "anilist": _video_info( - "anilist", 105333, title="流浪星球" - ), - } - called_sources = [] - - def recognize_source(module_kwargs, source, cache): - """使用线程屏障证明三个副源不是串行执行。""" - del module_kwargs, cache - called_sources.append(source) - if source == "themoviedb": - return _video_info( - "themoviedb", 999, title="完全不同的电影" - ) - fallback_barrier.wait() - return candidates[source] - - monkeypatch.setattr(chain, "_recognize_video_from_source", recognize_source) - - result = chain._run_native_media_recognize(_module_kwargs(meta), cache=True) - - assert result is candidates["douban"] - assert called_sources[0] == "themoviedb" - assert set(called_sources[1:]) == {"douban", "bangumi", "anilist"} - - -def test_async_video_auto_recognize_concurrently_scores_fallback_sources( - monkeypatch, -) -> None: - """异步自动识别应在 TMDB 失败后并发等待全部影视副源。""" - chain = object.__new__(MediaChain) - meta = _video_meta() - candidates = { - source: _video_info(source, index) - for index, source in enumerate(("douban", "bangumi", "anilist"), start=1) - } - started_sources = set() - all_started = asyncio.Event() - - async def recognize_source(module_kwargs, source, cache): - """等三个副源均开始后再放行，串行实现会触发超时。""" - del module_kwargs, cache - if source == "themoviedb": - return None - started_sources.add(source) - if len(started_sources) == 3: - all_started.set() - await asyncio.wait_for(all_started.wait(), timeout=1) - return candidates[source] - - monkeypatch.setattr( - chain, - "_async_recognize_video_from_source", - recognize_source, - ) - - result = asyncio.run( - chain._async_run_native_media_recognize(_module_kwargs(meta), cache=True) - ) - - assert result is candidates["douban"] - assert started_sources == {"douban", "bangumi", "anilist"} - - -def test_video_candidate_score_uses_requested_season_year() -> None: - """电视剧应使用请求季年份，而不是整部剧首播年份进行可信度判断。""" - meta = _video_meta("测试剧", "2024", MediaType.TV) - meta.begin_season = 2 - candidate = _video_info( - "themoviedb", - 42, - title="测试剧", - year="2020", - mtype=MediaType.TV, - seasons={1: [1], 2: [1]}, - season_years={1: "2020", 2: "2024"}, - ) - - score = MediaChain._video_candidate_score(meta, candidate, MediaType.TV) - - assert score == 100 - - -def test_video_candidate_score_rejects_type_and_year_conflicts() -> None: - """类型冲突或明显年份冲突必须在进入副源排名前淘汰。""" - meta = _video_meta() - wrong_type = _video_info( - "bangumi", 1, mtype=MediaType.TV - ) - wrong_year = _video_info( - "douban", 2, year="2023" - ) - - assert MediaChain._video_candidate_score(meta, wrong_type) is None - assert MediaChain._video_candidate_score(meta, wrong_year) is None - - -def test_video_explicit_source_bypasses_automatic_fallback(monkeypatch) -> None: - """显式影视来源应继续走通用严格单源分发，不进入自动策略。""" - chain = object.__new__(MediaChain) - meta = _video_meta() - expected = _video_info("douban", 26266893) - generic_calls = [] - def generic_recognize(_self, module_kwargs, cache): - """记录父类通用分发调用并返回显式来源结果。""" - generic_calls.append((module_kwargs, cache)) - return expected - - def unexpected_auto(*_args, **_kwargs): - """显式来源误入自动策略时立即使测试失败。""" - raise AssertionError("显式来源不应进入自动影视识别") + """记录父类模块分发参数并模拟 TMDB 未命中。""" + calls.append((module_kwargs, cache)) + return None monkeypatch.setattr( ChainBase, "_run_native_media_recognize", generic_recognize, ) - monkeypatch.setattr(chain, "_recognize_video_best", unexpected_auto) result = chain._run_native_media_recognize( - _module_kwargs(meta, source="douban"), + _module_kwargs(_video_meta()), + cache=True, + ) + + assert result is None + assert len(calls) == 1 + assert calls[0][0]["source"] == "themoviedb" + + +def test_async_video_auto_recognize_only_uses_tmdb(monkeypatch) -> None: + """异步未指定影视来源时也只委托 TMDB 原生识别入口。""" + chain = object.__new__(MediaChain) + expected = MediaInfo( + source="themoviedb", + media_id="550", + tmdb_id=550, + title="流浪地球", + year="2019", + type=MediaType.MOVIE, + ) + calls = [] + + async def generic_recognize(_self, module_kwargs, cache): + """记录异步父类模块分发参数并返回 TMDB 结果。""" + calls.append((module_kwargs, cache)) + return expected + + monkeypatch.setattr( + ChainBase, + "_async_run_native_media_recognize", + generic_recognize, + ) + + result = asyncio.run(chain._async_run_native_media_recognize( + _module_kwargs(_video_meta()), + cache=True, + )) + + assert result is expected + assert len(calls) == 1 + assert calls[0][0]["source"] == "themoviedb" + + +def test_video_explicit_source_is_preserved(monkeypatch) -> None: + """手工指定影视来源时必须保持严格单源分发。""" + chain = object.__new__(MediaChain) + expected = MediaInfo( + source="douban", + media_id="26266893", + douban_id="26266893", + title="流浪地球", + year="2019", + type=MediaType.MOVIE, + ) + calls = [] + + def generic_recognize(_self, module_kwargs, cache): + """记录显式来源并返回预设结果。""" + calls.append((module_kwargs, cache)) + return expected + + monkeypatch.setattr( + ChainBase, + "_run_native_media_recognize", + generic_recognize, + ) + + result = chain._run_native_media_recognize( + _module_kwargs(_video_meta(), source="douban"), cache=True, ) assert result is expected - assert generic_calls[0][0]["source"] == "douban" + assert calls[0][0]["source"] == "douban"