diff --git a/app/application/audio.py b/app/application/audio.py index d6cb7d14e..1c644a8b2 100644 --- a/app/application/audio.py +++ b/app/application/audio.py @@ -3,14 +3,20 @@ from typing import Any, Optional, Union from uuid import UUID from mutagen import File as MutagenFile +from mutagen import MutagenError +from mutagen.aiff import AIFF from mutagen.apev2 import APEBinaryValue, APETextValue +from mutagen.dsdiff import DSDIFF +from mutagen.dsf import DSF from mutagen.flac import FLAC, Picture from mutagen.id3 import APIC, ID3, SYLT, USLT from mutagen.monkeysaudio import MonkeysAudio +from mutagen.mp3 import MP3 from mutagen.mp4 import MP4, MP4Cover, MP4Tags +from mutagen.wave import WAVE from app.domain.context import MusicInfo, MusicLyrics -from app.domain.meta.metamusic import MetaMusic +from app.domain.meta.metamusic import MetaMusic, parse_music_release_types from app.runtime.log import logger from app.schemas.types import MUSIC_ENTITY_RECORDING, MediaSource @@ -27,28 +33,38 @@ def _read_mp4_tags(tags: MP4Tags) -> dict[str, list[str]]: values[key] = [f"{current}/{total}"] normalized = {key.casefold(): value for key, value in tags.items()} for name in ("MusicBrainz Track Id", "MusicBrainz Album Id", "MusicBrainz Release Group Id", - "MusicBrainz Release Track Id", "ISRC", "ORIGINALDATE", "ORIGINALYEAR", "VERSION", "SUBTITLE"): + "MusicBrainz Release Track Id", "MusicBrainz Album Type", "ISRC", "ORIGINALDATE", + "ORIGINALYEAR", "VERSION", "SUBTITLE", "RELEASETYPE"): raw = normalized.get(f"----:com.apple.itunes:{name.casefold()}", []) values[name.casefold().replace(" ", "_")] = [ value.decode("utf-8", errors="replace") if isinstance(value, bytes) else str(value) for value in raw ] + if "cpil" in tags: + values["compilation"] = ["1" if tags["cpil"] else "0"] return values -def _mark_audio_evidence(meta: MetaMusic, info: Any) -> MetaMusic: +def _mark_audio_evidence(meta: MetaMusic, audio: Any) -> MetaMusic: """记录实际标签和流参数的来源，名称推测不能获得标签级可信度。""" tag_fields = ( "title", "artists", "album", "album_artist", "year", "original_year", "release_year", "disc_number", "track_number", "total_discs", "total_tracks", "version", "isrc", "media_source", "media_id", "musicbrainz_release_id", "musicbrainz_release_group_id", - "musicbrainz_release_track_id", + "musicbrainz_release_track_id", "album_type", "secondary_types", ) meta.field_sources = {key: "tag" for key in tag_fields if getattr(meta, key) not in (None, "", [])} for key in ("bit_depth", "sample_rate", "bitrate", "duration"): if getattr(meta, key) is not None: meta.field_sources[key] = "stream" - if meta.audio_format: + info = getattr(audio, "info", None) + detected = next((name for kind, name in ( + (FLAC, "FLAC"), (MP3, "MP3"), (WAVE, "WAV"), (AIFF, "AIFF"), + (MonkeysAudio, "APE"), (DSF, "DSD"), (DSDIFF, "DSD"), + ) if isinstance(audio, kind)), None) + if detected: + meta.apply_audio_quality(detected, overwrite=True, evidence_source="stream") + elif meta.audio_format: source = "stream" if getattr(info, "codec", None) or getattr(info, "codec_description", None) else "filename" meta.field_sources.update(audio_format=source, audio_lossless=source) return meta @@ -87,7 +103,12 @@ class AudioMetadataHelper: def read_tags(cls, path: Path) -> Optional[MetaMusic]: """只读取本地音频标签和流参数，不使用文件名或目录补齐。""" try: - audio = MutagenFile(path, easy=False) + try: + audio = MutagenFile(path, easy=False) + except MutagenError: + # 错误后缀可能让 Mutagen 选错解析器；重试时保留流式读取，仅去掉扩展名提示。 + with path.open("rb") as stream: + audio = MutagenFile(fileobj=stream, filename="audio", easy=False) except Exception as err: logger.warning(f"读取音频标签失败：{path} - {err}") return None @@ -116,12 +137,17 @@ class AudioMetadataHelper: info = getattr(audio, "info", None) original_year = cls._year(cls._first_of(tags, "originaldate", "originalyear")) release_year = cls._year(cls._first_of(tags, "date", "year")) + release_types = [value for key in ("releasetype", "musicbrainz_albumtype", "musicbrainz_album_type") + for value in cls._values(tags, key)] + album_type, secondary_types = parse_music_release_types(release_types, cls._first(tags, "compilation")) meta = MetaMusic( org_string=path.name, title=cls._first(tags, "title"), artists=cls._values(tags, "artist"), album=cls._first(tags, "album"), album_artist=cls._first_of(tags, "albumartist", "album artist"), + album_type=album_type, + secondary_types=secondary_types, year=release_year or original_year, original_year=original_year, release_year=release_year, @@ -146,7 +172,7 @@ class AudioMetadataHelper: musicbrainz_release_track_id=cls._normalize_musicbrainz_id( cls._first_of(tags, "musicbrainz_releasetrackid", "musicbrainz_release_track_id")), ) - return _mark_audio_evidence(meta, info) + return _mark_audio_evidence(meta, audio) @staticmethod def _readable_tags(tags: Any) -> Any: @@ -163,7 +189,7 @@ class AudioMetadataHelper: "title": "TIT2", "artist": "TPE1", "album": "TALB", "albumartist": "TPE2", "date": "TDRC", "originaldate": "TDOR", "tracknumber": "TRCK", "discnumber": "TPOS", "isrc": "TSRC", - "subtitle": "TIT3", + "subtitle": "TIT3", "compilation": "TCMP", } values = { key: [str(value) for frame in tags.getall(frame_id) for value in frame.text] @@ -263,6 +289,7 @@ class AudioMetadataHelper: org_string=path.name, title=path.stem, audio_format=path.suffix.lstrip(".").upper() or None, + field_sources={"audio_format": "filename", "audio_lossless": "filename"} if path.suffix else None, ).apply_path_context(path) @classmethod diff --git a/app/chain/media/album.py b/app/chain/media/album.py index 7dc002683..c929baaba 100644 --- a/app/chain/media/album.py +++ b/app/chain/media/album.py @@ -30,6 +30,7 @@ def _album_directory_cache_key( regions: tuple[str, ...], scripts: tuple[str, ...], contextual_meta: Optional[MetaMusic] = None, + file_scope: Optional[list[str]] = None, ) -> str: """将发行偏好及资源证据纳入缓存键，同目录更换种子线索时重新识别。""" evidence = { @@ -40,7 +41,7 @@ def _album_directory_cache_key( evidence["artists"] = [evidence["album_artist"]] if evidence and not any(evidence.values()): evidence = None - return json.dumps([os.path.abspath(directory), regions, scripts, evidence], ensure_ascii=False, sort_keys=True) + return json.dumps([os.path.abspath(directory), regions, scripts, evidence, file_scope], ensure_ascii=False, sort_keys=True) def _album_context_with_resource( @@ -76,8 +77,15 @@ class MediaAlbumOwner(_MediaOwnerBase): """专辑目录扫描、曲目对齐与缓存编排 owner。""" @classmethod - def _directory_audio_files(cls, directory: Path) -> list[Path]: + def _directory_audio_files(cls, directory: Path, file_paths: Optional[list[Path]] = None) -> list[Path]: """收集专辑目录及其一级碟片子目录中的音频文件。""" + if file_paths is not None: + root = directory.absolute() + return sorted({ + path for raw_path in file_paths if ".." not in raw_path.parts + for path in (raw_path.absolute(),) if path.is_relative_to(root) + and path.is_file() and path.suffix.lower() in get_chain_runtime_config_snapshot().audio_extensions + }) files: list[Path] = [] def collect(current: Path) -> None: @@ -292,12 +300,13 @@ class MediaAlbumOwner(_MediaOwnerBase): music_release_regions: Optional[list[str]] = None, music_release_scripts: Optional[list[str]] = None, contextual_meta: Optional[MetaMusic] = None, + file_paths: Optional[list[Path]] = None, ) -> dict[str, MusicInfo]: """按目录级线索批量识别整张专辑并返回文件到曲目的映射。""" - directory = Path(path) + directory = Path(path).absolute() if not directory.is_dir(): return {} - files = self._directory_audio_files(directory) + files = self._directory_audio_files(directory, file_paths) if file_paths is not None else self._directory_audio_files(directory) if len(files) < self._album_match_min_files: return {} regions, scripts = self._music_release_preferences( @@ -305,7 +314,8 @@ class MediaAlbumOwner(_MediaOwnerBase): music_release_scripts, ) custom_preference = music_release_regions is not None or music_release_scripts is not None - key = _album_directory_cache_key(directory, regions, scripts, contextual_meta) + scope = [str(file.relative_to(directory)) for file in files] if file_paths is not None else None + key = _album_directory_cache_key(directory, regions, scripts, contextual_meta, scope) signature = self._album_directory_signature(directory, files) matched = self._album_dir_cache.resolve( key, @@ -333,12 +343,16 @@ class MediaAlbumOwner(_MediaOwnerBase): music_release_regions: Optional[list[str]] = None, music_release_scripts: Optional[list[str]] = None, contextual_meta: Optional[MetaMusic] = None, + file_paths: Optional[list[Path]] = None, ) -> dict[str, MusicInfo]: """异步按目录级线索批量识别整张专辑。""" - directory = Path(path) + directory = Path(path).absolute() if not await run_in_threadpool(_is_directory, directory): return {} - files = await run_in_threadpool(self._directory_audio_files, directory) + files = ( + await run_in_threadpool(self._directory_audio_files, directory, file_paths) + if file_paths is not None else await run_in_threadpool(self._directory_audio_files, directory) + ) if len(files) < self._album_match_min_files: return {} regions, scripts = self._music_release_preferences( @@ -346,7 +360,8 @@ class MediaAlbumOwner(_MediaOwnerBase): music_release_scripts, ) custom_preference = music_release_regions is not None or music_release_scripts is not None - key = _album_directory_cache_key(directory, regions, scripts, contextual_meta) + scope = [str(file.relative_to(directory)) for file in files] if file_paths is not None else None + key = _album_directory_cache_key(directory, regions, scripts, contextual_meta, scope) signature = self._album_directory_signature(directory, files) matched = await self._album_dir_cache.async_resolve( key, diff --git a/app/chain/media/contract.py b/app/chain/media/contract.py index 1d45d988f..fb7bfe526 100644 --- a/app/chain/media/contract.py +++ b/app/chain/media/contract.py @@ -172,6 +172,7 @@ if TYPE_CHECKING: music_release_regions: Optional[list[str]] = None, music_release_scripts: Optional[list[str]] = None, contextual_meta: Optional[MetaMusic] = None, + file_paths: Optional[list[Path]] = None, ) -> dict[str, MusicInfo]: """同步识别音乐专辑目录。""" ... @@ -192,6 +193,7 @@ if TYPE_CHECKING: music_release_regions: Optional[list[str]] = None, music_release_scripts: Optional[list[str]] = None, contextual_meta: Optional[MetaMusic] = None, + file_paths: Optional[list[Path]] = None, ) -> dict[str, MusicInfo]: """异步识别音乐专辑目录。""" ... diff --git a/app/chain/transfer/filter.py b/app/chain/transfer/filter.py index cd4e5e253..ffa7bf2f5 100644 --- a/app/chain/transfer/filter.py +++ b/app/chain/transfer/filter.py @@ -20,6 +20,7 @@ from app.chain.media import MediaChain from app.chain.storage import StorageChain from app.chain.transfer.contract import _TransferOwnerBase from app.chain.transfer.music import ( + MusicReleaseGroup, prepare_music_batch_context, resolve_music_batch_file_context, restore_music_resource_meta, @@ -180,6 +181,7 @@ class _MusicFileFilterBase(_TransferOwnerBase): file_meta: MetaMusic, music_release_regions: Optional[list[str]] = None, music_release_scripts: Optional[list[str]] = None, + release_group: Optional[MusicReleaseGroup] = None, ) -> tuple[MetaMusic, Optional[MusicInfo]]: """为缺少远端身份的本地音频尝试目录级专辑匹配，命中后回填文件元数据。 @@ -191,10 +193,11 @@ class _MusicFileFilterBase(_TransferOwnerBase): return file_meta, None try: matched = MediaChain().recognize_music_album_directory( - file_path.parent, + release_group.directory if release_group else file_path.parent, music_release_regions=music_release_regions, music_release_scripts=music_release_scripts, contextual_meta=file_meta, + **({"file_paths": release_group.paths} if release_group else {}), ) except Exception as err: logger.debug(f"音乐专辑目录匹配失败：{file_path} - {err}") diff --git a/app/chain/transfer/music.py b/app/chain/transfer/music.py index cf73dc08e..e83d55a00 100644 --- a/app/chain/transfer/music.py +++ b/app/chain/transfer/music.py @@ -25,9 +25,36 @@ class MusicBatchContext: related_main_keys: dict[Tuple[str, str], Tuple[str, str]] = field(default_factory=dict) single_main_keys: set[Tuple[str, str]] = field(default_factory=set) album_main_keys: set[Tuple[str, str]] = field(default_factory=set) - directory_evidence: dict[Tuple[str, str], MetaMusic] = field(default_factory=dict) album_evidence_by_main_key: dict[Tuple[str, str], MetaMusic] = field(default_factory=dict) resolved_contexts: dict[Tuple[str, str], tuple[MetaMusic, MusicInfo]] = field(default_factory=dict) + tags_by_file: dict[Tuple[str, str], Optional[MetaMusic]] = field(default_factory=dict) + release_by_main_key: dict[Tuple[str, str], "MusicReleaseGroup"] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class MusicReleaseGroup: + """同一物理发行单元的文件范围和仅供专辑级补缺的标签共识。""" + + directory: Path + files: tuple[FileItem, ...] + evidence: Optional[MetaMusic] + + @property + def paths(self) -> list[Path]: + """返回完整音轨路径列表，路径失效时拒绝悄悄缩小识别范围。""" + return [_release_item_path(item) for item in self.files] + + +def _music_file_key(item: FileItem) -> Tuple[str, str]: + """按存储与规范路径隔离标签快照，远端同名路径不能借用本机文件标签。""" + return item.storage or "local", str(Path(item.path)) if item.path else "" + + +def _release_item_path(item: FileItem) -> Path: + """发行分组只接受已经解析的实际文件路径，不能把空路径解释成当前目录。""" + if not item.path: + raise ValueError("发行分组中的音轨缺少路径") + return Path(item.path) def restore_music_resource_meta( @@ -85,10 +112,10 @@ def restore_music_resource_meta( return meta -def _music_directory_consensus(values: list[str]) -> Optional[str]: +def _music_directory_consensus(values: list[str], *, artist: bool = False) -> Optional[str]: """只在同目录至少两个标签高度一致时返回共识值。""" normalized = [ - (music_text_key(value), value.strip()) + ("variousartists" if artist and _collective_artist(value) else music_text_key(value), value.strip()) for value in values if value and music_text_key(value) ] @@ -101,72 +128,168 @@ def _music_directory_consensus(values: list[str]) -> Optional[str]: return next(value for current_key, value in normalized if current_key == key) -def _music_directory_evidence(items: list[FileItem]) -> Optional[MetaMusic]: +def _music_directory_evidence( + items: list[FileItem], tags_by_file: dict[Tuple[str, str], Optional[MetaMusic]], +) -> Optional[MetaMusic]: """从同一发行目录的原始音频标签提取可信艺人与专辑共识。""" artist_values: list[str] = [] + track_artists: list[str] = [] album_values: list[str] = [] - collective_keys = { - music_text_key(value) - for value in ("Various Artists", "Various", "VA", "群星", "众艺人", "眾藝人") - } + metas: list[MetaMusic] = [] for item in items: if getattr(item, "storage", "local") != "local" or not item.path: continue - tag_meta = AudioMetadataHelper.read_tags(Path(item.path)) + tag_meta = tags_by_file.get(_music_file_key(item)) if not tag_meta: continue - artist = tag_meta.album_artist - if not artist and len(tag_meta.artists) == 1: - artist = tag_meta.artists[0] - if artist and music_text_key(artist) not in collective_keys: + artist = _album_artist(tag_meta, _release_directory(item)) + if artist: artist_values.append(artist) if tag_meta.album: album_values.append(tag_meta.album) - artist = _music_directory_consensus(artist_values) + if len(tag_meta.artists) == 1: + track_artists.append(tag_meta.artists[0]) + metas.append(tag_meta) + artist = _music_directory_consensus(artist_values, artist=True) + track_artist = _music_directory_consensus(track_artists, artist=True) album = _music_directory_consensus(album_values) if not artist and not album: return None - return MetaMusic( - artists=[artist] if artist else None, + evidence = MetaMusic( + artists=[track_artist] if track_artist and not _collective_artist(track_artist) else None, album_artist=artist, album=album, ) - - -def _music_tagged_album_groups( - items: list[FileItem], -) -> list[tuple[list[FileItem], MetaMusic]]: - """按精确的专辑艺人和专辑标签划分同目录内的多碟/多发行分组。""" - grouped: dict[tuple[str, str], tuple[list[FileItem], str, str]] = {} - collective_keys = { - music_text_key(value) - for value in ("Various Artists", "Various", "VA", "群星", "众艺人", "眾藝人") - } - for item in items: - if getattr(item, "storage", "local") != "local" or not item.path: - continue - tag_meta = AudioMetadataHelper.read_tags(Path(item.path)) - if not tag_meta or not tag_meta.album: - continue - artist = tag_meta.album_artist - if not artist and len(tag_meta.artists) == 1: - artist = tag_meta.artists[0] - artist_key = music_text_key(artist) - album_key = music_text_key(tag_meta.album) - if not artist or artist_key in collective_keys or not album_key: - continue - key = (artist_key, album_key) - if key not in grouped: - grouped[key] = ([], artist.strip(), tag_meta.album.strip()) - grouped[key][0].append(item) - return [ - ( - grouped_items, - MetaMusic(artists=[artist], album_artist=artist, album=album), + for key in ("musicbrainz_release_id", "musicbrainz_release_group_id", "album_type", "original_year", "release_year"): + values = {getattr(meta, key) for meta in metas if getattr(meta, key)} + if len(values) == 1: + setattr(evidence, key, next(iter(values))) + secondary = {tuple(meta.secondary_types) for meta in metas if meta.secondary_types} + if len(secondary) == 1: + evidence.secondary_types = list(next(iter(secondary))) + discs = [meta.total_discs or meta.disc_number or 1 for meta in metas] + discs.extend(MetaMusic.parse_disc_dir(_release_item_path(item).parent.name) or 1 for item in items) + evidence.total_discs = max(discs, default=1) if max(discs, default=1) > 1 else None + if evidence.total_discs: + evidence.field_sources["total_discs"] = ( + "album_tags" if any(meta.total_discs == evidence.total_discs for meta in metas) else "directory" ) - for grouped_items, artist, album in grouped.values() - if len(grouped_items) >= 2 - ] + return evidence + + +def _collective_artist(value: Optional[str]) -> bool: + """识别专辑层的群星署名，不把它复制成每首歌曲的演唱者。""" + return music_text_key(value) in { + music_text_key(name) for name in ("Various Artists", "Various", "VA", "群星", "众艺人", "眾藝人") + } + + +def _album_artist(meta: MetaMusic, directory: Path) -> Optional[str]: + """专辑署名优先读 Album Artist，明确的合辑标志可补群星，但不改逐曲艺人。""" + if meta.album_artist: + return meta.album_artist + if "Compilation" in meta.secondary_types: + return "Various Artists" + parsed = MetaMusic.parse_album_dir(directory.name) + if _collective_artist(parsed.get("artist")) and music_text_key(parsed.get("album")) == music_text_key(meta.album): + return cast(str, parsed["artist"]) + return meta.artists[0] if len(meta.artists) == 1 else None + + +def _release_signature(meta: MetaMusic, directory: Path) -> tuple[str, str, Any]: + """缺少发行 ID 时，以专辑署名、标题和当前年份限定可关联的标签。""" + artist = _album_artist(meta, directory) + artist_key = "variousartists" if _collective_artist(artist) else music_text_key(artist) + return artist_key, music_text_key(meta.album), meta.release_year or meta.year + + +def _rendition_groups(group: list[FileItem], tags: dict[Tuple[str, str], Optional[MetaMusic]]) -> list[list[FileItem]]: + """同一曲序存在多份时按实际编码区分版本，正常混合介质的多碟仍保持一组。""" + positions = Counter( + (meta.disc_number or MetaMusic.parse_disc_dir(_release_item_path(item).parent.name) or 1, meta.track_number) + for item in group if (meta := tags.get(_music_file_key(item))) and meta.track_number + ) + if not any(count > 1 for count in positions.values()): + return [group] + renditions: dict[tuple[Any, ...], list[FileItem]] = {} + for item in group: + meta = tags.get(_music_file_key(item)) + key = (meta.audio_format, meta.bit_depth, meta.sample_rate) if meta else (None,) + renditions.setdefault(key, []).append(item) + return list(renditions.values()) + + +def _release_groups(items: list[FileItem], tags_by_file: dict[Tuple[str, str], Optional[MetaMusic]]) -> list[MusicReleaseGroup]: + """优先按具体发行 ID 分组，缺 ID 时按专辑署名、标题及发行年隔离同目录多专辑。""" + grouped: dict[tuple[Any, ...], list[FileItem]] = {} + unassigned: list[FileItem] = [] + known: dict[tuple[str, str, Any], set[str]] = {} + for item in items: + meta = tags_by_file.get(_music_file_key(item)) + if meta and meta.musicbrainz_release_id: + known.setdefault(_release_signature(meta, _release_directory(item)), set()).add(meta.musicbrainz_release_id) + for item in items: + meta = tags_by_file.get(_music_file_key(item)) + if not meta: + unassigned.append(item) + continue + signature = _release_signature(meta, _release_directory(item)) + matching_ids = known.get(signature, set()) + key: tuple[Any, ...] + if meta.musicbrainz_release_id: + key = ("release", meta.musicbrainz_release_id) + elif len(matching_ids) == 1: + key = ("release", next(iter(matching_ids))) + elif signature[0] and signature[1]: + key = ("tags", *signature) + else: + unassigned.append(item) + continue + grouped.setdefault(key, []).append(item) + if len(grouped) == 1: + existing = next(iter(grouped.values())) + album_names = {music_text_key(meta.album) for item in existing if (meta := tags_by_file.get(_music_file_key(item))) and meta.album} + album_artists = {_album_artist(meta, _release_directory(item)) for item in existing + if (meta := tags_by_file.get(_music_file_key(item)))} + artist_keys = {music_text_key(artist) for artist in album_artists if artist} + collective = any(_collective_artist(artist) for artist in album_artists) + directory = MetaMusic.parse_album_dir(_release_directory(existing[0]).name) + directory_matches = bool( + (directory.get("artist") or directory.get("year")) + and music_text_key(directory.get("album")) in album_names + ) + enough_tags = len(existing) * 2 >= len(items) or directory_matches + remaining = [] + for item in unassigned: + meta = tags_by_file.get(_music_file_key(item)) + if not enough_tags: + remaining.append(item) + continue + if meta and not meta.album and meta.artists and not collective and not ( + artist_keys & {music_text_key(artist) for artist in meta.artists} + ): + remaining.append(item) + continue + if not meta or not meta.album or music_text_key(meta.album) in album_names: + existing.append(item) + else: + remaining.append(item) + unassigned = remaining + if unassigned: + grouped[("unassigned",)] = unassigned + output = [] + for group in grouped.values(): + for rendition in _rendition_groups(group, tags_by_file): + root = _release_directory(rendition[0]) + evidence = _music_directory_evidence(rendition, tags_by_file) + output.append(MusicReleaseGroup(root, tuple(rendition), evidence)) + return output + + +def _release_directory(item: FileItem) -> Path: + """把 CD/Disc 子目录定位到所属专辑根，不跨越其它子专辑目录。""" + parent = _release_item_path(item).parent + return parent.parent if MetaMusic.parse_disc_dir(parent.name) is not None else parent def _apply_music_directory_evidence( @@ -186,11 +309,27 @@ def _apply_music_directory_evidence( if not merged.album and evidence.album: merged.album = evidence.album merged.field_sources["album"] = "album_tags" + for key in ("musicbrainz_release_id", "musicbrainz_release_group_id", "total_discs", "album_type", + "original_year", "release_year"): + if not getattr(merged, key) and getattr(evidence, key): + setattr(merged, key, getattr(evidence, key)) + merged.field_sources[key] = evidence.field_sources.get(key, "album_tags") + if not merged.secondary_types and evidence.secondary_types: + merged.secondary_types = list(evidence.secondary_types) + merged.field_sources["secondary_types"] = "album_tags" + if not merged.year and (merged.release_year or merged.original_year): + merged.year = cast(Any, merged.release_year or merged.original_year) + merged.field_sources["year"] = merged.field_sources.get("release_year", merged.field_sources.get("original_year", "album_tags")) return merged def _apply_music_directory_year(meta: MetaMusic, file_path: Path) -> MetaMusic: - """用最近发行目录开头的四位年份纠正合集内不可靠的音频年份标签。""" + """保留明确的标签发行年，仅用目录年份补正旧版未区分语义的 year。""" + if meta.release_year and meta.field_sources.get("release_year") in {"tag", "album_tags"}: + merged = deepcopy(meta) + merged.year = cast(Any, meta.release_year) + merged.field_sources["year"] = meta.field_sources["release_year"] + return merged for parent in list(file_path.parents)[:4]: match = re.match(r"^((?:19|20)\d{2})(?:\D|$)", parent.name) if not match: @@ -206,11 +345,16 @@ def _local_music_context( owner: _TransferOwnerBase, file_item: FileItem, file_path: Path, + batch_context: MusicBatchContext, ) -> Optional[tuple[MetaMusic, MusicInfo]]: """仅以本地真实标签建立快速整理上下文，不向在线来源确认身份。""" if file_item.storage != "local" or not owner._is_audio_file(file_item): return None - tags = AudioMetadataHelper.read_tags(file_path) + tags = batch_context.tags_by_file.get(_music_file_key(file_item)) if _music_file_key(file_item) in batch_context.tags_by_file else AudioMetadataHelper.read_tags(file_path) + if tags: + tags = deepcopy(tags) + if owner._get_file_key(file_item) in batch_context.album_main_keys: + tags = _apply_music_directory_evidence(tags, batch_context.album_evidence_by_main_key.get(owner._get_file_key(file_item))) if tags is None or not music_tags_are_usable(tags): return None meta = tags.apply_path_context(file_path) @@ -225,35 +369,34 @@ def prepare_music_batch_context( ) -> MusicBatchContext: """建立同目录音轨、歌词和单音轨目录的批次索引。""" context = MusicBatchContext() - if batch_mtype != MediaType.MUSIC: + if batch_mtype not in (None, MediaType.MUSIC): return context main_items_by_dir: dict[Tuple[str, str], list[FileItem]] = {} + release_items: dict[Tuple[str, str], list[FileItem]] = {} for current_item, current_bluray_dir in file_items: if not current_bluray_dir and owner._is_media_file(current_item, MediaType.MUSIC): main_items_by_dir.setdefault( owner._get_file_parent_key(current_item), [] ).append(current_item) + if current_item.path: + release_items.setdefault((current_item.storage or "local", str(_release_directory(current_item))), []).append(current_item) + if current_item.storage == "local": + context.tags_by_file[_music_file_key(current_item)] = AudioMetadataHelper.read_tags(Path(current_item.path)) context.single_main_keys = { owner._get_file_key(items[0]) - for items in main_items_by_dir.values() + for items in release_items.values() if len(items) == 1 } - for parent_key, items in main_items_by_dir.items(): - evidence = _music_directory_evidence(items) - if evidence: - context.directory_evidence[parent_key] = evidence - # 两首及以上音轨，且目录内艺人和专辑标签均达到高一致性时，足以在 - # 远端临时不可用时证明这是一个专辑目录；不能把整个艺术家合集根目录 - # 当成一张专辑，也不能仅凭文件夹名称猜测类别。 - if evidence and len(items) > 1 and evidence.artists and evidence.album: - context.album_main_keys.update( - owner._get_file_key(item) for item in items - ) - for grouped_items, grouped_evidence in _music_tagged_album_groups(items): - for item in grouped_items: + for items in release_items.values(): + for group in _release_groups(items, context.tags_by_file): + evidence = group.evidence + for item in group.files: item_key = owner._get_file_key(item) - context.album_main_keys.add(item_key) - context.album_evidence_by_main_key[item_key] = grouped_evidence + context.release_by_main_key[item_key] = group + if evidence: + context.album_evidence_by_main_key[item_key] = evidence + if evidence and len(group.files) > 1 and evidence.album and (evidence.album_artist or evidence.artists): + context.album_main_keys.add(item_key) for current_item, _current_bluray_dir in file_items: if not owner._is_music_lyrics_file(current_item): continue @@ -266,6 +409,55 @@ def prepare_music_batch_context( return context +def _finalize_music_group_context( + owner: _TransferOwnerBase, + context: MusicBatchContext, + item: FileItem, + meta: Any, + info: Optional[Union[MediaInfo, MusicInfo]], + *, + local: bool, + preserve_selection: bool, +) -> tuple[Any, Optional[Union[MediaInfo, MusicInfo]]]: + """统一同一发行的专辑名称与分类，保留各曲艺人及用户明确选择。""" + if not isinstance(meta, MetaMusic) or not isinstance(info, MusicInfo): + return meta, info + key = owner._get_file_key(item) + grouped = key in context.album_main_keys and not preserve_selection + if grouped: + meta, info = deepcopy(meta), deepcopy(info) + evidence = context.album_evidence_by_main_key.get(key) + for name in ("album", "album_artist", "total_discs"): + value = getattr(evidence, name, None) + if value: + setattr(meta, name, value) + setattr(info, name, value) + source = evidence.field_sources.get(name, "album_tags") if evidence else "album_tags" + meta.field_sources[name] = info.field_sources[name] = source + if meta.year: + info.year = meta.year + if meta.album_type: + info.album_type = meta.album_type + info.field_sources["album_type"] = meta.field_sources.get("album_type", "album_tags") + elif str(info.album_type or "").casefold() not in {"ep", "broadcast", "other"}: + # 未声明具体类型的多音轨标签共识沿用 Album 兜底，不能覆盖已声明的 EP/Single。 + info.album_type = "Album" + info.field_sources["album_type"] = "album_tags" + if meta.secondary_types: + info.secondary_types = list(meta.secondary_types) + info.field_sources["secondary_types"] = meta.field_sources.get("secondary_types", "album_tags") + if local or grouped: + finalized = owner._finalize_recognition_result( + info, **({"allow_enrichment": False} if local else {"refresh": True}), + ) + if isinstance(finalized, MusicInfo): + info = finalized + if not local and grouped and not info.classification and info.album_type == "Album": + # 没有装配分类服务的旧调用方保留兼容目录；真实策略结果不能被硬编码覆盖。 + info.set_library_category("Album") + return meta, info + + def resolve_music_batch_file_context( owner: _TransferOwnerBase, *, @@ -298,8 +490,9 @@ def resolve_music_batch_file_context( selected_tracks, fallback, ) + selected_context = task_mediainfo is not None local_context = ( - _local_music_context(owner, file_item, file_path) + _local_music_context(owner, file_item, file_path, batch_context) if not task_mediainfo and isinstance(file_meta, MetaMusic) else None ) @@ -309,8 +502,7 @@ def resolve_music_batch_file_context( file_key = owner._get_file_key(file_item) file_meta = _apply_music_directory_evidence( file_meta, - batch_context.album_evidence_by_main_key.get(file_key) - or batch_context.directory_evidence.get(owner._get_file_parent_key(file_item)), + batch_context.album_evidence_by_main_key.get(file_key), ) if discard_shared_identity and not local_context: file_meta = _apply_music_directory_year(file_meta, file_path) @@ -334,51 +526,10 @@ def resolve_music_batch_file_context( if isinstance(task_mediainfo, MusicInfo) and file_meta.year: task_mediainfo = deepcopy(task_mediainfo) task_mediainfo.year = file_meta.year - directory_evidence = ( - batch_context.album_evidence_by_main_key.get(owner._get_file_key(file_item)) - or batch_context.directory_evidence.get(owner._get_file_parent_key(file_item)) + file_meta, task_mediainfo = _finalize_music_group_context( + owner, batch_context, file_item, file_meta, task_mediainfo, + local=bool(local_context), preserve_selection=selected_context, ) - if ( - owner._is_audio_file(file_item) - and isinstance(file_meta, MetaMusic) - and isinstance(task_mediainfo, MusicInfo) - and owner._get_file_key(file_item) in batch_context.album_main_keys - and directory_evidence - ): - # 同一物理发行目录只能生成一个专辑目录。逐曲远端识别可能因服务 - # 波动而混用正式标题、别名或本地兜底；用已验证的目录标签共识统一 - # 专辑和专辑艺人，但保留每首曲目的远端 ID、曲名、封面等信息。 - file_meta = deepcopy(file_meta) - task_mediainfo = deepcopy(task_mediainfo) - if directory_evidence.album: - file_meta.album = directory_evidence.album - task_mediainfo.album = directory_evidence.album - file_meta.field_sources["album"] = task_mediainfo.field_sources["album"] = "album_tags" - if directory_evidence.album_artist: - file_meta.album_artist = directory_evidence.album_artist - task_mediainfo.album_artist = directory_evidence.album_artist - file_meta.field_sources["album_artist"] = task_mediainfo.field_sources["album_artist"] = "album_tags" - if file_meta.year: - task_mediainfo.year = file_meta.year - if ( - owner._is_audio_file(file_item) - and isinstance(task_mediainfo, MusicInfo) - and owner._get_file_key(file_item) in batch_context.album_main_keys - and str(task_mediainfo.album_type or "").casefold() - not in {"ep", "broadcast", "other"} - ): - # 曲目可能同时作为 Single 单独发行。整理完整多音轨目录时，目录内 - # 高一致性的专辑/专辑艺人标签是当前文件归属的更强证据，不能让某一 - # 首歌的单曲发行身份把同一张专辑拆进 Single 分类。 - task_mediainfo = deepcopy(task_mediainfo) - task_mediainfo.album_type = "Album" - if not local_context: - finalized = owner._finalize_recognition_result(task_mediainfo, refresh=True) - if isinstance(finalized, MusicInfo): - task_mediainfo = finalized - task_mediainfo.set_library_category("Album") - if local_context: - task_mediainfo = owner._finalize_recognition_result(task_mediainfo, allow_enrichment=False) if ( owner._is_audio_file(file_item) and isinstance(file_meta, MetaMusic) @@ -410,6 +561,7 @@ def _recognize_music_batch_file( file_meta, release_regions, release_scripts, + release_group=batch_context.release_by_main_key.get(owner._get_file_key(file_item)), ) if not task_mediainfo and discard_shared_identity and owner._is_audio_file(file_item): file_meta, task_mediainfo = owner._match_music_recording_context( @@ -425,23 +577,23 @@ def _recognize_music_batch_file( multi_track_batch and owner._get_file_key(file_item) in batch_context.album_main_keys ): - task_mediainfo.album_type = "Album" - finalized = owner._finalize_recognition_result(task_mediainfo) + task_mediainfo.album_type = file_meta.album_type or "Album" + finalized = owner._finalize_recognition_result(task_mediainfo, allow_enrichment=False) if isinstance(finalized, MusicInfo): task_mediainfo = finalized - if not task_mediainfo.library_category: - task_mediainfo.set_library_category("Album") + if not task_mediainfo.library_category and not task_mediainfo.classification: + task_mediainfo.set_library_category(task_mediainfo.album_type) return file_meta, task_mediainfo if owner._get_file_key(file_item) not in batch_context.single_main_keys: return file_meta, task_mediainfo # 远端识别均未命中时，仅单音轨子目录可安全按 Single 兜底； # 多音轨目录仍保持未识别，避免把缺失专辑误判为单曲。 - task_mediainfo.album_type = "Single" - finalized = owner._finalize_recognition_result(task_mediainfo) + task_mediainfo.album_type = file_meta.album_type or "Single" + finalized = owner._finalize_recognition_result(task_mediainfo, allow_enrichment=False) if isinstance(finalized, MusicInfo): task_mediainfo = finalized - if not task_mediainfo.library_category: + if not task_mediainfo.library_category and not task_mediainfo.classification: # 隔离测试可能没有装配分类服务，保留旧分类路径作为测试兼容。 - task_mediainfo.set_library_category("Single") + task_mediainfo.set_library_category(task_mediainfo.album_type) return file_meta, task_mediainfo diff --git a/app/domain/context.py b/app/domain/context.py index 96f59eff3..bb399fda4 100644 --- a/app/domain/context.py +++ b/app/domain/context.py @@ -846,6 +846,8 @@ class MusicInfo: media_source=normalize_media_source(meta.media_source), media_id=meta.media_id, music_type=meta.music_type or MUSIC_ENTITY_RECORDING, + album_type=meta.album_type, + secondary_types=list(meta.secondary_types), musicbrainz_release_id=meta.musicbrainz_release_id, musicbrainz_release_group_id=meta.musicbrainz_release_group_id, musicbrainz_release_track_id=meta.musicbrainz_release_track_id, diff --git a/app/domain/meta/metamusic.py b/app/domain/meta/metamusic.py index 58182d7a6..a6319a6e5 100644 --- a/app/domain/meta/metamusic.py +++ b/app/domain/meta/metamusic.py @@ -45,6 +45,28 @@ _LOSSY_AUDIO_FORMATS = frozenset({"MP3", "AAC", "OGG", "OPUS", "WMA"}) logger = logging.getLogger(__name__) +def parse_music_release_types(values: Any, compilation: Optional[str] = None) -> tuple[Optional[str], list[str]]: + """将标签中的发行主副类型归一，保留 EP/Single，不从曲目数推断类型。""" + primary_names = {"album": "Album", "studio album": "Album", "专辑": "Album", "專輯": "Album", + "single": "Single", "单曲": "Single", "單曲": "Single", "ep": "EP", + "broadcast": "Broadcast", "other": "Other"} + secondary_names = {"compilation": "Compilation", "精选集": "Compilation", "合辑": "Compilation", + "soundtrack": "Soundtrack", "ost": "Soundtrack", "原声": "Soundtrack", + "live": "Live", "现场": "Live", "remix": "Remix", "dj-mix": "DJ-mix", + "mixtape": "Mixtape/Street", "spokenword": "Spokenword"} + primary: Optional[str] = None + secondary: list[str] = [] + for value in _string_list(values): + for token in re.split(r"[;,/+\\]", value): + key = token.strip().casefold() + primary = primary or primary_names.get(key) + if key in secondary_names and secondary_names[key] not in secondary: + secondary.append(secondary_names[key]) + if str(compilation or "").strip().casefold() in {"1", "true", "yes"} and "Compilation" not in secondary: + secondary.append("Compilation") + return primary, secondary + + def normalize_audio_format(value: Any) -> Optional[str]: """将音频格式名称归一为订阅筛选和展示使用的规范值。""" text = str(value or "").strip().upper() @@ -735,6 +757,8 @@ class MetaMusic(MetaBase): release_year: Optional[int] = None, field_sources: Optional[dict[str, str]] = None, music_type: Optional[str] = None, + album_type: Optional[str] = None, + secondary_types: Optional[list[str]] = None, ): """初始化音乐标题、标签、音频规格和统一媒体身份。""" # 音乐无季集概念，仅复用 MetaBase 的基础字段初始化，不触发副标题季集识别 @@ -753,6 +777,8 @@ class MetaMusic(MetaBase): self.musicbrainz_release_track_id = musicbrainz_release_track_id self.field_sources = dict(field_sources or {}) self.music_type = music_type + self.album_type = album_type + self.secondary_types = list(secondary_types or []) self.disc_number = disc_number self.track_number = track_number self.total_discs = total_discs @@ -781,6 +807,8 @@ class MetaMusic(MetaBase): self.__dict__.setdefault(key, None) self.field_sources = dict(state.get("field_sources") or {}) self.music_type = state.get("music_type") + self.album_type = state.get("album_type") + self.secondary_types = list(state.get("secondary_types") or []) @classmethod def parse_query(cls, query: str) -> "MetaMusic": @@ -1006,6 +1034,8 @@ class MetaMusic(MetaBase): media_source=info.media_source, media_id=info.media_id, music_type=getattr(info, "music_type", None), + album_type=getattr(info, "album_type", None), + secondary_types=getattr(info, "secondary_types", None), musicbrainz_release_id=getattr(info, "musicbrainz_release_id", None), musicbrainz_release_group_id=getattr(info, "musicbrainz_release_group_id", None), musicbrainz_release_track_id=getattr(info, "musicbrainz_release_track_id", None), @@ -1710,7 +1740,7 @@ class MetaMusic(MetaBase): def parse_disc_dir(cls, name: str) -> Optional[int]: """识别 CD1、Disc 2 这类碟片子目录并返回碟号。""" match = _MUSIC_DISC_DIR_RE.match(str(name or "").strip()) - return int(match.group("num")) if match else None + return int(match.group("num")) if match and int(match.group("num")) > 0 else None @staticmethod def _normalize_album_year(value: str) -> str: @@ -1912,6 +1942,8 @@ class MetaMusic(MetaBase): "release_year": self.release_year, "field_sources": dict(self.field_sources), "music_type": self.music_type, + "album_type": self.album_type, + "secondary_types": list(self.secondary_types), } @classmethod @@ -1948,6 +1980,8 @@ class MetaMusic(MetaBase): release_year=_optional_int(data.get("release_year")), field_sources=data.get("field_sources"), music_type=data.get("music_type"), + album_type=data.get("album_type"), + secondary_types=_string_list(data.get("secondary_types")), ) meta.apply_words = data.get("apply_words") return meta diff --git a/app/schemas/music.py b/app/schemas/music.py index 2b5a72cae..a3d36de83 100644 --- a/app/schemas/music.py +++ b/app/schemas/music.py @@ -13,6 +13,8 @@ class MusicMeta(OptionalMediaIdentityMixin, BaseModel): type: Literal["音乐"] = "音乐" music_type: Optional[MusicEntityType] = Field(default=None, description="已知主身份的实体类型，未知时不推断") + album_type: Optional[str] = Field(default=None, description="标签声明的发行主类型，例如 Album、EP、Single") + secondary_types: list[str] = Field(default_factory=list, description="标签声明的 Compilation、Soundtrack 等副类型") org_string: Optional[str] = None apply_words: list[str] = Field(default_factory=list, description="资源解析时实际应用的自定义识别词") title: Optional[str] = None diff --git a/docs/mcp-api.md b/docs/mcp-api.md index e50528143..702681a6f 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -512,6 +512,8 @@ SSE 的 `candidate_items` 是站点原始返回数量，`match_counts` 记录身 音乐识别结果同时提供 `audio_format`、`audio_lossless`、`audio_quality`、`bit_depth`、`sample_rate`、`bitrate`、`audio_specs` 和 `audio_quality_score`。本地文件识别读取实际音频流参数，并使用 Chromaprint 的 `fpcalc` 在本地生成指纹后查询 AcoustID；音频文件本身不会上传。站点资源识别从标题和描述提取声明参数；码率、采样率的存储单位分别为 bps 和 Hz。 `MusicMeta.music_type` 保留已绑定主身份的实体类型；仅从名称推测时可为空。 +`MusicMeta.album_type` 和 `secondary_types` 保留标签声明的发行主副类型，例如 EP、Single、Compilation； +它们与 recording/album/artist 实体类型不同，不能根据曲目数量覆盖明确的标签类型。 `media_source`、`media_id` 与 `music_type` 必须一起解释，不能把专辑 ID 当作 Recording ID。 补充字段 `musicbrainz_release_id`、`musicbrainz_release_group_id`、`musicbrainz_release_track_id` 分别表示具体发行、发行组和发行中的曲目，不能替代主身份。`original_year` 与 `release_year` diff --git a/skills/moviepilot-api/api/music.md b/skills/moviepilot-api/api/music.md index 7f53a6c32..c5d43ce8a 100644 --- a/skills/moviepilot-api/api/music.md +++ b/skills/moviepilot-api/api/music.md @@ -37,6 +37,9 @@ Music recognition, exploration, album and artist navigation, and recognition-cac recording ID. The optional `musicbrainz_release_id`, `musicbrainz_release_group_id`, and `musicbrainz_release_track_id` refer to distinct MusicBrainz entities and never replace the primary identity. +- `MusicMeta.album_type` and `secondary_types` retain declared release types, + including EP, Single, and Compilation. These differ from the recording/album/ + artist entity type; track count alone must not override a declared release type. - `original_year` and `release_year` distinguish the original and current edition; `year` remains the display-compatible value. A release group's first release date does not prove the current edition. Preserve `total_discs` and diff --git a/tests/test_music_release_groups.py b/tests/test_music_release_groups.py new file mode 100644 index 000000000..286a3b05e --- /dev/null +++ b/tests/test_music_release_groups.py @@ -0,0 +1,223 @@ +"""按真实发行单元整理多碟、群星和全集，保持标签类型与文件查询范围。""" + +import shutil +from pathlib import Path +from unittest.mock import Mock + +import pytest +from mutagen.flac import FLAC +from mutagen.id3 import TPOS, TXXX + +from app.application.audio import AudioMetadataHelper +from app.chain.media import MediaChain +from app.chain.media.cache import AlbumDirectoryCache +from app.domain.meta.metamusic import MetaMusic +from app.schemas.types import MediaType +from tests.test_music_identity_evidence import _tag_audio +from tests.test_transfer_sync_extra_files import make_fileitem, make_transfer_chain + +RELEASE_A = "11111111-1111-4111-8111-111111111111" +RELEASE_B = "22222222-2222-4222-8222-222222222222" + + +def _audio(path: Path, *, artist="Artist", album="Record", release=None, release_type=None, disc=None, compilation=False): + """写入真实 FLAC 标签，省略总碟数以验证目录补缺。""" + path.parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(Path(__file__).parent / "fixtures/audio/silence.flac", path) + audio = FLAC(path) + audio["title"] = [path.stem] + audio["tracknumber"] = ["1"] + audio["date"] = ["2003"] + for key, value in (("artist", artist), ("album", album), ("discnumber", disc), + ("musicbrainz_albumid", release), ("releasetype", release_type)): + if value is not None: + audio[key] = [str(value)] + if compilation: + audio["compilation"] = ["1"] + audio.save() + return make_fileitem(str(path)) + + +def _prepare(items, mtype=MediaType.MUSIC): + """构建隔离整理批次，保留生产读取及分组函数。""" + owner = make_transfer_chain() + context = owner._prepare_music_batch_context([(item, False) for item in items], mtype) + return owner, context + + +def _resolve(owner, context, item): + """从批次读取真实文件的最终局部音乐上下文，不执行文件副作用。""" + return owner._resolve_music_batch_file_context( + batch_context=context, file_item=item, file_path=Path(item.path), + file_meta=AudioMetadataHelper.read(Path(item.path)), selected_tracks={}, fallback=None, + discard_shared_identity=True, multi_track_batch=True, release_regions=None, release_scripts=None, + ) + + +@pytest.mark.parametrize("mtype", [None, MediaType.MUSIC]) +def test_disc_directories_share_release_and_keep_disc_paths(tmp_path, mtype): + """自动与显式音乐目录的 CD1/CD2 属于同一发行，总碟数缺失也保留分碟。""" + items = [_audio(tmp_path / "Record" / f"CD{disc}" / "Song.flac", release=RELEASE_A) + for disc in (1, 2)] + owner, context = _prepare(items, mtype) + groups = [context.release_by_main_key[owner._get_file_key(item)] for item in items] + + assert groups[0] is groups[1] + assert groups[0].directory == tmp_path / "Record" + assert context.single_main_keys == set() + for disc, item in enumerate(items, 1): + meta, info = _resolve(owner, context, item) + assert meta.disc_number == info.disc_number == disc + assert meta.total_discs == info.total_discs == 2 + assert meta.field_sources["total_discs"] == "directory" + + +def test_compilation_groups_different_performers_without_replacing_them(tmp_path): + """明确合辑标签可补专辑艺人为群星，各曲真实艺人不得被多数票覆盖。""" + items = [_audio(tmp_path / "Compilation" / f"Song {index}.flac", artist=artist, compilation=True) + for index, artist in enumerate(("Artist A", "Artist B"), 1)] + owner, context = _prepare(items) + + for item, artist in zip(items, ("Artist A", "Artist B")): + meta, info = _resolve(owner, context, item) + assert meta.artists == info.artists == [artist] + assert meta.album_artist == info.album_artist == "Various Artists" + assert "Compilation" in info.secondary_types + + +@pytest.mark.parametrize("release_type,expected", [("EP", "EP"), ("single", "Single"), ("album / soundtrack", "Album")]) +def test_explicit_release_type_survives_local_batch(tmp_path, release_type, expected): + """多音轨目录不能把标签已明确的 EP 或 Single 强制改成 Album。""" + items = [_audio(tmp_path / "Record" / f"Song {index}.flac", release_type=release_type) for index in (1, 2)] + owner, context = _prepare(items) + + for item in items: + meta, info = _resolve(owner, context, item) + assert meta.album_type == info.album_type == expected + if "soundtrack" in release_type: + assert info.secondary_types == ["Soundtrack"] + + +def test_collection_child_albums_and_same_name_editions_stay_separate(tmp_path): + """全集子专辑和同目录不同 Release ID 都不能被合成一个发行单元。""" + items = [ + _audio(tmp_path / "Collection" / "2003 - First" / "Song.flac", album="First"), + _audio(tmp_path / "Collection" / "2004 - Second" / "Song.flac", album="Second"), + _audio(tmp_path / "Editions" / "A.flac", release=RELEASE_A), + _audio(tmp_path / "Editions" / "B.flac", release=RELEASE_B), + ] + owner, context = _prepare(items) + groups = [context.release_by_main_key[owner._get_file_key(item)] for item in items] + + assert len({tuple(item.path for item in group.files) for group in groups}) == 4 + + +def test_partial_tags_use_consistent_album_peers_without_online_lookup(tmp_path, monkeypatch): + """同发行有足够一致标签时，只有曲名的文件可用邻居标签补专辑而不重复联网。""" + items = [_audio(tmp_path / "Record" / f"Song {index}.flac") for index in (1, 2)] + items.append(_audio(tmp_path / "Record" / "Third Song.flac", artist=None, album=None)) + owner, context = _prepare(items) + online = Mock(side_effect=AssertionError("已确认的标签组不应重新联网")) + monkeypatch.setattr(owner, "_match_music_album_context", online) + + meta, info = _resolve(owner, context, items[-1]) + + assert meta.title == info.title == "Third Song" + assert meta.album == info.album == "Record" + assert meta.artists == ["Artist"] + assert meta.field_sources["album"] == "album_tags" + online.assert_not_called() + + +def test_tag_scan_is_shared_by_grouping_and_local_resolution(tmp_path, monkeypatch): + """同一批次的分组与快速整理复用纯标签快照，避免各分组重复读取。""" + items = [_audio(tmp_path / "Record" / f"Song {index}.flac") for index in (1, 2)] + reader = Mock(wraps=AudioMetadataHelper.read_tags) + monkeypatch.setattr(AudioMetadataHelper, "read_tags", reader) + owner, context = _prepare(items) + for item in items: + owner._resolve_music_batch_file_context( + batch_context=context, file_item=item, file_path=Path(item.path), + file_meta=MetaMusic(title=Path(item.path).stem), selected_tracks={}, fallback=None, + discard_shared_identity=True, multi_track_batch=True, release_regions=None, release_scripts=None, + ) + + assert reader.call_count == 2 + + +def test_album_cache_separates_file_scopes_in_one_directory(tmp_path, monkeypatch): + """同名同年的两组文件各自缓存结果，不能交替覆盖成重复查询或串用映射。""" + items = [_audio(tmp_path / f"Song {index}.flac") for index in range(4)] + chain = MediaChain() + monkeypatch.setattr(chain, "_album_dir_cache", AlbumDirectoryCache(16)) + source = Mock(match_music_album=Mock(return_value=None)) + monkeypatch.setattr("app.chain.media.album.MusicBrainzChain", Mock(return_value=source)) + first = [Path(item.path) for item in items[:2]] + second = [Path(item.path) for item in items[2:]] + for selected in (first, second, first): + chain.recognize_music_album_directory(tmp_path, file_paths=selected) + + assert source.match_music_album.call_count == 2 + assert [len(call.args[1]) for call in source.match_music_album.call_args_list] == [2, 2] + + +def test_container_header_overrides_misleading_extension(tmp_path): + """MP3 即使命名为 FLAC 也不能被标作无损音频。""" + path = tmp_path / "misnamed.flac" + shutil.copyfile(Path(__file__).parent / "fixtures/audio/silence.mp3", path) + + meta = AudioMetadataHelper.read(path) + + assert meta.audio_format == "MP3" + assert meta.audio_lossless is False + assert meta.field_sources["audio_format"] == "stream" + + +def test_tag_cache_never_crosses_storage_namespaces(tmp_path): + """云盘同名逻辑路径不能借用本地音频的标签或发行分组。""" + local = _audio(tmp_path / "Song.flac") + remote = local.model_copy(update={"storage": "alist"}) + owner, context = _prepare([local, remote]) + + assert len(context.tags_by_file) == 1 + assert context.release_by_main_key[owner._get_file_key(remote)].evidence is None + + +@pytest.mark.parametrize("different_disc,expected_groups", [(False, 2), (True, 1)]) +def test_duplicate_positions_split_encodings_but_mixed_discs_do_not(tmp_path, different_disc, expected_groups): + """重复曲序的两种编码独立匹配，混合编码但不同碟号的同一发行保留一组。""" + paths = [tmp_path / "Song.flac", tmp_path / "Song.mp3"] + for path in paths: + shutil.copyfile(Path(__file__).parent / f"fixtures/audio/silence{path.suffix}", path) + _tag_audio(path) + if different_disc: + from mutagen.mp3 import MP3 + + audio = MP3(paths[1]) + audio.tags.add(TPOS(encoding=3, text=["1/2"])) + audio.tags.add(TXXX(encoding=3, desc="MusicBrainz Release Track Id", text=["55555555-5555-4555-8555-555555555555"])) + audio.save() + items = [make_fileitem(str(path)) for path in paths] + owner, context = _prepare(items) + + groups = {tuple(file.path for file in context.release_by_main_key[owner._get_file_key(item)].files) for item in items} + assert len(groups) == expected_groups + + +def test_missing_album_with_conflicting_artist_does_not_gain_peer_tags(tmp_path): + """已知艺人冲突时，不能给未声明专辑的文件套上邻居的专辑标签。""" + items = [_audio(tmp_path / f"Song {index}.flac") for index in (1, 2)] + items.append(_audio(tmp_path / "Unrelated.flac", artist="Different Artist", album=None)) + owner, context = _prepare(items) + + assert owner._get_file_key(items[-1]) not in context.album_main_keys + assert context.release_by_main_key[owner._get_file_key(items[-1])].files == (items[-1],) + + +def test_sparse_tags_do_not_label_an_entire_mixed_inbox(tmp_path): + """普通收件目录的少数标签不能把大量未知音轨都绑定到同一专辑。""" + items = [_audio(tmp_path / "inbox" / f"Tagged {index}.flac") for index in (1, 2)] + items += [_audio(tmp_path / "inbox" / f"Unknown {index}.flac", artist=None, album=None) for index in range(6)] + owner, context = _prepare(items) + + assert all(owner._get_file_key(item) not in context.album_main_keys for item in items[2:])