diff --git a/app/chain/lyrics.py b/app/chain/lyrics.py index e0168c61b..804a641c5 100644 --- a/app/chain/lyrics.py +++ b/app/chain/lyrics.py @@ -1,76 +1,178 @@ +"""歌词候选聚合与择优处理链。""" + import time -from typing import Any, Optional, Union +from dataclasses import dataclass +from enum import Enum +from typing import Any, Generator, Optional, Union, cast from app.chain.base import ChainBase from app.domain.context import MusicInfo, MusicLyrics from app.domain.meta.metamusic import MetaMusic +class _LyricsAction(Enum): + """歌词状态机允许调用的模块接口。""" + + LEGACY = "music_lyrics" + CANDIDATES = "music_lyrics_candidates" + + +@dataclass(frozen=True, slots=True) +class _LyricsRequest: + """保存歌词聚合的初始候选、截止状态和本地兜底。""" + + candidates: tuple[MusicLyrics, ...] + fallback: Optional[MusicLyrics] + budget_exceeded: bool + + +@dataclass(frozen=True, slots=True) +class _LyricsOutcome: + """保存聚合后的稳定候选顺序及预算状态。""" + + candidates: tuple[MusicLyrics, ...] + budget_exceeded: bool = False + + +def _normalize_lyrics_results(value: Any) -> list[MusicLyrics]: + """把模块的单结果、列表或字典结果统一转换为候选列表。""" + values = value if isinstance(value, list) else [value] + normalized = [] + for item in values: + if isinstance(item, MusicLyrics): + normalized.append(item) + elif isinstance(item, dict): + normalized.append(MusicLyrics.from_dict(item)) + return normalized + + +def _deduplicate_lyrics(candidates: list[MusicLyrics]) -> list[MusicLyrics]: + """按来源身份和内容去重，并保留首次出现位置。""" + selected: dict[tuple[str, str, str], MusicLyrics] = {} + for candidate in candidates: + if not candidate or not (candidate.content or candidate.instrumental): + continue + current = selected.get(candidate.identity_key) + if current is None or ( + candidate.match_score, + candidate.quality_rank, + candidate.provider_priority, + ) > ( + current.match_score, + current.quality_rank, + current.provider_priority, + ): + selected[candidate.identity_key] = candidate + return list(selected.values()) + + +def _build_lyrics_request( + music: Union[MetaMusic, MusicInfo], + local_candidates: Optional[list[MusicLyrics]], + deadline: Optional[float], +) -> _LyricsRequest: + """构造一次歌词聚合请求，截止时间只在进入模块调用前判定。""" + fallback_text = str(getattr(music, "lyrics", None) or "").strip() + fallback = ( + MusicLyrics( + provider="theaudiodb", + plain_lyrics=fallback_text, + match_score=85, + provider_priority=10, + ) + if fallback_text + else None + ) + return _LyricsRequest( + candidates=tuple(local_candidates or ()), + fallback=fallback, + budget_exceeded=deadline is not None and time.monotonic() >= deadline, + ) + + +def _lyrics_plan( + request: _LyricsRequest, +) -> Generator[_LyricsAction, Any, _LyricsOutcome]: + """按旧接口、新接口和本地兜底的唯一顺序聚合歌词。""" + candidates = list(request.candidates) + if request.budget_exceeded: + return _LyricsOutcome( + candidates=tuple(_deduplicate_lyrics(candidates)), + budget_exceeded=True, + ) + legacy = yield _LyricsAction.LEGACY + candidates.extend(_normalize_lyrics_results(legacy)) + results = yield _LyricsAction.CANDIDATES + candidates.extend(_normalize_lyrics_results(results)) + if request.fallback: + candidates.append(request.fallback) + return _LyricsOutcome(candidates=tuple(_deduplicate_lyrics(candidates))) + + class LyricsChain(ChainBase): """聚合本地、插件和宿主歌词候选，并按匹配度与内容质量择优。""" def get_music_lyrics_candidates( - self, - music: Union[MetaMusic, MusicInfo], - local_candidates: Optional[list[MusicLyrics]] = None, + self, + music: Union[MetaMusic, MusicInfo], + local_candidates: Optional[list[MusicLyrics]] = None, ) -> list[MusicLyrics]: """获取所有标准化歌词候选，同时兼容旧版单结果插件接口。""" - candidates = list(local_candidates or []) - if self.deadline is not None and time.monotonic() >= self.deadline: + request = _build_lyrics_request(music, local_candidates, self.deadline) + plan = _lyrics_plan(request) + outcome = _LyricsOutcome(candidates=()) + try: + action = next(plan) + while True: + value = ( + self.run_module("music_lyrics", music=music) + if action is _LyricsAction.LEGACY + else self.run_module("music_lyrics_candidates", music=music) + ) + action = plan.send(value) + except StopIteration as completed: + outcome = cast(_LyricsOutcome, completed.value) + if outcome.budget_exceeded: self.budget_exceeded = True - return self._deduplicate(candidates) - legacy = self.run_module("music_lyrics", music=music) - candidates.extend(self._normalize_results(legacy)) - results = self.run_module("music_lyrics_candidates", music=music) - candidates.extend(self._normalize_results(results)) - - fallback = str(getattr(music, "lyrics", None) or "").strip() - if fallback: - candidates.append(MusicLyrics( - provider="theaudiodb", - plain_lyrics=fallback, - match_score=85, - provider_priority=10, - )) - return self._deduplicate(candidates) + return list(outcome.candidates) def get_music_lyrics( - self, - music: Union[MetaMusic, MusicInfo], - local_candidates: Optional[list[MusicLyrics]] = None, + self, + music: Union[MetaMusic, MusicInfo], + local_candidates: Optional[list[MusicLyrics]] = None, ) -> Optional[MusicLyrics]: """返回匹配可信且质量最高的歌词候选。""" candidates = self.get_music_lyrics_candidates(music, local_candidates) return self._select_best(candidates) async def async_get_music_lyrics_candidates( - self, - music: Union[MetaMusic, MusicInfo], - local_candidates: Optional[list[MusicLyrics]] = None, + self, + music: Union[MetaMusic, MusicInfo], + local_candidates: Optional[list[MusicLyrics]] = None, ) -> list[MusicLyrics]: """异步聚合新旧模块候选，并复用同步路径的本地兜底规则。""" - candidates = list(local_candidates or []) - if self.deadline is not None and time.monotonic() >= self.deadline: + request = _build_lyrics_request(music, local_candidates, self.deadline) + plan = _lyrics_plan(request) + outcome = _LyricsOutcome(candidates=()) + try: + action = next(plan) + while True: + value = ( + await self.async_run_module("music_lyrics", music=music) + if action is _LyricsAction.LEGACY + else await self.async_run_module("music_lyrics_candidates", music=music) + ) + action = plan.send(value) + except StopIteration as completed: + outcome = cast(_LyricsOutcome, completed.value) + if outcome.budget_exceeded: self.budget_exceeded = True - return self._deduplicate(candidates) - legacy = await self.async_run_module("music_lyrics", music=music) - candidates.extend(self._normalize_results(legacy)) - results = await self.async_run_module("music_lyrics_candidates", music=music) - candidates.extend(self._normalize_results(results)) - fallback = str(getattr(music, "lyrics", None) or "").strip() - if fallback: - candidates.append(MusicLyrics( - provider="theaudiodb", - plain_lyrics=fallback, - match_score=85, - provider_priority=10, - )) - return self._deduplicate(candidates) + return list(outcome.candidates) async def async_get_music_lyrics( - self, - music: Union[MetaMusic, MusicInfo], - local_candidates: Optional[list[MusicLyrics]] = None, + self, + music: Union[MetaMusic, MusicInfo], + local_candidates: Optional[list[MusicLyrics]] = None, ) -> Optional[MusicLyrics]: """异步返回质量最高的标准歌词候选。""" candidates = await self.async_get_music_lyrics_candidates(music, local_candidates) @@ -90,37 +192,6 @@ class LyricsChain(ChainBase): ), ) - @classmethod - def _normalize_results(cls, value: Any) -> list[MusicLyrics]: - """把模块的单结果、列表或字典结果统一转换为候选列表。""" - values = value if isinstance(value, list) else [value] - normalized = [] - for item in values: - if isinstance(item, MusicLyrics): - normalized.append(item) - elif isinstance(item, dict): - normalized.append(MusicLyrics.from_dict(item)) - return normalized - - @staticmethod - def _deduplicate(candidates: list[MusicLyrics]) -> list[MusicLyrics]: - """按来源身份和内容去重，重复项保留质量与匹配度更高的一条。""" - selected: dict[tuple[str, str, str], MusicLyrics] = {} - for candidate in candidates: - if not candidate or not (candidate.content or candidate.instrumental): - continue - current = selected.get(candidate.identity_key) - if current is None or ( - candidate.match_score, - candidate.quality_rank, - candidate.provider_priority, - ) > ( - current.match_score, - current.quality_rank, - current.provider_priority, - ): - selected[candidate.identity_key] = candidate - return list(selected.values()) def __init__(self, deadline: Optional[float] = None) -> None: """保存批次查询截止时间，防止单张专辑长期占用刮削任务。""" super().__init__() diff --git a/app/chain/media/path.py b/app/chain/media/path.py index cd7d0665f..5379eb4b9 100644 --- a/app/chain/media/path.py +++ b/app/chain/media/path.py @@ -1,7 +1,9 @@ """音频证据、单曲层级与统一路径识别 owner。""" +from dataclasses import dataclass +from enum import Enum from pathlib import Path -from typing import Optional, Tuple, Union +from typing import Any, Generator, Optional, Tuple, TypeGuard, Union, cast from app.application.audio import AudioMetadataHelper from app.application.configuration import get_chain_runtime_config_snapshot @@ -9,6 +11,7 @@ from app.chain.acoustid import AcoustIdChain from app.chain.media.contract import _MediaOwnerBase from app.domain.context import ( Context, + MediaInfo, MusicInfo, ) from app.domain.media import is_music_media_source @@ -28,6 +31,206 @@ def _is_regular_file(path: Path) -> bool: return path.exists() and path.is_file() +class _MusicTierActionKind(Enum): + """音乐证据层允许执行的外部识别动作。""" + + DIRECT = "direct" + SEARCH = "search" + + +@dataclass(frozen=True, slots=True) +class _MusicTierAction: + """描述一次音乐证据层 I/O，不携带可变执行状态。""" + + kind: _MusicTierActionKind + meta: MetaMusic + recording_id: Optional[str] = None + + +@dataclass(frozen=True, slots=True) +class _MusicTierOutcome: + """保存证据层命中结果及其日志说明。""" + + info: Optional[MusicInfo] = None + message: Optional[str] = None + + +class _MusicPathActionKind(Enum): + """音乐路径识别状态机的稳定动作顺序。""" + + FINGERPRINT = "fingerprint" + TAG = "tag" + FILENAME = "filename" + ALBUM = "album" + + +@dataclass(frozen=True, slots=True) +class _MusicPathAction: + """描述音乐路径状态机的一层识别动作。""" + + kind: _MusicPathActionKind + meta: Optional[MetaMusic] = None + tier_name: Optional[str] = None + + +class _PathRoute(Enum): + """统一路径识别入口的业务路由。""" + + MUSIC = "music" + VIDEO = "video" + + +@dataclass(frozen=True, slots=True) +class _PathRecognitionRequest: + """保存同步和异步路径识别共同使用的请求参数。""" + + path: str + route: _PathRoute + media_source: Optional[MediaSource] + episode_group: Optional[str] + obtain_images: bool + + +def _has_remote_music_identity( + info: Optional[MusicInfo], +) -> TypeGuard[MusicInfo]: + """判断音乐结果是否已获得可终止回退的远程身份。""" + return bool(info and info.media_source and info.media_id) + + +def _musicbrainz_recording_meta(meta: MetaMusic, recording_id: str) -> MetaMusic: + """构造 MusicBrainz Recording 直查使用的独立身份副本。""" + identity_meta = MetaMusic.from_dict(meta.to_dict()) + identity_meta.media_source = MediaSource.MusicBrainz + identity_meta.media_id = recording_id + return identity_meta + + +def _without_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 + + +def _merge_music_audio_quality(info: MusicInfo, meta: MetaMusic) -> MusicInfo: + """将本地文件的实际音频参数合并到音乐识别结果。""" + for key in ( + "audio_format", + "audio_lossless", + "bit_depth", + "sample_rate", + "bitrate", + ): + value = getattr(meta, key, None) + if value is not None: + setattr(info, key, value) + return info + + +def _finalize_music_path_info( + meta: MetaMusic, + info: Optional[MusicInfo], +) -> MusicInfo: + """统一远端命中和本地兜底的音频质量合并。""" + return _merge_music_audio_quality(info or MusicInfo.from_meta(meta), meta) + + +def _music_tier_plan( + meta: Optional[MetaMusic], + media_source: Optional[MediaSource], + tier_name: str, +) -> Generator[_MusicTierAction, Optional[MusicInfo], _MusicTierOutcome]: + """按 MBID 直查再标题搜索的固定顺序生成证据层动作。""" + if not meta: + return _MusicTierOutcome() + normalized_source = normalize_media_source(media_source) + search_meta = meta + if meta.media_source == MediaSource.MusicBrainz and meta.media_id: + if normalized_source in (None, MediaSource.MusicBrainz): + direct = yield _MusicTierAction( + kind=_MusicTierActionKind.DIRECT, + meta=meta, + recording_id=str(meta.media_id), + ) + if _has_remote_music_identity(direct): + return _MusicTierOutcome( + info=direct, + message=f"音乐识别命中{tier_name}层 MusicBrainz ID 直查", + ) + search_meta = _without_music_identity(meta) + if not search_meta.title: + return _MusicTierOutcome() + result = yield _MusicTierAction( + kind=_MusicTierActionKind.SEARCH, + meta=search_meta, + ) + if _has_remote_music_identity(result): + return _MusicTierOutcome( + info=result, + message=f"音乐识别命中{tier_name}层：{result.title}", + ) + return _MusicTierOutcome() + + +def _music_path_plan( + tag_meta: Optional[MetaMusic], + filename_meta: Optional[MetaMusic], + media_source: Optional[MediaSource], +) -> Generator[_MusicPathAction, Optional[MusicInfo], Optional[MusicInfo]]: + """生成“指纹→标签→文件名→专辑目录”的唯一回退状态机。""" + normalized_source = normalize_media_source(media_source) + if normalized_source in (None, MediaSource.MusicBrainz): + info = yield _MusicPathAction(kind=_MusicPathActionKind.FINGERPRINT) + if _has_remote_music_identity(info): + return info + info = yield _MusicPathAction( + kind=_MusicPathActionKind.TAG, + meta=tag_meta, + tier_name="文件标签", + ) + if _has_remote_music_identity(info): + return info + info = yield _MusicPathAction( + kind=_MusicPathActionKind.FILENAME, + meta=filename_meta, + tier_name="文件名", + ) + if _has_remote_music_identity(info): + return info + if normalized_source in (None, MediaSource.MusicBrainz): + return (yield _MusicPathAction(kind=_MusicPathActionKind.ALBUM)) + return None + + +def _build_path_recognition_request( + path: str, + media_source: Optional[MediaSource], + episode_group: Optional[str], + obtain_images: bool, + is_music: bool, +) -> _PathRecognitionRequest: + """将路径识别参数投影为同步和异步共用的稳定路由请求。""" + return _PathRecognitionRequest( + path=path, + route=_PathRoute.MUSIC if is_music else _PathRoute.VIDEO, + media_source=media_source, + episode_group=episode_group, + obtain_images=obtain_images, + ) + + +def _path_context( + meta: Any, + info: Optional[Union[MediaInfo, MusicInfo]], +) -> Context: + """统一成功与失败路径的 Context 投影。""" + if info is not None: + return Context(meta_info=meta, media_info=info) + return Context(meta_info=meta) + + class MediaPathOwner(_MediaOwnerBase): """音频证据、单曲层级与统一路径识别 owner。""" @@ -52,24 +255,17 @@ class MediaPathOwner(_MediaOwnerBase): @staticmethod def _merge_music_audio_quality(info: MusicInfo, meta: MetaMusic) -> MusicInfo: """将本地文件的实际音频参数合并到远端音乐身份识别结果。""" - for key in ("audio_format", "audio_lossless", "bit_depth", "sample_rate", "bitrate"): - value = getattr(meta, key, None) - if value is not None: - setattr(info, key, value) - return info + return _merge_music_audio_quality(info, meta) @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 + return _without_music_identity(meta) @staticmethod def _is_remote_music_info(info: Optional[MusicInfo]) -> bool: """判断音乐识别结果是否携带可复用的远程身份。""" - return bool(info and info.media_source and info.media_id) + return _has_remote_music_identity(info) def _recognize_musicbrainz_recording( self, @@ -77,9 +273,7 @@ class MediaPathOwner(_MediaOwnerBase): recording_id: str, ) -> Optional[MusicInfo]: """按已知 MusicBrainz Recording ID 直接读取单曲详情。""" - identity_meta = MetaMusic.from_dict(meta.to_dict()) - identity_meta.media_source = MediaSource.MusicBrainz - identity_meta.media_id = recording_id + identity_meta = _musicbrainz_recording_meta(meta, recording_id) return self.recognize_music_from_source( media_source=MediaSource.MusicBrainz, meta=identity_meta, @@ -93,9 +287,7 @@ class MediaPathOwner(_MediaOwnerBase): recording_id: str, ) -> Optional[MusicInfo]: """异步按已知 MusicBrainz Recording ID 直接读取单曲详情。""" - identity_meta = MetaMusic.from_dict(meta.to_dict()) - identity_meta.media_source = MediaSource.MusicBrainz - identity_meta.media_id = recording_id + identity_meta = _musicbrainz_recording_meta(meta, recording_id) return await self.async_recognize_music_from_source( media_source=MediaSource.MusicBrainz, meta=identity_meta, @@ -110,31 +302,29 @@ class MediaPathOwner(_MediaOwnerBase): tier_name: str, ) -> Optional[MusicInfo]: """识别单个音乐元数据证据层，标签中的 MBID 优先直查。""" - if not meta: - return None - normalized_source = normalize_media_source(media_source) - search_meta = meta - if meta.media_source == MediaSource.MusicBrainz and meta.media_id: - if normalized_source in (None, MediaSource.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, - media_source=media_source, - music_type=MUSIC_ENTITY_RECORDING, - ) - if isinstance(result, MusicInfo) and self._is_remote_music_info(result): - logger.info(f"音乐识别命中{tier_name}层：{result.title}") - return result - return None + plan = _music_tier_plan(meta, media_source, tier_name) + outcome = _MusicTierOutcome() + try: + action = next(plan) + while True: + if action.kind is _MusicTierActionKind.DIRECT: + result = self._recognize_musicbrainz_recording( + meta=action.meta, + recording_id=action.recording_id or "", + ) + else: + candidate = self.recognize_media( + meta=action.meta, + media_source=media_source, + music_type=MUSIC_ENTITY_RECORDING, + ) + result = candidate if isinstance(candidate, MusicInfo) else None + action = plan.send(result) + except StopIteration as completed: + outcome = cast(_MusicTierOutcome, completed.value) + if outcome.message: + logger.info(outcome.message) + return outcome.info async def _async_recognize_music_meta_tier( self, @@ -143,31 +333,29 @@ class MediaPathOwner(_MediaOwnerBase): tier_name: str, ) -> Optional[MusicInfo]: """异步识别单个音乐元数据证据层，标签中的 MBID 优先直查。""" - if not meta: - return None - normalized_source = normalize_media_source(media_source) - search_meta = meta - if meta.media_source == MediaSource.MusicBrainz and meta.media_id: - if normalized_source in (None, MediaSource.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, - media_source=media_source, - music_type=MUSIC_ENTITY_RECORDING, - ) - if isinstance(result, MusicInfo) and self._is_remote_music_info(result): - logger.info(f"音乐识别命中{tier_name}层：{result.title}") - return result - return None + plan = _music_tier_plan(meta, media_source, tier_name) + outcome = _MusicTierOutcome() + try: + action = next(plan) + while True: + if action.kind is _MusicTierActionKind.DIRECT: + result = await self._async_recognize_musicbrainz_recording( + meta=action.meta, + recording_id=action.recording_id or "", + ) + else: + candidate = await self.async_recognize_media( + meta=action.meta, + media_source=media_source, + music_type=MUSIC_ENTITY_RECORDING, + ) + result = candidate if isinstance(candidate, MusicInfo) else None + action = plan.send(result) + except StopIteration as completed: + outcome = cast(_MusicTierOutcome, completed.value) + if outcome.message: + logger.info(outcome.message) + return outcome.info def _music_album_dir_fallback( self, @@ -206,32 +394,28 @@ class MediaPathOwner(_MediaOwnerBase): ) -> Tuple[MetaMusic, MusicInfo]: """按指纹、文件标签、文件名三级顺序识别本地音乐。""" meta, tag_meta, filename_meta = AudioMetadataHelper.read_evidence(Path(path)) - info = None - normalized_source = normalize_media_source(media_source) - if normalized_source in (None, MediaSource.MusicBrainz): - recording_id = AcoustIdChain().identify_music_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, - media_source=media_source, - tier_name="文件标签", - ) - if not self._is_remote_music_info(info): - info = self._recognize_music_meta_tier( - meta=filename_meta, - media_source=media_source, - tier_name="文件名", - ) - result = self._merge_music_audio_quality(info or self._music_info_from_path_meta(meta), meta) - if not result.media_source and media_source in (None, MediaSource.MusicBrainz): - # 单曲搜索未命中时，按所在目录做专辑级匹配兑底 - matched = self._music_album_dir_fallback(path) - if matched: - result = self._merge_music_audio_quality(matched, meta) + plan = _music_path_plan(tag_meta, filename_meta, media_source) + info: Optional[MusicInfo] = None + try: + action = next(plan) + while True: + if action.kind is _MusicPathActionKind.FINGERPRINT: + recording_id = AcoustIdChain().identify_music_by_fingerprint(path) + info = self._recognize_musicbrainz_recording(meta, recording_id) if recording_id else None + if self._is_remote_music_info(info): + logger.info("音乐识别命中 AcoustID 指纹层，已跳过标签和文件名识别") + elif action.kind is _MusicPathActionKind.ALBUM: + info = self._music_album_dir_fallback(path) + else: + info = self._recognize_music_meta_tier( + meta=action.meta, + media_source=media_source, + tier_name=action.tier_name or "", + ) + action = plan.send(info) + except StopIteration as completed: + info = cast(Optional[MusicInfo], completed.value) + result = _finalize_music_path_info(meta, info) return meta, self._simplify_recognized_music_info(result) async def async_recognize_music_by_path( @@ -244,35 +428,35 @@ class MediaPathOwner(_MediaOwnerBase): AudioMetadataHelper.read_evidence, Path(path), ) - info = None - normalized_source = normalize_media_source(media_source) - if normalized_source in (None, MediaSource.MusicBrainz): - recording_id = await AcoustIdChain().async_identify_music_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, - media_source=media_source, - tier_name="文件标签", - ) - if not self._is_remote_music_info(info): - info = await self._async_recognize_music_meta_tier( - meta=filename_meta, - media_source=media_source, - tier_name="文件名", - ) - result = self._merge_music_audio_quality(info or self._music_info_from_path_meta(meta), meta) - if not result.media_source and media_source in (None, MediaSource.MusicBrainz): - # 单曲搜索未命中时，按所在目录做专辑级匹配兑底 - matched = await self._async_music_album_dir_fallback(path) - if matched: - result = self._merge_music_audio_quality(matched, meta) + plan = _music_path_plan(tag_meta, filename_meta, media_source) + info: Optional[MusicInfo] = None + try: + action = next(plan) + while True: + if action.kind is _MusicPathActionKind.FINGERPRINT: + recording_id = await AcoustIdChain().async_identify_music_by_fingerprint(path) + info = ( + await self._async_recognize_musicbrainz_recording( + meta, + recording_id, + ) + if recording_id + else None + ) + if self._is_remote_music_info(info): + logger.info("音乐识别命中 AcoustID 指纹层，已跳过标签和文件名识别") + elif action.kind is _MusicPathActionKind.ALBUM: + info = await self._async_music_album_dir_fallback(path) + else: + info = await self._async_recognize_music_meta_tier( + meta=action.meta, + media_source=media_source, + tier_name=action.tier_name or "", + ) + action = plan.send(info) + except StopIteration as completed: + info = cast(Optional[MusicInfo], completed.value) + result = _finalize_music_path_info(meta, info) return meta, self._simplify_recognized_music_info(result) def _is_music_path_request(self, path: str, media_source: Optional[MediaSource]) -> bool: @@ -295,25 +479,30 @@ class MediaPathOwner(_MediaOwnerBase): :param obtain_images: 是否补充图片 :return: 识别上下文 """ - logger.info(f"开始识别媒体信息，文件：{path} ...") - # 音频文件直接在本链完成标签读取、搜索匹配与专辑目录兜底，封面等图片由刮削环节补充 - if self._is_music_path_request(path, media_source): - music_meta, music_info = self.recognize_music_by_path(path, media_source=media_source) - return Context(meta_info=music_meta, media_info=music_info) - file_path = Path(path) - # 元数据 - file_meta = MetaInfoPath(file_path) - mediainfo = self._recognize_with_fallback_by_meta( - metainfo=file_meta, + request = _build_path_recognition_request( + path=path, media_source=media_source, episode_group=episode_group, obtain_images=obtain_images, + is_music=self._is_music_path_request(path, media_source), + ) + logger.info(f"开始识别媒体信息，文件：{request.path} ...") + if request.route is _PathRoute.MUSIC: + music_meta, music_info = self.recognize_music_by_path( + request.path, + media_source=request.media_source, + ) + return _path_context(music_meta, music_info) + file_meta = MetaInfoPath(Path(request.path)) + mediainfo = self._recognize_with_fallback_by_meta( + metainfo=file_meta, + media_source=request.media_source, + episode_group=request.episode_group, + obtain_images=request.obtain_images, ) if not mediainfo: - logger.warn(f"{path} 未识别到媒体信息") - return Context(meta_info=file_meta) - # 返回上下文 - return Context(meta_info=file_meta, media_info=mediainfo) + logger.warn(f"{request.path} 未识别到媒体信息") + return _path_context(file_meta, mediainfo) async def async_recognize_by_path( self, @@ -331,22 +520,27 @@ class MediaPathOwner(_MediaOwnerBase): :param obtain_images: 是否补充图片 :return: 识别上下文 """ - logger.info(f"开始识别媒体信息，文件：{path} ...") - # 音频文件直接在本链完成标签读取、搜索匹配与专辑目录兜底，封面等图片由刮削环节补充 - if self._is_music_path_request(path, media_source): - music_meta, music_info = await self.async_recognize_music_by_path(path, media_source=media_source) - return Context(meta_info=music_meta, media_info=music_info) - file_path = Path(path) - # 元数据 - file_meta = MetaInfoPath(file_path) - mediainfo = await self._async_recognize_with_fallback_by_meta( - metainfo=file_meta, + request = _build_path_recognition_request( + path=path, media_source=media_source, episode_group=episode_group, obtain_images=obtain_images, + is_music=self._is_music_path_request(path, media_source), + ) + logger.info(f"开始识别媒体信息，文件：{request.path} ...") + if request.route is _PathRoute.MUSIC: + music_meta, music_info = await self.async_recognize_music_by_path( + request.path, + media_source=request.media_source, + ) + return _path_context(music_meta, music_info) + file_meta = MetaInfoPath(Path(request.path)) + mediainfo = await self._async_recognize_with_fallback_by_meta( + metainfo=file_meta, + media_source=request.media_source, + episode_group=request.episode_group, + obtain_images=request.obtain_images, ) if not mediainfo: - logger.warn(f"{path} 未识别到媒体信息") - return Context(meta_info=file_meta) - # 返回上下文 - return Context(meta_info=file_meta, media_info=mediainfo) + logger.warn(f"{request.path} 未识别到媒体信息") + return _path_context(file_meta, mediainfo) diff --git a/app/chain/media/search.py b/app/chain/media/search.py index 2819b6c51..8d6fac37e 100644 --- a/app/chain/media/search.py +++ b/app/chain/media/search.py @@ -1,5 +1,6 @@ """媒体搜索入口 owner。""" +from dataclasses import dataclass from typing import List, Optional, Tuple from app.chain.media.contract import _MediaOwnerBase @@ -15,6 +16,44 @@ from app.schemas.types import ( ) +@dataclass(frozen=True, slots=True) +class _MediaSearchRequest: + """保存同步和异步搜索共同使用的输入投影。""" + + content: str + meta: MetaBase + + +def _build_media_search_request(title: str) -> _MediaSearchRequest: + """将搜索文本一次性投影为规范元数据，避免双入口规则漂移。""" + mtype, _, season_num, episode_num, year, content = title_rules.parse_search_keyword(title) + content = content or title + meta = MetaInfo(content) + if not meta.name: + meta.cn_name = content + if mtype: + meta.type = mtype + if season_num: + meta.begin_season = season_num + if episode_num: + meta.begin_episode = episode_num + if year: + meta.year = year + return _MediaSearchRequest(content=content, meta=meta) + + +def _finish_media_search( + request: _MediaSearchRequest, + medias: Optional[List[MediaInfo]], +) -> Tuple[MetaBase, List[MediaInfo]]: + """统一空结果与成功结果投影，并保持既有日志语义。""" + if not medias: + logger.warn(f"{request.meta.name} 没有找到对应的媒体信息！") + return request.meta, [] + logger.info(f"{request.content} 搜索到 {len(medias)} 条相关媒体信息") + return request.meta, medias + + class MediaSearchOwner(_MediaOwnerBase): """媒体搜索入口 owner。""" @@ -28,31 +67,13 @@ class MediaSearchOwner(_MediaOwnerBase): :param media_source: 请求级搜索数据源 :return: 识别元数据，媒体信息列表 """ - # 提取要素 - mtype, key_word, season_num, episode_num, year, content = title_rules.parse_search_keyword(title) - # 识别 - content = content or title - meta = MetaInfo(content) - if not meta.name: - meta.cn_name = content - # 合并信息 - if mtype: - meta.type = mtype - if season_num: - meta.begin_season = season_num - if episode_num: - meta.begin_episode = episode_num - if year: - meta.year = year - # 开始搜索 - logger.info(f"开始搜索媒体信息：{meta.name}") - medias: Optional[List[MediaInfo]] = self.search_medias(meta=meta, media_source=media_source) - if not medias: - logger.warn(f"{meta.name} 没有找到对应的媒体信息！") - return meta, [] - logger.info(f"{content} 搜索到 {len(medias)} 条相关媒体信息") - # 识别的元数据，媒体信息列表 - return meta, medias + request = _build_media_search_request(title) + logger.info(f"开始搜索媒体信息：{request.meta.name}") + medias: Optional[List[MediaInfo]] = self.search_medias( + meta=request.meta, + media_source=media_source, + ) + return _finish_media_search(request, medias) async def async_search( self, title: str, media_source: Optional[MediaSourceSelection] = None @@ -64,28 +85,10 @@ class MediaSearchOwner(_MediaOwnerBase): :param media_source: 请求级搜索数据源 :return: 识别元数据，媒体信息列表 """ - # 提取要素 - mtype, key_word, season_num, episode_num, year, content = title_rules.parse_search_keyword(title) - # 识别 - content = content or title - meta = MetaInfo(content) - if not meta.name: - meta.cn_name = content - # 合并信息 - if mtype: - meta.type = mtype - if season_num: - meta.begin_season = season_num - if episode_num: - meta.begin_episode = episode_num - if year: - meta.year = year - # 开始搜索 - logger.info(f"开始搜索媒体信息：{meta.name}") - medias: Optional[List[MediaInfo]] = await self.async_search_medias(meta=meta, media_source=media_source) - if not medias: - logger.warn(f"{meta.name} 没有找到对应的媒体信息！") - return meta, [] - logger.info(f"{content} 搜索到 {len(medias)} 条相关媒体信息") - # 识别的元数据，媒体信息列表 - return meta, medias + request = _build_media_search_request(title) + logger.info(f"开始搜索媒体信息：{request.meta.name}") + medias: Optional[List[MediaInfo]] = await self.async_search_medias( + meta=request.meta, + media_source=media_source, + ) + return _finish_media_search(request, medias) diff --git a/tests/test_media_path_search_lyrics_parity.py b/tests/test_media_path_search_lyrics_parity.py new file mode 100644 index 000000000..ed13c90f0 --- /dev/null +++ b/tests/test_media_path_search_lyrics_parity.py @@ -0,0 +1,555 @@ +"""媒体搜索、路径识别与歌词聚合的同步异步同形回归。""" + +import asyncio +from pathlib import Path +from unittest.mock import AsyncMock, Mock + +import pytest + +from app.chain.acoustid import AcoustIdChain +from app.chain.lyrics import LyricsChain +from app.chain.media import MediaChain +from app.domain.context import MediaInfo, MusicInfo, MusicLyrics +from app.domain.meta.metamusic import MetaMusic +from app.schemas.types import MediaSource, MediaType + + +def _remote_music(stage: str) -> MusicInfo: + """构造指定回退层命中的标准远端音乐结果。""" + return MusicInfo( + media_source=MediaSource.MusicBrainz, + media_id=f"recording-{stage}", + title=f"Track {stage}", + ) + + +@pytest.mark.parametrize("empty", [False, True]) +def test_media_search_sync_async_share_projection_and_empty_result( + monkeypatch: pytest.MonkeyPatch, + empty: bool, +) -> None: + """搜索双入口应共享要素投影，并将所有空结果规范为新列表。""" + chain = MediaChain() + source = (MediaSource.TMDB,) + medias = [] if empty else [MediaInfo(type=MediaType.TV, title="Result")] + sync_search = Mock(return_value=medias or None) + async_search = AsyncMock(return_value=medias or None) + monkeypatch.setattr( + "app.chain.media.search.title_rules.parse_search_keyword", + Mock( + return_value=( + MediaType.TV, + "ignored", + 2, + 3, + 2024, + "Projected Title", + ) + ), + ) + monkeypatch.setattr(MediaChain, "search_medias", sync_search) + monkeypatch.setattr(MediaChain, "async_search_medias", async_search) + + sync_meta, sync_result = chain.search("raw title", media_source=source) + async_meta, async_result = asyncio.run(chain.async_search("raw title", media_source=source)) + + assert sync_meta is not None and async_meta is not None + assert ( + ( + sync_meta.name, + sync_meta.type, + sync_meta.begin_season, + sync_meta.begin_episode, + sync_meta.year, + ) + == ( + async_meta.name, + async_meta.type, + async_meta.begin_season, + async_meta.begin_episode, + async_meta.year, + ) + == ("Projected Title", MediaType.TV, 2, 3, 2024) + ) + assert sync_result == async_result == medias + assert (sync_result is not async_result) is empty + assert sync_search.call_args.kwargs["media_source"] == source + assert async_search.await_args.kwargs["media_source"] == source + + +def test_media_search_sync_async_propagate_provider_errors( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """搜索提供方异常不得被任一入口伪装为空结果。""" + chain = MediaChain() + monkeypatch.setattr( + MediaChain, + "search_medias", + Mock(side_effect=RuntimeError("sync search failed")), + ) + monkeypatch.setattr( + MediaChain, + "async_search_medias", + AsyncMock(side_effect=RuntimeError("async search failed")), + ) + + with pytest.raises(RuntimeError, match="sync search failed"): + chain.search("Title") + with pytest.raises(RuntimeError, match="async search failed"): + asyncio.run(chain.async_search("Title")) + + +@pytest.mark.parametrize( + ("hit_stage", "expected_order"), + [ + ("fingerprint", ["fingerprint"]), + ("tag", ["fingerprint", "tag"]), + ("filename", ["fingerprint", "tag", "filename"]), + ("album", ["fingerprint", "tag", "filename", "album"]), + (None, ["fingerprint", "tag", "filename", "album"]), + ], +) +def test_music_path_sync_async_follow_one_fallback_state_machine( + monkeypatch: pytest.MonkeyPatch, + hit_stage: str | None, + expected_order: list[str], +) -> None: + """路径识别双入口应在同一层终止，并保持每层调用次数一致。""" + chain = MediaChain() + path = Path("Track.flac") + merged = MetaMusic(title="Track", audio_format="FLAC") + tag_meta = MetaMusic(title="Tagged") + filename_meta = MetaMusic(title="Filename") + sync_order: list[str] = [] + async_order: list[str] = [] + monkeypatch.setattr( + "app.chain.media.path.AudioMetadataHelper.read_evidence", + Mock(return_value=(merged, tag_meta, filename_meta)), + ) + + def sync_fingerprint(_self: AcoustIdChain, _path: Path) -> str | None: + """记录同步指纹层并按场景返回 Recording ID。""" + sync_order.append("fingerprint") + return "fingerprint-id" if hit_stage == "fingerprint" else None + + async def async_fingerprint(_self: AcoustIdChain, _path: Path) -> str | None: + """记录异步指纹层并按场景返回 Recording ID。""" + async_order.append("fingerprint") + return "fingerprint-id" if hit_stage == "fingerprint" else None + + def sync_tier( + _chain: MediaChain, + *, + meta: MetaMusic, + media_source: MediaSource | None, + tier_name: str, + ): + """模拟同步标签或文件名层的远端命中。""" + del meta, media_source + stage = "tag" if tier_name == "文件标签" else "filename" + sync_order.append(stage) + return _remote_music(stage) if hit_stage == stage else None + + async def async_tier( + _chain: MediaChain, + *, + meta: MetaMusic, + media_source: MediaSource | None, + tier_name: str, + ): + """模拟异步标签或文件名层的远端命中。""" + del meta, media_source + stage = "tag" if tier_name == "文件标签" else "filename" + async_order.append(stage) + return _remote_music(stage) if hit_stage == stage else None + + def sync_album(_chain: MediaChain, _path: Path) -> MusicInfo | None: + """记录同步专辑目录兜底。""" + sync_order.append("album") + return _remote_music("album") if hit_stage == "album" else None + + async def async_album(_chain: MediaChain, _path: Path) -> MusicInfo | None: + """记录异步专辑目录兜底。""" + async_order.append("album") + return _remote_music("album") if hit_stage == "album" else None + + monkeypatch.setattr(AcoustIdChain, "identify_music_by_fingerprint", sync_fingerprint) + monkeypatch.setattr(AcoustIdChain, "async_identify_music_by_fingerprint", async_fingerprint) + monkeypatch.setattr( + MediaChain, + "_recognize_musicbrainz_recording", + Mock(return_value=_remote_music("fingerprint")), + ) + monkeypatch.setattr( + MediaChain, + "_async_recognize_musicbrainz_recording", + AsyncMock(return_value=_remote_music("fingerprint")), + ) + monkeypatch.setattr(MediaChain, "_recognize_music_meta_tier", sync_tier) + monkeypatch.setattr(MediaChain, "_async_recognize_music_meta_tier", async_tier) + monkeypatch.setattr(MediaChain, "_music_album_dir_fallback", sync_album) + monkeypatch.setattr(MediaChain, "_async_music_album_dir_fallback", async_album) + + _, sync_info = chain.recognize_music_by_path(path) + _, async_info = asyncio.run(chain.async_recognize_music_by_path(path)) + + assert sync_order == async_order == expected_order + expected_id = f"recording-{hit_stage}" if hit_stage else None + assert sync_info.media_id == async_info.media_id == expected_id + assert sync_info.audio_format == async_info.audio_format == "FLAC" + + +@pytest.mark.parametrize( + ("failed_stage", "expected_order"), + [ + ("fingerprint", ["fingerprint"]), + ("tag", ["fingerprint", "tag"]), + ("filename", ["fingerprint", "tag", "filename"]), + ], +) +def test_music_path_sync_async_stop_at_same_failing_tier( + monkeypatch: pytest.MonkeyPatch, + failed_stage: str, + expected_order: list[str], +) -> None: + """指纹、标签或文件名 I/O 异常时双入口应在相同层停止。""" + chain = MediaChain() + meta = MetaMusic(title="Track") + sync_order: list[str] = [] + async_order: list[str] = [] + monkeypatch.setattr( + "app.chain.media.path.AudioMetadataHelper.read_evidence", + Mock(return_value=(meta, meta, meta)), + ) + + def sync_fingerprint(_self: AcoustIdChain, _path: Path) -> None: + """记录同步指纹动作并按场景抛错。""" + sync_order.append("fingerprint") + if failed_stage == "fingerprint": + raise RuntimeError("sync fingerprint failed") + + async def async_fingerprint(_self: AcoustIdChain, _path: Path) -> None: + """记录异步指纹动作并按场景抛错。""" + async_order.append("fingerprint") + if failed_stage == "fingerprint": + raise RuntimeError("async fingerprint failed") + + def sync_tier( + _chain: MediaChain, + *, + meta: MetaMusic, + media_source: MediaSource | None, + tier_name: str, + ): + """记录同步元数据层并在指定层抛错。""" + del meta, media_source + stage = "tag" if tier_name == "文件标签" else "filename" + sync_order.append(stage) + if failed_stage == stage: + raise RuntimeError(f"sync {stage} failed") + return None + + async def async_tier( + _chain: MediaChain, + *, + meta: MetaMusic, + media_source: MediaSource | None, + tier_name: str, + ): + """记录异步元数据层并在指定层抛错。""" + del meta, media_source + stage = "tag" if tier_name == "文件标签" else "filename" + async_order.append(stage) + if failed_stage == stage: + raise RuntimeError(f"async {stage} failed") + return None + + monkeypatch.setattr(AcoustIdChain, "identify_music_by_fingerprint", sync_fingerprint) + monkeypatch.setattr(AcoustIdChain, "async_identify_music_by_fingerprint", async_fingerprint) + monkeypatch.setattr(MediaChain, "_recognize_music_meta_tier", sync_tier) + monkeypatch.setattr(MediaChain, "_async_recognize_music_meta_tier", async_tier) + + with pytest.raises(RuntimeError, match=f"sync {failed_stage} failed"): + chain.recognize_music_by_path("Track.flac") + with pytest.raises(RuntimeError, match=f"async {failed_stage} failed"): + asyncio.run(chain.async_recognize_music_by_path("Track.flac")) + + assert sync_order == async_order == expected_order + + +@pytest.mark.parametrize( + ("source", "direct_hit", "has_title", "expected_calls"), + [ + (None, True, True, ["direct"]), + (None, False, True, ["direct", "search"]), + (MediaSource.TMDB, False, True, ["search"]), + (None, False, False, ["direct"]), + ], +) +def test_music_meta_tier_sync_async_share_direct_and_search_decisions( + monkeypatch: pytest.MonkeyPatch, + source: MediaSource | None, + direct_hit: bool, + has_title: bool, + expected_calls: list[str], +) -> None: + """标签 MBID 直查、身份清理和标题回退在双入口中必须同形。""" + chain = MediaChain() + meta = MetaMusic( + title="Tagged" if has_title else None, + media_source=MediaSource.MusicBrainz, + media_id="recording-tagged", + ) + sync_calls: list[str] = [] + async_calls: list[str] = [] + direct_result = _remote_music("direct") if direct_hit else None + search_result = _remote_music("search") + + def sync_direct(_chain: MediaChain, **_kwargs) -> MusicInfo | None: + """记录同步 MBID 直查。""" + sync_calls.append("direct") + return direct_result + + async def async_direct(_chain: MediaChain, **_kwargs) -> MusicInfo | None: + """记录异步 MBID 直查。""" + async_calls.append("direct") + return direct_result + + def sync_search(_chain: MediaChain, **kwargs) -> MusicInfo: + """记录同步标题搜索并验证身份已清理。""" + sync_calls.append("search") + assert kwargs["meta"].media_source is None + assert kwargs["meta"].media_id is None + return search_result + + async def async_search(_chain: MediaChain, **kwargs) -> MusicInfo: + """记录异步标题搜索并验证身份已清理。""" + async_calls.append("search") + assert kwargs["meta"].media_source is None + assert kwargs["meta"].media_id is None + return search_result + + monkeypatch.setattr(MediaChain, "_recognize_musicbrainz_recording", sync_direct) + monkeypatch.setattr(MediaChain, "_async_recognize_musicbrainz_recording", async_direct) + monkeypatch.setattr(MediaChain, "recognize_media", sync_search) + monkeypatch.setattr(MediaChain, "async_recognize_media", async_search) + + sync_result = chain._recognize_music_meta_tier(meta, source, "文件标签") + async_result = asyncio.run(chain._async_recognize_music_meta_tier(meta, source, "文件标签")) + + assert sync_calls == async_calls == expected_calls + expected = direct_result if direct_hit else (search_result if has_title else None) + assert sync_result is expected + assert async_result is expected + + +def test_music_album_fallback_sync_async_isolate_directory_errors( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """专辑目录识别异常在同步和异步入口都应降级为空结果。""" + path = tmp_path / "Track.flac" + path.write_bytes(b"audio") + chain = MediaChain() + monkeypatch.setattr( + MediaChain, + "recognize_music_album_directory", + Mock(side_effect=RuntimeError("sync album failed")), + ) + monkeypatch.setattr( + MediaChain, + "async_recognize_music_album_directory", + AsyncMock(side_effect=RuntimeError("async album failed")), + ) + + assert chain._music_album_dir_fallback(path) is None + assert asyncio.run(chain._async_music_album_dir_fallback(path)) is None + + +@pytest.mark.parametrize("recognized", [False, True]) +def test_video_path_sync_async_share_route_and_failure_context( + monkeypatch: pytest.MonkeyPatch, + recognized: bool, +) -> None: + """影视路径双入口应共享路由参数，并在失败时都保留元数据 Context。""" + chain = MediaChain() + info = MediaInfo(type=MediaType.MOVIE, title="Movie") if recognized else None + sync_recognize = Mock(return_value=info) + async_recognize = AsyncMock(return_value=info) + monkeypatch.setattr(MediaChain, "_is_music_path_request", Mock(return_value=False)) + monkeypatch.setattr(MediaChain, "_recognize_with_fallback_by_meta", sync_recognize) + monkeypatch.setattr(MediaChain, "_async_recognize_with_fallback_by_meta", async_recognize) + + sync_context = chain.recognize_by_path( + "Movie.2024.mkv", + media_source=MediaSource.TMDB, + episode_group="group-1", + obtain_images=True, + ) + async_context = asyncio.run( + chain.async_recognize_by_path( + "Movie.2024.mkv", + media_source=MediaSource.TMDB, + episode_group="group-1", + obtain_images=True, + ) + ) + + assert sync_context is not None and async_context is not None + assert sync_context.media_info is async_context.media_info is info + assert sync_context.meta_info.title == async_context.meta_info.title + for call in (sync_recognize.call_args, async_recognize.await_args): + assert call.kwargs["media_source"] == MediaSource.TMDB + assert call.kwargs["episode_group"] == "group-1" + assert call.kwargs["obtain_images"] is True + + +def test_music_path_route_sync_async_skip_video_recognition( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """音乐路由双入口应直接返回音乐 Context，且不调用影视识别。""" + chain = MediaChain() + meta = MetaMusic(title="Track") + info = _remote_music("route") + sync_music = Mock(return_value=(meta, info)) + async_music = AsyncMock(return_value=(meta, info)) + sync_video = Mock() + async_video = AsyncMock() + monkeypatch.setattr(MediaChain, "_is_music_path_request", Mock(return_value=True)) + monkeypatch.setattr(MediaChain, "recognize_music_by_path", sync_music) + monkeypatch.setattr(MediaChain, "async_recognize_music_by_path", async_music) + monkeypatch.setattr(MediaChain, "_recognize_with_fallback_by_meta", sync_video) + monkeypatch.setattr(MediaChain, "_async_recognize_with_fallback_by_meta", async_video) + + sync_context = chain.recognize_by_path("Track.flac", media_source=MediaSource.MusicBrainz) + async_context = asyncio.run(chain.async_recognize_by_path("Track.flac", media_source=MediaSource.MusicBrainz)) + + assert sync_context is not None and async_context is not None + assert sync_context.meta_info is async_context.meta_info is meta + assert sync_context.media_info is async_context.media_info is info + sync_music.assert_called_once_with("Track.flac", media_source=MediaSource.MusicBrainz) + async_music.assert_awaited_once_with("Track.flac", media_source=MediaSource.MusicBrainz) + sync_video.assert_not_called() + async_video.assert_not_awaited() + + +def test_lyrics_sync_async_share_deadline_fallback_dedup_and_call_order( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """歌词双入口应按旧、新、本地顺序聚合，并稳定替换重复候选。""" + low = MusicLyrics(provider="same", plain_lyrics="duplicate", match_score=10) + high = MusicLyrics(provider="same", plain_lyrics="duplicate", match_score=90) + new = MusicLyrics(provider="new", synced_lyrics="[00:01]new", match_score=80) + music = MusicInfo(title="Track", lyrics="fallback") + sync_chain = LyricsChain() + async_chain = LyricsChain() + sync_calls: list[str] = [] + async_calls: list[str] = [] + + def sync_module(method: str, **_kwargs): + """记录同步歌词接口顺序并返回对应候选。""" + sync_calls.append(method) + return high if method == "music_lyrics" else [new] + + async def async_module(method: str, **_kwargs): + """记录异步歌词接口顺序并返回对应候选。""" + async_calls.append(method) + return high if method == "music_lyrics" else [new] + + monkeypatch.setattr(sync_chain, "run_module", sync_module) + monkeypatch.setattr(async_chain, "async_run_module", async_module) + + sync_result = sync_chain.get_music_lyrics_candidates(music, [low]) + async_result = asyncio.run(async_chain.async_get_music_lyrics_candidates(music, [low])) + + assert sync_calls == async_calls == ["music_lyrics", "music_lyrics_candidates"] + assert [item.provider for item in sync_result] == ["same", "new", "theaudiodb"] + assert [item.provider for item in async_result] == ["same", "new", "theaudiodb"] + assert sync_result[0] is async_result[0] is high + + +def test_lyrics_sync_async_deadline_skips_modules_and_local_fallback( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """预算耗尽时双入口只返回现有本地候选，不再调用模块或追加来源兜底。""" + monkeypatch.setattr("app.chain.lyrics.time.monotonic", Mock(return_value=10.0)) + local = MusicLyrics(provider="embedded", plain_lyrics="local") + music = MusicInfo(title="Track", lyrics="fallback") + sync_chain = LyricsChain(deadline=10.0) + async_chain = LyricsChain(deadline=10.0) + sync_run = Mock() + async_run = AsyncMock() + monkeypatch.setattr(sync_chain, "run_module", sync_run) + monkeypatch.setattr(async_chain, "async_run_module", async_run) + + sync_result = sync_chain.get_music_lyrics_candidates(music, [local]) + async_result = asyncio.run(async_chain.async_get_music_lyrics_candidates(music, [local])) + + assert sync_result == async_result == [local] + assert sync_chain.budget_exceeded is async_chain.budget_exceeded is True + sync_run.assert_not_called() + async_run.assert_not_awaited() + + +def test_lyrics_sync_async_empty_modules_without_fallback_return_empty( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """新旧接口均为空且无本地歌词时，双入口应稳定返回空列表。""" + music = MusicInfo(title="Track") + sync_chain = LyricsChain() + async_chain = LyricsChain() + sync_run = Mock(side_effect=[None, []]) + async_run = AsyncMock(side_effect=[None, []]) + monkeypatch.setattr(sync_chain, "run_module", sync_run) + monkeypatch.setattr(async_chain, "async_run_module", async_run) + + assert sync_chain.get_music_lyrics_candidates(music) == [] + assert asyncio.run(async_chain.async_get_music_lyrics_candidates(music)) == [] + assert sync_run.call_count == async_run.await_count == 2 + + +@pytest.mark.parametrize( + ("failed_method", "expected_calls"), + [ + ("music_lyrics", ["music_lyrics"]), + ( + "music_lyrics_candidates", + ["music_lyrics", "music_lyrics_candidates"], + ), + ], +) +def test_lyrics_sync_async_propagate_same_module_exception( + monkeypatch: pytest.MonkeyPatch, + failed_method: str, + expected_calls: list[str], +) -> None: + """旧或新模块接口异常时双入口应在同一步停止且不伪造兜底成功。""" + music = MusicInfo(title="Track", lyrics="fallback") + sync_chain = LyricsChain() + async_chain = LyricsChain() + sync_calls: list[str] = [] + async_calls: list[str] = [] + + def sync_module(method: str, **_kwargs): + """在指定同步歌词接口抛出异常。""" + sync_calls.append(method) + if method == failed_method: + raise RuntimeError("sync failed") + return None + + async def async_module(method: str, **_kwargs): + """在指定异步歌词接口抛出异常。""" + async_calls.append(method) + if method == failed_method: + raise RuntimeError("async failed") + return None + + monkeypatch.setattr(sync_chain, "run_module", sync_module) + monkeypatch.setattr(async_chain, "async_run_module", async_module) + + with pytest.raises(RuntimeError, match="sync failed"): + sync_chain.get_music_lyrics_candidates(music) + with pytest.raises(RuntimeError, match="async failed"): + asyncio.run(async_chain.async_get_music_lyrics_candidates(music)) + + assert sync_calls == async_calls == expected_calls