From 07ea99442e3126784198837e9c5b4e6f5e543516 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Mon, 10 Aug 2026 20:11:23 +0800 Subject: [PATCH] =?UTF-8?q?feat(music):=20=E5=A2=9E=E5=BC=BA=E6=97=A0?= =?UTF-8?q?=E6=A0=87=E7=AD=BE=E9=9F=B3=E9=A2=91=E8=AF=86=E5=88=AB=EF=BC=8C?= =?UTF-8?q?=E6=94=AF=E6=8C=81=E6=96=87=E4=BB=B6=E5=90=8D/=E7=9B=AE?= =?UTF-8?q?=E5=BD=95=E8=A7=A3=E6=9E=90=E4=B8=8E=E4=B8=93=E8=BE=91=E7=BA=A7?= =?UTF-8?q?=E5=8C=B9=E9=85=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 MusicNameParser：剥离曲序/碟号前缀、拆分歌手与曲名、解析专辑目录名（歌手/专辑/年份/音质）和 CD1 等碟片目录 - MetaInfoPath 与 MusicChain.read_path_meta 接入路径上下文，WAV 及标签不全的 FLAC/MP3 可补齐识别线索 - MusicBrainz 新增 match_music_album：按专辑名/歌手/曲名搜索候选发行版本，用曲目数、总时长、逐曲时长和曲名重合度打分对位 - MusicChain 新增 recognize_album_directory 目录级批量识别（按目录缓存），单曲识别未命中时自动兜底 - transfer 整理链路接入专辑匹配，命中后回填曲目身份用于重命名与刮削 --- app/chain/music.py | 193 ++++++++++++++++++++- app/chain/transfer.py | 65 +++++++- app/core/metainfo.py | 5 +- app/helper/music_name.py | 173 +++++++++++++++++++ app/modules/musicbrainz/__init__.py | 249 ++++++++++++++++++++++++++++ tests/test_metainfo.py | 9 +- tests/test_music_album_match.py | 210 +++++++++++++++++++++++ tests/test_music_name_parser.py | 131 +++++++++++++++ 8 files changed, 1022 insertions(+), 13 deletions(-) create mode 100644 app/helper/music_name.py create mode 100644 tests/test_music_album_match.py create mode 100644 tests/test_music_name_parser.py diff --git a/app/chain/music.py b/app/chain/music.py index d9b771edb..facca39b0 100644 --- a/app/chain/music.py +++ b/app/chain/music.py @@ -17,6 +17,7 @@ from app.core.context import ( ) from app.core.meta import MetaMusic from app.helper.audio import AudioMetadataHelper +from app.helper.music_name import MusicNameParser from app.log import logger @@ -25,6 +26,11 @@ class MusicChain(ChainBase): _artist_title_pattern = re.compile(r"^\s*(?P.+?)\s+[-–—]\s+(?P.+?)\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 def parse_query(cls, query: str) -> MetaMusic: @@ -317,11 +323,14 @@ class MusicChain(ChainBase): @classmethod def read_path_meta(cls, path: str | Path) -> MetaMusic: - """读取本地音频标签，不可访问时按文件名构造最小音乐元数据。""" + """读取本地音频标签，标签缺失时用文件名和目录线索补齐。""" file_path = Path(path) if file_path.exists() and file_path.is_file(): - return AudioMetadataHelper.read(file_path) - return cls.parse_query(file_path.stem) + meta = AudioMetadataHelper.read(file_path) + 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( self, @@ -331,10 +340,16 @@ class MusicChain(ChainBase): """根据音频标签和文件名识别音乐，远端不可用时仍返回最小音乐信息。""" # Mutagen 会同步读取本地文件，异步识别入口需要移出事件循环。 meta = await run_in_threadpool(self.read_path_meta, path) - # 统一识别入口分发到音乐模块，模块负责详情/搜索/匹配/兜底 + # 统一识别入口分发到音乐模块，模块负责详情/搜索/匹配/兑底 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( self, path: str | Path, @@ -342,9 +357,171 @@ class MusicChain(ChainBase): ) -> tuple[MetaMusic, MusicInfo]: """同步根据音频标签和文件名识别音乐，并保留离线最小结果。""" meta = self.read_path_meta(path) - # 统一识别入口分发到音乐模块，模块负责详情/搜索/匹配/兜底 + # 统一识别入口分发到音乐模块，模块负责详情/搜索/匹配/兑底 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 def to_meta(cls, info: MusicInfo) -> MetaMusic: diff --git a/app/chain/transfer.py b/app/chain/transfer.py index 56adfc473..de2067640 100755 --- a/app/chain/transfer.py +++ b/app/chain/transfer.py @@ -1063,6 +1063,64 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): 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 def _restore_music_download_context( cls, @@ -3626,7 +3684,12 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): # 自动整理预载的媒体信息来自整条下载历史；电影合集内文件年份冲突时逐文件识别。 task_mediainfo = mediainfo or history_music_info 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 ( not manual and self._is_movie_year_conflict(file_meta, task_mediainfo) diff --git a/app/core/metainfo.py b/app/core/metainfo.py index db8dbc69c..fa1b8711a 100644 --- a/app/core/metainfo.py +++ b/app/core/metainfo.py @@ -12,6 +12,7 @@ from app.core.meta.infopath import ( should_use_parent_title_for_file_stem, ) from app.core.meta.words import WordsMatcher +from app.helper.music_name import MusicNameParser from app.log import logger from app.schemas.types import MediaType 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() if not force_video and audio_suffix in settings.RMT_AUDIOEXT: - return MetaMusic( + music_meta = MetaMusic( org_string=path.name, title=path.stem, audio_format=audio_suffix.lstrip(".").upper() or None, ) + # 无标签音频只能依靠文件名和目录结构，补充曲序、碟号、歌手和专辑线索 + return MusicNameParser.apply_path_context(music_meta, path) path_context = " ".join( [path.name, path.parent.name, path.parent.parent.name] ) diff --git a/app/helper/music_name.py b/app/helper/music_name.py new file mode 100644 index 000000000..15685f395 --- /dev/null +++ b/app/helper/music_name.py @@ -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() diff --git a/app/modules/musicbrainz/__init__.py b/app/modules/musicbrainz/__init__.py index d3ea28c1d..9a41427d7 100644 --- a/app/modules/musicbrainz/__init__.py +++ b/app/modules/musicbrainz/__init__.py @@ -1,6 +1,7 @@ import re import threading import time +from difflib import SequenceMatcher from typing import Any, Iterable, Optional, Tuple, Union from fastapi.concurrency import run_in_threadpool @@ -183,6 +184,254 @@ class MusicBrainzModule(_ModuleBase): index += 1 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( self, meta: MetaBase = None, diff --git a/tests/test_metainfo.py b/tests/test_metainfo.py index d8e0975e6..8b885e922 100644 --- a/tests/test_metainfo.py +++ b/tests/test_metainfo.py @@ -234,13 +234,15 @@ def test_metainfo_routes_audio_filename_to_music(): def test_metainfo_routes_audio_path_to_music_without_parent_merge(): - """音频路径应直接构造音乐元数据，不与父目录季集合并。""" + """音频路径应直接构造音乐元数据，不参与影视季集合并，并拆分歌手与曲名。""" meta = MetaInfoPath(Path("/music/叶惠美/周杰伦 - 晴天.flac")) assert isinstance(meta, MetaMusic) assert meta.type == MediaType.MUSIC assert meta.org_string == "周杰伦 - 晴天.flac" - assert meta.title == "周杰伦 - 晴天" + # 文件名中的歌手与曲名应拆分，便于无标签音频搜索识别 + assert meta.title == "晴天" + assert meta.artists == ["周杰伦"] assert meta.audio_format == "FLAC" @@ -259,7 +261,8 @@ def test_metainfo_music_round_trip_preserves_fields(): restored = MetaMusic.from_dict(payload) assert restored.type == MediaType.MUSIC - assert restored.title == "周杰伦 - 晴天" + assert restored.title == "晴天" + assert restored.artists == ["周杰伦"] assert restored.audio_format == "FLAC" assert payload["type"] == "音乐" diff --git a/tests/test_music_album_match.py b/tests/test_music_album_match.py new file mode 100644 index 000000000..23624b05f --- /dev/null +++ b/tests/test_music_album_match.py @@ -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" diff --git a/tests/test_music_name_parser.py b/tests/test_music_name_parser.py new file mode 100644 index 000000000..0b95a5149 --- /dev/null +++ b/tests/test_music_name_parser.py @@ -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