feat(music): preserve release identities and metadata evidence

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 []
 

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
 
 

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(

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

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 (

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),

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"(?<![A-Z])(?P<format>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

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]:

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

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` 实体过滤 |

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
 

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 标准日期字段。

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