diff --git a/app/application/audio.py b/app/application/audio.py index d17d9341a..d6cb7d14e 100644 --- a/app/application/audio.py +++ b/app/application/audio.py @@ -3,11 +3,11 @@ from typing import Any, Optional, Union from uuid import UUID from mutagen import File as MutagenFile -from mutagen.apev2 import APEBinaryValue +from mutagen.apev2 import APEBinaryValue, APETextValue from mutagen.flac import FLAC, Picture from mutagen.id3 import APIC, ID3, SYLT, USLT from mutagen.monkeysaudio import MonkeysAudio -from mutagen.mp4 import MP4, MP4Cover +from mutagen.mp4 import MP4, MP4Cover, MP4Tags from app.domain.context import MusicInfo, MusicLyrics from app.domain.meta.metamusic import MetaMusic @@ -15,6 +15,45 @@ from app.runtime.log import logger from app.schemas.types import MUSIC_ENTITY_RECORDING, MediaSource +def _read_mp4_tags(tags: MP4Tags) -> dict[str, list[str]]: + """读取标准 MP4 atom 与明确的文本 freeform，覆盖 EasyMP4 未注册的发行 ID。""" + atoms = {"title": "\xa9nam", "artist": "\xa9ART", "album": "\xa9alb", + "albumartist": "aART", "date": "\xa9day", "genre": "\xa9gen"} + values = {key: [str(value) for value in tags.get(atom, [])] for key, atom in atoms.items()} + for key, atom in (("tracknumber", "trkn"), ("discnumber", "disk")): + positions = tags.get(atom) or [] + if positions: + current, total = positions[0] + 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"): + 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 + ] + return values + + +def _mark_audio_evidence(meta: MetaMusic, info: 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", + ) + 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: + 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 + + class AudioMetadataHelper: """读取和写入音频标签，并转换为标准音乐元数据。""" @@ -48,7 +87,7 @@ class AudioMetadataHelper: def read_tags(cls, path: Path) -> Optional[MetaMusic]: """只读取本地音频标签和流参数，不使用文件名或目录补齐。""" try: - audio = MutagenFile(path, easy=True) + audio = MutagenFile(path, easy=False) except Exception as err: logger.warning(f"读取音频标签失败：{path} - {err}") return None @@ -70,16 +109,22 @@ class AudioMetadataHelper: tags, "musicbrainz_trackid", "musicbrainz_recordingid", + "musicbrainz_track_id", + "musicbrainz_recording_id", ) ) info = getattr(audio, "info", None) - return MetaMusic( + original_year = cls._year(cls._first_of(tags, "originaldate", "originalyear")) + release_year = cls._year(cls._first_of(tags, "date", "year")) + 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"), - year=cls._year(cls._first_of(tags, "date", "year", "originaldate")), + year=release_year or original_year, + original_year=original_year, + release_year=release_year, disc_number=disc_number, track_number=track_number, total_discs=total_discs, @@ -93,15 +138,25 @@ class AudioMetadataHelper: isrc=cls._first(tags, "isrc"), media_source=MediaSource.MusicBrainz if musicbrainz_id else None, media_id=musicbrainz_id, + music_type=MUSIC_ENTITY_RECORDING, + musicbrainz_release_id=cls._normalize_musicbrainz_id( + cls._first_of(tags, "musicbrainz_albumid", "musicbrainz_album_id")), + musicbrainz_release_group_id=cls._normalize_musicbrainz_id( + cls._first_of(tags, "musicbrainz_releasegroupid", "musicbrainz_release_group_id")), + musicbrainz_release_track_id=cls._normalize_musicbrainz_id( + cls._first_of(tags, "musicbrainz_releasetrackid", "musicbrainz_release_track_id")), ) + return _mark_audio_evidence(meta, info) @staticmethod def _readable_tags(tags: Any) -> Any: - """统一 WAV/DSF 的原生 ID3 与其它容器的 Easy 标签，不修改文件标签。 + """将原生 ID3/MP4 及 Vorbis/APEv2 映射到统一键，不修改音频。 - Mutagen 的 easy=True 只适用于提供 Easy 包装的容器；WAV/DSF 仍返回 - ID3 帧。Recording 的身份只取专用字段，不能误用 Release Track ID。 + 原生读取可保留 Easy 包装未注册的发行字段；Recording 的身份只取 + 专用字段，不能误用 Release Track ID。 """ + if isinstance(tags, MP4Tags): + return _read_mp4_tags(tags) if not isinstance(tags, ID3): return tags or {} fields = { @@ -286,7 +341,7 @@ class AudioMetadataHelper: """仅将 MusicBrainz 单曲身份写入 recording 标签，避免误写专辑 ID。""" if ( getattr(music, "media_source", None) == MediaSource.MusicBrainz - and getattr(music, "music_type", MUSIC_ENTITY_RECORDING) + and (getattr(music, "music_type", None) or MUSIC_ENTITY_RECORDING) == MUSIC_ENTITY_RECORDING ): media_id = getattr(music, "media_id", None) @@ -368,9 +423,9 @@ class AudioMetadataHelper: def _values(tags: Any, key: str) -> list[str]: """从 Mutagen Easy 标签中提取非空字符串列表。""" value = tags.get(key) if hasattr(tags, "get") else None - if value is None: + if value is None or isinstance(value, APEBinaryValue): return [] - if isinstance(value, (list, tuple)): + if isinstance(value, (list, tuple, APETextValue)): return [str(item).strip() for item in value if str(item).strip()] return [str(value).strip()] if str(value).strip() else [] diff --git a/app/chain/media/path.py b/app/chain/media/path.py index 1e13ed777..f6af6f962 100644 --- a/app/chain/media/path.py +++ b/app/chain/media/path.py @@ -119,6 +119,7 @@ def _musicbrainz_recording_meta(meta: MetaMusic, recording_id: str) -> MetaMusic identity_meta = MetaMusic.from_dict(meta.to_dict()) identity_meta.media_source = MediaSource.MusicBrainz identity_meta.media_id = recording_id + identity_meta.music_type = MUSIC_ENTITY_RECORDING return identity_meta @@ -184,6 +185,8 @@ def _merge_music_audio_quality(info: MusicInfo, meta: MetaMusic) -> MusicInfo: value = getattr(meta, key, None) if value is not None: setattr(info, key, value) + if key in meta.field_sources: + info.field_sources[key] = meta.field_sources[key] return info diff --git a/app/chain/media/recognition.py b/app/chain/media/recognition.py index 82c9886fb..f0e61cbeb 100644 --- a/app/chain/media/recognition.py +++ b/app/chain/media/recognition.py @@ -87,8 +87,8 @@ class MediaRecognitionOwner(_MediaOwnerBase): "media_id": module_kwargs.get("media_id"), "cache": cache, } - if "music_type" in module_kwargs: - recognize_kwargs["music_type"] = module_kwargs["music_type"] + if "music_type" in module_kwargs or (isinstance(meta, MetaMusic) and meta.music_type): + recognize_kwargs["music_type"] = module_kwargs.get("music_type") or getattr(meta, "music_type", None) return _NativeRecognitionPlan( action=_NativeRecognitionAction.MUSIC, kwargs=recognize_kwargs, @@ -101,7 +101,7 @@ class MediaRecognitionOwner(_MediaOwnerBase): "media_source": self._music_primary_source, "meta": meta, "cache": cache, - "music_type": module_kwargs.get("music_type") or MUSIC_ENTITY_RECORDING, + "music_type": module_kwargs.get("music_type") or meta.music_type or MUSIC_ENTITY_RECORDING, }, ) return _NativeRecognitionPlan( diff --git a/app/chain/transfer/filter.py b/app/chain/transfer/filter.py index 161549897..cd4e5e253 100644 --- a/app/chain/transfer/filter.py +++ b/app/chain/transfer/filter.py @@ -246,8 +246,10 @@ class _MusicFileFilterBase(_TransferOwnerBase): cls, file_meta: MetaMusic, info: MusicInfo, + *, + selected_release: bool = False, ) -> tuple[MetaMusic, MusicInfo]: - """以已选发行版的曲目身份更新文件元数据，同时保留本地音频参数。""" + """合并曲目身份与实际音频；用户显式选发行时替换旧发行标识，自动识别只补缺。""" merged_meta = deepcopy(file_meta) # 保留本地音频的实际技术参数，仅回填身份和名称字段 if info.title: @@ -260,18 +262,42 @@ class _MusicFileFilterBase(_TransferOwnerBase): merged_meta.album_artist = info.album_artist if info.year: merged_meta.year = cast(Any, info.year) + merged_meta.field_sources.update({ + key: "manual" if selected_release else info.field_sources.get(key, "remote") + for key in ("title", "artists", "album", "album_artist", "year") + if getattr(info, key) + }) + for key in ("musicbrainz_release_id", "musicbrainz_release_group_id", "musicbrainz_release_track_id", + "original_year", "release_year"): + value = getattr(info, key) + if selected_release or (not getattr(merged_meta, key) and value): + setattr(merged_meta, key, value) + if value: + merged_meta.field_sources[key] = "manual" if selected_release else info.field_sources.get(key, "remote") + else: + merged_meta.field_sources.pop(key, None) # 曲序和碟号描述的是当前物理文件在本地发行目录中的位置。MusicBrainz # Recording 可能同时属于多个发行版，候选 release 的曲序并不一定对应 # 用户手里的这一版；只有本地没有这些字段时才用远端值补齐，避免同一 # 专辑的多首歌被错误写到相同目标文件名并相互覆盖。 if not merged_meta.disc_number and info.disc_number: merged_meta.disc_number = info.disc_number + merged_meta.field_sources["disc_number"] = info.field_sources.get("disc_number", "remote") if not merged_meta.track_number and info.track_number: merged_meta.track_number = info.track_number + merged_meta.field_sources["track_number"] = info.field_sources.get("track_number", "remote") if not merged_meta.total_tracks and info.total_tracks: merged_meta.total_tracks = info.total_tracks + merged_meta.field_sources["total_tracks"] = info.field_sources.get("total_tracks", "remote") + if not merged_meta.total_discs and info.total_discs: + merged_meta.total_discs = info.total_discs + merged_meta.field_sources["total_discs"] = info.field_sources.get("total_discs", "remote") merged_meta.media_source = info.media_source merged_meta.media_id = info.media_id + merged_meta.music_type = info.music_type + if info.media_id: + source = "manual" if selected_release else info.field_sources.get("media_id", "remote") + merged_meta.field_sources.update(media_id=source, media_source=source) merged_info = cls._music_info_from_meta(merged_meta) # 补齐曲目级远端信息，供后续刮削和展示使用 merged_info.music_type = info.music_type @@ -375,7 +401,7 @@ class FileFilterMixin(_TransferOwnerBase): if selected and isinstance(file_meta, MetaMusic): return cast( tuple[Any, Optional[Union[MediaInfo, MusicInfo]]], - self._merge_music_track_context(file_meta, selected), + self._merge_music_track_context(file_meta, selected, selected_release=True), ) return file_meta, fallback @@ -494,8 +520,23 @@ class FileFilterMixin(_TransferOwnerBase): ): if getattr(file_tags, field_name, None): setattr(file_meta, field_name, getattr(file_tags, field_name)) + for field_name in ("musicbrainz_release_id", "musicbrainz_release_group_id", "musicbrainz_release_track_id", + "original_year", "release_year"): + if getattr(file_tags, field_name): + setattr(file_meta, field_name, getattr(file_tags, field_name)) + file_meta.field_sources.update({ + key: source for key, source in file_tags.field_sources.items() + if getattr(file_meta, key, None) == getattr(file_tags, key, None) + }) file_meta.media_source = saved_info.media_source or saved_meta.media_source file_meta.media_id = saved_info.media_id or saved_meta.media_id + file_meta.music_type = saved_info.music_type + for key in ("media_id", "media_source"): + source = saved_info.field_sources.get(key) or saved_meta.field_sources.get(key) + if source: + file_meta.field_sources[key] = source + else: + file_meta.field_sources.pop(key, None) file_info = cls._music_info_from_meta(file_meta) file_info.media_source = saved_info.media_source diff --git a/app/chain/transfer/music.py b/app/chain/transfer/music.py index d4c6383a2..cf73dc08e 100644 --- a/app/chain/transfer/music.py +++ b/app/chain/transfer/music.py @@ -65,18 +65,23 @@ def restore_music_resource_meta( artists = [artist for artist in resource.artists if music_text_key(artist) not in collective] if not meta.artists and artists: meta.artists = list(artists) + meta.field_sources["artists"] = "torrent" if not meta.album_artist and len(resource.artists) == 1: meta.album_artist = resource.artists[0] + meta.field_sources["album_artist"] = "torrent" if direct_release and not collection: album = resource.album if root != path and not resource.track_number: album = album or resource.title if not meta.album and album: meta.album = album + meta.field_sources["album"] = "torrent" if not meta.year and resource.year: meta.year = resource.year + meta.field_sources["year"] = "torrent" if not meta.version and resource.version: meta.version = resource.version + meta.field_sources["version"] = "torrent" return meta @@ -174,10 +179,13 @@ def _apply_music_directory_evidence( merged = deepcopy(meta) if not merged.artists and evidence.artists: merged.artists = list(evidence.artists) + merged.field_sources["artists"] = "album_tags" if not merged.album_artist and evidence.album_artist: merged.album_artist = evidence.album_artist + merged.field_sources["album_artist"] = "album_tags" if not merged.album and evidence.album: merged.album = evidence.album + merged.field_sources["album"] = "album_tags" return merged @@ -189,6 +197,7 @@ def _apply_music_directory_year(meta: MetaMusic, file_path: Path) -> MetaMusic: continue merged = deepcopy(meta) merged.year = cast(Any, int(match.group(1))) + merged.field_sources["year"] = "directory" return merged return meta @@ -344,9 +353,11 @@ def resolve_music_batch_file_context( 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 ( diff --git a/app/domain/context.py b/app/domain/context.py index 9544aa54a..96f59eff3 100644 --- a/app/domain/context.py +++ b/app/domain/context.py @@ -565,16 +565,23 @@ class MusicInfo: album_artist: str | None = None # 所属专辑标准 ID（MusicBrainz Release Group） album_id: str | None = None + # 发行、发行组和发行曲目身份独立于 Recording 主身份，不能相互代用。 + musicbrainz_release_id: str | None = field(default=None, kw_only=True) + musicbrainz_release_group_id: str | None = field(default=None, kw_only=True) + musicbrainz_release_track_id: str | None = field(default=None, kw_only=True) # 专辑主类型：Album、EP、Single 等 album_type: str | None = None # 专辑副类型：Live、Compilation、Soundtrack 等 secondary_types: list[str] = field(default_factory=list) year: int | None = None + original_year: int | None = field(default=None, kw_only=True) + release_year: int | None = field(default=None, kw_only=True) release_date: str | None = None release_status: str | None = None disc_number: int | None = None track_number: int | None = None total_tracks: int | None = None + total_discs: int | None = field(default=None, kw_only=True) duration: int | None = None isrc: str | None = None cover_url: str | None = None @@ -605,6 +612,7 @@ class MusicInfo: detail_link: str | None = None listen_count: int | None = None raw_data: dict[str, Any] = field(default_factory=dict) + field_sources: dict[str, str] = field(default_factory=dict, kw_only=True) # 内部标记：是否命中本地识别缓存，不参与序列化；与 MediaInfo 保持一致， # 显式声明以保留 getattr(..., False) 的默认值语义（__getattr__ 兜底会覆盖它） recognize_cache_hit = False @@ -614,9 +622,12 @@ class MusicInfo: self.__dict__.update(state) for name in ("title_aliases", "album_aliases", "artist_aliases"): self.__dict__.setdefault(name, []) + self.__dict__.setdefault("field_sources", {}) def __post_init__(self) -> None: """规范化媒体身份，并兼容拆分旧音乐分类字段。""" + self.field_sources = dict(self.field_sources or {}) + self.year = self.year if self.year is not None else self.release_year or self.original_year self.media_source, self.media_id = resolve_media_identity(media=self) classification = _classification_result(self.classification) library_category, metadata_category = _resolve_music_categories( @@ -787,6 +798,7 @@ class MusicInfo: values.get("audio_format"), values.get("audio_lossless") ) values["raw_data"] = dict(values.get("raw_data") or {}) + values["field_sources"] = dict(values.get("field_sources") or {}) if "library_category" in data or "metadata_category" in data: library_value = ( data.get("library_category") @@ -812,9 +824,12 @@ class MusicInfo: values["metadata_category"] = metadata_category for key in ( "year", + "original_year", + "release_year", "disc_number", "track_number", "total_tracks", + "total_discs", "duration", "listen_count", "bit_depth", @@ -830,6 +845,13 @@ class MusicInfo: return cls( media_source=normalize_media_source(meta.media_source), media_id=meta.media_id, + music_type=meta.music_type or MUSIC_ENTITY_RECORDING, + 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, + original_year=meta.original_year, + release_year=meta.release_year, + field_sources=dict(meta.field_sources), title=meta.title, artists=list(meta.artists), album=meta.album, @@ -838,6 +860,7 @@ class MusicInfo: disc_number=meta.disc_number, track_number=meta.track_number, total_tracks=meta.total_tracks, + total_discs=meta.total_discs, duration=meta.duration, isrc=meta.isrc, version=meta.version, @@ -892,6 +915,10 @@ class MusicAlbumInfo: music_type: str = field(default=MUSIC_ENTITY_ALBUM, init=False) media_source: MediaSource | None = None media_id: str | None = None + musicbrainz_release_id: str | None = field(default=None, kw_only=True) + musicbrainz_release_group_id: str | None = field(default=None, kw_only=True) + original_year: int | None = field(default=None, kw_only=True) + total_discs: int | None = field(default=None, kw_only=True) title: str | None = None artists: list[str] = field(default_factory=list) artist_ids: list[str] = field(default_factory=list) @@ -921,15 +948,18 @@ class MusicAlbumInfo: # 同一专辑下的其它发行版本 releases: list[MusicRelease] = field(default_factory=list) raw_data: dict[str, Any] = field(default_factory=dict) + field_sources: dict[str, str] = field(default_factory=dict, kw_only=True) def __setstate__(self, state: dict[str, Any]) -> None: """恢复旧版专辑缓存时补齐别名列表，保留既有媒体身份及曲目数据。""" self.__dict__.update(state) for name in ("title_aliases", "artist_aliases"): self.__dict__.setdefault(name, []) + self.__dict__.setdefault("field_sources", {}) def __post_init__(self) -> None: """规范化媒体身份，并补全专辑描述分类和分类结果。""" + self.field_sources = dict(self.field_sources or {}) self.media_source, self.media_id = resolve_media_identity(media=self) self.library_category = str(self.library_category or "").strip() self.metadata_category = str(self.metadata_category or "").strip() or ( @@ -1056,6 +1086,9 @@ class MusicAlbumInfo: for item in data.get("releases") or [] ] values["raw_data"] = dict(values.get("raw_data") or {}) + values["field_sources"] = dict(values.get("field_sources") or {}) + values["original_year"] = _music_optional_int(values.get("original_year")) + values["total_discs"] = _music_optional_int(values.get("total_discs")) if "library_category" in data or "metadata_category" in data: library_value = ( data.get("library_category") @@ -1084,6 +1117,12 @@ class MusicAlbumInfo: return MusicInfo( media_source=self.media_source, media_id=self.media_id, + musicbrainz_release_id=self.musicbrainz_release_id, + musicbrainz_release_group_id=self.musicbrainz_release_group_id, + original_year=self.original_year, + release_year=self.year if self.musicbrainz_release_id else None, + total_discs=self.total_discs, + field_sources=dict(self.field_sources), music_type=MUSIC_ENTITY_ALBUM, title=self.title, artists=list(self.artists), diff --git a/app/domain/meta/metamusic.py b/app/domain/meta/metamusic.py index fd35a49b3..58182d7a6 100644 --- a/app/domain/meta/metamusic.py +++ b/app/domain/meta/metamusic.py @@ -11,7 +11,7 @@ from Pinyin2Hanzi import DefaultHmmParams, is_pinyin from app.domain.meta.metabase import MetaBase from app.domain.meta.runtime import get_metainfo_accelerator from app.schemas.media import resolve_media_identity -from app.schemas.types import MediaSource, MediaType +from app.schemas.types import MUSIC_ENTITY_ALBUM, MediaSource, MediaType _AUDIO_FORMAT_PATTERN = re.compile( r"(?DSD(?:64|128|256|512)?|DSF|DFF|SACD|FLAC|ALAC|APE|WAV|WAVE|AIFF?|PCM|" @@ -727,6 +727,14 @@ class MetaMusic(MetaBase): media_source: Optional[MediaSource] = None, media_id: Optional[str] = None, parse_title: bool = False, + *, + musicbrainz_release_id: Optional[str] = None, + musicbrainz_release_group_id: Optional[str] = None, + musicbrainz_release_track_id: Optional[str] = None, + original_year: Optional[int] = None, + release_year: Optional[int] = None, + field_sources: Optional[dict[str, str]] = None, + music_type: Optional[str] = None, ): """初始化音乐标题、标签、音频规格和统一媒体身份。""" # 音乐无季集概念，仅复用 MetaBase 的基础字段初始化，不触发副标题季集识别 @@ -737,7 +745,14 @@ class MetaMusic(MetaBase): self.artists = list(artists) if artists else [] self.album = album self.album_artist = album_artist - self.year = year + self.year = year if year is not None else release_year or original_year + self.original_year = original_year + self.release_year = release_year + self.musicbrainz_release_id = musicbrainz_release_id + self.musicbrainz_release_group_id = musicbrainz_release_group_id + self.musicbrainz_release_track_id = musicbrainz_release_track_id + self.field_sources = dict(field_sources or {}) + self.music_type = music_type self.disc_number = disc_number self.track_number = track_number self.total_discs = total_discs @@ -758,6 +773,15 @@ class MetaMusic(MetaBase): # 种子/文件名字符串场景：解析艺术家、曲名、年份并补充音质参数 self.apply_title(self.title or org_string or "") + def __setstate__(self, state: dict[str, Any]) -> None: + """恢复旧版解析缓存时补齐发行事实和字段来源，避免丢失旧字段或共享字典。""" + self.__dict__.update(state) + for key in ("musicbrainz_release_id", "musicbrainz_release_group_id", "musicbrainz_release_track_id", + "original_year", "release_year"): + self.__dict__.setdefault(key, None) + self.field_sources = dict(state.get("field_sources") or {}) + self.music_type = state.get("music_type") + @classmethod def parse_query(cls, query: str) -> "MetaMusic": """把用户输入或资源标题解析为音乐元数据。""" @@ -829,6 +853,11 @@ class MetaMusic(MetaBase): meta.title = native_title if not meta.version: meta.version = cls._resource_version(title, subtitle) + meta.field_sources.update({ + key: "torrent" for key in ("title", "artists", "album", "year", "version", + "audio_format", "audio_lossless", "bit_depth", "sample_rate", "bitrate") + if getattr(meta, key) not in (None, "", []) + }) return meta @classmethod @@ -976,6 +1005,13 @@ class MetaMusic(MetaBase): isrc=info.isrc, media_source=info.media_source, media_id=info.media_id, + music_type=getattr(info, "music_type", 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), + original_year=getattr(info, "original_year", None), + release_year=getattr(info, "release_year", None), + field_sources=getattr(info, "field_sources", None), ) @classmethod @@ -1001,6 +1037,7 @@ class MetaMusic(MetaBase): artist = majority_artist if majority_artist and artist_votes[majority_artist] >= threshold else None return cls( org_string=directory_name, + music_type=MUSIC_ENTITY_ALBUM, title=album or directory.get("album") or directory_name, album=album or directory.get("album"), artists=[artist or directory.get("artist")] @@ -1051,12 +1088,16 @@ class MetaMusic(MetaBase): self.audio_format, self.audio_lossless, self.bit_depth, self.sample_rate, self.bitrate ) - def apply_audio_quality(self, value: Any, overwrite: bool = False) -> None: + def apply_audio_quality( + self, value: Any, overwrite: bool = False, evidence_source: Optional[str] = None, + ) -> None: """从资源文本补充音质参数，默认保留文件标签读取到的实际值。""" parsed = parse_audio_quality(value) for key, parsed_value in parsed.items(): if parsed_value is not None and (overwrite or getattr(self, key, None) is None): setattr(self, key, parsed_value) + if evidence_source: + self.field_sources[key] = evidence_source self.audio_format = normalize_audio_format(self.audio_format) self.audio_lossless = infer_audio_lossless(self.audio_format, self.audio_lossless) @@ -1740,7 +1781,7 @@ class MetaMusic(MetaBase): """ file_path = Path(path) stem = file_path.stem - title_from_name = not self.title or self.title == stem + title_from_name = not self.title or (self.title == stem and self.field_sources.get("title") != "tag") # 曲序/碟号前缀是文件路径的强结构，先于通用艺术家-标题模式剥离， # 避免「01 - One More Time」把 01 误判为艺术家。 @@ -1758,6 +1799,7 @@ class MetaMusic(MetaBase): filename_meta.disc_number = disc_number if title_from_name and filename_meta.title: self.title = filename_meta.title + self.field_sources["title"] = "filename" for field_name in ( "artists", "album", @@ -1778,6 +1820,7 @@ class MetaMusic(MetaBase): parsed_value = getattr(filename_meta, field_name, None) if current_value in (None, "", []) and parsed_value not in (None, "", []): setattr(self, field_name, parsed_value) + self.field_sources[field_name] = "filename" # 目录结构：父目录可能是碟片目录，专辑目录再往上一级 parent = file_path.parent @@ -1786,6 +1829,7 @@ class MetaMusic(MetaBase): if parent_disc is not None: if self.disc_number is None: self.disc_number = parent_disc + self.field_sources["disc_number"] = "directory" album_dir = parent.parent dir_info = self.parse_album_dir(album_dir.name) if dir_info: @@ -1793,18 +1837,23 @@ class MetaMusic(MetaBase): if dir_info.get("artist") or dir_info.get("year"): if not self.album and dir_info.get("album"): self.album = dir_info["album"] + self.field_sources["album"] = "directory" if not self.artists and dir_info.get("artist"): self.artists = [dir_info["artist"]] + self.field_sources["artists"] = "directory" if not self.album_artist and dir_info.get("artist"): self.album_artist = dir_info["artist"] + self.field_sources["album_artist"] = "directory" if self.year is None and dir_info.get("year"): self.year = dir_info["year"] + self.field_sources["year"] = "directory" if title_from_name and not self.version and dir_info.get("version"): # 目录版本可补无标签曲目，真实歌曲标签不能被整张专辑的版本推测覆盖。 self.version = dir_info["version"] + self.field_sources["version"] = "directory" # 目录名里的格式、位深、采样率可补齐本地标签未声明的音质参数 if dir_info.get("quality_text"): - self.apply_audio_quality(dir_info["quality_text"]) + self.apply_audio_quality(dir_info["quality_text"], evidence_source="directory") return self @property @@ -1856,6 +1905,13 @@ class MetaMusic(MetaBase): "isrc": self.isrc, "media_source": self.media_source, "media_id": self.media_id, + "musicbrainz_release_id": self.musicbrainz_release_id, + "musicbrainz_release_group_id": self.musicbrainz_release_group_id, + "musicbrainz_release_track_id": self.musicbrainz_release_track_id, + "original_year": self.original_year, + "release_year": self.release_year, + "field_sources": dict(self.field_sources), + "music_type": self.music_type, } @classmethod @@ -1885,6 +1941,13 @@ class MetaMusic(MetaBase): isrc=data.get("isrc"), media_source=data.get("media_source"), media_id=data.get("media_id"), + musicbrainz_release_id=data.get("musicbrainz_release_id"), + musicbrainz_release_group_id=data.get("musicbrainz_release_group_id"), + musicbrainz_release_track_id=data.get("musicbrainz_release_track_id"), + original_year=_optional_int(data.get("original_year")), + release_year=_optional_int(data.get("release_year")), + field_sources=data.get("field_sources"), + music_type=data.get("music_type"), ) meta.apply_words = data.get("apply_words") return meta diff --git a/app/modules/musicbrainz/__init__.py b/app/modules/musicbrainz/__init__.py index 31ec551ae..8e461614f 100644 --- a/app/modules/musicbrainz/__init__.py +++ b/app/modules/musicbrainz/__init__.py @@ -4,7 +4,7 @@ import threading import time from dataclasses import dataclass from difflib import SequenceMatcher -from typing import Any, Iterable, Optional, Tuple, Union +from typing import Any, Iterable, Optional, Tuple, TypeVar, Union from app.adapters.network.http import AsyncRequestUtils, RequestUtils from app.domain.classification.evaluator import read_fact @@ -55,6 +55,20 @@ from app.schemas.types import ( MusicEntityType, ) +_MusicEvidenceModel = TypeVar("_MusicEvidenceModel", MusicInfo, MusicAlbumInfo) + + +def _remote_music_evidence(info: _MusicEvidenceModel) -> _MusicEvidenceModel: + """只为响应中实际存在的标准字段标记远端来源，不给空值制造证据。""" + fields = ("title", "artists", "album", "album_artist", "year", "original_year", "release_year", + "release_date", "disc_number", "track_number", "total_tracks", "total_discs", + "version", "media_id", "musicbrainz_release_id", "musicbrainz_release_group_id", + "musicbrainz_release_track_id") + info.field_sources.update({ + key: "remote" for key in fields if getattr(info, key, None) not in (None, "", []) + }) + return info + @dataclass(frozen=True, slots=True) class _MusicBrainzRecognitionPlan: @@ -1162,8 +1176,12 @@ class MusicBrainzModule(_ModuleBase): artists, artist_ids = cls._artist_credits(detail.get("artist-credit")) album = MusicAlbumInfo( media_source=cls._source, - # 优先使用 Release Group ID，与专辑详情和封面入口保持一致 - media_id=str(group_id or release_id), + # Release 与 Release Group 是不同实体，缺少 Group 时不能用 Release ID 补位。 + media_id=str(group_id) if group_id else None, + musicbrainz_release_id=str(release_id), + musicbrainz_release_group_id=str(group_id) if group_id else None, + original_year=cls._year(release_group.get("first-release-date")), + total_discs=len(detail.get("media") or []) or None, title=str(title), artists=artists, artist_ids=artist_ids, @@ -1185,7 +1203,7 @@ class MusicBrainzModule(_ModuleBase): for track in medium.get("tracks") or [] if (info := cls._track_to_info(album, medium, track)) ] - return album + return _remote_music_evidence(album) def recognize_media( self, @@ -2014,7 +2032,7 @@ class MusicBrainzModule(_ModuleBase): if (value := cls._stripped(item)) ] category_parts = [cls._stripped(release_group.get("primary-type")), *secondary_types] - return MusicInfo( + return _remote_music_evidence(MusicInfo( media_source=cls._source, media_id=str(media_id), title=str(title), @@ -2023,6 +2041,10 @@ class MusicBrainzModule(_ModuleBase): album=album, album_artist=" / ".join(album_artists) if album_artists else None, album_id=str(release_group["id"]) if release_group.get("id") else None, + musicbrainz_release_id=str(release["id"]) if release.get("id") else None, + musicbrainz_release_group_id=str(release_group["id"]) if release_group.get("id") else None, + original_year=cls._year(recording.get("first-release-date") or release_group.get("first-release-date")), + release_year=cls._year(release.get("date")), album_type=cls._stripped(release_group.get("primary-type")), secondary_types=secondary_types, year=cls._year(release_date), @@ -2040,7 +2062,7 @@ class MusicBrainzModule(_ModuleBase): artist_aliases=cls._credit_aliases(recording.get("artist-credit")), detail_link=f"{cls._detail_url}/{media_id}", raw_data=recording, - ) + )) @classmethod def _release_group_to_album(cls, release_group: dict[str, Any]) -> Optional[MusicAlbumInfo]: @@ -2051,9 +2073,11 @@ class MusicBrainzModule(_ModuleBase): return None artists, artist_ids = cls._artist_credits(release_group.get("artist-credit")) rating = release_group.get("rating") or {} - return MusicAlbumInfo( + return _remote_music_evidence(MusicAlbumInfo( media_source=cls._source, media_id=str(media_id), + musicbrainz_release_group_id=str(media_id), + original_year=cls._year(release_group.get("first-release-date")), title=str(title), artists=artists, artist_ids=artist_ids, @@ -2070,7 +2094,7 @@ class MusicBrainzModule(_ModuleBase): rating_votes=rating.get("votes-count"), detail_link=f"{cls._album_detail_url}/{media_id}", raw_data=release_group, - ) + )) @classmethod def _release_variants(cls, releases: list[dict[str, Any]]) -> list[MusicRelease]: @@ -2202,12 +2226,15 @@ class MusicBrainzModule(_ModuleBase): album.title = title if release_date := cls._stripped(payload.get("date")): album.release_date = release_date + album.musicbrainz_release_id = cls._stripped(payload.get("id")) + album.total_discs = len(payload.get("media") or []) or None album.raw_data = { **(album.raw_data or {}), "release_id": cls._stripped(payload.get("id")), "release_country": cls._stripped(payload.get("country")), "release_script": cls._release_script(payload) or None, } + _remote_music_evidence(album) @classmethod def _project_album_tracks( @@ -2240,7 +2267,7 @@ class MusicBrainzModule(_ModuleBase): artists, artist_ids = cls._artist_credits( track.get("artist-credit") or recording.get("artist-credit") ) - return MusicInfo( + return _remote_music_evidence(MusicInfo( media_source=cls._source, media_id=str(media_id), title=str(title), @@ -2249,9 +2276,15 @@ class MusicBrainzModule(_ModuleBase): album=album.title, album_artist=album.artist or None, album_id=album.media_id, + musicbrainz_release_id=album.musicbrainz_release_id, + musicbrainz_release_group_id=album.musicbrainz_release_group_id or album.media_id, + musicbrainz_release_track_id=cls._stripped(track.get("id")), + original_year=cls._year(recording.get("first-release-date")) or album.original_year, + release_year=album.year if album.musicbrainz_release_id else None, + total_discs=album.total_discs, album_type=album.album_type, year=album.year, - release_date=recording.get("first-release-date") or album.release_date, + release_date=album.release_date, disc_number=cls._optional_int(medium.get("position")), track_number=cls._optional_int(track.get("position")), total_tracks=cls._optional_int(medium.get("track-count")), @@ -2266,7 +2299,7 @@ class MusicBrainzModule(_ModuleBase): release_status=album.release_status, names=[str(title)], detail_link=f"{cls._detail_url}/{media_id}", - ) + )) @classmethod def _select_release(cls, releases: list[dict[str, Any]]) -> dict[str, Any]: diff --git a/app/schemas/music.py b/app/schemas/music.py index 11728baa5..2b5a72cae 100644 --- a/app/schemas/music.py +++ b/app/schemas/music.py @@ -12,6 +12,7 @@ class MusicMeta(OptionalMediaIdentityMixin, BaseModel): """音乐名称及音频文件解析结果。""" type: Literal["音乐"] = "音乐" + music_type: Optional[MusicEntityType] = Field(default=None, description="已知主身份的实体类型，未知时不推断") org_string: Optional[str] = None apply_words: list[str] = Field(default_factory=list, description="资源解析时实际应用的自定义识别词") title: Optional[str] = None @@ -37,6 +38,12 @@ class MusicMeta(OptionalMediaIdentityMixin, BaseModel): isrc: Optional[str] = None media_source: Optional[MediaSource] = None media_id: Optional[str] = None + musicbrainz_release_id: Optional[str] = Field(default=None, description="具体发行版 MBID") + musicbrainz_release_group_id: Optional[str] = Field(default=None, description="发行组 MBID") + musicbrainz_release_track_id: Optional[str] = Field(default=None, description="发行版内曲目 MBID，不是 Recording ID") + original_year: Optional[int] = Field(default=None, description="原始发行年份") + release_year: Optional[int] = Field(default=None, description="当前发行版年份") + field_sources: dict[str, str] = Field(default_factory=dict, description="字段的 tag/album_tags/stream/filename/directory/torrent/remote/manual 证据来源") class MusicInfo(OptionalMediaIdentityMixin, BaseModel): @@ -54,13 +61,19 @@ class MusicInfo(OptionalMediaIdentityMixin, BaseModel): album: Optional[str] = None album_artist: Optional[str] = None album_id: Optional[str] = None + musicbrainz_release_id: Optional[str] = None + musicbrainz_release_group_id: Optional[str] = None + musicbrainz_release_track_id: Optional[str] = None album_type: Optional[str] = None secondary_types: list[str] = Field(default_factory=list) year: Optional[int] = None + original_year: Optional[int] = None + release_year: Optional[int] = None release_date: Optional[str] = None disc_number: Optional[int] = None track_number: Optional[int] = None total_tracks: Optional[int] = None + total_discs: Optional[int] = None duration: Optional[int] = None isrc: Optional[str] = None cover_url: Optional[str] = None @@ -90,6 +103,7 @@ class MusicInfo(OptionalMediaIdentityMixin, BaseModel): detail_link: Optional[str] = None listen_count: Optional[int] = None raw_data: dict[str, JsonData] = Field(default_factory=dict) + field_sources: dict[str, str] = Field(default_factory=dict) title_year: Optional[str] = None poster_path: Optional[str] = None backdrop_path: Optional[str] = None @@ -186,6 +200,11 @@ class MusicAlbumInfo(OptionalMediaIdentityMixin, BaseModel): music_type: Literal["album"] = "album" media_source: Optional[MediaSource] = None media_id: Optional[str] = None + musicbrainz_release_id: Optional[str] = None + musicbrainz_release_group_id: Optional[str] = None + original_year: Optional[int] = None + total_discs: Optional[int] = None + field_sources: dict[str, str] = Field(default_factory=dict) title: Optional[str] = None artists: list[str] = Field(default_factory=list) artist: Optional[str] = None diff --git a/docs/mcp-api.md b/docs/mcp-api.md index 3d3001331..e50528143 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -511,6 +511,15 @@ 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` 保留已绑定主身份的实体类型；仅从名称推测时可为空。 +`media_source`、`media_id` 与 `music_type` 必须一起解释，不能把专辑 ID 当作 Recording ID。 +补充字段 `musicbrainz_release_id`、`musicbrainz_release_group_id`、`musicbrainz_release_track_id` +分别表示具体发行、发行组和发行中的曲目，不能替代主身份。`original_year` 与 `release_year` +分别保留原始发行年份和当前发行年份，`year` 继续用于兼容展示；发行组尚未选定具体 Release 时 +不能凭首发年份填充 `release_year`。`total_discs` 保留真实碟数。 +`field_sources` 按字段记录 `tag`、`album_tags`、`stream`、`filename`、`directory`、`torrent`、 +`remote` 或 `manual` 等来源，缺失表示来源未记录；该说明不等于远端身份已验证，也不用于绕过整理准入。 + | 方法 | 路径 | 说明 | | :--- | :--- | :--- | | GET | `/api/v1/media/search` | 当 `type=music` 或指定音乐 `media_source` 时按歌曲或专辑关键词搜索音乐元数据；艺术家统一由 `type=person` 搜索，支持 TMDB 与 MusicBrainz 来源。参数：`title`、`type`、`count`、可重复的 `media_source` 枚举，以及可选的 `music_type` 实体过滤 | diff --git a/skills/moviepilot-api/api/music.md b/skills/moviepilot-api/api/music.md index cfd187469..7f53a6c32 100644 --- a/skills/moviepilot-api/api/music.md +++ b/skills/moviepilot-api/api/music.md @@ -32,6 +32,17 @@ Music recognition, exploration, album and artist navigation, and recognition-cac rules. Album lookup uses the album credit without replacing the track's performer. Conflicting explicit recording dates are version mismatches; missing dates alone do not reject a candidate. +- Preserve `MusicMeta.music_type` when metadata carries a bound identity. Read + `media_source`, `media_id`, and `music_type` together; an album ID is not a + 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. +- `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 + `field_sources` in round trips. Sources such as `tag`, `album_tags`, `stream`, + `filename`, `directory`, `torrent`, `remote`, and `manual` explain field origins; + they do not independently prove that an online identity was verified. ## Operations diff --git a/tests/fixtures/audio/README.md b/tests/fixtures/audio/README.md index f592833a0..dd7d7d4f3 100644 --- a/tests/fixtures/audio/README.md +++ b/tests/fixtures/audio/README.md @@ -6,3 +6,8 @@ 生成过程：标准库 `wave` 写入三秒零值 PCM，通过 ffmpeg 的 `flac` 编码器转换，再使用 `mutagen.flac.FLAC.clear()` 和 `save(padding=lambda _: 0)` 清除标签及填充。 + +`silence.m4a`（ALAC）与 `silence.mp3`（32 kbps MP3）由同一静音 FLAC 本地转换， +移除了输入元数据，用于验证原生 MP4 atom/freeform 与 MP3 ID3 字段。 +测试直接通过 Mutagen 写入独立 MBID 和日期，CI 不依赖编码器。 +M4A 的 ORIGINALDATE 是可选自定义 freeform 兼容样本，不声明它是 MP4 标准日期字段。 diff --git a/tests/fixtures/audio/silence.m4a b/tests/fixtures/audio/silence.m4a new file mode 100644 index 000000000..b61f5716c Binary files /dev/null and b/tests/fixtures/audio/silence.m4a differ diff --git a/tests/fixtures/audio/silence.mp3 b/tests/fixtures/audio/silence.mp3 new file mode 100644 index 000000000..8fd0fa979 Binary files /dev/null and b/tests/fixtures/audio/silence.mp3 differ diff --git a/tests/test_music_identity_evidence.py b/tests/test_music_identity_evidence.py new file mode 100644 index 000000000..e088ba084 --- /dev/null +++ b/tests/test_music_identity_evidence.py @@ -0,0 +1,298 @@ +"""真实容器、来源响应及 JSON 快照分别保留录音与发行身份和字段来源。""" + +import asyncio +import json +import pickle +import shutil +from pathlib import Path +from unittest.mock import AsyncMock, Mock + +import pytest +from mutagen import File +from mutagen.flac import FLAC +from mutagen.id3 import TALB, TDOR, TDRC, TIT2, TPE1, TPE2, TPOS, TRCK, TXXX, UFID +from mutagen.mp4 import MP4 + +from app.application.audio import AudioMetadataHelper +from app.chain.download import DownloadChain +from app.chain.media import MediaChain +from app.chain.transfer.facade import TransferChain +from app.domain.context import MusicAlbumInfo, MusicInfo +from app.domain.meta.metamusic import MetaMusic +from app.modules.musicbrainz import MusicBrainzModule +from app.schemas.music import MusicInfo as MusicInfoSchema +from app.schemas.music import MusicMeta as MusicMetaSchema +from tests.test_audio_containers import _write_dsf, _write_wave +from tests.test_transfer_sync_extra_files import make_fileitem + +RECORDING = "11111111-1111-4111-8111-111111111111" +RELEASE = "22222222-2222-4222-8222-222222222222" +GROUP = "33333333-3333-4333-8333-333333333333" +TRACK = "44444444-4444-4444-8444-444444444444" +IDENTITIES = { + "musicbrainz_release_id": RELEASE, + "musicbrainz_release_group_id": GROUP, + "musicbrainz_release_track_id": TRACK, +} + + +def _tag_audio(path: Path) -> None: + """按容器的实际规范写入不同层级 ID，不调用产品标签写入实现。""" + if path.suffix == ".flac": + audio = FLAC(path) + audio.update({ + "title": ["Song"], "artist": ["Artist"], "albumartist": ["Album Artist"], "album": ["Record"], + "date": ["2020-06-01"], "originaldate": ["2003-01-01"], "tracknumber": ["3/12"], "discnumber": ["2/2"], + "musicbrainz_trackid": [RECORDING], "musicbrainz_albumid": [RELEASE], + "musicbrainz_releasegroupid": [GROUP], "musicbrainz_releasetrackid": [TRACK], + }) + elif path.suffix == ".m4a": + audio = MP4(path) + audio.update({"\xa9nam": ["Song"], "\xa9ART": ["Artist"], "\xa9alb": ["Record"], + "aART": ["Album Artist"], "\xa9day": ["2020-06-01"], "trkn": [(3, 12)], "disk": [(2, 2)]}) + for name, value in (("MusicBrainz Track Id", RECORDING), ("MusicBrainz Album Id", RELEASE), + ("MusicBrainz Release Group Id", GROUP), ("MusicBrainz Release Track Id", TRACK), + ("ORIGINALDATE", "2003-01-01")): + audio[f"----:com.apple.iTunes:{name}"] = [value.encode("utf-8")] + else: + audio = File(path) + if audio.tags is None: + audio.add_tags() + for frame in ( + TIT2(encoding=3, text=["Song"]), TPE1(encoding=3, text=["Artist"]), + TALB(encoding=3, text=["Record"]), TPE2(encoding=3, text=["Album Artist"]), + TDRC(encoding=3, text=["2020-06-01"]), TDOR(encoding=3, text=["2003-01-01"]), + TRCK(encoding=3, text=["3/12"]), TPOS(encoding=3, text=["2/2"]), + UFID(owner="http://musicbrainz.org", data=RECORDING.encode("ascii")), + TXXX(encoding=3, desc="MusicBrainz Album Id", text=[RELEASE]), + TXXX(encoding=3, desc="MusicBrainz Release Group Id", text=[GROUP]), + TXXX(encoding=3, desc="MusicBrainz Release Track Id", text=[TRACK]), + ): + audio.tags.add(frame) + audio.save() + + +@pytest.mark.parametrize("extension", ["flac", "wav", "dsf", "m4a", "mp3"]) +def test_real_container_keeps_distinct_musicbrainz_entities(tmp_path, extension): + """实际文件中四种 MBID 和两种年份必须读入各自字段，不混用录音与发行曲目。""" + path = tmp_path / f"03.{extension}" + if extension == "wav": + _write_wave(path) + elif extension == "dsf": + _write_dsf(path) + else: + shutil.copyfile(Path(__file__).parent / f"fixtures/audio/silence.{extension}", path) + _tag_audio(path) + before = path.read_bytes() + + meta = AudioMetadataHelper.read_tags(path) + + assert meta.media_id == RECORDING + for key, expected in IDENTITIES.items(): + assert getattr(meta, key) == expected + assert meta.field_sources[key] == "tag" + assert (meta.year, meta.release_year, meta.original_year) == (2020, 2020, 2003) + assert (meta.track_number, meta.total_tracks, meta.disc_number, meta.total_discs) == (3, 12, 2, 2) + assert meta.title == "Song" + assert meta.field_sources["title"] == "tag" + assert path.read_bytes() == before + + +def test_release_only_tag_never_becomes_recording_identity(tmp_path): + """只有发行 ID 时保留补充身份，不能凭空创建 Recording 主身份。""" + path = tmp_path / "01.flac" + shutil.copyfile(Path(__file__).parent / "fixtures/audio/silence.flac", path) + audio = FLAC(path) + audio["musicbrainz_albumid"] = [RELEASE] + audio["musicbrainz_releasetrackid"] = ["invalid-id"] + audio.save() + + meta = AudioMetadataHelper.read_tags(path) + + assert meta.musicbrainz_release_id == RELEASE + assert meta.musicbrainz_release_track_id is None + assert meta.media_source is None + assert meta.media_id is None + + +def test_audio_evidence_roundtrip_keeps_ids_and_independent_sources(): + """领域、API、下载历史和旧缓存读写不会丢弃补充身份或共享可变来源字典。""" + meta = MetaMusic(title="Song", artists=["Artist"], album="Record", media_source="musicbrainz", + media_id=RECORDING, original_year=2003, release_year=2020, total_discs=2, + field_sources={"title": "tag"}, **IDENTITIES) + info = MusicInfo.from_meta(meta) + note = DownloadChain._build_download_note("Manual", info, meta) + decoded = json.loads(json.dumps(note)) + restored_meta = MetaMusic.from_dict(MusicMetaSchema.model_validate(decoded["music"]["meta"]).model_dump()) + restored_info = MusicInfo.from_dict(MusicInfoSchema.model_validate(decoded["music"]["media"]).model_dump()) + + for value in (restored_meta, restored_info, pickle.loads(pickle.dumps(meta)), pickle.loads(pickle.dumps(info))): + assert value.media_id == RECORDING + assert value.original_year == 2003 + assert value.release_year == 2020 + assert value.total_discs == 2 + for key, expected in IDENTITIES.items(): + assert getattr(value, key) == expected + meta.field_sources["title"] = "manual" + assert info.field_sources["title"] == restored_meta.field_sources["title"] == "tag" + + +def test_legacy_meta_pickle_defaults_new_evidence_fields(): + """旧版解析对象没有新字段时仍可还原、套用路径和序列化。""" + original = MetaMusic(title="Song", artists=["Artist"]) + state = dict(original.__dict__) + for key in (*IDENTITIES, "original_year", "release_year", "field_sources"): + state.pop(key) + restored = MetaMusic.__new__(MetaMusic) + restored.__setstate__(state) + + restored.apply_path_context("/music/Artist - Record (2003)/01 - Song.flac") + + assert restored.to_dict()["musicbrainz_release_id"] is None + assert restored.field_sources["album"] == "directory" + assert restored.title == "Song" + + +def test_tag_title_equal_to_filename_is_not_reparsed(tmp_path): + """标题恰好等于文件名仍是实际标签，不能错误当作文件名兜底而清洗掉内容。""" + path = tmp_path / "03 - Song.flac" + shutil.copyfile(Path(__file__).parent / "fixtures/audio/silence.flac", path) + audio = FLAC(path) + audio["title"] = [path.stem] + audio.save() + + meta = AudioMetadataHelper.read(path) + + assert meta.title == "03 - Song" + assert meta.track_number == 3 + assert meta.field_sources["title"] == "tag" + assert meta.field_sources["track_number"] == "filename" + assert meta.field_sources["duration"] == "stream" + + +def _release_payload() -> dict: + """构造同一录音在具体再版及第二碟中的 MusicBrainz 响应。""" + return { + "id": RELEASE, "title": "Record", "date": "2020-06-01", + "artist-credit": [{"artist": {"id": "artist", "name": "Artist"}}], + "release-group": {"id": GROUP, "primary-type": "Album", "first-release-date": "2003-01-01"}, + "media": [{"position": 1, "tracks": []}, {"position": 2, "track-count": 1, "tracks": [{ + "id": TRACK, "position": 1, "title": "Song", "length": 3000, + "recording": {"id": RECORDING, "title": "Song", "first-release-date": "2003-01-01"}, + }]}], + } + + +def test_selected_release_preserves_original_year_and_track_namespace(): + """具体再版日期不能被录音首发日期覆盖，发行曲目 ID 必须独立保存。""" + album = MusicBrainzModule._release_to_album(_release_payload()) + track = album.tracks[0] + + assert album.media_id == GROUP + assert album.musicbrainz_release_id == RELEASE + assert track.media_id == RECORDING + assert track.musicbrainz_release_track_id == TRACK + assert track.musicbrainz_release_group_id == GROUP + assert track.musicbrainz_release_id == RELEASE + assert (track.year, track.release_year, track.original_year) == (2020, 2020, 2003) + assert track.release_date == "2020-06-01" + assert track.total_discs == 2 + assert track.field_sources["musicbrainz_release_id"] == "remote" + card = MusicAlbumInfo.from_dict(album.to_dict()).to_music_info() + assert card.musicbrainz_release_id == RELEASE + assert card.release_year == 2020 + assert card.original_year == 2003 + + +def test_release_without_group_does_not_forge_album_group_id(): + """残缺响应缺少发行组时不能把发行 ID 填进发行组主身份。""" + payload = _release_payload() + payload.pop("release-group") + + album = MusicBrainzModule._release_to_album(payload) + + assert album.media_id is None + assert album.musicbrainz_release_group_id is None + assert album.musicbrainz_release_id == RELEASE + assert album.tracks[0].media_id == RECORDING + + +def test_group_without_selected_release_has_no_release_year(): + """发行组的首发年份不等于已经确认当前文件的发行年份。""" + album = MusicBrainzModule._release_group_to_album({ + "id": GROUP, "title": "Record", "first-release-date": "2003-01-01", + }) + + card = album.to_music_info() + + assert card.original_year == 2003 + assert card.release_year is None + assert card.musicbrainz_release_id is None + + +def test_album_meta_retains_entity_type_and_cannot_write_recording_tag(): + """专辑上下文往返 MetaMusic 后仍是专辑，不能把组 ID 写进录音标签。""" + album = MusicAlbumInfo(media_source="musicbrainz", media_id=GROUP, title="Record") + + meta = MetaMusic.from_music_info(album.to_music_info()) + restored = MetaMusic.from_dict(meta.to_dict()) + info = MusicInfo.from_meta(restored) + + assert restored.music_type == info.music_type == "album" + assert restored.media_id == info.media_id == GROUP + assert AudioMetadataHelper._musicbrainz_recording_id(restored) is None + + +@pytest.mark.parametrize("source", [None, "musicbrainz"]) +@pytest.mark.parametrize("asynchronous", [False, True]) +def test_bound_album_meta_routes_as_album_with_or_without_source_override(source, asynchronous, monkeypatch): + """指定或继承数据源时，已绑定专辑身份都不能回到默认 Recording 查询。""" + meta = MetaMusic.from_music_info(MusicAlbumInfo( + media_source="musicbrainz", media_id=GROUP, title="Record", + ).to_music_info()) + + chain = MediaChain() + operation = AsyncMock(return_value=None) if asynchronous else Mock(return_value=None) + name = "async_recognize_music_from_source" if asynchronous else "recognize_music_from_source" + monkeypatch.setattr(chain, name, operation) + if asynchronous: + asyncio.run(chain._async_run_native_media_recognize({"meta": meta, "media_source": source}, cache=True)) + else: + chain._run_native_media_recognize({"meta": meta, "media_source": source}, cache=True) + + assert operation.call_args.kwargs["music_type"] == "album" + assert operation.call_args.kwargs["meta"].media_id == GROUP + + +def test_new_identity_fields_preserve_legacy_positional_constructors(): + """补充事实只能新增关键字参数，插件既有位置参数不能被悄悄错位。""" + track = MusicInfo("musicbrainz", RECORDING, "recording", "Song", ["Artist"], [], + "Record", "Artist", GROUP, "Album", [], 2003) + album = MusicAlbumInfo("musicbrainz", GROUP, "Record", ["Artist"]) + + assert (track.title, track.album_type, track.year) == ("Song", "Album", 2003) + assert album.title == "Record" + assert album.artists == ["Artist"] + + +def test_explicit_release_selection_replaces_stale_local_ids(tmp_path): + """手选发行不能留下另一个发行的标签 ID，实际音频参数仍来自当前文件。""" + path = tmp_path / "03.flac" + item = make_fileitem(str(path)) + local = MetaMusic(title="Old Song", album="Old Record", musicbrainz_release_id=GROUP, + musicbrainz_release_group_id=RELEASE, musicbrainz_release_track_id=RECORDING, + sample_rate=96000, field_sources={"sample_rate": "stream"}) + selected = MusicInfo(media_source="musicbrainz", media_id=RECORDING, title="Song", + artists=["Artist"], album="Record", release_year=2020, **IDENTITIES) + + meta, info = TransferChain()._selected_music_task_context( + item, path, local, {str(path.resolve()): selected}, None, + ) + + for key, expected in IDENTITIES.items(): + assert getattr(meta, key) == getattr(info, key) == expected + assert meta.field_sources[key] == "manual" + assert meta.music_type == "recording" + assert meta.sample_rate == info.sample_rate == 96000 + assert meta.field_sources["sample_rate"] == "stream" + assert local.musicbrainz_release_id == GROUP