nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-11-01 04:47:03 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(music): validate bounded fingerprint candidates and recover failed lookups

Browse Source
This commit is contained in:
jxxghp
2026-09-30 07:39:57 +08:00
parent 7e43ea9676
commit 3a7b0c1efc
11 changed files with 681 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
app/application/music/observation.py
View File

@@ -21,6 +21,7 @@ class MusicRecognitionObservation:
status: str = "not_found"
message: str = "没有匹配到足够可靠的音乐信息"
candidates: list[dict[str, Any]] = field(default_factory=list)
fingerprint_candidates: list[dict[str, Any]] = field(default_factory=list)
lock: Any = field(default_factory=RLock, repr=False)
@property
@@ -82,6 +83,14 @@ def music_recognition_diagnostics() -> dict[str, Any]:
return observation.to_dict() if observation is not None else {}
def report_music_fingerprints(candidates: list[dict[str, Any]]) -> None:
"""保留指纹原始候选证据，不把指纹关联提前宣布为已确认录音。"""
observation = _current_observation.get()
if observation is not None:
with observation.lock:
observation.fingerprint_candidates = [dict(item) for item in candidates[:5]]
def music_wait_allowed(delay: float) -> bool:
"""限流和重试等待也计入总预算，不为无法执行的请求继续占用未来时隙。"""
observation = _current_observation.get()

262
app/chain/media/path.py
View File

@@ -5,10 +5,15 @@ from copy import deepcopy
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Any, Generator, Optional, Tuple, TypeGuard, Union, cast
from typing import Any, Awaitable, Callable, Generator, Optional, Tuple, TypeGuard, Union, cast
from app.application.audio import AudioMetadataHelper
from app.application.configuration import get_chain_runtime_config_snapshot
from app.application.music.observation import (
capture_music_recognition,
music_recognition_failed,
report_music_fingerprints,
)
from app.chain.acoustid import AcoustIdChain
from app.chain.media.contract import _MediaOwnerBase
from app.domain.context import (
@@ -19,6 +24,7 @@ from app.domain.context import (
from app.domain.media import is_music_media_source
from app.domain.meta.metamusic import MetaMusic
from app.domain.metainfo import MetaInfoPath
from app.domain.music import music_album_title_is_weak, music_track_title_is_weak
from app.runtime.execution import run_in_threadpool
from app.runtime.log import logger
from app.schemas.media import normalize_media_source
@@ -210,42 +216,46 @@ def _fingerprint_info_matches_evidence(
info: Optional[MusicInfo],
tag_meta: Optional[MetaMusic],
filename_meta: Optional[MetaMusic],
*,
fingerprint_score: Optional[float] = None,
fingerprint_duration: Optional[int] = None,
) -> bool:
"""Require an AcoustID candidate to agree with local textual evidence.
Public AcoustID mappings can point to the wrong MusicBrainz recording even
at a high score. A hit is therefore only authoritative when its title and
version, plus any locally available artist credit, agree with the tags or
parsed filename.
"""
from difflib import SequenceMatcher
"""保留强标签冲突检查；原生高分指纹可用真实时长弥补占位名称或缺失署名。"""
from app.domain.music import ( # pylint: disable=import-outside-toplevel
music_album_matches,
music_artist_matches,
music_base_title,
music_text_key,
music_title_matches,
music_titles,
music_usable_artists,
music_version_matches,
music_year_matches,
)
if not _has_remote_music_identity(info):
return False
primary = tag_meta if tag_meta and tag_meta.title else filename_meta
if not primary or not primary.title:
primary = tag_meta if tag_meta and not music_track_title_is_weak(tag_meta) else filename_meta or tag_meta
if not primary:
return False
artist_evidence = primary.artists or (
weak_title = music_track_title_is_weak(primary)
local_duration = fingerprint_duration or (tag_meta.duration if tag_meta else None) or primary.duration
close_duration = bool(local_duration and info.duration and abs(local_duration - info.duration) <= max(3, min(8, local_duration * 0.025)))
native_strong = fingerprint_score is not None and fingerprint_score >= 0.98 and close_duration
if local_duration and info.duration and abs(local_duration - info.duration) > max(10, local_duration * 0.08):
return False
artist_meta = tag_meta if tag_meta and tag_meta.artists else primary
artist_evidence = music_usable_artists(artist_meta.artists) or music_usable_artists(
filename_meta.artists if filename_meta else []
)
# 只有曲名时，公开 AcoustID 映射中的同名录音无法排除。
# 必须由文件标签、文件名或同目录共识提供艺人证据。
if not artist_evidence:
collective = {music_text_key(value) for value in ("Various Artists", "Various", "VA", "群星", "众艺人", "眾藝人")}
if (fingerprint_score is not None and artist_meta.field_sources.get("artists") in {"directory", "torrent", "album_tags"}
and all(music_text_key(artist) in collective for artist in artist_evidence)):
artist_evidence = []
if not artist_evidence and not native_strong:
return False
if not music_artist_matches(info, artist_evidence):
if artist_evidence and not music_artist_matches(info, artist_evidence):
return False
if not music_version_matches(info, primary):
if (not weak_title or primary.version) and not music_version_matches(info, primary):
return False
if tag_meta and tag_meta is not primary and tag_meta.version and not music_version_matches(info, tag_meta):
return False
standalone_single = _is_standalone_single_evidence(primary)
if (
@@ -253,16 +263,31 @@ def _fingerprint_info_matches_evidence(
and info.album
and not music_album_matches(info, primary.album)
and not standalone_single
and not native_strong
):
return False
if not standalone_single and not music_year_matches(info, primary):
if not native_strong and not standalone_single and not music_year_matches(info, primary):
return False
if music_title_matches(info, primary.title):
if weak_title:
return native_strong
return _fingerprint_title_matches(info, primary.title)
def _fingerprint_title_matches(info: MusicInfo, title_evidence: str) -> bool:
"""在音频与署名已核验后比较曲名别称及展示差异，避免混入发行身份判断。"""
from difflib import SequenceMatcher
from app.domain.music import ( # pylint: disable=import-outside-toplevel
music_base_title,
music_text_key,
music_title_matches,
music_titles,
)
if music_title_matches(info, title_evidence):
return True
# AcoustID is strong audio evidence once the artist agrees. Allow common
# radio/edit/remaster suffixes and very small legacy-tag typos, while still
# rejecting unrelated recordings.
evidence_key = music_text_key(music_base_title(primary.title))
# 上层已核验音频、艺人及版本，允许常见展示后缀与轻微标签拼写差异。
evidence_key = music_text_key(music_base_title(title_evidence))
version_suffix = re.compile(
r"(?:radio|single|version|edit|mix|remix|remaster(?:ed)?|live|acoustic|"
r"demo|mono|stereo|recorded)+"
@@ -275,7 +300,7 @@ def _fingerprint_info_matches_evidence(
if shorter and longer.startswith(shorter) and version_suffix.fullmatch(longer[len(shorter):]):
return True
qualified_evidence_key = music_text_key(
music_base_title(_FINGERPRINT_TITLE_QUALIFIER.sub("", primary.title))
music_base_title(_FINGERPRINT_TITLE_QUALIFIER.sub("", title_evidence))
)
qualified_candidate_key = music_text_key(
music_base_title(_FINGERPRINT_TITLE_QUALIFIER.sub("", title))
@@ -317,12 +342,14 @@ def _reconcile_fingerprint_release(
info: MusicInfo,
tag_meta: Optional[MetaMusic],
filename_meta: Optional[MetaMusic],
*,
verified_audio: bool = False,
) -> MusicInfo:
"""用明确的本地单曲标签校正指纹录音所选中的任意关联发行版。
"""指纹只确认录音，用本地发行证据校正远端随意选取的关联专辑。
MusicBrainz Recording 可以同时收录于单曲和原声专辑。指纹证明的是录音
身份，而不是具体发行版；当本地标题和专辑同名时，保留录音 MBID，并把
发行层字段收敛为本地单曲证据，避免错误归入远端返回的另一张专辑。
身份，不能确认实际发行；采用本地专辑字段时移除没有证明的发行标识，
保留字段来源，旧插件的单ID回执仍只适用原有同名单曲纠正。
"""
from app.domain.music import ( # pylint: disable=import-outside-toplevel
music_album_matches,
@@ -331,9 +358,9 @@ def _reconcile_fingerprint_release(
primary = tag_meta if tag_meta and tag_meta.title else filename_meta
if (
not primary
or not _is_standalone_single_evidence(primary)
or not (_is_standalone_single_evidence(primary) or verified_audio)
or not primary.album
or not info.album
or music_album_title_is_weak(primary)
or (
music_album_matches(info, primary.album)
and music_year_matches(info, primary)
@@ -345,21 +372,137 @@ def _reconcile_fingerprint_release(
reconciled.album = primary.album
reconciled.album_artist = primary.album_artist or info.artist
reconciled.album_id = None
reconciled.album_type = "Single"
reconciled.secondary_types = []
reconciled.musicbrainz_release_id = primary.musicbrainz_release_id
reconciled.musicbrainz_release_group_id = primary.musicbrainz_release_group_id
reconciled.musicbrainz_release_track_id = primary.musicbrainz_release_track_id
reconciled.release_year = primary.release_year
reconciled.original_year = primary.original_year or info.original_year
reconciled.album_type = primary.album_type or ("Single" if _is_standalone_single_evidence(primary) else None)
reconciled.secondary_types = list(primary.secondary_types)
reconciled.year = int(primary.year) if primary.year is not None else None
reconciled.release_date = None
reconciled.release_status = None
reconciled.disc_number = primary.disc_number
reconciled.track_number = primary.track_number
reconciled.total_tracks = primary.total_tracks
reconciled.total_discs = primary.total_discs
reconciled.album_aliases = []
reconciled.cover_url = None
reconciled.category = ""
reconciled.metadata_category = "Single"
reconciled.metadata_category = reconciled.album_type or ""
reconciled.classification = None
for key in ("album", "album_artist", "year", "release_year", "disc_number", "track_number", "total_tracks", "total_discs",
"musicbrainz_release_id", "musicbrainz_release_group_id", "musicbrainz_release_track_id"):
if getattr(reconciled, key, None) is not None:
reconciled.field_sources[key] = primary.field_sources.get(key, "tag" if primary is tag_meta else "filename")
else:
reconciled.field_sources.pop(key, None)
return reconciled
def _select_fingerprint_info(
meta: MetaMusic, matches: list[tuple[Optional[float], MusicInfo]],
tag_meta: Optional[MetaMusic], filename_meta: Optional[MetaMusic],
*, truncated: bool = False,
) -> Optional[MusicInfo]:
"""只采用唯一可解释的录音候选，近分多录音保留待确认，不用文本回退掩盖歧义。"""
if not matches:
return None
if tag_meta and tag_meta.media_source == MediaSource.MusicBrainz and tag_meta.media_id:
tagged = next(((score, info) for score, info in matches if info.media_id == tag_meta.media_id), None)
if tagged is None:
return None
score, info = tagged
return _finish_fingerprint_recording(score, info, tag_meta, filename_meta)
ranked = sorted(matches, key=lambda item: item[0] if item[0] is not None else 1.0, reverse=True)
if truncated or (len(ranked) > 1 and round((ranked[0][0] or 0) - (ranked[1][0] or 0), 6) <= 0.05):
pending = MusicInfo.from_meta(meta)
pending.media_source, pending.media_id = None, None
pending.raw_data["recognition"] = {"status": "ambiguous", "message": "指纹对应多个合理的录音，请手动选择歌曲或专辑", "candidates": [
{"media_source": str(info.media_source), "media_id": info.media_id, "title": info.title, "artist": info.artist, "score": score}
for score, info in ranked[:5]
]}
return pending
score, info = ranked[0]
return _finish_fingerprint_recording(score, info, tag_meta, filename_meta)
def _finish_fingerprint_recording(
score: Optional[float], info: MusicInfo, tag_meta: Optional[MetaMusic], filename_meta: Optional[MetaMusic],
) -> MusicInfo:
"""仅确认Recording；发行ID只能沿用实际标签，不能从指纹详情所选关联发行推断。"""
if score is None:
# 旧插件的单ID回执沿用既有对象及文本核验契约，不伪造原生指纹结论。
return _reconcile_fingerprint_release(info, tag_meta, filename_meta)
result = deepcopy(_reconcile_fingerprint_release(info, tag_meta, filename_meta,
verified_audio=score is not None and score >= 0.98))
result.album_id = tag_meta.musicbrainz_release_group_id if tag_meta else None
result.field_sources.pop("album_id", None)
for key in ("musicbrainz_release_id", "musicbrainz_release_group_id", "musicbrainz_release_track_id"):
value = getattr(tag_meta, key, None)
setattr(result, key, value)
if value:
result.field_sources[key] = tag_meta.field_sources.get(key, "tag") if tag_meta else "tag"
else:
result.field_sources.pop(key, None)
result.raw_data["recognition"] = {"status": "matched", "method": "fingerprint", "score": score,
"identity_type": "recording", "release_verified": False}
return result
def _fingerprint_candidate_list(recording_id: Optional[str], observed: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""旧插件仍只给ID时保持原核验要求，不给它制造AcoustID分数。"""
if observed:
return observed[:5]
return [{"recording_id": recording_id, "score": None, "duration": None}] if recording_id else []
def _matching_fingerprint(
candidate: dict[str, Any], info: Optional[MusicInfo],
tag_meta: Optional[MetaMusic], filename_meta: Optional[MetaMusic],
) -> bool:
"""同步和异步候选采用同一证据核验规则。"""
return _fingerprint_info_matches_evidence(info, tag_meta, filename_meta,
fingerprint_score=candidate.get("score"),
fingerprint_duration=candidate.get("duration"))
def _recognize_fingerprints(
path: Union[str, Path], meta: MetaMusic, tag_meta: Optional[MetaMusic], filename_meta: Optional[MetaMusic],
recognize: Callable[[MetaMusic, str], Optional[MusicInfo]],
) -> Optional[MusicInfo]:
"""在一次有界调用中逐个核验原生候选，保留旧指纹插件的调用顺序。"""
with capture_music_recognition(seconds=90, request_limit=16) as observation:
report_music_fingerprints([])
recording_id = AcoustIdChain().identify_music_by_fingerprint(path)
matches: list[tuple[Optional[float], MusicInfo]] = []
for candidate in _fingerprint_candidate_list(recording_id, observation.fingerprint_candidates):
info = recognize(meta, str(candidate["recording_id"]))
if info and _matching_fingerprint(candidate, info, tag_meta, filename_meta):
matches.append((candidate.get("score"), info))
return None if music_recognition_failed() else _select_fingerprint_info(
meta, matches, tag_meta, filename_meta, truncated=any(item.get("truncated") for item in observation.fingerprint_candidates),
)
async def _async_recognize_fingerprints(
path: Union[str, Path], meta: MetaMusic, tag_meta: Optional[MetaMusic], filename_meta: Optional[MetaMusic],
recognize: Callable[[MetaMusic, str], Awaitable[Optional[MusicInfo]]],
) -> Optional[MusicInfo]:
"""异步路径使用相同候选集合、预算和最终歧义判断。"""
with capture_music_recognition(seconds=90, request_limit=16) as observation:
report_music_fingerprints([])
recording_id = await AcoustIdChain().async_identify_music_by_fingerprint(path)
matches: list[tuple[Optional[float], MusicInfo]] = []
for candidate in _fingerprint_candidate_list(recording_id, observation.fingerprint_candidates):
info = await recognize(meta, str(candidate["recording_id"]))
if info and _matching_fingerprint(candidate, info, tag_meta, filename_meta):
matches.append((candidate.get("score"), info))
return None if music_recognition_failed() else _select_fingerprint_info(
meta, matches, tag_meta, filename_meta, truncated=any(item.get("truncated") for item in observation.fingerprint_candidates),
)
def _music_info_matches_text_evidence(
info: Optional[MusicInfo],
meta: Optional[MetaMusic],
@@ -421,7 +564,7 @@ def _music_path_plan(
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):
if _has_remote_music_identity(info) or (info and info.raw_data.get("recognition", {}).get("status") == "ambiguous"):
return info
info = yield _MusicPathAction(
kind=_MusicPathActionKind.TAG,
@@ -665,24 +808,7 @@ class MediaPathOwner(_MediaOwnerBase):
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):
if not _fingerprint_info_matches_evidence(
info, tag_meta, filename_meta,
):
logger.warning(
"AcoustID 候选与本地标签/文件名不符，"
f"已回退文本识别：{Path(path).name} -> {info.artist} - {info.title}"
)
info = None
if self._is_remote_music_info(info):
info = _reconcile_fingerprint_release(
info,
tag_meta,
filename_meta,
)
logger.info("音乐识别命中 AcoustID 指纹层，已跳过标签和文件名识别")
info = _recognize_fingerprints(path, meta, tag_meta, filename_meta, self._recognize_musicbrainz_recording)
elif action.kind is _MusicPathActionKind.ALBUM:
info = self._music_album_dir_fallback(path)
if info and not _music_info_matches_text_evidence(info, meta):
@@ -735,31 +861,7 @@ class MediaPathOwner(_MediaOwnerBase):
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):
if not _fingerprint_info_matches_evidence(
info, tag_meta, filename_meta,
):
logger.warning(
"AcoustID 候选与本地标签/文件名不符，"
f"已回退文本识别：{Path(path).name} -> {info.artist} - {info.title}"
)
info = None
if self._is_remote_music_info(info):
info = _reconcile_fingerprint_release(
info,
tag_meta,
filename_meta,
)
logger.info("音乐识别命中 AcoustID 指纹层，已跳过标签和文件名识别")
info = await _async_recognize_fingerprints(path, meta, tag_meta, filename_meta, self._async_recognize_musicbrainz_recording)
elif action.kind is _MusicPathActionKind.ALBUM:
info = await self._async_music_album_dir_fallback(path)
if info and not _music_info_matches_text_evidence(info, meta):

2
app/chain/transfer/filter.py
View File

@@ -264,6 +264,8 @@ class _MusicFileFilterBase(_TransferOwnerBase):
except Exception as err:
logger.debug(f"音乐曲目兜底识别失败：{file_path} - {err}")
return file_meta, None
if info and (info.raw_data.get("recognition") or {}).get("status") in BLOCKING_MUSIC_RECOGNITION_STATES:
return file_meta, info
if not info or not info.media_id:
return file_meta, None
logger.info(f"{file_path.name} 通过曲目证据识别为：{info.artist} - {info.title}")

13
app/domain/music.py
View File

@@ -208,7 +208,7 @@ def music_tags_are_usable(meta: Optional[MetaMusic]) -> bool:
年份和远端 ID 不影响本地可整理性；占位、乱码或宣传 URL 不能成为
歌曲/专辑身份。调用方必须传入未经路径补写的原始标签证据。
"""
if meta is None or re.fullmatch(r"\d{1,3}", str(meta.title or "").strip()):
if meta is None or music_track_title_is_weak(meta):
return False
artist = meta.album_artist or next(iter(meta.artists), None)
required = [meta.title, meta.album, artist]
@@ -218,13 +218,22 @@ def music_tags_are_usable(meta: Optional[MetaMusic]) -> bool:
def music_track_title_is_weak(meta: MetaMusic) -> bool:
"""编号或占位文件名不能否决实际曲目，但真实标签中的数字歌曲名仍是证据。"""
title = str(meta.title or "").strip()
if not title:
if not _usable_music_tag(title):
return True
if title.isdigit():
if len(title) > 1 and title.startswith("0") and not meta.media_id and meta.music_type != MUSIC_ENTITY_ALBUM:
return True
return meta.field_sources.get("title") not in {"tag", "cue", "manual", "remote"}
if meta.field_sources.get("title") not in {"tag", "cue", "manual", "remote"} and re.fullmatch(r"[a-f\d]{16,64}", title, re.I):
return True
return bool(re.fullmatch(r"(?:(?:track|audio|音轨|曲目)[\s._-]*\d*|unknown|untitled)", title, re.I))
def music_usable_artists(artists: Iterable[str]) -> list[str]:
"""剔除抓轨占位、宣传链接和乱码署名；缺失艺人不是与指纹候选冲突的证据。"""
return [artist for artist in artists if _usable_music_tag(artist)]
def music_album_title_is_weak(meta: MetaMusic) -> bool:
"""普通收件目录不是专辑证据，标签或种子明确提供的同名专辑仍有效。"""
title = meta.album or meta.title

214
app/modules/acoustid/__init__.py
View File

@@ -1,5 +1,6 @@
import asyncio
import json
import math
import shutil
import subprocess
import threading
@@ -11,6 +12,14 @@ from typing import Any, Optional, Tuple, Union
from uuid import UUID
from app.adapters.network.http import AsyncRequestUtils, RequestUtils
from app.application.music.observation import (
capture_music_recognition,
music_recognition_diagnostics,
music_request_timeout,
music_wait_allowed,
report_music_fingerprints,
report_music_recognition,
)
from app.modules import _ModuleBase
from app.runtime.execution import run_in_threadpool
from app.runtime.log import logger
@@ -26,6 +35,17 @@ class _AcoustIdLookupPlan:
data: dict[str, Union[str, int]]
@dataclass(frozen=True, slots=True)
class _FingerprintCacheEntry:
"""文件指纹的有界缓存，保留候选和故障语义，避免仅缓存首个ID或永久空结果。"""
recording_id: Optional[str]
candidates: tuple[dict[str, Any], ...]
expires_at: float
status: str
message: str
class AcoustIdModule(_ModuleBase):
"""通过 Chromaprint 本地指纹和 AcoustID API 识别 MusicBrainz Recording ID。"""
@@ -43,7 +63,7 @@ class AcoustIdModule(_ModuleBase):
"""初始化 fpcalc 路径和进程内文件指纹识别缓存。"""
super().__init__()
self._fpcalc_path: Optional[str] = None
self._cache: OrderedDict[tuple[str, int, int], Optional[str]] = OrderedDict()
self._cache: OrderedDict[tuple[str, int, int], _FingerprintCacheEntry] = OrderedDict()
self._cache_lock = threading.Lock()
@staticmethod
@@ -123,47 +143,55 @@ class AcoustIdModule(_ModuleBase):
def identify_music_by_fingerprint(self, path: Path) -> Optional[str]:
"""读取本地音频指纹并返回高置信匹配的 MusicBrainz Recording ID。"""
file_path = Path(path)
if not self._fpcalc_path or not file_path.is_file():
return None
cache_key = self._file_cache_key(file_path)
if cache_key:
found, cached_id = self._get_cached(cache_key)
if found:
return cached_id
fingerprint = self._generate_fingerprint(file_path)
recording_id = (
self._lookup_recording_id(*fingerprint) if fingerprint else None
)
if cache_key:
self._set_cached(cache_key, recording_id)
return recording_id
with capture_music_recognition(seconds=90) as observed:
report_music_fingerprints([])
file_path = Path(path)
if not self._fpcalc_path or not file_path.is_file():
return None
cache_key = self._file_cache_key(file_path)
if cache_key:
found, cached_id = self._get_cached(cache_key)
if found:
return cached_id
fingerprint = self._generate_fingerprint(file_path)
recording_id = (
self._lookup_recording_id(*fingerprint) if fingerprint else None
)
if fingerprint and observed.fingerprint_candidates:
report_music_fingerprints([{**item, "duration": fingerprint[0]} for item in observed.fingerprint_candidates])
if cache_key:
self._set_cached(cache_key, recording_id, observed.fingerprint_candidates)
return recording_id
async def async_identify_music_by_fingerprint(
self,
path: Path,
) -> Optional[str]:
"""异步读取音频指纹并返回高置信匹配的 MusicBrainz Recording ID。"""
file_path = Path(path)
if not self._fpcalc_path or not await run_in_threadpool(
Path.is_file,
file_path,
):
return None
cache_key = self._file_cache_key(file_path)
if cache_key:
found, cached_id = self._get_cached(cache_key)
if found:
return cached_id
fingerprint = await self._async_generate_fingerprint(file_path)
recording_id = (
await self._async_lookup_recording_id(*fingerprint)
if fingerprint
else None
)
if cache_key:
self._set_cached(cache_key, recording_id)
return recording_id
with capture_music_recognition(seconds=90) as observed:
report_music_fingerprints([])
file_path = Path(path)
if not self._fpcalc_path or not await run_in_threadpool(
Path.is_file,
file_path,
):
return None
cache_key = self._file_cache_key(file_path)
if cache_key:
found, cached_id = self._get_cached(cache_key)
if found:
return cached_id
fingerprint = await self._async_generate_fingerprint(file_path)
recording_id = (
await self._async_lookup_recording_id(*fingerprint)
if fingerprint
else None
)
if fingerprint and observed.fingerprint_candidates:
report_music_fingerprints([{**item, "duration": fingerprint[0]} for item in observed.fingerprint_candidates])
if cache_key:
self._set_cached(cache_key, recording_id, observed.fingerprint_candidates)
return recording_id
@staticmethod
def _file_cache_key(path: Path) -> Optional[tuple[str, int, int]]:
@@ -183,38 +211,54 @@ class AcoustIdModule(_ModuleBase):
if cache_key not in self._cache:
return False, None
value = self._cache.pop(cache_key)
if value.expires_at <= time.monotonic():
return False, None
self._cache[cache_key] = value
return True, value
report_music_fingerprints(list(value.candidates))
if value.status in {"service_error", "budget_exhausted"}:
report_music_recognition(value.status, value.message)
return True, value.recording_id
def _set_cached(
self,
cache_key: tuple[str, int, int],
recording_id: Optional[str],
candidates: Optional[list[dict[str, Any]]] = None,
) -> None:
"""写入文件识别 LRU 缓存并淘汰最早使用的条目。"""
diagnostics = music_recognition_diagnostics()
status = str(diagnostics.get("status") or "not_found")
ttl = 3600 if recording_id else 15 if status in {"service_error", "budget_exhausted"} else 300
entry = _FingerprintCacheEntry(recording_id, tuple(dict(item) for item in candidates or []),
time.monotonic() + ttl, status, str(diagnostics.get("message") or ""))
with self._cache_lock:
self._cache.pop(cache_key, None)
self._cache[cache_key] = recording_id
self._cache[cache_key] = entry
while len(self._cache) > self._cache_max:
self._cache.popitem(last=False)
def _generate_fingerprint(self, path: Path) -> Optional[tuple[int, str]]:
"""调用 fpcalc 生成 AcoustID 查询所需的完整时长和压缩指纹。"""
timeout = music_request_timeout(claim=False, maximum=self._fingerprint_timeout)
if timeout is None:
return None
try:
result = subprocess.run(
[self._fpcalc_path, "-json", str(path)],
capture_output=True,
text=True,
check=False,
timeout=self._fingerprint_timeout,
timeout=timeout,
)
except (OSError, subprocess.TimeoutExpired) as err:
logger.warning(f"生成音频指纹失败：{path} - {err}")
report_music_recognition("service_error", "音频指纹暂时无法生成，请检查音频解码支持")
return None
if result.returncode not in self._usable_fpcalc_returncodes:
logger.warning(
f"生成音频指纹失败：{path} - fpcalc 退出码 {result.returncode}"
)
report_music_recognition("service_error", "音频指纹解码失败")
return None
return self._parse_fingerprint_output(path, result.stdout)
@@ -223,6 +267,9 @@ class AcoustIdModule(_ModuleBase):
path: Path,
) -> Optional[tuple[int, str]]:
"""异步调用 fpcalc 生成 AcoustID 查询所需的指纹。"""
timeout = music_request_timeout(claim=False, maximum=self._fingerprint_timeout)
if timeout is None:
return None
process = None
try:
process = await asyncio.create_subprocess_exec(
@@ -234,13 +281,14 @@ class AcoustIdModule(_ModuleBase):
)
stdout, _ = await asyncio.wait_for(
process.communicate(),
timeout=self._fingerprint_timeout,
timeout=timeout,
)
except asyncio.TimeoutError:
if process and process.returncode is None:
process.kill()
await process.communicate()
logger.warning(f"生成音频指纹超时：{path}")
report_music_recognition("service_error", "音频指纹暂时无法生成，请检查音频解码支持")
return None
except asyncio.CancelledError:
if process and process.returncode is None:
@@ -249,11 +297,13 @@ class AcoustIdModule(_ModuleBase):
raise
except OSError as err:
logger.warning(f"生成音频指纹失败：{path} - {err}")
report_music_recognition("service_error", "音频指纹暂时无法生成，请检查音频解码支持")
return None
if process.returncode not in self._usable_fpcalc_returncodes:
logger.warning(
f"生成音频指纹失败：{path} - fpcalc 退出码 {process.returncode}"
)
report_music_recognition("service_error", "音频指纹解码失败")
return None
return self._parse_fingerprint_output(
path,
@@ -270,11 +320,13 @@ class AcoustIdModule(_ModuleBase):
payload = json.loads(output)
duration = round(float(payload.get("duration") or 0))
fingerprint = str(payload.get("fingerprint") or "").strip()
except (AttributeError, TypeError, ValueError) as err:
except (AttributeError, OverflowError, TypeError, ValueError) as err:
logger.warning(f"fpcalc 输出解析失败：{path} - {err}")
report_music_recognition("service_error", "音频指纹暂时无法生成，请检查音频解码支持")
return None
if duration <= 0 or not fingerprint:
logger.warning(f"fpcalc 未返回有效音频指纹：{path}")
report_music_recognition("service_error", "音频指纹暂时无法生成，请检查音频解码支持")
return None
return duration, fingerprint
@@ -284,20 +336,30 @@ class AcoustIdModule(_ModuleBase):
with cls._request_lock:
now = time.monotonic()
request_at = max(now, cls._last_request_at + cls._request_interval)
if not music_wait_allowed(request_at - now):
return -1
cls._last_request_at = request_at
return max(0.0, request_at - now)
@classmethod
def _wait_for_rate_limit(cls) -> None:
def _wait_for_rate_limit(cls) -> bool:
"""同步等待 AcoustID 公共接口的已预留请求时间。"""
if delay := cls._reserve_request_delay():
delay = cls._reserve_request_delay()
if delay < 0:
return False
if delay:
time.sleep(delay)
return True
@classmethod
async def _async_wait_for_rate_limit(cls) -> None:
async def _async_wait_for_rate_limit(cls) -> bool:
"""异步等待 AcoustID 公共接口的已预留请求时间。"""
if delay := cls._reserve_request_delay():
delay = cls._reserve_request_delay()
if delay < 0:
return False
if delay:
await asyncio.sleep(delay)
return True
@classmethod
def _lookup_plan(
@@ -328,9 +390,13 @@ class AcoustIdModule(_ModuleBase):
payload: Any,
) -> Optional[str]:
"""按统一 HTTP 状态和 AcoustID 规则投影 Recording ID。"""
if status_code != 200:
if (status_code != 200 or not isinstance(payload, dict) or payload.get("status") != "ok"
or not isinstance(payload.get("results", []), list)):
report_music_recognition("service_error", "音频指纹服务暂时不可用")
return None
return cls._select_recording_id(payload)
candidates = cls._select_recording_candidates(payload)
report_music_fingerprints(candidates)
return str(candidates[0]["recording_id"]) if candidates else None
def _lookup_recording_id(
self,
@@ -345,7 +411,11 @@ class AcoustIdModule(_ModuleBase):
)
if not plan:
return None
self._wait_for_rate_limit()
if music_request_timeout(maximum=30) is None or self._wait_for_rate_limit() is False:
return None
timeout = music_request_timeout(claim=False, maximum=30)
if timeout is None:
return None
response = RequestUtils(
ua=get_runtime_setting('USER_AGENT'),
proxies=get_runtime_setting('PROXY'),
@@ -353,19 +423,23 @@ class AcoustIdModule(_ModuleBase):
).post_res(
url=plan.url,
data=plan.data,
timeout=timeout,
)
if response is None:
logger.warning("AcoustID 指纹查询失败：无响应")
report_music_recognition("service_error", "音频指纹服务连接失败")
return None
try:
if response.status_code != 200:
logger.warning(f"AcoustID 指纹查询失败：HTTP {response.status_code}")
report_music_recognition("service_error", "音频指纹服务暂时不可用")
return None
return self._project_lookup_response(
response.status_code, response.json()
)
except (TypeError, ValueError) as err:
logger.warning(f"AcoustID 响应解析失败：{err}")
report_music_recognition("service_error", "音频指纹服务返回了无效响应")
return None
finally:
response.close()
@@ -383,7 +457,11 @@ class AcoustIdModule(_ModuleBase):
)
if not plan:
return None
await self._async_wait_for_rate_limit()
if music_request_timeout(maximum=30) is None or await self._async_wait_for_rate_limit() is False:
return None
timeout = music_request_timeout(claim=False, maximum=30)
if timeout is None:
return None
response = await AsyncRequestUtils(
ua=get_runtime_setting('USER_AGENT'),
proxies=get_runtime_setting('PROXY'),
@@ -391,29 +469,43 @@ class AcoustIdModule(_ModuleBase):
).post_res(
url=plan.url,
data=plan.data,
timeout=timeout,
)
if response is None:
logger.warning("AcoustID 指纹查询失败：无响应")
report_music_recognition("service_error", "音频指纹服务连接失败")
return None
try:
if response.status_code != 200:
logger.warning(f"AcoustID 指纹查询失败：HTTP {response.status_code}")
report_music_recognition("service_error", "音频指纹服务暂时不可用")
return None
return self._project_lookup_response(
response.status_code, response.json()
)
except (TypeError, ValueError) as err:
logger.warning(f"AcoustID 响应解析失败：{err}")
report_music_recognition("service_error", "音频指纹服务返回了无效响应")
return None
finally:
await response.aclose()
@classmethod
def _select_recording_id(cls, payload: Any) -> Optional[str]:
"""按匹配分从 AcoustID 响应中选择首个有效 MusicBrainz Recording ID。"""
"""兼容旧调用方的单ID视图，内部识别使用完整有界候选集合。"""
candidates = cls._select_recording_candidates(payload)
return str(candidates[0]["recording_id"]) if candidates else None
@classmethod
def _select_recording_candidates(cls, payload: Any) -> list[dict[str, Any]]:
"""按实际匹配分保留最多五个不同Recording ID，不用数组中的首个ID消除歧义。"""
if not isinstance(payload, dict) or payload.get("status") != "ok":
return None
return []
candidates: list[dict[str, Any]] = []
seen: set[str] = set()
results = payload.get("results") or []
if not isinstance(results, list):
return []
ranked = sorted(
(item for item in results if isinstance(item, dict)),
key=lambda item: cls._score(item.get("score")),
@@ -423,22 +515,26 @@ class AcoustIdModule(_ModuleBase):
score = cls._score(item.get("score"))
if score < cls._minimum_score:
break
for recording in item.get("recordings") or []:
recordings = item.get("recordings") or []
if not isinstance(recordings, list):
continue
for recording in recordings:
recording_id = cls._normalize_recording_id(
recording.get("id") if isinstance(recording, dict) else None
)
if recording_id:
logger.info(
f"AcoustID 指纹命中 MusicBrainz：{recording_id}，匹配度 {score:.3f}"
)
return recording_id
return None
if recording_id and recording_id not in seen:
seen.add(recording_id)
candidates.append({"recording_id": recording_id, "score": score})
if len(candidates) == 6:
return [{**candidate, "truncated": True} for candidate in candidates[:5]]
return candidates
@staticmethod
def _score(value: Any) -> float:
"""将 AcoustID 匹配分安全转换为零到一之间的浮点数。"""
try:
return max(0.0, min(float(value), 1.0))
score = float(value)
return score if math.isfinite(score) and 0 <= score <= 1 else 0.0
except (TypeError, ValueError):
return 0.0

4
docs/architecture-overview.md
View File

@@ -763,7 +763,7 @@ flowchart LR
| 指标 | 当前值 |
|---|---:|
| Python 模块 | 1055 |
| 内部导入边 | 8,985 |
| 内部导入边 | 8,990 |
| 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
| Application / Chain 具体 Adapter 直连 | 0 / 0 |
| Direct egress | 53（债务已清零，53 条精确 containment） |
@@ -773,7 +773,7 @@ flowchart LR
| Model/Oper 自动事务与自建 Session | 0 |
| 组合根外 `SystemConfigOper()` | 0 |
整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责；宿主依赖边现为 8,985，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。当前 `app.startup.lifecycle` 为 575、`app.factory` 为 587、`app.main` 为 589。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。
整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责；宿主依赖边现为 8,990，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。当前 `app.startup.lifecycle` 为 575、`app.factory` 为 587、`app.main` 为 589。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。
架构专项验证分为两个 CI 投影：`Check event semantic policy` 先运行依赖、Adapter、出口和 Event
语义门禁，`Check host architecture snapshot` 再执行快照测试及一次

10
docs/mcp-api.md
View File

@@ -511,6 +511,16 @@ SSE 的 `candidate_items` 是站点原始返回数量，`match_counts` 记录身
音乐识别结果同时提供 `audio_format`、`audio_lossless`、`audio_quality`、`bit_depth`、`sample_rate`、`bitrate`、`audio_specs` 和 `audio_quality_score`。本地文件识别读取实际音频流参数，并使用 Chromaprint 的 `fpcalc` 在本地生成指纹后查询 AcoustID；音频文件本身不会上传。站点资源识别从标题和描述提取声明参数；码率、采样率的存储单位分别为 bps 和 Hz。
原生 AcoustID 识别会核验最多五个 Recording 候选，以实际时长、有效曲名、艺人和版本排除冲突。
唯一高分且时长吻合的指纹可补齐无标签或纯编号文件；`Unknown Artist` 等占位不是艺人冲突。
真实数字歌曲标签仍可整理，`01`、`Track 01` 等抓轨占位不会仅凭完整字段认定为可信歌曲。
近分歧义或候选被截断时保留 `raw_data.recognition.status=ambiguous`，不能由返回顺序自动选中。
原生指纹命中的诊断含 `method=fingerprint`、`identity_type=recording`、`release_verified=false`；
指纹不证明实际发行，补充发行 ID 只保留实际标签提供的值，`album_id` 只沿用标签发行组 ID。
旧插件的单 ID 接口保持兼容，不人为添加 AcoustID 分数或降低其文本核验要求。
指纹生成和候选详情共用最多 90 秒、16 次 HTTP 尝试的预算；嵌套调用服从已有更早的预算。
指纹缓存保留全部候选，成功、无匹配、临时故障分别在 3600、300、15 秒后过期。
`MusicMeta.music_type` 保留已绑定主身份的实体类型；仅从名称推测时可为空。
`MusicMeta.album_type` 和 `secondary_types` 保留标签声明的发行主副类型，例如 EP、Single、Compilation；
它们与 recording/album/artist 实体类型不同，不能根据曲目数量覆盖明确的标签类型。

2
docs/refactor/optimization-checklist.md
View File

@@ -96,7 +96,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
| 指标 | 当前值 | 解释 |
|---|---:|---|
| 宿主 Python 模块 / 内部依赖边 | 1055 / 8,985 | `dependency-baseline.json` 当前快照；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档 |
| 宿主 Python 模块 / 内部依赖边 | 1055 / 8,990 | `dependency-baseline.json` 当前快照；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档 |
| 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
| 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
| Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |

15
skills/moviepilot-api/api/music.md
View File

@@ -79,6 +79,21 @@ Music recognition, exploration, album and artist navigation, and recognition-cac
- Track-title discovery uses the Recording search index and its related
releases. Release search has no `recording` field. At similar relevance,
releases supported by multiple recordings precede editions of one single.
- Native AcoustID recognition verifies at most five Recording candidates using
duration, meaningful titles, artists, and versions. A unique high-score match
with consistent duration can identify untagged or numbered files. Placeholder
artist tags are missing evidence, while explicit conflicting artists or live
versions still reject a candidate. Real numeric song tags remain meaningful;
zero-padded rip numbers and `Track 01` do not establish a trusted local title.
Near ties or truncated candidates remain `ambiguous`. A native fingerprint
result has `raw_data.recognition.method=fingerprint`,
`identity_type=recording`, and `release_verified=false`; supplemental release
IDs and `album_id` must come from actual tags, with `album_id` retaining only
a release-group ID. Legacy single-ID plugins keep their text-validation
contract and are not assigned an invented AcoustID score. Fingerprint work
shares a 90-second, 16-HTTP-attempt budget, or the existing enclosing budget.
Cached fingerprints retain candidates and expire after 3600 seconds for a
hit, 300 seconds for no match, and 15 seconds for a transient failure.
- Recognition has an eight-attempt HTTP budget with 45 seconds for waits and
request timeouts. Cached HTTP results cost no request slots. Album-directory
success, no-match, and transient failures expire after 3600, 300, and 15

9
tests/fixtures/architecture/dependency-baseline.json vendored
View File

@@ -1097,8 +1097,8 @@
"runtime_only": true
}
},
"edge_count": 8985,
"edge_sha256": "8b77394a0e5714e776ab92c5bdbfbea6a253e530e8288c79c24cd4e8e40a4c46",
"edge_count": 8990,
"edge_sha256": "0f702bc0a98c671186cf06117094b842de518f45935f4b98096577ee78483dfe",
"edges": [
"app -> app.foundation",
"app -> app.foundation.environment",
@@ -4708,6 +4708,8 @@
"app.chain.media.path -> app.application",
"app.chain.media.path -> app.application.audio",
"app.chain.media.path -> app.application.configuration",
"app.chain.media.path -> app.application.music",
"app.chain.media.path -> app.application.music.observation",
"app.chain.media.path -> app.chain",
"app.chain.media.path -> app.chain.acoustid",
"app.chain.media.path -> app.chain.media",
@@ -6687,6 +6689,9 @@
"app.modules.acoustid -> app.adapters",
"app.modules.acoustid -> app.adapters.network",
"app.modules.acoustid -> app.adapters.network.http",
"app.modules.acoustid -> app.application",
"app.modules.acoustid -> app.application.music",
"app.modules.acoustid -> app.application.music.observation",
"app.modules.acoustid -> app.modules",
"app.modules.acoustid -> app.runtime",
"app.modules.acoustid -> app.runtime.execution",

287
tests/test_music_fingerprint_candidates.py Normal file
View File

@@ -0,0 +1,287 @@
"""指纹多候选、占位文件名、失败缓存与录音/发行边界回归。"""
import asyncio
import shutil
from pathlib import Path
from unittest.mock import AsyncMock, Mock
import pytest
from mutagen.flac import FLAC
from app.application.audio import AudioMetadataHelper
from app.application.music.observation import (
capture_music_recognition,
report_music_fingerprints,
report_music_recognition,
)
from app.chain.acoustid import AcoustIdChain
from app.chain.media import MediaChain
from app.chain.media.path import (
_async_recognize_fingerprints,
_fingerprint_info_matches_evidence,
_recognize_fingerprints,
_reconcile_fingerprint_release,
_select_fingerprint_info,
)
from app.chain.transfer.facade import TransferChain
from app.domain.context import MusicInfo
from app.domain.meta.metamusic import MetaMusic
from app.domain.music import music_tags_are_usable, music_track_title_is_weak
from app.modules.acoustid import AcoustIdModule
from tests.test_acoustid_module import RECORDING_ID, FakeResponse
from tests.test_transfer_sync_extra_files import make_fileitem
OTHER_ID = "70b71113-b078-4f07-9d24-374482016a41"
def _audio(tmp_path, name="01.flac"):
"""创建只有时长证据、没有名称标签的真实音频。"""
path = tmp_path / name
shutil.copyfile(Path(__file__).parent / "fixtures/audio/silence.flac", path)
return path
@pytest.mark.parametrize("asynchronous", [False, True])
def test_all_native_candidates_are_checked_before_selecting_recording(tmp_path, monkeypatch, asynchronous):
"""首个高分映射不符合时长时，后续正确候选仍可命中，纯编号不能否决真实指纹。"""
path = _audio(tmp_path)
chain = object.__new__(MediaChain)
candidates = [{"recording_id": RECORDING_ID, "score": 0.99, "duration": 3},
{"recording_id": OTHER_ID, "score": 0.99, "duration": 3}]
def identify(_path):
"""模拟旧指纹接口，同时通过观察上下文保留原生多候选证据。"""
report_music_fingerprints(candidates)
return RECORDING_ID
monkeypatch.setattr(AcoustIdChain, "async_identify_music_by_fingerprint" if asynchronous else "identify_music_by_fingerprint",
AsyncMock(side_effect=identify) if asynchronous else Mock(side_effect=identify))
infos = {RECORDING_ID: MusicInfo(media_source="musicbrainz", media_id=RECORDING_ID, title="Wrong Recording", duration=200),
OTHER_ID: MusicInfo(media_source="musicbrainz", media_id=OTHER_ID, title="Actual Song", artists=["Artist"], duration=3)}
recognize = AsyncMock(side_effect=lambda _meta, key: infos[key]) if asynchronous else Mock(side_effect=lambda _meta, key: infos[key])
monkeypatch.setattr(chain, "_async_recognize_musicbrainz_recording" if asynchronous else "_recognize_musicbrainz_recording", recognize)
monkeypatch.setattr(chain, "_async_finalize_recognition_result", AsyncMock(side_effect=lambda info, **_kwargs: info))
monkeypatch.setattr(chain, "_finalize_recognition_result", lambda info, **_kwargs: info)
text_lookup = Mock(side_effect=AssertionError("唯一指纹已命中，不应继续文本查询"))
monkeypatch.setattr(chain, "recognize_media", text_lookup)
monkeypatch.setattr(chain, "async_recognize_media", AsyncMock(side_effect=text_lookup))
_meta, info = asyncio.run(chain.async_recognize_music_by_path(path)) if asynchronous else chain.recognize_music_by_path(path)
assert info.media_id == OTHER_ID and info.title == "Actual Song"
assert recognize.call_count == 2
text_lookup.assert_not_called()
def test_equal_fingerprint_candidates_stay_pending_and_do_not_fall_through(tmp_path, monkeypatch):
"""两个合理录音不能由返回顺序决定，也不能用后续弱搜索掩盖歧义。"""
path = _audio(tmp_path)
chain = object.__new__(MediaChain)
def identify(_path):
"""提供两个同样高分且时长相同的录音关联。"""
report_music_fingerprints([{"recording_id": item, "score": 0.99, "duration": 3} for item in (RECORDING_ID, OTHER_ID)])
return RECORDING_ID
monkeypatch.setattr(AcoustIdChain, "identify_music_by_fingerprint", Mock(side_effect=identify))
monkeypatch.setattr(chain, "_recognize_musicbrainz_recording", lambda _meta, key: MusicInfo(
media_source="musicbrainz", media_id=key, title="Song", artists=["Artist"], duration=3))
monkeypatch.setattr(chain, "_finalize_recognition_result", lambda info, **_kwargs: info)
text_lookup = Mock(side_effect=AssertionError("不得继续弱搜索"))
monkeypatch.setattr(chain, "recognize_media", text_lookup)
meta, info = chain.recognize_music_by_path(path)
monkeypatch.setattr(MediaChain, "recognize_music_by_path", Mock(return_value=(meta, info)))
_meta, transfer_info = TransferChain._match_music_recording_context(make_fileitem(str(path)), path, meta)
assert info.media_id is None
assert info.raw_data["recognition"]["status"] == "ambiguous"
assert transfer_info.raw_data["recognition"]["status"] == "ambiguous"
text_lookup.assert_not_called()
def test_fingerprint_bound_marks_unchecked_candidates_instead_of_claiming_unique():
"""超过五个关联时保留截断事实，不能将未查完的候选称为唯一匹配。"""
payload = {"status": "ok", "results": [{"score": 1.0, "recordings": [
{"id": f"00000000-0000-4000-8000-{index:012d}"} for index in range(10)
]}]}
candidates = AcoustIdModule._select_recording_candidates(payload)
assert len(candidates) == 5
assert all(item["truncated"] for item in candidates)
info = MusicInfo(media_source="musicbrainz", media_id=RECORDING_ID, title="Song")
pending = _select_fingerprint_info(MetaMusic(title="01"), [(1.0, info)], None, None, truncated=True)
assert pending.media_id is None and pending.raw_data["recognition"]["status"] == "ambiguous"
def test_tagged_recording_identity_disambiguates_supported_fingerprint_mapping():
"""当音频和文本均支持已存Recording ID时，保留标签身份，不任意替换为重复记录。"""
tagged = MetaMusic(title="Song", artists=["Artist"], media_source="musicbrainz", media_id=OTHER_ID)
infos = [MusicInfo(media_source="musicbrainz", media_id=value, title="Song", artists=["Artist"]) for value in (RECORDING_ID, OTHER_ID)]
chosen = _select_fingerprint_info(tagged, [(0.99, info) for info in infos], tagged, None)
assert chosen.media_id == OTHER_ID
assert _select_fingerprint_info(tagged, [(0.99, infos[0])], tagged, None) is None
def test_fingerprint_keeps_tagged_release_ids_and_does_not_invent_missing_ones():
"""同名同年的远端发行也不证明实际版本；标签发行ID保留，缺失时不猜。"""
info = MusicInfo(media_source="musicbrainz", media_id=RECORDING_ID, title="Song", album="Album", year=2004,
album_id="remote-group", musicbrainz_release_id="remote-edition", musicbrainz_release_group_id="remote-group")
tags = MetaMusic(title="Song", album="Album", year=2004, musicbrainz_release_id="tagged-edition")
selected = _select_fingerprint_info(tags, [(0.99, info)], tags, None)
assert selected.musicbrainz_release_id == "tagged-edition"
assert selected.musicbrainz_release_group_id is None
assert selected.album_id is None
assert selected.raw_data["recognition"]["release_verified"] is False
assert info.musicbrainz_release_id == "remote-edition"
@pytest.mark.parametrize("asynchronous", [False, True])
def test_incomplete_fingerprint_details_do_not_select_first_success(tmp_path, monkeypatch, asynchronous):
"""后续候选查询故障时，首个成功详情不能冒充已经排除全部歧义。"""
path = _audio(tmp_path)
def identify(_path):
"""返回两个需要核验的高分录音。"""
report_music_fingerprints([{"recording_id": item, "score": 0.99, "duration": 3} for item in (RECORDING_ID, OTHER_ID)])
return RECORDING_ID
def recognize(_meta, key):
"""只完成第一个详情，第二个模拟上游临时故障。"""
if key == OTHER_ID:
report_music_recognition("service_error", "详情服务临时失败")
return None
return MusicInfo(media_source="musicbrainz", media_id=key, title="Song", duration=3)
monkeypatch.setattr(AcoustIdChain, "async_identify_music_by_fingerprint" if asynchronous else "identify_music_by_fingerprint",
AsyncMock(side_effect=identify) if asynchronous else Mock(side_effect=identify))
meta = MetaMusic(title="01")
result = asyncio.run(_async_recognize_fingerprints(path, meta, None, meta, AsyncMock(side_effect=recognize))) if asynchronous else (
_recognize_fingerprints(path, meta, None, meta, recognize))
assert result is None
@pytest.mark.parametrize("artist", ["Unknown Artist", "未知艺术家", "www.example.com", "\ufffd"])
def test_placeholder_artist_does_not_reject_unique_native_fingerprint(artist):
"""占位和宣传署名不是强证据，真实唯一指纹可补齐缺失艺人。"""
tags = MetaMusic(title="01", artists=[artist], field_sources={"title": "tag", "artists": "tag"})
info = MusicInfo(media_source="musicbrainz", media_id=RECORDING_ID, title="Song", artists=["Artist"], duration=180)
assert _fingerprint_info_matches_evidence(info, tags, MetaMusic(title="01"), fingerprint_score=0.99, fingerprint_duration=180)
def test_placeholder_title_does_not_hide_explicit_live_version_conflict():
"""即使文件名和曲名是占位，标签明确声明的现场版本仍必须核验。"""
tags = MetaMusic(title="01", version="Live", artists=["Artist"], field_sources={"title": "tag"})
info = MusicInfo(media_source="musicbrainz", media_id=RECORDING_ID, title="Song (Studio Version)", artists=["Artist"], duration=180)
assert not _fingerprint_info_matches_evidence(info, tags, MetaMusic(title="01"), fingerprint_score=0.99, fingerprint_duration=180)
def test_fingerprint_budget_prevents_queued_requests_and_cleans_context(monkeypatch):
"""超出时限不占据未来限流时隙，预算退出后不污染下一次独立识别。"""
module = AcoustIdModule()
monkeypatch.setattr(AcoustIdModule, "_last_request_at", 10.0)
monkeypatch.setattr("app.modules.acoustid.time.monotonic", lambda: 10.0)
monkeypatch.setattr("app.application.music.observation.monotonic", lambda: 10.0)
with capture_music_recognition(seconds=0.1) as first:
assert module._reserve_request_delay() == -1
assert first.status == "budget_exhausted"
assert AcoustIdModule._last_request_at == 10.0
with capture_music_recognition(seconds=2) as second:
assert 0 < module._reserve_request_delay() < 1
assert second.status == "not_found"
@pytest.mark.parametrize("payload", [
{"status": "ok", "results": 12},
{"status": "ok", "results": [{"score": 1, "recordings": 12}]},
])
def test_malformed_candidate_collections_are_not_recordings(payload):
"""响应结构异常不应崩溃或伪造有效指纹候选。"""
assert AcoustIdModule._select_recording_candidates(payload) == []
def test_whole_album_numeric_cue_title_is_not_a_track_placeholder():
"""整轨专辑的数字标题与抓轨序号不同，不能因前导零丢掉专辑名。"""
meta = MetaMusic(title="01", album="01", artists=["Artist"], music_type="album", field_sources={"title": "cue"})
assert music_tags_are_usable(meta)
@pytest.mark.parametrize("score", [float("nan"), float("inf"), -0.1, 1.1, "invalid"])
def test_invalid_fingerprint_score_cannot_become_confidence(score):
"""非有限值及越界分数不能被截成满分录音证据。"""
assert AcoustIdModule._select_recording_candidates({"status": "ok", "results": [
{"score": score, "recordings": [{"id": RECORDING_ID}]},
]}) == []
def test_candidates_survive_cache_and_failures_expire(tmp_path, monkeypatch):
"""缓存完整候选，并让服务故障在短冷却后恢复，不能永久缓存无结果。"""
path = _audio(tmp_path)
module = AcoustIdModule()
module._fpcalc_path = "/fpcalc"
now = [100.0]
monkeypatch.setattr("app.modules.acoustid.time.monotonic", lambda: now[0])
monkeypatch.setattr(module, "_generate_fingerprint", Mock(return_value=(3, "fingerprint")))
monkeypatch.setattr(module, "_wait_for_rate_limit", lambda: None)
post = Mock(return_value=FakeResponse({}, status_code=503))
monkeypatch.setattr("app.modules.acoustid.RequestUtils.post_res", post)
with capture_music_recognition() as first:
assert module.identify_music_by_fingerprint(path) is None
assert first.status == "service_error"
with capture_music_recognition():
assert module.identify_music_by_fingerprint(path) is None
assert post.call_count == 1
now[0] += 16
post.return_value = FakeResponse({"status": "ok", "results": [{"score": 0.99, "recordings": [
{"id": RECORDING_ID}, {"id": OTHER_ID},
]}]})
for _ in range(2):
with capture_music_recognition() as recovered:
assert module.identify_music_by_fingerprint(path) == RECORDING_ID
assert [item["recording_id"] for item in recovered.fingerprint_candidates] == [RECORDING_ID, OTHER_ID]
assert all(item["duration"] == 3 for item in recovered.fingerprint_candidates)
assert post.call_count == 2
def test_legacy_plugin_id_without_score_does_not_gain_audio_only_trust():
"""旧插件回执保持兼容，但没有原生分数和时长时不能冒充强指纹证据。"""
info = MusicInfo(media_source="musicbrainz", media_id=RECORDING_ID, title="Actual Song", duration=180)
assert not _fingerprint_info_matches_evidence(info, None, MetaMusic(title="01", duration=180))
@pytest.mark.parametrize("title,usable", [("22", True), ("1979", True), ("01", False), ("Track 01", False)])
def test_real_numeric_tags_and_placeholders_use_consistent_rules(tmp_path, title, usable):
"""真实数字歌名可凭完整标签整理，抓轨序号和Track占位仍需补充识别。"""
path = _audio(tmp_path)
tags = FLAC(path)
tags.update(title=[title], artist=["Artist"], album=["Album"], tracknumber=["1"])
tags.save()
meta = AudioMetadataHelper.read_tags(path)
assert music_tags_are_usable(meta) is usable
assert music_track_title_is_weak(meta) is not usable
def test_recording_fingerprint_cannot_assign_an_unverified_compilation_release():
"""确认录音不等于确认发行；本地专辑保留来源，移除无关远端发行标识。"""
info = MusicInfo(media_source="musicbrainz", media_id=RECORDING_ID, title="Song", artists=["Artist"],
album="Random Compilation", album_id="wrong-group", musicbrainz_release_id="wrong-release",
musicbrainz_release_group_id="wrong-group", musicbrainz_release_track_id="wrong-slot",
year=2023, release_year=2023, original_year=2004)
tags = MetaMusic(title="Song", artists=["Artist"], album="Actual Album", album_artist="Artist", year=2004,
field_sources={"title": "tag", "artists": "tag", "album": "tag", "year": "tag"})
corrected = _reconcile_fingerprint_release(info, tags, None, verified_audio=True)
assert corrected.media_id == RECORDING_ID
assert corrected.album == "Actual Album" and corrected.year == 2004
assert corrected.album_id is None and corrected.musicbrainz_release_id is None
assert corrected.musicbrainz_release_group_id is None and corrected.musicbrainz_release_track_id is None
assert corrected.field_sources["album"] == "tag"
assert info.album == "Random Compilation"