feat(music): 增强无标签音频识别,支持文件名/目录解析与专辑级匹配

- 新增 MusicNameParser:剥离曲序/碟号前缀、拆分歌手与曲名、解析专辑目录名(歌手/专辑/年份/音质)和 CD1 等碟片目录
- MetaInfoPath 与 MusicChain.read_path_meta 接入路径上下文,WAV 及标签不全的 FLAC/MP3 可补齐识别线索
- MusicBrainz 新增 match_music_album:按专辑名/歌手/曲名搜索候选发行版本,用曲目数、总时长、逐曲时长和曲名重合度打分对位
- MusicChain 新增 recognize_album_directory 目录级批量识别(按目录缓存),单曲识别未命中时自动兜底
- transfer 整理链路接入专辑匹配,命中后回填曲目身份用于重命名与刮削
This commit is contained in:
jxxghp
2026-08-10 20:11:23 +08:00
parent 85a7f98a94
commit 07ea99442e
8 changed files with 1022 additions and 13 deletions
+184 -7
View File
@@ -17,6 +17,7 @@ from app.core.context import (
) )
from app.core.meta import MetaMusic from app.core.meta import MetaMusic
from app.helper.audio import AudioMetadataHelper from app.helper.audio import AudioMetadataHelper
from app.helper.music_name import MusicNameParser
from app.log import logger from app.log import logger
@@ -25,6 +26,11 @@ class MusicChain(ChainBase):
_artist_title_pattern = re.compile(r"^\s*(?P<artist>.+?)\s+[-–—]\s+(?P<title>.+?)\s*$") _artist_title_pattern = re.compile(r"^\s*(?P<artist>.+?)\s+[-–—]\s+(?P<title>.+?)\s*$")
_spaces_pattern = re.compile(r"\s+") _spaces_pattern = re.compile(r"\s+")
# 专辑目录匹配结果缓存:{目录路径: (音频文件数, 匹配结果)},避免逐文件整理时重复请求远端
_album_dir_cache: dict[str, tuple[int, dict[str, MusicInfo]]] = {}
_album_dir_cache_max = 128
# 目录级匹配至少需要两个音频文件,单文件由单曲搜索链路处理
_album_match_min_files = 2
@classmethod @classmethod
def parse_query(cls, query: str) -> MetaMusic: def parse_query(cls, query: str) -> MetaMusic:
@@ -317,11 +323,14 @@ class MusicChain(ChainBase):
@classmethod @classmethod
def read_path_meta(cls, path: str | Path) -> MetaMusic: def read_path_meta(cls, path: str | Path) -> MetaMusic:
"""读取本地音频标签,不可访问时按文件名构造最小音乐元数据""" """读取本地音频标签,标签缺失时用文件名和目录线索补齐"""
file_path = Path(path) file_path = Path(path)
if file_path.exists() and file_path.is_file(): if file_path.exists() and file_path.is_file():
return AudioMetadataHelper.read(file_path) meta = AudioMetadataHelper.read(file_path)
return cls.parse_query(file_path.stem) else:
meta = cls.parse_query(file_path.stem)
# WAV 无标签、FLAC/MP3 标签不全时,依靠文件名和目录结构补充识别线索
return MusicNameParser.apply_path_context(meta, file_path)
async def async_recognize_by_path( async def async_recognize_by_path(
self, self,
@@ -331,9 +340,15 @@ class MusicChain(ChainBase):
"""根据音频标签和文件名识别音乐,远端不可用时仍返回最小音乐信息。""" """根据音频标签和文件名识别音乐,远端不可用时仍返回最小音乐信息。"""
# Mutagen 会同步读取本地文件,异步识别入口需要移出事件循环。 # Mutagen 会同步读取本地文件,异步识别入口需要移出事件循环。
meta = await run_in_threadpool(self.read_path_meta, path) meta = await run_in_threadpool(self.read_path_meta, path)
# 统一识别入口分发到音乐模块,模块负责详情/搜索/匹配/ # 统一识别入口分发到音乐模块,模块负责详情/搜索/匹配/
info = await self.async_recognize_media(meta=meta, source=source) info = await self.async_recognize_media(meta=meta, source=source)
return meta, self._merge_audio_quality(info or self._info_from_meta(meta), meta) result = self._merge_audio_quality(info or self._info_from_meta(meta), meta)
if not result.source:
# 单曲搜索未命中时,按所在目录做专辑级匹配兑底
matched = await run_in_threadpool(self._album_dir_fallback, path)
if matched:
result = self._merge_audio_quality(matched, meta)
return meta, result
def recognize_by_path( def recognize_by_path(
self, self,
@@ -342,9 +357,171 @@ class MusicChain(ChainBase):
) -> tuple[MetaMusic, MusicInfo]: ) -> tuple[MetaMusic, MusicInfo]:
"""同步根据音频标签和文件名识别音乐,并保留离线最小结果。""" """同步根据音频标签和文件名识别音乐,并保留离线最小结果。"""
meta = self.read_path_meta(path) meta = self.read_path_meta(path)
# 统一识别入口分发到音乐模块,模块负责详情/搜索/匹配/ # 统一识别入口分发到音乐模块,模块负责详情/搜索/匹配/
info = self.recognize_media(meta=meta, source=source) info = self.recognize_media(meta=meta, source=source)
return meta, self._merge_audio_quality(info or self._info_from_meta(meta), meta) result = self._merge_audio_quality(info or self._info_from_meta(meta), meta)
if not result.source:
# 单曲搜索未命中时,按所在目录做专辑级匹配兑底
matched = self._album_dir_fallback(path)
if matched:
result = self._merge_audio_quality(matched, meta)
return meta, result
def _album_dir_fallback(self, path: str | Path) -> Optional[MusicInfo]:
"""单曲识别无远端身份时,查找所在目录专辑匹配中属于当前文件的结果。"""
file_path = Path(path)
if not file_path.exists() or not file_path.is_file():
return None
try:
matched = self.recognize_album_directory(file_path.parent)
except Exception as err:
logger.debug(f"专辑目录匹配失败:{file_path.parent} - {err}")
return None
return matched.get(str(file_path.resolve()))
def recognize_album_directory(self, path: str | Path) -> dict[str, MusicInfo]:
"""按目录级线索批量识别整目录音频,返回 文件路径 到标准音乐信息的映射。
适用于 WAV 无标签或标签不全的整专目录:先用目录名和文件标签构造专辑线索,
再交给音乐元数据模块用曲目数、时长等特征对位到具体发行版本。
"""
dir_path = Path(path)
if not dir_path.is_dir():
return {}
files = self._directory_audio_files(dir_path)
if len(files) < self._album_match_min_files:
return {}
cache_key = str(dir_path)
cached = self._album_dir_cache.get(cache_key)
# 目录内音频数量变化时视为内容更新,需要重新匹配
if cached and cached[0] == len(files):
return cached[1]
matched = self._match_album_directory(dir_path, files)
if len(self._album_dir_cache) >= self._album_dir_cache_max:
self._album_dir_cache.clear()
self._album_dir_cache[cache_key] = (len(files), matched)
return matched
async def async_recognize_album_directory(self, path: str | Path) -> dict[str, MusicInfo]:
"""目录级批量识别的异步版本,本地文件读取移出事件循环。"""
return await run_in_threadpool(self.recognize_album_directory, path)
@classmethod
def _directory_audio_files(cls, dir_path: Path) -> list[Path]:
"""收集目录及其一级子目录(如 CD1/CD2)内的音频文件。"""
audio_exts = settings.RMT_AUDIOEXT
files: list[Path] = []
def collect(current: Path) -> None:
try:
entries = sorted(current.iterdir())
except OSError:
return
for entry in entries:
if entry.name.startswith("."):
continue
if entry.is_file() and entry.suffix.lower() in audio_exts:
files.append(entry)
collect(dir_path)
try:
subdirs = sorted(entry for entry in dir_path.iterdir()
if entry.is_dir() and not entry.name.startswith("."))
except OSError:
subdirs = []
for subdir in subdirs:
collect(subdir)
return files
def _match_album_directory(
self,
dir_path: Path,
files: list[Path],
) -> dict[str, MusicInfo]:
"""执行目录级专辑匹配,并把专辑曲目对位到具体音频文件。"""
metas = [self.read_path_meta(file) for file in files]
album_meta = self._album_meta_from_context(dir_path, metas)
if not (album_meta.album or album_meta.title or album_meta.artists):
logger.debug(f"目录缺少专辑识别线索,跳过专辑匹配:{dir_path}")
return {}
candidates = self.run_module("match_music_album", meta=album_meta, tracks=metas)
album = next(
(item for item in candidates or [] if isinstance(item, MusicAlbumInfo) and item.tracks),
None,
)
if not album:
return {}
logger.info(f"目录 {dir_path.name} 匹配到专辑:{album.title_year}{album.source}")
matched: dict[str, MusicInfo] = {}
for file, info in self._align_album_tracks(files, metas, album.tracks).items():
matched[str(file.resolve())] = info
return matched
@classmethod
def _album_meta_from_context(cls, dir_path: Path, metas: list[MetaMusic]) -> MetaMusic:
"""汇总目录名和文件标签中的专辑线索,作为专辑搜索条件。"""
dir_info = MusicNameParser.parse_album_dir(dir_path.name)
# 文件标签中的专辑信息比目录名更可靠,多数文件一致时优先采用
album_votes: dict[str, int] = {}
artist_votes: dict[str, int] = {}
for meta in metas:
if meta.album:
album_votes[meta.album] = album_votes.get(meta.album, 0) + 1
if meta.album_artist:
artist_votes[meta.album_artist] = artist_votes.get(meta.album_artist, 0) + 1
elif meta.artists:
artist_votes[meta.artists[0]] = artist_votes.get(meta.artists[0], 0) + 1
majority_album = max(album_votes, key=album_votes.get) if album_votes else None
majority_artist = max(artist_votes, key=artist_votes.get) if artist_votes else None
# 多数文件共享同一专辑标签才可信,避免杂集目录的个别错误标签带偏搜索
album = majority_album if majority_album and album_votes[majority_album] >= max(2, len(metas) // 2) else None
artist = majority_artist if majority_artist and artist_votes[majority_artist] >= max(2, len(metas) // 2) else None
return MetaMusic(
org_string=dir_path.name,
title=album or dir_info.get("album") or dir_path.name,
album=album or dir_info.get("album"),
artists=[artist or dir_info.get("artist")] if (artist or dir_info.get("artist")) else [],
album_artist=artist or dir_info.get("artist"),
year=dir_info.get("year"),
)
@classmethod
def _align_album_tracks(
cls,
files: list[Path],
metas: list[MetaMusic],
tracks: list[MusicInfo],
) -> dict[Path, MusicInfo]:
"""把专辑曲目对位到目录内的音频文件。
带曲序标签的文件按(碟号, 曲序)精确对位,其余文件按排序顺序依次补齐。
"""
matched: dict[Path, MusicInfo] = {}
used_keys: set[tuple[int, int]] = set()
by_position: dict[tuple[int, int], MusicInfo] = {}
for track in tracks:
if track.track_number:
by_position[(track.disc_number or 1, track.track_number)] = track
pending: list[tuple[Path, MetaMusic]] = []
for file, meta in zip(files, metas):
key = (meta.disc_number or 1, meta.track_number)
track = by_position.get(key) if meta.track_number else None
if track and key not in used_keys:
matched[file] = track
used_keys.add(key)
else:
pending.append((file, meta))
if not pending:
return matched
remaining = [
track for track in tracks
if (track.disc_number or 1, track.track_number or 0) not in used_keys
]
# 无曲序线索的文件按碟号和文件名排序,与剩余曲目顺序对位
pending.sort(key=lambda item: (item[1].disc_number or 1, item[0].name.casefold()))
for (file, _meta), track in zip(pending, remaining):
matched[file] = track
return matched
@classmethod @classmethod
def to_meta(cls, info: MusicInfo) -> MetaMusic: def to_meta(cls, info: MusicInfo) -> MetaMusic:
+64 -1
View File
@@ -1063,6 +1063,64 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
names=[name for name in (meta.title, meta.album) if name], names=[name for name in (meta.title, meta.album) if name],
) )
@classmethod
def _match_music_album_context(
cls,
file_item: FileItem,
file_path: Path,
file_meta: MetaMusic,
) -> tuple[MetaMusic, Optional[MusicInfo]]:
"""为缺少远端身份的本地音频尝试目录级专辑匹配,命中后回填文件元数据。
WAV 等无标签文件只能依靠目录结构和曲目特征识别;匹配结果在 MusicChain
内按目录缓存,同一专辑目录内的后续文件不会重复请求远端。
"""
# 目录级匹配需要读取本地音频时长,远端存储文件无法参与
if file_meta.media_id or getattr(file_item, "storage", "local") != "local":
return file_meta, None
try:
from app.chain.music import MusicChain
matched = MusicChain().recognize_album_directory(file_path.parent)
except Exception as err:
logger.debug(f"音乐专辑目录匹配失败:{file_path} - {err}")
return file_meta, None
info = matched.get(str(file_path.resolve()))
if not info or not info.media_id:
return file_meta, None
logger.info(f"{file_path.name} 通过专辑目录匹配识别为:{info.artist} - {info.title}")
merged_meta = deepcopy(file_meta)
# 保留本地音频的实际技术参数,仅回填身份和名称字段
if info.title:
merged_meta.title = info.title
if info.artists:
merged_meta.artists = list(info.artists)
if info.album:
merged_meta.album = info.album
if info.album_artist:
merged_meta.album_artist = info.album_artist
if info.year:
merged_meta.year = info.year
if info.disc_number:
merged_meta.disc_number = info.disc_number
if info.track_number:
merged_meta.track_number = info.track_number
if info.total_tracks:
merged_meta.total_tracks = info.total_tracks
merged_meta.media_source = info.source
merged_meta.media_id = info.media_id
merged_info = cls._music_info_from_meta(merged_meta)
# 补齐曲目级远端信息,供后续刮削和展示使用
merged_info.music_type = info.music_type
merged_info.artist_ids = list(info.artist_ids)
merged_info.album_id = info.album_id
merged_info.album_type = info.album_type
merged_info.release_date = info.release_date
merged_info.cover_url = info.cover_url
merged_info.category = info.category
merged_info.genres = list(info.genres)
merged_info.detail_link = info.detail_link
return merged_meta, merged_info
@classmethod @classmethod
def _restore_music_download_context( def _restore_music_download_context(
cls, cls,
@@ -3626,7 +3684,12 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
# 自动整理预载的媒体信息来自整条下载历史;电影合集内文件年份冲突时逐文件识别。 # 自动整理预载的媒体信息来自整条下载历史;电影合集内文件年份冲突时逐文件识别。
task_mediainfo = mediainfo or history_music_info task_mediainfo = mediainfo or history_music_info
if not task_mediainfo and isinstance(file_meta, MetaMusic): if not task_mediainfo and isinstance(file_meta, MetaMusic):
task_mediainfo = self._music_info_from_meta(file_meta) # 无标签音频按目录级专辑匹配补齐曲目身份,命中结果带缓存不会逐文件重复请求
file_meta, task_mediainfo = self._match_music_album_context(
file_item, file_path, file_meta
)
if not task_mediainfo:
task_mediainfo = self._music_info_from_meta(file_meta)
if ( if (
not manual not manual
and self._is_movie_year_conflict(file_meta, task_mediainfo) and self._is_movie_year_conflict(file_meta, task_mediainfo)
+4 -1
View File
@@ -12,6 +12,7 @@ from app.core.meta.infopath import (
should_use_parent_title_for_file_stem, should_use_parent_title_for_file_stem,
) )
from app.core.meta.words import WordsMatcher from app.core.meta.words import WordsMatcher
from app.helper.music_name import MusicNameParser
from app.log import logger from app.log import logger
from app.schemas.types import MediaType from app.schemas.types import MediaType
from app.utils import rust_accel from app.utils import rust_accel
@@ -463,11 +464,13 @@ def MetaInfoPath(path: Path, custom_words: List[str] = None, force_video: bool =
# 音频文件直接构造音乐元数据,不参与父目录季集合并,影视附加音轨强制走视频解析 # 音频文件直接构造音乐元数据,不参与父目录季集合并,影视附加音轨强制走视频解析
audio_suffix = path.suffix.lower() audio_suffix = path.suffix.lower()
if not force_video and audio_suffix in settings.RMT_AUDIOEXT: if not force_video and audio_suffix in settings.RMT_AUDIOEXT:
return MetaMusic( music_meta = MetaMusic(
org_string=path.name, org_string=path.name,
title=path.stem, title=path.stem,
audio_format=audio_suffix.lstrip(".").upper() or None, audio_format=audio_suffix.lstrip(".").upper() or None,
) )
# 无标签音频只能依靠文件名和目录结构,补充曲序、碟号、歌手和专辑线索
return MusicNameParser.apply_path_context(music_meta, path)
path_context = " ".join( path_context = " ".join(
[path.name, path.parent.name, path.parent.parent.name] [path.name, path.parent.name, path.parent.parent.name]
) )
+173
View File
@@ -0,0 +1,173 @@
import re
from pathlib import Path
from typing import Any, Optional
from app.core.meta import MetaMusic
class MusicNameParser:
"""解析音频文件名和目录名,在音频标签缺失时补充音乐识别线索。
WAV 等容器不带标签、FLAC/MP3 标签不全时,唯一线索来自文件名和目录结构。
这里只负责从文本中提取结构化信息(曲序、碟号、歌手、专辑、年份),
不访问文件系统以外的任何资源,解析结果按"标签 > 文件名 > 目录名"优先级回填。
"""
# 碟号-曲序前缀:1-02、CD1.03、Disc2-05 等,后面可跟分隔符和曲名
_disc_track_prefix_pattern = re.compile(
r"^\s*(?:(?:cd|disc|disk)\s*)?(?P<disc>\d{1,2})\s*[-._]\s*(?P<num>\d{1,3})"
r"\s*[-–—.。、) ]*\s*(?P<rest>.*\S)?\s*$",
re.IGNORECASE,
)
# 曲序前缀:01.、01 -、01)、01 晴天、Track 01 - 等
_track_prefix_pattern = re.compile(
r"^\s*(?:track\s*)?(?P<num>\d{1,3})\s*[-–—.。、) ]+\s*(?P<rest>.*\S)\s*$",
re.IGNORECASE,
)
# 纯数字文件名:01.wav、Track 12.flac,只能得到曲序没有曲名
_number_only_pattern = re.compile(
r"^\s*(?:(?:track|cd|disc|disk)\s*)?(?P<num>\d{1,3})\s*$",
re.IGNORECASE,
)
# 碟片目录名:CD1、Disc 2、Disk01
_disc_dir_pattern = re.compile(
r"^\s*(?:cd|disc|disk)\s*(?P<num>\d{1,2})\s*$",
re.IGNORECASE,
)
# 目录名中的年份:(2004)、[2004]
_year_pattern = re.compile(r"[(\[]\s*(?P<year>(?:19|20)\d{2})\s*[)\]]")
# 目录名中的括号补充说明(格式、音质、厂牌等),如 [FLAC 24bit-96kHz]
_bracket_pattern = re.compile(r"\[[^\]]*\]|【[^】]*】|\([^)]*\)")
# 歌手与标题/专辑的分隔符
_artist_title_pattern = re.compile(
r"^\s*(?P<artist>.+?)\s+[-–—]\s+(?P<title>.+?)\s*$"
)
_spaces_pattern = re.compile(r"\s+")
@classmethod
def strip_track_prefix(cls, stem: str) -> tuple[Optional[int], Optional[int], Optional[str]]:
"""剥离文件名中的曲序和碟号前缀。
:param stem: 不含扩展名的文件名
:return: (曲序, 碟号, 剥离前缀后的曲名),无法剥离的字段返回 None;
曲名为 None 表示文件名没有携带曲名信息
"""
text = str(stem or "").strip()
if not text:
return None, None, None
match = cls._disc_track_prefix_pattern.match(text)
if match:
return (
int(match.group("num")),
int(match.group("disc")),
cls._clean(match.group("rest")),
)
match = cls._track_prefix_pattern.match(text)
if match:
return int(match.group("num")), None, cls._clean(match.group("rest"))
match = cls._number_only_pattern.match(text)
if match:
# 纯数字文件名保留原始文本作为兜底标题,只提取曲序
return int(match.group("num")), None, None
return None, None, None
@classmethod
def split_artist_title(cls, text: str) -> tuple[Optional[str], str]:
"""拆分 `歌手 - 标题` 结构,未命中时原文作为标题返回。"""
match = cls._artist_title_pattern.match(str(text or "").strip())
if match:
return cls._clean(match.group("artist")), cls._clean(match.group("title"))
return None, cls._clean(text)
@classmethod
def parse_disc_dir(cls, name: str) -> Optional[int]:
"""识别 CD1、Disc 2 这类碟片子目录并返回碟号。"""
match = cls._disc_dir_pattern.match(str(name or "").strip())
return int(match.group("num")) if match else None
@classmethod
def parse_album_dir(cls, name: str) -> dict[str, Any]:
"""解析专辑目录名,提取歌手、专辑名、年份和音质描述。
支持 `歌手 - 专辑 (2004) [FLAC 24bit-96kHz]` 等常见命名。
"""
text = cls._clean(name)
if not text:
return {}
year = None
year_match = cls._year_pattern.search(text)
if year_match:
year = int(year_match.group("year"))
text = cls._year_pattern.sub(" ", text)
# 括号内的格式/音质描述先剥离出专辑名,但仍可用于音质解析
brackets = " ".join(
fragment
for fragment in cls._bracket_pattern.findall(text)
)
album_text = cls._clean(cls._bracket_pattern.sub(" ", text))
if not album_text:
return {}
artist, album = cls.split_artist_title(album_text)
return {
"artist": artist,
"album": album,
"year": year,
"quality_text": cls._clean(f"{album_text} {brackets}"),
}
@classmethod
def apply_path_context(cls, meta: MetaMusic, path: Path) -> MetaMusic:
"""用文件名和目录线索回填音乐元数据中缺失的字段。
仅补充空字段,音频标签中已读取到的内容不会被目录猜测覆盖;
标题来自文件名兜底(等于文件主干名)时视为缺失,允许用解析结果替换。
"""
file_path = Path(path)
stem = file_path.stem
title_from_name = not meta.title or meta.title == stem
# 文件名前缀:曲序、碟号、曲名
track_number, disc_number, parsed_title = cls.strip_track_prefix(stem)
if meta.track_number is None and track_number is not None:
meta.track_number = track_number
if meta.disc_number is None and disc_number is not None:
meta.disc_number = disc_number
if title_from_name:
base_title = parsed_title or stem
if not meta.artists:
# `歌手 - 曲名` 文件名在无艺术家标签时继续拆分
artist, title = cls.split_artist_title(base_title)
if artist:
meta.artists = [artist]
base_title = title
meta.title = base_title
# 目录结构:父目录可能是碟片目录,专辑目录再往上一级
parent = file_path.parent
album_dir = parent
parent_disc = cls.parse_disc_dir(parent.name)
if parent_disc is not None:
if meta.disc_number is None:
meta.disc_number = parent_disc
album_dir = parent.parent
dir_info = cls.parse_album_dir(album_dir.name)
if dir_info:
# 目录名同时带歌手或年份才视为有意的专辑命名,避免把监控根目录误当专辑
if dir_info.get("artist") or dir_info.get("year"):
if not meta.album and dir_info.get("album"):
meta.album = dir_info["album"]
if not meta.artists and dir_info.get("artist"):
meta.artists = [dir_info["artist"]]
if not meta.album_artist and dir_info.get("artist"):
meta.album_artist = dir_info["artist"]
if meta.year is None and dir_info.get("year"):
meta.year = dir_info["year"]
# 目录名里的格式、位深、采样率可补齐本地标签未声明的音质参数
if dir_info.get("quality_text"):
meta.apply_audio_quality(dir_info["quality_text"])
return meta
@classmethod
def _clean(cls, value: Optional[str]) -> str:
"""压缩多余空白,返回可用于匹配和展示的文本。"""
return cls._spaces_pattern.sub(" ", str(value or "")).strip()
+249
View File
@@ -1,6 +1,7 @@
import re import re
import threading import threading
import time import time
from difflib import SequenceMatcher
from typing import Any, Iterable, Optional, Tuple, Union from typing import Any, Iterable, Optional, Tuple, Union
from fastapi.concurrency import run_in_threadpool from fastapi.concurrency import run_in_threadpool
@@ -183,6 +184,254 @@ class MusicBrainzModule(_ModuleBase):
index += 1 index += 1
return results return results
def match_music_album(
self,
meta: MetaMusic,
tracks: list[MetaMusic],
limit: int = 5,
) -> Optional[MusicAlbumInfo]:
"""按目录线索和曲目特征把本地音频集合对位到 MusicBrainz 发行版本。
适用于无标签整专目录:用专辑名、歌手搜索候选发行版本,再用曲目数、
总时长和逐曲时长相似度打分,选出最可信的版本并返回其曲目表。
"""
if not tracks:
return None
best_album: Optional[MusicAlbumInfo] = None
best_score = 0.0
for release in self._search_release_candidates(meta, tracks, limit=limit):
release_id = release.get("id")
if not release_id:
continue
detail = self._request_json(
f"/release/{release_id}",
params={"inc": "recordings+media+artist-credits", "fmt": "json"},
)
if not detail:
continue
summary = self._release_track_summary(detail)
score = self._score_release(meta, tracks, detail, summary)
if score > best_score:
best_score = score
best_album = self._release_to_album(detail)
# 得分低于阈值时宁可不匹配,避免把曲目写到错误的专辑上
if best_score < self._album_match_threshold:
return None
return best_album
_album_match_threshold = 60.0
def _search_release_candidates(
self,
meta: MetaMusic,
tracks: list[MetaMusic],
limit: int,
) -> list[dict[str, Any]]:
"""按专辑名和曲名线索搜索候选发行版本,多个查询按命中顺序去重。"""
releases: list[dict[str, Any]] = []
seen: set[str] = set()
for query in self._release_queries(meta, tracks):
payload = self._request_json(
"/release",
params={"query": query, "limit": max(1, min(limit, 25)), "fmt": "json"},
)
for item in (payload or {}).get("releases") or []:
release_id = item.get("id")
if release_id and release_id not in seen:
seen.add(release_id)
releases.append(item)
if len(releases) >= limit:
break
return releases[:limit]
@classmethod
def _release_queries(cls, meta: MetaMusic, tracks: list[MetaMusic]) -> list[str]:
"""构造专辑搜索表达式:优先专辑名+歌手,无专辑线索时用曲名兜底。"""
queries: list[str] = []
album_title = meta.album or meta.title
artist = meta.artists[0] if meta.artists else meta.album_artist
if album_title:
if artist:
queries.append(
f'release:"{cls._escape_query(album_title)}" AND artist:"{cls._escape_query(artist)}"'
)
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()]
)[:3]
if titles:
recording_clause = " OR ".join(
f'recording:"{cls._escape_query(title)}"' for title in titles
)
query = f"({recording_clause})"
if artist:
query += f' AND artist:"{cls._escape_query(artist)}"'
queries.append(query)
return queries
@classmethod
def _release_track_summary(cls, detail: dict[str, Any]) -> list[dict[str, Any]]:
"""提取发行版本的曲目概要(碟号、曲序、时长、标题)供打分使用。"""
summary: list[dict[str, Any]] = []
for medium in detail.get("media") or []:
disc = cls._optional_int(medium.get("position")) or 1
for track in medium.get("tracks") or []:
recording = track.get("recording") or {}
summary.append({
"disc": disc,
"position": cls._optional_int(track.get("position")),
"length": cls._duration_seconds(
track.get("length") or recording.get("length")
),
"title": track.get("title") or recording.get("title"),
})
return summary
@classmethod
def _score_release(
cls,
meta: MetaMusic,
tracks: list[MetaMusic],
detail: dict[str, Any],
summary: list[dict[str, Any]],
) -> float:
"""给候选发行版本打分(0-100),综合标题、歌手、曲目数和时长相似度。"""
local_count = len(tracks)
release_count = len(summary)
if not release_count:
return 0.0
# 曲目数差异过大直接排除,避免单曲误命中整专或反之
diff = abs(local_count - release_count)
if diff > max(4, int(local_count * 0.5)):
return 0.0
# 本地文件比发行版本多出的曲目无法被覆盖,超出容忍范围视为错误候选
if local_count > release_count and diff > max(1, int(release_count * 0.25)):
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
@staticmethod
def _duration_similarity(left: int, right: int) -> float:
"""比较两个时长的接近程度,完全一致为 1,差异越大越接近 0。"""
if not left or not right:
return 0.0
return max(0.0, 1 - abs(left - right) / max(left, right))
@classmethod
def _text_similarity(cls, left: Optional[str], right: Optional[str]) -> float:
"""忽略大小写和标点后比较两段音乐文本的相似度。"""
normalized_left = cls._match_text(left)
normalized_right = cls._match_text(right)
if not normalized_left or not normalized_right:
return 0.0
return SequenceMatcher(None, normalized_left, normalized_right).ratio()
@staticmethod
def _match_text(value: Optional[str]) -> str:
"""移除大小写、空白和标点差异,生成相似度比较使用的紧凑文本。"""
return re.sub(r"[\W_]+", "", str(value or "").casefold(), flags=re.UNICODE)
@classmethod
def _unique_texts(cls, values: Iterable[Optional[str]]) -> list[str]:
"""按规范化文本去重并保留原始顺序。"""
results: list[str] = []
seen: set[str] = set()
for value in values:
normalized = cls._normalize_text(value)
identity = normalized.casefold()
if not normalized or identity in seen:
continue
seen.add(identity)
results.append(normalized)
return results
@classmethod
def _release_to_album(cls, detail: dict[str, Any]) -> Optional[MusicAlbumInfo]:
"""将 MusicBrainz Release 详情转换为带曲目表的标准化专辑信息。"""
release_id = detail.get("id")
title = detail.get("title")
if not release_id or not title:
return None
release_group = detail.get("release-group") or {}
group_id = release_group.get("id")
artists, artist_ids = cls._artist_credits(detail.get("artist-credit"))
album = MusicAlbumInfo(
source=cls._source,
# 优先使用 Release Group ID,与专辑详情和封面入口保持一致
media_id=str(group_id or release_id),
title=str(title),
artists=artists,
artist_ids=artist_ids,
album_type=release_group.get("primary-type"),
secondary_types=[str(item) for item in release_group.get("secondary-types") or []],
release_date=detail.get("date") or None,
cover_url=cls._build_cover_url(group_id),
genres=cls._names_of(detail.get("genres")),
detail_link=f"https://musicbrainz.org/release/{release_id}",
raw_data={"release_id": str(release_id)},
)
album.tracks = [
info
for medium in detail.get("media") or []
for track in medium.get("tracks") or []
if (info := cls._track_to_info(album, medium, track))
]
return album
def recognize_media( def recognize_media(
self, self,
meta: MetaBase = None, meta: MetaBase = None,
+6 -3
View File
@@ -234,13 +234,15 @@ def test_metainfo_routes_audio_filename_to_music():
def test_metainfo_routes_audio_path_to_music_without_parent_merge(): def test_metainfo_routes_audio_path_to_music_without_parent_merge():
"""音频路径应直接构造音乐元数据,不与父目录季集合并""" """音频路径应直接构造音乐元数据,不参与影视季集合并,并拆分歌手与曲名"""
meta = MetaInfoPath(Path("/music/叶惠美/周杰伦 - 晴天.flac")) meta = MetaInfoPath(Path("/music/叶惠美/周杰伦 - 晴天.flac"))
assert isinstance(meta, MetaMusic) assert isinstance(meta, MetaMusic)
assert meta.type == MediaType.MUSIC assert meta.type == MediaType.MUSIC
assert meta.org_string == "周杰伦 - 晴天.flac" assert meta.org_string == "周杰伦 - 晴天.flac"
assert meta.title == "周杰伦 - 晴天" # 文件名中的歌手与曲名应拆分,便于无标签音频搜索识别
assert meta.title == "晴天"
assert meta.artists == ["周杰伦"]
assert meta.audio_format == "FLAC" assert meta.audio_format == "FLAC"
@@ -259,7 +261,8 @@ def test_metainfo_music_round_trip_preserves_fields():
restored = MetaMusic.from_dict(payload) restored = MetaMusic.from_dict(payload)
assert restored.type == MediaType.MUSIC assert restored.type == MediaType.MUSIC
assert restored.title == "周杰伦 - 晴天" assert restored.title == "晴天"
assert restored.artists == ["周杰伦"]
assert restored.audio_format == "FLAC" assert restored.audio_format == "FLAC"
assert payload["type"] == "音乐" assert payload["type"] == "音乐"
+210
View File
@@ -0,0 +1,210 @@
import pytest
from app.chain.music import MusicChain
from app.core.context import MusicAlbumInfo, MusicInfo
from app.core.meta import MetaMusic
from app.modules.musicbrainz import MusicBrainzModule
def _release_detail(release_id: str, title: str, artist: str, tracks: list[tuple[str, int]]):
"""构造 MusicBrainz Release 详情响应,tracks 为 (曲名, 时长秒) 列表。"""
return {
"id": release_id,
"title": title,
"date": "2004-08-03",
"artist-credit": [{"artist": {"id": "artist-1", "name": artist}}],
"release-group": {"id": "rg-1", "primary-type": "Album", "secondary-types": []},
"media": [
{
"position": 1,
"track-count": len(tracks),
"tracks": [
{
"position": index + 1,
"length": length * 1000,
"title": name,
"recording": {"id": f"rec-{index + 1}", "title": name, "length": length * 1000},
}
for index, (name, length) in enumerate(tracks)
],
}
],
}
ALBUM_TRACKS = [("我的地盘", 215), ("七里香", 299), ("借口", 265)]
def _local_tracks():
"""构造无标签整专目录读取得到的本地曲目元数据。"""
return [
MetaMusic(title=name, track_number=index + 1, duration=length, audio_format="WAV")
for index, (name, length) in enumerate(ALBUM_TRACKS)
]
def test_match_music_album_selects_release_by_count_and_duration(monkeypatch):
"""曲目数和时长一致的发行版本应被选中并返回曲目表。"""
module = MusicBrainzModule()
detail = _release_detail("release-1", "七里香", "周杰伦", ALBUM_TRACKS)
def fake_request(path, params=None):
if path == "/release":
return {"releases": [{"id": "release-1", "title": "七里香"}]}
if path == "/release/release-1":
return detail
return None
monkeypatch.setattr(module, "_request_json", fake_request)
album = module.match_music_album(
MetaMusic(album="七里香", artists=["周杰伦"]),
_local_tracks(),
)
assert album is not None
assert album.media_id == "rg-1"
assert album.title == "七里香"
assert album.artists == ["周杰伦"]
assert [track.media_id for track in album.tracks] == ["rec-1", "rec-2", "rec-3"]
assert album.tracks[0].track_number == 1
assert album.tracks[0].album == "七里香"
def test_match_music_album_rejects_mismatched_trackset(monkeypatch):
"""曲目数和时长都对不上的候选应被拒绝,避免写错标签。"""
module = MusicBrainzModule()
# 候选只有 1 首歌且时长差异巨大
detail = _release_detail("release-1", "七里香", "周杰伦", [("七里香", 60)])
def fake_request(path, params=None):
if path == "/release":
return {"releases": [{"id": "release-1", "title": "七里香"}]}
if path == "/release/release-1":
return detail
return None
monkeypatch.setattr(module, "_request_json", fake_request)
album = module.match_music_album(
MetaMusic(album="七里香", artists=["周杰伦"]),
_local_tracks(),
)
assert album is None
def test_release_queries_fallback_to_track_titles():
"""目录名没有专辑线索时应使用代表性曲名反查发行版本。"""
queries = MusicBrainzModule._release_queries(
MetaMusic(title="Various"),
[MetaMusic(title="晴天"), MetaMusic(title="七里香"), MetaMusic(title="03")],
)
assert any("recording:" in query for query in queries)
# 纯数字文件名不能作为曲名线索
assert all('"03"' not in query for query in queries)
@pytest.fixture()
def music_chain():
"""构造绕过重量级初始化的 MusicChain,并清理目录匹配缓存。"""
chain = MusicChain.__new__(MusicChain)
MusicChain._album_dir_cache.clear()
yield chain
MusicChain._album_dir_cache.clear()
def test_recognize_album_directory_maps_files(tmp_path, music_chain, monkeypatch):
"""目录级匹配应把每个音频文件对位到专辑曲目并缓存结果。"""
album_dir = tmp_path / "周杰伦 - 七里香 (2004)"
album_dir.mkdir()
files = []
for index, (name, length) in enumerate(ALBUM_TRACKS):
file = album_dir / f"{index + 1:02d}.{name}.wav"
file.write_bytes(b"RIFF")
files.append(file)
album = MusicAlbumInfo(
source="musicbrainz",
media_id="rg-1",
title="七里香",
artists=["周杰伦"],
tracks=[
MusicInfo(
source="musicbrainz",
media_id=f"rec-{index + 1}",
title=name,
artists=["周杰伦"],
album="七里香",
track_number=index + 1,
duration=length,
)
for index, (name, length) in enumerate(ALBUM_TRACKS)
],
)
calls = {"count": 0}
def fake_run_module(method, **kwargs):
calls["count"] += 1
return [album] if method == "match_music_album" else []
monkeypatch.setattr(music_chain, "run_module", fake_run_module)
matched = music_chain.recognize_album_directory(album_dir)
assert len(matched) == len(files)
for index, file in enumerate(files):
info = matched[str(file.resolve())]
assert info.media_id == f"rec-{index + 1}"
assert info.title == ALBUM_TRACKS[index][0]
# 同一目录再次识别直接命中缓存,不重复请求模块
assert music_chain.recognize_album_directory(album_dir) == matched
assert calls["count"] == 1
def test_recognize_album_directory_skips_single_file(tmp_path, music_chain, monkeypatch):
"""单文件目录不走专辑匹配,交给单曲识别链路。"""
album_dir = tmp_path / "单曲"
album_dir.mkdir()
(album_dir / "晴天.wav").write_bytes(b"RIFF")
def fake_run_module(method, **kwargs):
raise AssertionError("单文件目录不应触发专辑匹配")
monkeypatch.setattr(music_chain, "run_module", fake_run_module)
assert music_chain.recognize_album_directory(album_dir) == {}
def test_recognize_by_path_falls_back_to_album_match(tmp_path, music_chain, monkeypatch):
"""单曲识别无远端身份时应用目录级匹配结果兜底。"""
album_dir = tmp_path / "周杰伦 - 七里香 (2004)"
album_dir.mkdir()
file = album_dir / "01.我的地盘.wav"
file.write_bytes(b"RIFF")
matched_info = MusicInfo(
source="musicbrainz",
media_id="rec-1",
title="我的地盘",
artists=["周杰伦"],
album="七里香",
track_number=1,
)
monkeypatch.setattr(
music_chain, "recognize_media", lambda **kwargs: None
)
monkeypatch.setattr(
music_chain,
"recognize_album_directory",
lambda path: {str(file.resolve()): matched_info},
)
meta, info = music_chain.recognize_by_path(file)
assert info.media_id == "rec-1"
assert info.title == "我的地盘"
assert info.album == "七里香"
# 本地音频参数应保留在识别结果中
assert meta.audio_format == "WAV"
+131
View File
@@ -0,0 +1,131 @@
from app.core.meta import MetaMusic
from app.helper.music_name import MusicNameParser
def test_strip_track_prefix_handles_dot_separator():
"""曲序前缀 01. 应剥离并返回曲名。"""
track, disc, title = MusicNameParser.strip_track_prefix("01.晴天")
assert (track, disc, title) == (1, None, "晴天")
def test_strip_track_prefix_handles_dash_and_space():
"""常见 rip 命名 01 - 曲名 和 01 曲名 都应识别曲序。"""
assert MusicNameParser.strip_track_prefix("03 - 七里香") == (3, None, "七里香")
assert MusicNameParser.strip_track_prefix("05 借口") == (5, None, "借口")
def test_strip_track_prefix_handles_disc_track_number():
"""碟号-曲序前缀 1-02 应同时提取碟号和曲序。"""
track, disc, title = MusicNameParser.strip_track_prefix("1-02 半岛铁盒")
assert (track, disc, title) == (2, 1, "半岛铁盒")
def test_strip_track_prefix_handles_number_only_name():
"""纯数字文件名只能得到曲序,曲名返回 None 由调用方兜底。"""
track, disc, title = MusicNameParser.strip_track_prefix("07")
assert (track, disc, title) == (7, None, None)
def test_strip_track_prefix_keeps_normal_title():
"""普通曲名不应被误判为曲序前缀。"""
assert MusicNameParser.strip_track_prefix("晴天") == (None, None, None)
assert MusicNameParser.strip_track_prefix("2002") == (None, None, None)
def test_split_artist_title():
"""歌手 - 曲名结构应拆分,无分隔符时原文作为标题。"""
artist, title = MusicNameParser.split_artist_title("周杰伦 - 晴天")
assert artist == "周杰伦"
assert title == "晴天"
assert MusicNameParser.split_artist_title("晴天") == (None, "晴天")
def test_parse_disc_dir():
"""CD1、Disc 2 等碟片目录应识别碟号。"""
assert MusicNameParser.parse_disc_dir("CD1") == 1
assert MusicNameParser.parse_disc_dir("Disc 2") == 2
assert MusicNameParser.parse_disc_dir("disk03") == 3
assert MusicNameParser.parse_disc_dir("无损音乐") is None
def test_parse_album_dir_extracts_artist_album_year():
"""专辑目录名应提取歌手、专辑、年份和音质描述。"""
info = MusicNameParser.parse_album_dir("周杰伦 - 七里香 (2004) [FLAC 24bit-96kHz]")
assert info["artist"] == "周杰伦"
assert info["album"] == "七里香"
assert info["year"] == 2004
assert "FLAC" in info["quality_text"]
def test_parse_album_dir_without_artist():
"""没有歌手分隔的目录名整体作为专辑名。"""
info = MusicNameParser.parse_album_dir("Random Access Memories (2013)")
assert info["artist"] is None
assert info["album"] == "Random Access Memories"
assert info["year"] == 2013
def test_apply_path_context_fills_wav_meta(tmp_path):
"""无标签 WAV 应从文件名和目录结构补齐曲序、专辑和歌手。"""
album_dir = tmp_path / "周杰伦 - 七里香 (2004) [FLAC]"
album_dir.mkdir()
wav_file = album_dir / "01.我的地盘.wav"
wav_file.write_bytes(b"RIFF")
meta = MetaMusic(org_string=wav_file.name, title=wav_file.stem, audio_format="WAV")
MusicNameParser.apply_path_context(meta, wav_file)
assert meta.track_number == 1
assert meta.title == "我的地盘"
assert meta.album == "七里香"
assert meta.artists == ["周杰伦"]
assert meta.album_artist == "周杰伦"
assert meta.year == 2004
def test_apply_path_context_uses_disc_subdir(tmp_path):
"""CD1 子目录内的文件应继承碟号并向上找到专辑目录。"""
album_dir = tmp_path / "Daft Punk - Discovery (2001)"
disc_dir = album_dir / "CD1"
disc_dir.mkdir(parents=True)
wav_file = disc_dir / "01 - One More Time.flac"
wav_file.write_bytes(b"fLaC")
meta = MetaMusic(org_string=wav_file.name, title=wav_file.stem, audio_format="FLAC")
MusicNameParser.apply_path_context(meta, wav_file)
assert meta.disc_number == 1
assert meta.track_number == 1
assert meta.title == "One More Time"
assert meta.album == "Discovery"
assert meta.artists == ["Daft Punk"]
def test_apply_path_context_keeps_existing_tags(tmp_path):
"""已有标签字段不应被目录猜测覆盖。"""
album_dir = tmp_path / "周杰伦 - 七里香 (2004)"
album_dir.mkdir()
audio_file = album_dir / "01.我的地盘.mp3"
audio_file.write_bytes(b"")
meta = MetaMusic(
org_string=audio_file.name,
title="我的地盘",
artists=["周杰伦"],
album="七里香",
year=2004,
track_number=1,
audio_format="MP3",
)
MusicNameParser.apply_path_context(meta, audio_file)
assert meta.title == "我的地盘"
assert meta.artists == ["周杰伦"]
assert meta.album == "七里香"
assert meta.year == 2004