fix(music): align album tracks by unique evidence instead of file order

This commit is contained in:
jxxghp
2026-09-30 03:29:05 +08:00
parent 983d3e12bc
commit 47303a1be6
10 changed files with 291 additions and 65 deletions

View File

@@ -16,6 +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.foundation.text import convert as zhconv_convert
from app.runtime.execution import run_in_threadpool
@@ -152,62 +153,16 @@ class MediaAlbumOwner(_MediaOwnerBase):
files: list[Path],
metas: list[MetaMusic],
tracks: list[MusicInfo],
*,
allow_title_override: bool = False,
) -> dict[Path, MusicInfo]:
"""优先按精确曲名、再按碟号和曲序把专辑曲目对位到本地文件。"""
matched: dict[Path, MusicInfo] = {}
used: set[int] = set()
indexed_tracks = list(enumerate(tracks))
by_title: dict[str, list[int]] = {}
for index, track in indexed_tracks:
title_key = cls._music_track_title_key(track.title)
if title_key:
by_title.setdefault(title_key, []).append(index)
pending: list[tuple[Path, MetaMusic]] = []
for file, meta in zip(files, metas):
candidates = [
index for index in by_title.get(cls._music_track_title_key(meta.title), []) if index not in used
]
if candidates:
local_position = (meta.disc_number or 1, meta.track_number or 0)
index = next(
(
candidate
for candidate in candidates
if (
tracks[candidate].disc_number or 1,
tracks[candidate].track_number or 0,
)
== local_position
),
candidates[0],
)
matched[file] = tracks[index]
used.add(index)
else:
pending.append((file, meta))
by_position: dict[tuple[int, int], list[int]] = {}
for index, track in indexed_tracks:
if track.track_number:
by_position.setdefault((track.disc_number or 1, track.track_number), []).append(index)
unresolved: list[tuple[Path, MetaMusic]] = []
for file, meta in pending:
position_key = (meta.disc_number or 1, meta.track_number or 0)
candidates = (
[index for index in by_position.get(position_key, []) if index not in used] if meta.track_number else []
)
if candidates:
index = candidates[0]
matched[file] = tracks[index]
used.add(index)
else:
unresolved.append((file, meta))
remaining = [track for index, track in indexed_tracks if index not in used]
unresolved.sort(key=lambda item: (item[1].disc_number or 1, item[0].name.casefold()))
for (file, _), track in zip(unresolved, remaining):
matched[file] = track
return matched
"""只输出有唯一曲目证据的文件,不对剩余文件进行排序猜测。"""
if len(files) != len(metas):
return {}
return {
files[index]: tracks[remote]
for index, remote in align_music_tracks(metas, tracks, allow_title_override=allow_title_override).items()
}
@classmethod
def _align_selected_music_album(
@@ -223,6 +178,7 @@ class MediaAlbumOwner(_MediaOwnerBase):
files,
metas,
album.tracks,
allow_title_override=True,
).items()
}

View File

@@ -33,6 +33,7 @@ from app.domain.music import (
music_album_matches,
music_artist_matches,
music_title_matches,
music_track_title_is_weak,
music_version_matches,
music_year_matches,
)
@@ -218,11 +219,12 @@ class _MusicFileFilterBase(_TransferOwnerBase):
info = matched.get(str(file_path.resolve()))
if not info or not info.media_id:
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))
and (not file_meta.title or music_title_matches(info, file_meta.title))
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 music_version_matches(info, file_meta)
and ((weak_title and not file_meta.version) or music_version_matches(info, file_meta))
and music_year_matches(info, file_meta)
)
if not evidence_matches:
@@ -392,8 +394,8 @@ class FileFilterMixin(_TransferOwnerBase):
if len(aligned_tracks) != len(audio_paths):
return {}, (
f"所选专辑只能对齐 {len(aligned_tracks)} / {len(audio_paths)} "
"个音频文件,目录中可能包含重复版本或额外曲目;"
"请分别选择单个版本后再整理"
"个音频文件,存在重复版本、曲目信息冲突或无法唯一对位的曲目;"
"请按单个发行版本选择文件并核对曲目"
)
selected_tracks: dict[str, MusicInfo] = {}
for resolved_path, track in aligned_tracks.items():

View File

@@ -205,6 +205,103 @@ def music_tags_are_usable(meta: Optional[MetaMusic]) -> bool:
return all(_usable_music_tag(value) for value in required)
def music_track_title_is_weak(meta: MetaMusic) -> bool:
"""编号或占位文件名不能否决实际曲目,但真实标签中的数字歌曲名仍是证据。"""
title = str(meta.title or "").strip()
if not title:
return True
if title.isdigit():
return meta.field_sources.get("title") not in {"tag", "cue", "manual", "remote"}
return bool(re.fullmatch(r"(?:(?:track|audio|音轨|曲目)[\s._-]*\d*|unknown|untitled)", title, re.I))
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()
def _music_track_pair_score(meta: MetaMusic, track: MusicInfo, allow_title_override: bool) -> float:
"""评估一条文件与发行曲目的证据;明确身份或时长冲突不能被其它分数抵消。"""
identity_match = False
for field in ("musicbrainz_release_id", "musicbrainz_release_track_id"):
local_id, remote_id = getattr(meta, field), getattr(track, field)
if local_id and remote_id:
if local_id != remote_id and not allow_title_override:
return 0.0
if field == "musicbrainz_release_track_id" and local_id == remote_id:
identity_match = True
if meta.media_id and track.media_id and meta.media_source == track.media_source and meta.music_type != MUSIC_ENTITY_ALBUM:
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
duration_close = False
if meta.duration and track.duration:
delta = abs(meta.duration - track.duration)
longest = max(meta.duration, track.duration)
if delta > max(10, longest * 0.08):
return 0.0
duration_close = delta <= max(3, min(8, longest * 0.025))
title_key = _alignment_title_key(meta.title)
weak_title = music_track_title_is_weak(meta)
title_match = not weak_title and bool(title_key) and any(
title_key == _alignment_title_key(value) for value in music_titles(track)
)
position_match = bool(
meta.track_number and meta.track_number == track.track_number
and (not meta.disc_number or meta.disc_number == (track.disc_number or 1))
)
if not identity_match and not title_match and not weak_title and not (allow_title_override and position_match):
return 0.0
if not identity_match and not title_match and meta.duration and track.duration and not duration_close:
return 0.0
if not (identity_match or title_match or position_match or (weak_title and duration_close)):
return 0.0
# 同一容差内的多个时长候选保持同分,不能以一秒的测量差异打破歧义。
return 1000 * identity_match + 60 * title_match + 30 * position_match + 20 * duration_close
def _unique_track_choice(scores: dict[int, float]) -> Optional[int]:
"""只选择有唯一最强证据的候选,不用列表顺序消除歧义。"""
if not scores:
return None
maximum = max(scores.values())
choices = [index for index, score in scores.items() if score == maximum]
return choices[0] if len(choices) == 1 else None
def align_music_tracks(
metas: list[MetaMusic], tracks: list[MusicInfo], *, allow_title_override: bool = False,
) -> dict[int, int]:
"""以双向唯一证据对位曲目,返回本地索引到远端索引;无证据项保持未匹配。
手选发行可依据唯一位置纠正旧曲名及旧身份,但仍拒绝显著时长冲突。
双向检查使重复版本不能抢占一个位置,输入排序不会改变歌曲身份。
"""
scores = {
index: {
remote: score for remote, track in enumerate(tracks)
if (score := _music_track_pair_score(meta, track, allow_title_override)) > 0
}
for index, meta in enumerate(metas)
}
matched: dict[int, int] = {}
while scores:
choices = {index: choice for index, values in scores.items() if (choice := _unique_track_choice(values)) is not None}
accepted = {
index: remote for index, remote in choices.items()
if _unique_track_choice({local: values[remote] for local, values in scores.items() if remote in values}) == index
}
if not accepted:
break
matched.update(accepted)
used = set(accepted.values())
scores = {
index: {remote: score for remote, score in values.items() if remote not in used}
for index, values in scores.items() if index not in accepted
}
return matched
def _usable_music_tag(value: Optional[str]) -> bool:
"""保守排除占位值及乱码,不误删数字专辑名和包含 Unknown 的正常标题。"""
text = str(value or "").strip()

View File

@@ -763,7 +763,7 @@ flowchart LR
| 指标 | 当前值 |
|---|---:|
| Python 模块 | 1054 |
| 内部导入边 | 8,962 |
| 内部导入边 | 8,963 |
| 非平凡 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,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`,各启动入口增加一个模块;性能基线只同步模块数量,原有耗时预算、历史采样和生命周期资源约束保持有效。
整理失败反馈由 `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`,各启动入口增加一个模块;性能基线只同步模块数量,原有耗时预算、历史采样和生命周期资源约束保持有效。
架构专项验证分为两个 CI 投影:`Check event semantic policy` 先运行依赖、Adapter、出口和 Event
语义门禁,`Check host architecture snapshot` 再执行快照测试及一次

View File

@@ -534,6 +534,11 @@ SSE 的 `candidate_items` 是站点原始返回数量,`match_counts` 记录身
明确按音乐整理的镜像(如 ISO)及压缩包会返回先提取/解压的失败提示,不执行自动转换。
明确按影视整理的附加音轨不继承旧音乐历史身份。
专辑曲目按身份、曲名、碟号/曲序与时长的唯一证据对位,不按文件排序补齐未匹配曲目。
编号文件名可由位置或唯一时长补全,真实标签中的数字曲名仍作为名称证据。
重复版本、同分候选和明显时长冲突保留未匹配;手选发行允许按唯一位置纠正旧曲名,
但不绕过时长冲突。手选批次不能完整对位时会在文件操作前返回核对提示。
| 方法 | 路径 | 说明 |
| :--- | :--- | :--- |
| GET | `/api/v1/media/search` | 当 `type=music` 或指定音乐 `media_source` 时按歌曲或专辑关键词搜索音乐元数据;艺术家统一由 `type=person` 搜索,支持 TMDB 与 MusicBrainz 来源。参数:`title`、`type`、`count`、可重复的 `media_source` 枚举,以及可选的 `music_type` 实体过滤 |

View File

@@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
| 指标 | 当前值 | 解释 |
|---|---:|---|
| 宿主 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` 归档 |
| 宿主 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` 归档 |
| 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
| 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
| Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |

View File

@@ -60,6 +60,12 @@ Music recognition, exploration, album and artist navigation, and recognition-cac
Explicit music transfers reject disc images and archives with an extraction
message. Explicit movie/TV transfers do not inherit old music identities for
accompanying audio tracks.
- Album track alignment requires unique identity, title, disc/track position,
or duration evidence; file order never fills unresolved tracks. Numeric file
names are weak evidence, while numeric titles in tags remain meaningful.
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.
## Operations

View File

@@ -1097,8 +1097,8 @@
"runtime_only": true
}
},
"edge_count": 8962,
"edge_sha256": "22c4168cb2468ec6a446bab6366a95dc12984608ab0ce34f40bf7aa0851c4dfe",
"edge_count": 8963,
"edge_sha256": "dd3d4f1797a5e9d4094a118a571323123d70086ff4e26597f6bb7dbe85b21493",
"edges": [
"app -> app.foundation",
"app -> app.foundation.environment",
@@ -4620,6 +4620,7 @@
"app.chain.media.album -> app.domain.context",
"app.chain.media.album -> app.domain.meta",
"app.chain.media.album -> app.domain.meta.metamusic",
"app.chain.media.album -> app.domain.music",
"app.chain.media.album -> app.foundation",
"app.chain.media.album -> app.foundation.text",
"app.chain.media.album -> app.runtime",

View File

@@ -0,0 +1,159 @@
"""发行曲目对位不依赖排序,并在名称、身份、曲序和时长之间保留证据边界。"""
from itertools import permutations
from pathlib import Path
from unittest.mock import Mock
import pytest
from app.chain.media import MediaChain
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 align_music_tracks
from tests.test_transfer_sync_extra_files import make_fileitem
def _track(title, number, duration=None, **kwargs):
"""构造带独立录音身份和实际发行位置的来源曲目。"""
return MusicInfo(title=title, media_source="musicbrainz", media_id=f"recording-{number}",
track_number=number, duration=duration, **kwargs)
def test_unmatched_remainder_is_never_zipped_to_remaining_song():
"""前曲匹配不代表剩余文件自动属于最后一曲。"""
files = [Path("01.flac"), Path("bonus.flac")]
metas = [MetaMusic(title="First", track_number=1), MetaMusic(title="Unrelated Bonus")]
tracks = [_track("First", 1), _track("Second", 2)]
assert MediaChain._align_music_album_tracks(files, metas, tracks) == {files[0]: tracks[0]}
@pytest.mark.parametrize("title", ["01", "Track 01", "audio_01", "音轨 01", ""])
def test_placeholder_name_can_use_unique_position(title):
"""不完整的抓轨命名可用发行位置补全,不要求占位曲名命中。"""
assert align_music_tracks([MetaMusic(title=title, track_number=1)], [_track("Actual Song", 1)]) == {0: 0}
def test_numeric_title_from_real_tags_is_not_a_placeholder():
"""真实数字歌曲名会与候选比较,不能被错误曲序吞掉。"""
local = MetaMusic(title="22", track_number=1, field_sources={"title": "tag"})
tracks = [_track("Different Song", 1), _track("22", 2)]
assert align_music_tracks([local], tracks) == {0: 1}
def test_placeholder_stored_in_tags_is_still_weak():
"""抓轨器把 Track 01 写进标签也不会使它成为真实歌曲名。"""
local = MetaMusic(title="Track 01", track_number=1, field_sources={"title": "tag"})
assert align_music_tracks([local], [_track("Actual Song", 1)]) == {0: 0}
def test_same_title_and_missing_disc_remains_ambiguous():
"""多碟重复曲名及曲序需要碟号或时长区分,不能默认分配到第一碟。"""
tracks = [_track("Song", 1, disc_number=disc) for disc in (1, 2)]
assert align_music_tracks([MetaMusic(title="Song", track_number=1)], tracks) == {}
assert align_music_tracks([MetaMusic(title="Song", track_number=1, disc_number=2)], tracks) == {0: 1}
def test_duplicate_local_editions_cannot_race_for_one_track():
"""相同证据的两份音源不能由遍历顺序决定谁冒充对应发行曲目。"""
metas = [MetaMusic(title="First", track_number=1) for _ in range(2)]
assert align_music_tracks(metas, [_track("First", 1)]) == {}
def test_unlabeled_tracks_can_use_unique_durations_and_ignore_filename_sort():
"""缺少曲序时,彼此明显不同的时长仍可恢复顺序;两个方向均需唯一。"""
local = [MetaMusic(title="Track", duration=seconds) for seconds in (240, 180, 300)]
remote = [_track("First", 1, 180), _track("Second", 2, 240), _track("Third", 3, 300)]
for metas in permutations(local):
for tracks in permutations(remote):
matched = align_music_tracks(list(metas), list(tracks))
assert {metas[index].duration: tracks[other].duration for index, other in matched.items()} == {
180: 180, 240: 240, 300: 300,
}
def test_near_equal_durations_do_not_break_ambiguity():
"""测量取整的一秒差异不能在两个同样合理的候选之间制造唯一性。"""
local = MetaMusic(title="01", duration=180)
assert align_music_tracks([local], [_track("First", 1, 179), _track("Second", 2, 181)]) == {}
@pytest.mark.parametrize("manual", [False, True])
def test_duration_conflict_blocks_even_exact_title_and_position(manual):
"""曲名和位置都相同也不能覆盖实际时长指示的不同内容。"""
local = MetaMusic(title="First", track_number=1, duration=300)
assert align_music_tracks([local], [_track("First", 1, 180)], allow_title_override=manual) == {}
def test_conflicting_position_does_not_beat_exact_title_or_duration():
"""错标曲序应由明确的曲名或唯一时长纠正,不能先按序号绑定。"""
tracks = [_track("First", 1, 180), _track("Second", 2, 195)]
assert align_music_tracks([MetaMusic(title="Second", track_number=1, duration=195)], tracks) == {0: 1}
assert align_music_tracks([MetaMusic(title="01", track_number=1, duration=195)], tracks) == {0: 1}
def test_manual_release_can_correct_titles_without_enabling_automatic_guessing():
"""明确手选发行可用唯一曲序纠名;自动匹配对同一冲突仍然拒绝。"""
local = MetaMusic(title="Wrong Name", track_number=1)
tracks = [_track("Correct Name", 1)]
assert align_music_tracks([local], tracks) == {}
assert align_music_tracks([local], tracks, allow_title_override=True) == {0: 0}
def test_explicit_release_identity_rejects_different_release():
"""具体发行 ID 不能被相同曲名及曲序抵消,手选新发行可纠正旧标签身份。"""
local = MetaMusic(title="First", track_number=1, musicbrainz_release_id="release-a")
track = _track("First", 1, musicbrainz_release_id="release-b")
assert align_music_tracks([local], [track]) == {}
assert align_music_tracks([local], [track], allow_title_override=True) == {0: 0}
def test_release_track_id_resolves_repeated_recording():
"""同一录音多次出现在发行时,Release Track ID 可确定实际槽位。"""
local = MetaMusic(title="Song", musicbrainz_release_track_id="slot-b")
tracks = [_track("Song", number, musicbrainz_release_track_id=slot) for number, slot in ((1, "slot-a"), (2, "slot-b"))]
assert align_music_tracks([local], tracks) == {0: 1}
def test_recording_id_conflict_is_not_ignored():
"""相同标题不能覆盖已知的另一个录音身份。"""
local = MetaMusic(title="First", media_source="musicbrainz", media_id="other-recording")
assert align_music_tracks([local], [_track("First", 1)]) == {}
def test_symbolic_names_and_recording_versions_keep_distinct_identities():
"""纯符号歌曲名不能归一为空,同名现场/混音也不能当作普通录音。"""
assert align_music_tracks([MetaMusic(title="!!!")], [_track("???", 1), _track("!!!", 2)]) == {0: 1}
assert align_music_tracks([MetaMusic(title="Song (Live)", track_number=1)], [_track("Song", 1)]) == {}
def test_mismatched_file_metadata_lengths_are_rejected():
"""元数据读取丢项时不能用截断 zip 悄悄改变文件对应关系。"""
assert MediaChain._align_music_album_tracks([Path("first.flac"), Path("second.flac")],
[MetaMusic(title="First")], [_track("First", 1)]) == {}
def test_numbered_filename_survives_transfer_album_evidence_check(monkeypatch):
"""专辑已经按位置对位后,整理层不能再次用占位曲名把正确曲目拒绝。"""
path = Path("/music/Artist - Album/01.flac")
track = _track("Actual Song (Live)", 1, 180, album="Album", artists=["Artist"])
monkeypatch.setattr(MediaChain, "recognize_music_album_directory", Mock(return_value={str(path.resolve()): track}))
local = MetaMusic(title="01", track_number=1, album="Album", artists=["Artist"], duration=180,
field_sources={"title": "filename"})
meta, info = TransferChain._match_music_album_context(make_fileitem(str(path)), path, local)
assert info is not None
assert meta.title == info.title == "Actual Song (Live)"
assert info.media_id == track.media_id

View File

@@ -702,5 +702,5 @@ def test_selected_album_rejects_duplicate_local_editions(tmp_path, monkeypatch):
)
assert state is False
assert "只能对齐 2 / 4" in message
assert "只能对齐 0 / 4" in message
assert "重复版本" in message