mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-01 04:47:03 +08:00
refactor(media): unify path search lyrics decisions
This commit is contained in:
jxxghp
@@ -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__()
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
555
tests/test_media_path_search_lyrics_parity.py
Normal file
555
tests/test_media_path_search_lyrics_parity.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user