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

feat(music): verify release candidates with identities and complete track evidence

Browse Source
This commit is contained in:
jxxghp
2026-09-30 04:00:48 +08:00
parent 47303a1be6
commit c15b4bf5e5
11 changed files with 567 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
app/chain/media/album.py
View File

@@ -16,7 +16,7 @@ from app.domain.context import (
MusicInfo,
)
from app.domain.meta.metamusic import MetaMusic
from app.domain.music import align_music_tracks
from app.domain.music import align_music_tracks, expand_music_tracks, music_text_key
from app.foundation.text import convert as zhconv_convert
from app.runtime.execution import run_in_threadpool
@@ -36,7 +36,8 @@ def _album_directory_cache_key(
"""将发行偏好及资源证据纳入缓存键，同目录更换种子线索时重新识别。"""
evidence = {
key: getattr(contextual_meta, key, None)
for key in ("album", "artists", "album_artist", "year", "version")
for key in ("album", "artists", "album_artist", "year", "version", "musicbrainz_release_id",
"musicbrainz_release_group_id", "original_year", "release_year", "field_sources")
} if contextual_meta else None
if evidence and evidence["album_artist"]:
evidence["artists"] = [evidence["album_artist"]]
@@ -52,11 +53,38 @@ def _album_context_with_resource(
) -> MetaMusic:
"""以文件自身专辑线索为先，用绑定到本次文件的资源证据补足目录查询。"""
album_meta = MetaMusic.from_album_context(directory.name, metas)
for key in ("album", "album_artist"):
tagged = {
music_text_key(str(getattr(meta, key))) for meta in metas
if getattr(meta, key) and meta.field_sources.get(key) in {"tag", "cue", "manual", "album_tags"}
}
if len(tagged) > 1:
album_meta.organization_error = "同一发行范围内的专辑标签冲突，请分别选择发行版本"
elif tagged:
album_meta.field_sources[key] = "album_tags"
if len(metas) == 1:
setattr(album_meta, key, getattr(metas[0], key))
if key == "album":
album_meta.title = metas[0].album or album_meta.title
elif key == "album_artist":
album_meta.artists = [str(metas[0].album_artist)]
for key in ("musicbrainz_release_id", "musicbrainz_release_group_id", "release_year", "original_year"):
values = {getattr(meta, key) for meta in metas if getattr(meta, key)}
if len(values) > 1 and key != "original_year":
album_meta.organization_error = "同一发行范围内的发行身份或年份冲突，请核对标签"
elif len(values) == 1:
setattr(album_meta, key, next(iter(values)))
album_meta.field_sources[key] = "album_tags"
elif contextual_meta:
setattr(album_meta, key, getattr(contextual_meta, key))
if key in contextual_meta.field_sources:
album_meta.field_sources[key] = contextual_meta.field_sources[key]
if not contextual_meta:
return album_meta
if contextual_meta.album and not any(meta.album for meta in metas):
album_meta.album = contextual_meta.album
album_meta.title = contextual_meta.album
album_meta.field_sources["album"] = contextual_meta.field_sources.get("album", "torrent")
if not album_meta.artists:
album_meta.artists = (
[contextual_meta.album_artist] if contextual_meta.album_artist
@@ -120,9 +148,16 @@ class MediaAlbumOwner(_MediaOwnerBase):
directory: Path,
files: list[Path],
) -> AlbumSignature:
"""按相对路径、大小和纳秒修改时间生成目录缓存签名。"""
"""把音频及同目录 CUE 的路径、大小和修改时间纳入缓存签名。"""
signature: list[tuple[str, int, int]] = []
for path in files:
paths = set(files)
for parent in {path.parent for path in files}:
try:
paths.update(sorted(path for path in parent.iterdir()
if not path.name.startswith(".") and path.suffix.casefold() == ".cue" and path.is_file())[:32])
except OSError:
continue
for path in sorted(paths):
try:
stat = path.stat()
except OSError:
@@ -171,16 +206,7 @@ class MediaAlbumOwner(_MediaOwnerBase):
album: MusicAlbumInfo,
) -> dict[str, MusicInfo]:
"""读取本地标签，并把手动选择发行版的曲目对齐到文件。"""
metas = AudioMetadataHelper.read_many(files)
return {
str(file.resolve()): info
for file, info in cls._align_music_album_tracks(
files,
metas,
album.tracks,
allow_title_override=True,
).items()
}
return cls._album_track_map(files, AudioMetadataHelper.read_many(files), album, allow_title_override=True)
@classmethod
def _album_track_map(
@@ -188,17 +214,28 @@ class MediaAlbumOwner(_MediaOwnerBase):
files: list[Path],
metas: list[MetaMusic],
album: MusicAlbumInfo,
*,
allow_title_override: bool = False,
) -> dict[str, MusicInfo]:
"""将专辑分类与识别事实传递给每条已对位曲目。"""
aligned = cls._align_music_album_tracks(files, metas, album.tracks)
for info in aligned.values():
"""逻辑曲目全部对位后才输出对应物理文件，整轨始终保持 Album 身份。"""
if len(files) != len(metas):
return {}
logical, owners = expand_music_tracks(metas)
aligned = align_music_tracks(logical, album.tracks, allow_title_override=allow_title_override)
result: dict[str, MusicInfo] = {}
for index, path in enumerate(files):
positions = [position for position, owner in enumerate(owners) if owner == index]
if not positions or any(position not in aligned for position in positions):
continue
info = album.to_music_info() if metas[index].music_layout == "image_cue" else deepcopy(album.tracks[aligned[positions[0]]])
for key in ("match_score", "match_coverage", "match_basis"):
if key in album.raw_data:
info.raw_data[key] = album.raw_data[key]
info.set_library_category(album.library_category)
info.classification = deepcopy(album.classification)
info.classification_facts = dict(album.classification_facts)
return {
str(file.resolve()): info
for file, info in aligned.items()
}
result[str(path.resolve())] = info
return result
def _match_music_album_directory(
self,
@@ -211,13 +248,19 @@ class MediaAlbumOwner(_MediaOwnerBase):
"""同步汇总本地专辑证据并委托 MusicBrainz 来源链匹配。"""
metas = AudioMetadataHelper.read_many(files)
album_meta = _album_context_with_resource(directory, metas, contextual_meta)
logical, _owners = expand_music_tracks(metas)
if not logical or album_meta.organization_error or (
len(logical) < self._album_match_min_files
and not (album_meta.musicbrainz_release_id or album_meta.musicbrainz_release_group_id)
):
return {}
regions, scripts = self._music_release_preferences(
list(music_release_regions) if music_release_regions is not None else None,
list(music_release_scripts) if music_release_scripts is not None else None,
)
album = MusicBrainzChain().match_music_album(
album_meta,
metas,
logical,
music_release_regions=list(regions),
music_release_scripts=list(scripts),
)
@@ -236,13 +279,19 @@ class MediaAlbumOwner(_MediaOwnerBase):
"""异步汇总本地专辑证据并委托 MusicBrainz 来源链匹配。"""
metas = await run_in_threadpool(AudioMetadataHelper.read_many, files)
album_meta = _album_context_with_resource(directory, metas, contextual_meta)
logical, _owners = expand_music_tracks(metas)
if not logical or album_meta.organization_error or (
len(logical) < self._album_match_min_files
and not (album_meta.musicbrainz_release_id or album_meta.musicbrainz_release_group_id)
):
return {}
regions, scripts = self._music_release_preferences(
list(music_release_regions) if music_release_regions is not None else None,
list(music_release_scripts) if music_release_scripts is not None else None,
)
album = await MusicBrainzChain().async_match_music_album(
album_meta,
metas,
logical,
music_release_regions=list(regions),
music_release_scripts=list(scripts),
)
@@ -263,7 +312,7 @@ class MediaAlbumOwner(_MediaOwnerBase):
if not directory.is_dir():
return {}
files = self._directory_audio_files(directory, file_paths) if file_paths is not None else self._directory_audio_files(directory)
if len(files) < self._album_match_min_files:
if not files:
return {}
regions, scripts = self._music_release_preferences(
music_release_regions,
@@ -309,7 +358,7 @@ class MediaAlbumOwner(_MediaOwnerBase):
await run_in_threadpool(self._directory_audio_files, directory, file_paths)
if file_paths is not None else await run_in_threadpool(self._directory_audio_files, directory)
)
if len(files) < self._album_match_min_files:
if not files:
return {}
regions, scripts = self._music_release_preferences(
music_release_regions,

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

@@ -4,7 +4,6 @@ from copy import deepcopy
from pathlib import Path
from typing import Any, Dict, Optional, Protocol, Union, cast
from app.application.audio import AudioMetadataHelper
from app.application.configuration import (
get_chain_runtime_config_snapshot,
)
@@ -32,10 +31,10 @@ from app.domain.meta.metamusic import MetaMusic
from app.domain.music import (
music_album_matches,
music_artist_matches,
music_release_year_matches,
music_title_matches,
music_track_title_is_weak,
music_version_matches,
music_year_matches,
)
from app.runtime.log import logger
from app.schemas.transfer import TransferInfo
@@ -203,7 +202,7 @@ class _MusicFileFilterBase(_TransferOwnerBase):
按目录缓存，同一专辑目录内的后续文件不会重复请求远端。
"""
# 目录级匹配需要读取本地音频时长，远端存储文件无法参与
if file_meta.media_id or getattr(file_item, "storage", "local") != "local":
if (file_meta.media_id and not file_meta.musicbrainz_release_id) or getattr(file_item, "storage", "local") != "local":
return file_meta, None
try:
matched = MediaChain().recognize_music_album_directory(
@@ -221,11 +220,12 @@ class _MusicFileFilterBase(_TransferOwnerBase):
return file_meta, None
weak_title = music_track_title_is_weak(file_meta)
evidence_matches = (
(not file_meta.artists or music_artist_matches(info, file_meta.artists))
(not file_meta.artists or file_meta.field_sources.get("artists") in {"directory", "torrent", "album_tags"}
or music_artist_matches(info, file_meta.artists))
and (weak_title or music_title_matches(info, file_meta.title))
and (not file_meta.album or not info.album or music_album_matches(info, file_meta.album))
and ((weak_title and not file_meta.version) or music_version_matches(info, file_meta))
and music_year_matches(info, file_meta)
and music_release_year_matches(info, file_meta)
)
if not evidence_matches:
logger.warning(
@@ -388,9 +388,7 @@ class FileFilterMixin(_TransferOwnerBase):
]
if not audio_paths:
return {}, None
image_paths = {path for path in audio_paths if AudioMetadataHelper.read(path).music_layout == "image_cue"}
aligned_tracks = MediaChain._align_selected_music_album([path for path in audio_paths if path not in image_paths], album)
aligned_tracks.update({str(path.resolve()): album.to_music_info() for path in image_paths})
aligned_tracks = MediaChain._align_selected_music_album(audio_paths, album)
if len(aligned_tracks) != len(audio_paths):
return {}, (
f"所选专辑只能对齐 {len(aligned_tracks)} / {len(audio_paths)} "

70
app/domain/music.py
View File

@@ -215,6 +215,63 @@ def music_track_title_is_weak(meta: MetaMusic) -> bool:
return bool(re.fullmatch(r"(?:(?:track|audio|音轨|曲目)[\s._-]*\d*|unknown|untitled)", title, re.I))
def music_album_title_is_weak(meta: MetaMusic) -> bool:
"""普通收件目录不是专辑证据，标签或种子明确提供的同名专辑仍有效。"""
title = meta.album or meta.title
if not title:
return True
source = meta.field_sources.get("album" if meta.album else "title")
if source in {"tag", "album_tags", "cue", "torrent", "manual", "remote"}:
return False
return music_text_key(title) in {
"music", "musics", "audio", "downloads", "download", "inbox", "unknown", "untitled",
"various", "variousartists", "va", "音乐", "音樂", "下载", "下載", "未分类", "未分類",
}
def expand_music_tracks(metas: list[MetaMusic]) -> tuple[list[MetaMusic], list[int]]:
"""把整轨 CUE 投影为用于匹配的逻辑曲目，并保留各曲所属物理文件索引。"""
tracks: list[MetaMusic] = []
owners: list[int] = []
for owner, meta in enumerate(metas):
if meta.organization_error:
return [], []
if meta.music_layout != "image_cue":
tracks.append(meta)
owners.append(owner)
continue
if not meta.cue_tracks:
return [], []
for index, cue in enumerate(meta.cue_tracks):
start, number = cue.get("start_frame"), cue.get("number")
end = meta.cue_tracks[index + 1].get("start_frame") if index + 1 < len(meta.cue_tracks) else (
meta.duration * 75 if meta.duration else None
)
if not isinstance(start, int) or start < 0 or not isinstance(number, int) or not 1 <= number <= 999:
return [], []
if end is not None and (not isinstance(end, (int, float)) or end <= start):
return [], []
track = MetaMusic.from_dict(meta.to_dict())
track.title = str(cue.get("title") or f"Track {number}")
track.field_sources["title"] = "cue" if cue.get("title") else "filename"
track.artists = [str(cue["artist"])] if cue.get("artist") else list(meta.artists)
track.track_number = number
track.duration = round((end - start) / 75) if end is not None else None
track.isrc = str(cue["isrc"]) if cue.get("isrc") else None
track.field_sources["track_number"] = "cue"
for key in ("duration", "isrc"):
if getattr(track, key) is not None:
track.field_sources[key] = "cue"
else:
track.field_sources.pop(key, None)
track.music_type = "recording"
track.music_layout, track.cue_filename, track.cue_tracks = None, None, []
track.media_id, track.media_source, track.musicbrainz_release_track_id = None, None, None
tracks.append(track)
owners.append(owner)
return tracks, owners
def _alignment_title_key(title: Optional[str]) -> str:
"""对位仅忽略名称排版差异，保留 live/remix 等版本正文和纯符号歌曲名。"""
return music_text_key(title) or re.sub(r"\s+", "", normalize("NFKC", str(title or ""))).casefold()
@@ -234,6 +291,10 @@ def _music_track_pair_score(meta: MetaMusic, track: MusicInfo, allow_title_overr
if meta.media_id != track.media_id and not allow_title_override:
return 0.0
identity_match = identity_match or meta.media_id == track.media_id
if (not allow_title_override and meta.artists and track.artists
and meta.field_sources.get("artists") not in {"directory", "torrent", "album_tags"}
and not music_artist_matches(track, meta.artists)):
return 0.0
duration_close = False
if meta.duration and track.duration:
delta = abs(meta.duration - track.duration)
@@ -436,6 +497,15 @@ def music_year_matches(music: MusicInfo, meta: MetaMusic) -> bool:
return str(music.year).strip() == str(meta.year).strip()
def music_release_year_matches(music: MusicInfo, meta: MetaMusic) -> bool:
"""具体发行按当前发行年核验；已分离的原始年不能再冒充当前年制造再版冲突。"""
if meta.release_year:
return not music.release_year or meta.release_year == music.release_year
if meta.original_year or (meta.musicbrainz_release_id and meta.musicbrainz_release_id == music.musicbrainz_release_id):
return True
return music_year_matches(music, meta) or bool(meta.year and str(meta.year) == str(music.original_year))
def _isrc_key(value: Optional[str]) -> Optional[str]:
"""校验 12 位 ISRC 结构，兼容展示前缀、空白和分隔符，不接受占位值。"""
code = re.sub(r"[\s-]+", "", str(value or ""))

198
app/modules/musicbrainz/__init__.py
View File

@@ -19,7 +19,9 @@ from app.domain.media import is_media_source_enabled, is_media_source_selected
from app.domain.meta.metabase import MetaBase
from app.domain.meta.metamusic import MetaMusic
from app.domain.music import (
align_music_tracks,
music_album_matches,
music_album_title_is_weak,
music_artist_affix_matches,
music_artist_matches,
music_base_title,
@@ -27,6 +29,7 @@ from app.domain.music import (
music_text_key,
music_title_matches,
music_titles,
music_track_title_is_weak,
music_version_matches,
music_year_matches,
unique_music_texts,
@@ -806,14 +809,14 @@ class MusicBrainzModule(_ModuleBase):
适用于无标签整专目录：用专辑名、歌手搜索候选发行版本，再用曲目数、
总时长和逐曲时长相似度打分，选出最可信的版本并返回其曲目表。
"""
if not tracks:
if not tracks or meta.organization_error:
return None
preference = self._release_preference(
music_release_regions,
music_release_scripts,
)
details: list[dict[str, Any]] = []
releases = self._search_release_candidates(
releases = [{"id": meta.musicbrainz_release_id}] if meta.musicbrainz_release_id else self._search_release_candidates(
meta,
tracks,
limit=limit,
@@ -835,14 +838,14 @@ class MusicBrainzModule(_ModuleBase):
music_release_scripts: Optional[list[str]] = None,
) -> Optional[MusicAlbumInfo]:
"""异步按目录线索和曲目特征匹配 MusicBrainz 发行版本。"""
if not tracks:
if not tracks or meta.organization_error:
return None
preference = self._release_preference(
music_release_regions,
music_release_scripts,
)
details: list[dict[str, Any]] = []
releases = await self._async_search_release_candidates(
releases = [{"id": meta.musicbrainz_release_id}] if meta.musicbrainz_release_id else await self._async_search_release_candidates(
meta,
tracks,
limit=limit,
@@ -866,24 +869,40 @@ class MusicBrainzModule(_ModuleBase):
preference: Optional[_MusicReleasePreference] = None,
) -> Optional[MusicAlbumInfo]:
"""对已获取的发行详情统一打分并投影最佳专辑。"""
best_album: Optional[MusicAlbumInfo] = None
best_score = 0.0
best_key: Optional[tuple[float, int, int]] = None
selected_preference = preference or cls._release_preference()
ranked: list[tuple[float, tuple[int, int], MusicAlbumInfo]] = []
seen: set[str] = set()
for detail in details:
release_id = str(detail.get("id") or "")
if not release_id or release_id in seen:
continue
seen.add(release_id)
summary = cls._release_track_summary(detail)
score = cls._score_release(meta, tracks, detail, summary)
region_rank, script_rank = cls._release_preference_sort_key(
detail,
selected_preference,
)
candidate_key = (score, -region_rank, -script_rank)
if best_key is None or candidate_key > best_key:
best_score = score
best_key = candidate_key
best_album = cls._release_to_album(detail)
# 得分低于阈值时宁可不匹配，避免把曲目写到错误的专辑上
return best_album if best_score >= cls._album_match_threshold else None
album = cls._release_to_album(detail)
if album and score > 0:
ranked.append((score, cls._release_preference_sort_key(detail, selected_preference), album))
if not ranked:
return None
ranked.sort(key=lambda entry: (-entry[0], entry[1]))
best_score, _preference, best = ranked[0]
if best_score < cls._album_match_threshold:
return None
equivalent = cls._release_content_key(best)
for score, _rank, candidate in ranked[1:]:
if best_score - score <= 5 and cls._release_content_key(candidate) != equivalent:
return None
best.raw_data.update(match_score=round(best_score, 2), match_coverage=1.0,
match_basis="release_id" if meta.musicbrainz_release_id else "track_evidence")
return best
@staticmethod
def _release_content_key(album: MusicAlbumInfo) -> tuple[Optional[str], tuple[tuple[int, int, str], ...]]:
"""同发行组且录音序列一致的文字/地区版本可按偏好选取，其他近分结果保持歧义。"""
return (album.musicbrainz_release_group_id or album.musicbrainz_release_id, tuple(sorted(
(track.disc_number or 1, track.track_number or 0, track.media_id or "")
for track in album.tracks
)))
_album_match_threshold = 60.0
@@ -903,8 +922,6 @@ class MusicBrainzModule(_ModuleBase):
request.path, params=request.params
)
self._merge_release_candidates(releases, seen, payload)
if len(releases) >= search_limit:
break
return self._rank_release_candidates(
releases,
preference or self._release_preference(),
@@ -926,8 +943,6 @@ class MusicBrainzModule(_ModuleBase):
request.path, params=request.params
)
self._merge_release_candidates(releases, seen, payload)
if len(releases) >= search_limit:
break
return self._rank_release_candidates(
releases,
preference or self._release_preference(),
@@ -941,8 +956,8 @@ class MusicBrainzModule(_ModuleBase):
) -> list[dict[str, Any]]:
"""在相近搜索相关度内优先目标地区与字形，保留强相关性边界。"""
def sort_key(release: dict[str, Any]) -> tuple[int, int, int, int]:
"""相关度每五分成组，组内应用发行偏好并保留原始得分。"""
def sort_key(release: dict[str, Any]) -> tuple[int, int, int, int, int]:
"""同等相关度先看多首录音的共同支持，再应用地区和字形偏好。"""
try:
score = int(release.get("score") or 0)
except (TypeError, ValueError):
@@ -951,7 +966,7 @@ class MusicBrainzModule(_ModuleBase):
release,
preference,
)
return -(score // 5), region_rank, script_rank, -score
return -(score // 5), -len(release.get("_recording_hits") or []), region_rank, script_rank, -score
return sorted(releases, key=sort_key)
@@ -961,12 +976,27 @@ class MusicBrainzModule(_ModuleBase):
seen: set[str],
payload: Optional[dict[str, Any]],
) -> None:
"""按首次命中顺序合并并去重 Release 搜索响应。"""
for item in (payload or {}).get("releases") or []:
"""合并发行搜索与录音搜索关联的发行，保留不同录音共同支持的证据。"""
candidates = list((payload or {}).get("releases") or [])
for recording in (payload or {}).get("recordings") or []:
candidates.extend({**release, "score": recording.get("score", 0),
"_recording_hits": [recording["id"]] if recording.get("id") else []}
for release in recording.get("releases") or [])
existing = {item["id"]: item for item in releases}
for item in candidates:
release_id = item.get("id")
if release_id and release_id not in seen:
seen.add(release_id)
releases.append(item)
copied = dict(item)
releases.append(copied)
existing[release_id] = copied
elif release_id and release_id in existing:
previous = existing[release_id]
previous["_recording_hits"] = sorted(set(previous.get("_recording_hits") or []) | set(item.get("_recording_hits") or []))
try:
previous["score"] = max(int(previous.get("score") or 0), int(item.get("score") or 0))
except (TypeError, ValueError):
pass
@classmethod
def _release_search_requests(
@@ -975,11 +1005,11 @@ class MusicBrainzModule(_ModuleBase):
tracks: list[MetaMusic],
limit: int,
) -> list[_MusicBrainzRequestPlan]:
"""构造发行候选查询计划，统一同步与异步的限额和参数。"""
"""曲名必须查询 recording 索引，再从响应提取发行；release 索引不支持 recording 字段。"""
normalized_limit = max(1, min(limit, 25))
return [
_MusicBrainzRequestPlan(
path="/release",
path="/recording" if query.startswith("(") else "/release",
params={
"query": query,
"limit": normalized_limit,
@@ -1013,7 +1043,9 @@ class MusicBrainzModule(_ModuleBase):
def _release_queries(cls, meta: MetaMusic, tracks: list[MetaMusic]) -> list[str]:
"""构造专辑搜索表达式：优先专辑名+歌手，无专辑线索时用曲名兜底。"""
queries: list[str] = []
album_title = meta.album or meta.title
if meta.musicbrainz_release_group_id:
return [f'rgid:{cls._escape_query(meta.musicbrainz_release_group_id)}']
album_title = None if music_album_title_is_weak(meta) else meta.album or meta.title
artist = meta.artists[0] if meta.artists else meta.album_artist
if album_title:
if artist:
@@ -1023,7 +1055,7 @@ class MusicBrainzModule(_ModuleBase):
queries.append(f'release:"{cls._escape_query(album_title)}"')
# 目录名无意义时（如 Various Artists 合集），用代表性曲名反查所属发行版本
titles = cls._unique_texts(
[track.title for track in tracks if track.title and not track.title.strip().isdigit()]
[track.title for track in tracks if track.title and not music_track_title_is_weak(track)]
)[:3]
if titles:
recording_clause = " OR ".join(
@@ -1062,75 +1094,49 @@ class MusicBrainzModule(_ModuleBase):
summary: list[dict[str, Any]],
) -> float:
"""给候选发行版本打分（0-100），综合标题、歌手、曲目数和时长相似度。"""
local_count = len(tracks)
release_count = len(summary)
if not release_count:
album = cls._release_to_album(detail)
if not album or not tracks or len(album.tracks) != len(summary):
return 0.0
# 曲目数差异过大直接排除，避免单曲误命中整专或反之
diff = abs(local_count - release_count)
if diff > max(4, int(local_count * 0.5)):
for field in ("musicbrainz_release_id", "musicbrainz_release_group_id"):
expected = getattr(meta, field)
if expected and expected != getattr(album, field):
return 0.0
if meta.release_year and album.year and meta.release_year != album.year:
return 0.0
# 本地文件比发行版本多出的曲目无法被覆盖，超出容忍范围视为错误候选
if local_count > release_count and diff > max(1, int(release_count * 0.25)):
known_title = not music_album_title_is_weak(meta)
title_sim = cls._text_similarity(meta.album or meta.title, album.title) if known_title else 0.0
if known_title and title_sim < 0.7 and not music_album_matches(album.to_music_info(), meta.album or meta.title):
return 0.0
score = 0.0
# 标题相似度：专辑目录名或文件标签中的专辑名/曲名
title_hints = cls._unique_texts([meta.album, meta.title])
title_sim = max(
(cls._text_similarity(hint, detail.get("title")) for hint in title_hints),
default=0.0,
)
artist_names = cls._artist_credits(detail.get("artist-credit"))[0]
if meta.artists and artist_names:
artist_sim = max(
cls._text_similarity(meta.artists[0], name) for name in artist_names
)
score += 40 * title_sim + 15 * artist_sim
else:
# 缺少歌手线索时把权重让给标题
score += 50 * title_sim
# 曲目数：完全一致是最强信号
if diff == 0:
score += 15
elif diff == 1:
score += 8
elif diff <= max(2, int(local_count * 0.15)):
score += 2
# 曲名重合度：部分曲目目录（只下载了整专的一部分）依靠曲名对位确认
release_titles = {cls._match_text(item["title"]) for item in summary}
named_tracks = [track for track in tracks if track.title and not track.title.strip().isdigit()]
if named_tracks and release_titles:
overlap = sum(
1 for track in named_tracks if cls._match_text(track.title) in release_titles
)
score += 15 * overlap / len(named_tracks)
# 总时长：无损整专 rip 的总时长与 MusicBrainz 记录高度接近
local_total = sum(track.duration or 0 for track in tracks)
release_total = sum(item["length"] or 0 for item in summary)
local_durations = [track.duration for track in tracks if track.duration]
if local_durations and release_total:
delta = abs(local_total - release_total) / max(local_total, release_total)
if delta <= 0.02:
score += 15
elif delta <= 0.05:
score += 10
elif delta <= 0.10:
score += 5
# 逐曲时长对位：曲目数一致时逐首比较
if diff == 0 and len(local_durations) == local_count:
similarities = []
for track, item in zip(
sorted(tracks, key=lambda item: (item.disc_number or 1, item.track_number or 0)),
summary,
):
if track.duration and item["length"]:
similarities.append(cls._duration_similarity(track.duration, item["length"]))
if similarities:
score += 15 * sum(similarities) / len(similarities)
return score
artists = [meta.album_artist] if meta.album_artist else meta.artists
artist_match = bool(artists and music_artist_matches(album.to_music_info(), artists))
if artists and not artist_match:
return 0.0
if meta.version and not music_version_matches(album.to_music_info(), meta):
return 0.0
aligned = align_music_tracks(tracks, album.tracks)
if len(aligned) != len(tracks):
return 0.0
# 曲序只是槽位。无标签曲名且没有可比时长时，除非提供具体发行ID，否则不能确认内容。
named = sum(not music_track_title_is_weak(track) for track in tracks)
durations = [
cls._duration_similarity(tracks[local].duration, album.tracks[remote].duration)
for local, remote in aligned.items() if tracks[local].duration and album.tracks[remote].duration
]
if not meta.musicbrainz_release_id and (not named and not durations or not known_title and len(tracks) < 2):
return 0.0
score = 40.0 + 25 * title_sim + 15 * artist_match
year_hint = meta.original_year or meta.year
score += (10 if year_hint else 15) * sum(durations) / len(tracks)
# 原始年份也可能描述合辑中歌曲的首发年，仅用于排名；当前发行年份才是版本约束。
year_matches = str(year_hint) == str(album.original_year) or (not meta.original_year and str(year_hint) == str(album.year))
score += 5 if year_hint and year_matches else 0
score += 5 if len(tracks) == len(album.tracks) else 0
if meta.musicbrainz_release_id:
score += 40
return min(score, 100.0)
@staticmethod
def _duration_similarity(left: int, right: int) -> float:
def _duration_similarity(left: Optional[int], right: Optional[int]) -> float:
"""比较两个时长的接近程度，完全一致为 1，差异越大越接近 0。"""
if not left or not right:
return 0.0

4
docs/architecture-overview.md
View File

@@ -763,7 +763,7 @@ flowchart LR
| 指标 | 当前值 |
|---|---:|
| Python 模块 | 1054 |
| 内部导入边 | 8,963 |
| 内部导入边 | 8,962 |
| 非平凡 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,963，音乐交互包未新增 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` 为 574、`app.factory` 为 586、`app.main` 为 588。站点图片域名快照新增 `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,962，音乐交互包未新增 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` 为 574、`app.factory` 为 586、`app.main` 为 588。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。
架构专项验证分为两个 CI 投影：`Check event semantic policy` 先运行依赖、Adapter、出口和 Event
语义门禁，`Check host architecture snapshot` 再执行快照测试及一次

10
docs/mcp-api.md
View File

@@ -539,6 +539,16 @@ SSE 的 `candidate_items` 是站点原始返回数量，`match_counts` 记录身
重复版本、同分候选和明显时长冲突保留未匹配；手选发行允许按唯一位置纠正旧曲名，
但不绕过时长冲突。手选批次不能完整对位时会在文件操作前返回核对提示。
目录匹配优先直查明确的 `musicbrainz_release_id`，`musicbrainz_release_group_id` 只约束发行搜索。
普通 `Music`/`inbox` 等目录名不作为专辑名；真实标签或种子明确给出的同名作品仍可搜索。
曲名反查使用 MusicBrainz 的 Recording 搜索及其关联发行；Release 搜索不支持 `recording` 字段。
同等搜索相关度优先检查被多首录音共同命中的发行，避免单曲的多个版本占满候选预算。
候选必须覆盖全部本地逻辑曲目；部分下载不要求覆盖远端专辑的全部歌曲。
身份、当前发行年和明确艺人冲突不能靠其它分数抵消；原始年份只辅助排名。
不同录音序列的近分候选不自动采用，同发行组且录音序列相同的地区/文字版本沿用用户偏好。
匹配成功时 `raw_data.match_score` 是 0–100 的排序分数，不是概率；`match_coverage=1` 仅表示当前输入文件全覆盖。
整轨 CUE 以逻辑歌曲数和相邻索引时长参与匹配，返回仍为物理文件的专辑身份；修改 CUE 会使目录缓存失效。
| 方法 | 路径 | 说明 |
| :--- | :--- | :--- |
| GET | `/api/v1/media/search` | 当 `type=music` 或指定音乐 `media_source` 时按歌曲或专辑关键词搜索音乐元数据；艺术家统一由 `type=person` 搜索，支持 TMDB 与 MusicBrainz 来源。参数：`title`、`type`、`count`、可重复的 `media_source` 枚举，以及可选的 `music_type` 实体过滤 |

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

@@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
| 指标 | 当前值 | 解释 |
|---|---:|---|
| 宿主 Python 模块 / 内部依赖边 | 1054 / 8,963 | `dependency-baseline.json` 当前快照；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档 |
| 宿主 Python 模块 / 内部依赖边 | 1054 / 8,962 | `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 等基线均为零 |

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

@@ -66,6 +66,19 @@ Music recognition, exploration, album and artist navigation, and recognition-cac
Manual album selection may correct names using unique positions, but still
rejects clear duration conflicts. An incomplete manual alignment stops before
file operations and asks the caller to check the selected edition and tracks.
- A concrete `musicbrainz_release_id` uses direct release lookup; a release-group
ID constrains release search. Generic inbox directory names are not album
evidence. All local logical tracks must align, even for a partial download.
Conflicting identities, current edition years, and performer tags reject a
candidate. Near ties with different recording sequences remain unmatched;
equivalent editions in one release group retain region/script preferences.
Successful `raw_data.match_score` is a ranking score, not a probability, and
`match_coverage=1` covers the supplied files, not necessarily the full album.
CUE images contribute logical tracks/durations and keep an album identity;
changing the CUE invalidates the directory match cache.
- 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.
## Operations

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

@@ -1097,8 +1097,8 @@
"runtime_only": true
}
},
"edge_count": 8963,
"edge_sha256": "dd3d4f1797a5e9d4094a118a571323123d70086ff4e26597f6bb7dbe85b21493",
"edge_count": 8962,
"edge_sha256": "5498df1985d898c3224ff07cea295817da004a04bb08283bd642a32f143c9354",
"edges": [
"app -> app.foundation",
"app -> app.foundation.environment",
@@ -5516,7 +5516,6 @@
"app.chain.transfer.facade -> app.runtime",
"app.chain.transfer.facade -> app.runtime.reload",
"app.chain.transfer.filter -> app.application",
"app.chain.transfer.filter -> app.application.audio",
"app.chain.transfer.filter -> app.application.configuration",
"app.chain.transfer.filter -> app.application.directory",
"app.chain.transfer.filter -> app.application.history",

9
tests/test_music_album_match.py
View File

@@ -417,17 +417,16 @@ def test_async_album_fallback_propagates_cancellation_during_path_check(
def test_recognize_album_directory_skips_single_file(tmp_path, media_chain, monkeypatch):
"""单文件目录不走专辑匹配，交给单曲识别链路。"""
"""没有专辑证据的普通单文件目录不发起整专查询，整轨 CUE 另按逻辑曲数判断。"""
album_dir = tmp_path / "单曲"
album_dir.mkdir()
(album_dir / "晴天.wav").write_bytes(b"RIFF")
def fake_run_module(method, **kwargs):
raise AssertionError("单文件目录不应触发专辑匹配")
monkeypatch.setattr(media_chain, "_match_music_album_directory", fake_run_module)
source = Mock(match_music_album=Mock(side_effect=AssertionError("无专辑证据的单文件不应发起查询")))
monkeypatch.setattr("app.chain.media.album.MusicBrainzChain", Mock(return_value=source))
assert media_chain.recognize_music_album_directory(album_dir) == {}
source.match_music_album.assert_not_called()
def test_recognize_album_directory_invalidates_cache_after_same_count_rename(

283
tests/test_music_release_matching.py Normal file
View File

@@ -0,0 +1,283 @@
"""实际发行身份、曲目覆盖率、候选歧义及 CUE 整专匹配的回归。"""
import asyncio
from copy import deepcopy
from unittest.mock import AsyncMock, Mock
import pytest
from app.application.audio import AudioMetadataHelper
from app.chain.media import MediaChain
from app.chain.media.album import _album_context_with_resource, _album_directory_cache_key
from app.chain.media.cache import AlbumDirectoryCache
from app.domain.context import MusicAlbumInfo
from app.domain.meta.metamusic import MetaMusic
from app.domain.music import expand_music_tracks, music_release_year_matches
from app.modules.musicbrainz import MusicBrainzModule
from tests.test_music_album_match import ALBUM_TRACKS, _local_tracks, _release_detail
from tests.test_music_cue import _image_pair
@pytest.mark.parametrize("asynchronous", [False, True])
def test_release_id_uses_direct_lookup_without_search(monkeypatch, asynchronous):
"""真实发行ID进入同步/异步直查，不重复按文件名搜索候选。"""
module = MusicBrainzModule()
detail = _release_detail("release-direct", "七里香", "周杰伦", ALBUM_TRACKS)
request = AsyncMock(return_value=detail) if asynchronous else Mock(return_value=detail)
monkeypatch.setattr(module, "_async_request_json" if asynchronous else "_request_json", request)
meta = MetaMusic(album="七里香", artists=["周杰伦"], musicbrainz_release_id="release-direct")
album = asyncio.run(module.async_match_music_album(meta, _local_tracks())) if asynchronous else module.match_music_album(meta, _local_tracks())
assert album is not None
assert album.musicbrainz_release_id == "release-direct"
request.assert_called_once()
assert request.call_args.args[0] == "/release/release-direct"
assert album.raw_data["match_basis"] == "release_id"
@pytest.mark.parametrize("field,value", [("musicbrainz_release_id", "other"), ("musicbrainz_release_group_id", "other"), ("release_year", 2023)])
def test_identity_and_current_release_year_conflicts_are_not_offset_by_score(field, value):
"""曲目完全一致也不能抵消明确发行身份或当前发行年的冲突。"""
meta = MetaMusic(album="七里香", artists=["周杰伦"], **{field: value})
detail = _release_detail("release-1", "七里香", "周杰伦", ALBUM_TRACKS)
assert MusicBrainzModule._select_release_match(meta, _local_tracks(), [detail]) is None
def test_different_artist_with_identical_titles_and_durations_is_rejected():
"""同名专辑的另一位艺人不能凭相同结构获得高分通过。"""
detail = _release_detail("release-1", "七里香", "Different Artist", ALBUM_TRACKS)
assert MusicBrainzModule._select_release_match(MetaMusic(album="七里香", artists=["周杰伦"]), _local_tracks(), [detail]) is None
def test_different_recordings_with_tied_scores_require_confirmation():
"""实际录音序列不同的近分候选不能按搜索返回顺序自动选中。"""
first = _release_detail("release-1", "七里香", "周杰伦", ALBUM_TRACKS)
other = deepcopy(first)
other["id"] = "release-2"
other["media"][0]["tracks"][1]["recording"]["id"] = "alternate-recording"
meta = MetaMusic(album="七里香", artists=["周杰伦"])
assert MusicBrainzModule._select_release_match(meta, _local_tracks(), [first, other]) is None
assert MusicBrainzModule._select_release_match(meta, _local_tracks(), [other, first]) is None
def test_equivalent_release_content_ignores_response_order():
"""同发行组的录音和位置不变时，响应数组顺序不产生虚假歧义。"""
first = _release_detail("release-1", "七里香", "周杰伦", ALBUM_TRACKS)
other = deepcopy(first)
other["id"] = "release-2"
other["media"][0]["tracks"].reverse()
assert MusicBrainzModule._select_release_match(MetaMusic(album="七里香", artists=["周杰伦"]), _local_tracks(), [first, other]) is not None
def test_actual_performer_conflict_is_rejected_but_album_credit_is_not_performer():
"""合辑逐曲艺人必须核对，目录继承的专辑艺人不能冒充每首歌的演唱者。"""
detail = _release_detail("release-1", "七里香", "Various Artists", ALBUM_TRACKS)
for track in detail["media"][0]["tracks"]:
track["recording"]["artist-credit"] = [{"artist": {"id": "guest", "name": "Guest"}}]
local = _local_tracks()
for track in local:
track.artists = ["Another Singer"]
track.field_sources["artists"] = "tag"
meta = MetaMusic(album="七里香", album_artist="Various Artists")
assert MusicBrainzModule._select_release_match(meta, local, [detail]) is None
for track in local:
track.artists = ["Various Artists"]
track.field_sources["artists"] = "directory"
assert MusicBrainzModule._select_release_match(meta, local, [detail]) is not None
def test_track_coverage_and_matching_are_order_independent():
"""评分实际使用曲目对位，乱序文件不改变匹配结果或分数。"""
detail = _release_detail("release-1", "七里香", "周杰伦", ALBUM_TRACKS)
meta = MetaMusic(album="七里香", artists=["周杰伦"])
first = MusicBrainzModule._select_release_match(meta, _local_tracks(), [detail])
shuffled = MusicBrainzModule._select_release_match(meta, list(reversed(_local_tracks())), [detail])
assert first is not None and shuffled is not None
assert first.raw_data["match_score"] == shuffled.raw_data["match_score"] == 100
assert first.raw_data["match_coverage"] == 1
def test_unmatched_bonus_and_duplicate_files_reject_entire_candidate():
"""不能先接受高分专辑，再让额外文件随排序绑定或部分冒认身份。"""
detail = _release_detail("release-1", "七里香", "周杰伦", ALBUM_TRACKS)
meta = MetaMusic(album="七里香", artists=["周杰伦"])
for extras in ([MetaMusic(title="Unrelated Bonus")], [_local_tracks()[0]]):
assert MusicBrainzModule._select_release_match(meta, [*_local_tracks(), *extras], [detail]) is None
def test_partial_album_with_known_titles_still_matches():
"""只下载部分歌曲时要求本地文件全覆盖，不要求必须凑齐完整专辑。"""
detail = _release_detail("release-1", "七里香", "周杰伦", ALBUM_TRACKS)
assert MusicBrainzModule._select_release_match(MetaMusic(album="七里香", artists=["周杰伦"]), _local_tracks()[1:], [detail]) is not None
def test_absence_of_bonus_tracks_does_not_prove_standard_edition():
"""已有曲目也可能来自豪华版的部分下载，少了附赠曲本身不能消除发行歧义。"""
standard = _release_detail("standard", "七里香", "周杰伦", ALBUM_TRACKS)
deluxe = _release_detail("deluxe", "七里香", "周杰伦", [*ALBUM_TRACKS, ("Bonus", 200)])
assert MusicBrainzModule._select_release_match(MetaMusic(album="七里香", artists=["周杰伦"]), _local_tracks(), [standard, deluxe]) is None
def test_explicit_live_version_cannot_select_studio_album():
"""已知现场录音版本先参与准入，不能先选录音室版再逐曲失败。"""
studio = _release_detail("studio", "七里香", "周杰伦", ALBUM_TRACKS)
live = deepcopy(studio)
live["id"] = "live"
live["release-group"]["secondary-types"] = ["Live"]
meta = MetaMusic(album="七里香", title="七里香", artists=["周杰伦"], version="Live")
album = MusicBrainzModule._select_release_match(meta, _local_tracks(), [studio, live])
assert album is not None and album.musicbrainz_release_id == "live"
@pytest.mark.parametrize("name", ["Music", "inbox", "Downloads", "Various Artists"])
def test_generic_directory_does_not_generate_album_search(name):
"""普通目录只能使用实际曲名反查，不能重复查询伪专辑。"""
queries = MusicBrainzModule._release_queries(MetaMusic(title=name), _local_tracks())
assert queries
assert all("release:" not in query for query in queries)
def test_real_tag_album_and_numeric_song_are_searchable():
"""真正名为 Music 的专辑和数字歌曲不会被目录占位规则丢弃。"""
queries = MusicBrainzModule._release_queries(
MetaMusic(album="Music", field_sources={"album": "tag"}),
[MetaMusic(title="22", field_sources={"title": "tag"})],
)
assert any('release:"Music"' in query for query in queries)
assert any('recording:"22"' in query for query in queries)
def test_release_group_id_limits_queries_to_its_namespace():
"""发行组ID只约束发行搜索，不能冒充具体发行或录音ID。"""
assert MusicBrainzModule._release_queries(MetaMusic(musicbrainz_release_group_id="group-1"), _local_tracks()) == ["rgid:group-1"]
@pytest.mark.parametrize("asynchronous", [False, True])
def test_track_title_lookup_uses_recording_index_then_related_releases(monkeypatch, asynchronous):
"""按官方 Recording 搜索响应形状回放，曲名反查不能向 Release 索引发送 recording 字段。"""
module = MusicBrainzModule()
detail = _release_detail("related-release", "七里香", "周杰伦", ALBUM_TRACKS)
recording_payload = {"recordings": [{"id": "rec-1", "score": 100, "title": "我的地盘",
"releases": [{"id": "related-release", "title": "七里香"}]}]}
def respond(path, params=None):
"""重放搜索和详情，任何错误的实体路由都直接失败。"""
if path == "/recording":
assert 'recording:"我的地盘"' in params["query"]
return recording_payload
assert path == "/release/related-release"
return detail
request = AsyncMock(side_effect=respond) if asynchronous else Mock(side_effect=respond)
monkeypatch.setattr(module, "_async_request_json" if asynchronous else "_request_json", request)
meta = MetaMusic(title="Music")
album = asyncio.run(module.async_match_music_album(meta, _local_tracks())) if asynchronous else module.match_music_album(meta, _local_tracks())
assert album is not None and album.musicbrainz_release_id == "related-release"
assert request.call_count == 2
def test_multiple_recording_hits_rank_shared_album_ahead_of_first_songs_single():
"""有限详情预算先用于多首歌共同出现的发行，不被第一首歌的多个单曲版占满。"""
releases, seen = [], set()
payload = {"recordings": [
{"id": "one", "score": "100", "releases": [{"id": "single"}, {"id": "album"}]},
{"id": "two", "score": "100", "releases": [{"id": "album"}]},
]}
original = deepcopy(payload)
MusicBrainzModule._merge_release_candidates(releases, seen, {"releases": [{"id": "album", "score": "80"}]})
MusicBrainzModule._merge_release_candidates(releases, seen, payload)
ranked = MusicBrainzModule._rank_release_candidates(releases, MusicBrainzModule._release_preference())
assert ranked[0]["id"] == "album"
assert ranked[0]["score"] == 100
assert len(ranked) == 2
assert payload == original
def test_reissue_keeps_original_and_current_year_distinct():
"""原始2004年与2023再版并不冲突，明确的当前年份仍必须匹配。"""
detail = _release_detail("reissue", "七里香", "周杰伦", ALBUM_TRACKS)
detail["date"] = "2023-01-01"
detail["release-group"]["first-release-date"] = "2004-08-03"
meta = MetaMusic(album="七里香", artists=["周杰伦"], year=2004, original_year=2004, release_year=2023)
album = MusicBrainzModule._select_release_match(meta, _local_tracks(), [detail])
assert album is not None
assert album.original_year == 2004 and album.year == 2023
assert music_release_year_matches(album.tracks[0], meta)
meta.release_year = 2022
assert not music_release_year_matches(album.tracks[0], meta)
def test_context_and_cache_include_release_identity_and_years(tmp_path):
"""标签和种子传播发行证据，手动改变发行上下文不会命中旧目录缓存。"""
tracks = [MetaMusic(album="七里香", musicbrainz_release_id="release-1", original_year=2004, release_year=2023)]
context = _album_context_with_resource(tmp_path, tracks, None)
assert context.musicbrainz_release_id == "release-1"
assert (context.original_year, context.release_year) == (2004, 2023)
first = _album_directory_cache_key(tmp_path, (), (), context)
context.musicbrainz_release_id = "release-2"
assert first != _album_directory_cache_key(tmp_path, (), (), context)
def test_conflicting_release_context_never_reaches_network(tmp_path, monkeypatch):
"""直接目录识别遇到多发行强标签冲突时明确保留未识别，不能按多数投票选错版。"""
files = [tmp_path / f"{index}.flac" for index in (1, 2)]
metas = [MetaMusic(title=str(index), musicbrainz_release_id=f"release-{index}") for index in (1, 2)]
monkeypatch.setattr(AudioMetadataHelper, "read_many", Mock(return_value=metas))
factory = Mock(side_effect=AssertionError("冲突证据不应请求网络"))
monkeypatch.setattr("app.chain.media.album.MusicBrainzChain", factory)
assert MediaChain()._match_music_album_directory(tmp_path, files) == {}
factory.assert_not_called()
@pytest.mark.parametrize("asynchronous", [False, True])
def test_single_cue_image_matches_as_two_logical_tracks(tmp_path, monkeypatch, asynchronous):
"""一个整轨文件使用两首逻辑歌曲验证，输出必须仍是专辑而非第一首录音。"""
audio, cue = _image_pair(tmp_path)
detail = _release_detail("image-release", "专辑示例", "测试歌手", [("第一首歌曲", 1), ("第二首歌曲", 2)])
album = MusicBrainzModule._release_to_album(detail)
request = AsyncMock(return_value=album) if asynchronous else Mock(return_value=album)
source = Mock(**{"async_match_music_album" if asynchronous else "match_music_album": request})
monkeypatch.setattr("app.chain.media.album.MusicBrainzChain", Mock(return_value=source))
chain = MediaChain()
monkeypatch.setattr(chain, "_album_dir_cache", AlbumDirectoryCache(16))
result = asyncio.run(chain.async_recognize_music_album_directory(audio.parent)) if asynchronous else chain.recognize_music_album_directory(audio.parent)
assert [track.title for track in request.call_args.args[1]] == ["第一首歌曲", "第二首歌曲"]
assert [track.duration for track in request.call_args.args[1]] == [1, 2]
assert request.call_args.args[0].album == "专辑示例"
info = result[str(audio.resolve())]
assert info.music_type == "album" and info.media_id == "rg-1"
assert info.title == "专辑示例"
signature = chain._album_directory_signature(audio.parent, [audio])
cue.write_text(cue.read_text().replace("第二首歌曲", "另一首歌曲"))
assert signature != chain._album_directory_signature(audio.parent, [audio])
def test_manual_cue_selection_validates_tracks_before_binding_album(tmp_path):
"""手选发行也不能给曲目数量或时长不符的整轨直接绑定任意专辑。"""
audio, _cue = _image_pair(tmp_path)
wrong = MusicBrainzModule._release_to_album(_release_detail("wrong", "Wrong", "Other", [("First", 180)]))
assert isinstance(wrong, MusicAlbumInfo)
assert MediaChain._align_selected_music_album([audio], wrong) == {}
logical, owners = expand_music_tracks([AudioMetadataHelper.read(audio)])
assert len(logical) == 2 and owners == [0, 0]