fix(music): read and write native ASF metadata safely

app/application/audio.py

@@ -1,5 +1,6 @@
 import re
 import shutil
+import struct
 from collections import OrderedDict
 from contextlib import contextmanager
 from contextvars import ContextVar
@@ -17,6 +18,7 @@ from mutagen import File as MutagenFile
 from mutagen import MutagenError
 from mutagen.aiff import AIFF
 from mutagen.apev2 import APEBinaryValue, APETextValue, APEv2
+from mutagen.asf import ASF, ASFBoolAttribute, ASFByteArrayAttribute, ASFTags
 from mutagen.dsdiff import DSDIFF
 from mutagen.dsf import DSF
 from mutagen.flac import FLAC, Picture
@@ -34,6 +36,23 @@ from app.schemas.types import MUSIC_ENTITY_RECORDING, MediaSource
 
 _AudioSignature = tuple[str, int, int, int, int]
 
+# 首项是写入键，其余是只读兼容别名；独立乐团/演奏者是自定义字段。
+_ASF_FIELDS = {
+    "title": ("Title",), "artist": ("Author",), "album": ("WM/AlbumTitle",),
+    "albumartist": ("WM/AlbumArtist", "album artist"), "date": ("WM/Year", "year"),
+    "originaldate": ("WM/OriginalReleaseTime", "WM/OriginalReleaseYear", "originalyear"),
+    "tracknumber": ("WM/TrackNumber", "track"), "discnumber": ("WM/PartOfSet", "disc"),
+    "isrc": ("WM/ISRC",), "subtitle": ("WM/SubTitle",),
+    "composer": ("WM/Composer",), "conductor": ("WM/Conductor",),
+    "orchestra": ("WM/Orchestra", "WM/Ensemble", "ensemble"), "performer": ("WM/Performer",),
+    "compilation": ("WM/IsCompilation",), "lyrics": ("WM/Lyrics",),
+    "musicbrainz_trackid": ("MusicBrainz/Track Id", "musicbrainz_track_id"),
+    "musicbrainz_albumid": ("MusicBrainz/Album Id", "musicbrainz_album_id"),
+    "musicbrainz_releasegroupid": ("MusicBrainz/Release Group Id", "musicbrainz_release_group_id"),
+    "musicbrainz_releasetrackid": ("MusicBrainz/Release Track Id", "musicbrainz_releasetrack_id"),
+    "musicbrainz_albumtype": ("MusicBrainz/Album Type",),
+}
+
 
 @dataclass(slots=True)
 class _AudioScan:
@@ -281,6 +300,72 @@ def _read_mp4_tags(tags: MP4Tags) -> dict[str, list[str]]:
     return values
 
 
+def _asf_text_values(values: Any) -> list[str]:
+    """ASF文本/整数/布尔属性均有合法文本含义，二进制属性不能伪装成标题或MBID。"""
+    return [str(int(item.value)) if isinstance(item, ASFBoolAttribute) else str(item).strip()
+            for item in values or [] if not isinstance(item, (ASFByteArrayAttribute, bytes)) and str(item).strip()]
+
+
+def _read_asf_tags(tags: ASFTags) -> dict[str, list[str]]:
+    """优先读取Windows Media原生键，再兼容旧小写写入；WM/Track为零基编号。"""
+    values = {key.casefold(): _asf_text_values(items) for key, items in tags.items()}
+    for key, aliases in _ASF_FIELDS.items():
+        values[key] = next((items for alias in (*aliases, key)
+                            if (items := _asf_text_values(tags.get(alias)) or values.get(alias.casefold()))), [])
+    current = next(iter(values.get("tracknumber", [])), "").split("/", 1)[0]
+    if not current.isdecimal() or int(current) < 1:
+        legacy = next(iter(values.get("wm/track", [])), "")
+        if legacy.isdecimal():
+            values["tracknumber"] = [str(int(legacy) + 1)]
+    return values
+
+
+def _write_asf_tags(tags: ASFTags, values: dict[str, list[str]]) -> bool:
+    """差异预检后只替换所需字段及其兼容别名，保留用户其它ASF属性。"""
+    for key, value in values.items():
+        names = _ASF_FIELDS.get(key, (key,))
+        aliases = {name.casefold() for name in (*names, key)}
+        if key == "tracknumber":
+            aliases.add("wm/track")
+        if key in {"tracknumber", "discnumber"} and "/" in value[0]:
+            aliases.update(("tracktotal", "totaltracks") if key == "tracknumber" else ("disctotal", "totaldiscs"))
+        for existing in list(tags.keys()):
+            if existing.casefold() in aliases or (key == "performer" and existing.casefold().startswith("performer:")):
+                del tags[existing]
+        tags[names[0]] = value
+    return bool(values)
+
+
+def _mark_asf_quality(meta: MetaMusic, audio: ASF) -> None:
+    """ASF编码类型来自音频流的Codec List；不能从码率、扩展名或描述中的宣传词推断无损。"""
+    codec = audio.info.codec_type.casefold()
+    if codec.startswith("windows media audio"):
+        meta.apply_audio_quality("WMA lossless" if "lossless" in codec else "WMA", overwrite=True, evidence_source="stream")
+    else:
+        # 未知编码的ASF不沿用可能错误的.flac/.mp3扩展名声明实际流格式。
+        meta.audio_format = None
+        meta.audio_lossless = None
+        meta.field_sources.pop("audio_format", None)
+        meta.field_sources.pop("audio_lossless", None)
+
+
+def _write_binary_cover(audio: Any, path: Path, data: bytes, mime: str, overwrite: bool) -> None:
+    """ASF与APE使用专用二进制属性保存封面，保留缺失才补写的策略。"""
+    if audio.tags is None:
+        audio.add_tags()
+    key = "WM/Picture" if isinstance(audio, ASF) else "Cover Art (Front)"
+    if audio.tags.get(key) and not overwrite:
+        return
+    if isinstance(audio, ASF):
+        # ASF_FLAT_PICTURE：类型、图像字节数、两个UTF-16LE终止字符串及图像数据。
+        payload = struct.pack("<BI", 3, len(data)) + (mime + "\0\0").encode("utf-16-le") + data
+        audio.tags[key] = [ASFByteArrayAttribute(payload)]
+    else:
+        filename = "cover.png" if mime == "image/png" else "cover.jpg"
+        audio.tags[key] = APEBinaryValue(filename.encode("ascii") + b"\x00" + data)
+    audio.save(path)
+
+
 def _performer_pair(value: str) -> tuple[str, str]:
     """解析PERFORMER专用的“人名 (乐器)”标签，不拆分人名内部的逗号或斜线。"""
     match = re.fullmatch(r"(.+?)\s+\(([^()]+)\)", value.strip())
@@ -371,7 +456,9 @@ def _mark_audio_evidence(meta: MetaMusic, audio: Any) -> MetaMusic:
         (FLAC, "FLAC"), (MP3, "MP3"), (WAVE, "WAV"), (AIFF, "AIFF"),
         (MonkeysAudio, "APE"), (DSF, "DSD"), (DSDIFF, "DSD"),
     ) if isinstance(audio, kind)), None)
-    if detected:
+    if isinstance(audio, ASF):
+        _mark_asf_quality(meta, audio)
+    elif detected:
         meta.apply_audio_quality(detected, overwrite=True, evidence_source="stream")
     elif meta.audio_format:
         source = "stream" if getattr(info, "codec", None) or getattr(info, "codec_description", None) else "filename"
@@ -513,13 +600,15 @@ class AudioMetadataHelper:
 
     @staticmethod
     def _readable_tags(tags: Any) -> Any:
-        """将原生 ID3/MP4 及 Vorbis/APEv2 映射到统一键，不修改音频。
+        """将原生ID3/MP4/ASF及Vorbis/APEv2映射到统一键，不修改音频。
 
         原生读取可保留 Easy 包装未注册的发行字段；Recording 的身份只取
         专用字段，不能误用 Release Track ID。
         """
         if isinstance(tags, MP4Tags):
             return _read_mp4_tags(tags)
+        if isinstance(tags, ASFTags):
+            return _read_asf_tags(tags)
         if isinstance(tags, APEv2):
             values = {key.casefold(): value for key, value in tags.items()}
             values.update({key.replace(" ", "_"): value for key, value in list(values.items())})
@@ -587,6 +676,8 @@ class AudioMetadataHelper:
             str(key).casefold(): value
             for key, value in getattr(tags, "items", lambda: [])()
         }
+        if isinstance(tags, ASFTags):
+            normalized.update(_read_asf_tags(tags))
         synced = synced or cls._tag_text(
             normalized,
             "syncedlyrics",
@@ -706,6 +797,7 @@ class AudioMetadataHelper:
                    "musicbrainz_albumid": ("musicbrainz_albumid", "musicbrainz_album_id"),
                    "musicbrainz_releasegroupid": ("musicbrainz_releasegroupid", "musicbrainz_release_group_id"),
                    "musicbrainz_releasetrackid": ("musicbrainz_releasetrackid", "musicbrainz_release_track_id")}
+        totals = {"tracknumber": ("tracktotal", "totaltracks"), "discnumber": ("disctotal", "totaldiscs")}
         updates = {}
         for key, value in cls._tag_values(music).items():
             if value in (None, "", []):
@@ -716,8 +808,10 @@ class AudioMetadataHelper:
                 current = credits[key]
                 if set(current) == set(expected):
                     continue
-            if current and key in {"tracknumber", "discnumber"} and cls._number_pair(current[0]) == cls._number_pair(expected[0]):
-                # MP4用(1, 0)表达未知总数，与文本标签的1具有相同位置语义。
+            if current and key in totals and cls._number_pair(
+                    current[0], cls._first_of(tags, *totals[key]),
+            ) == cls._number_pair(expected[0]):
+                # 组合与独立总数按相同位置语义比较，避免为等价标签复制整个音频。
                 continue
             if key in {"date", "originaldate"} and current and len(expected[0]) == 4 and current[0].startswith(f"{expected[0]}-"):
                 continue
@@ -737,6 +831,8 @@ class AudioMetadataHelper:
             return bool(tags.get("Cover Art (Front)"))
         if isinstance(audio, MP4):
             return bool(tags.get("covr"))
+        if isinstance(audio, ASF):
+            return bool(tags.get("WM/Picture"))
         return bool(hasattr(tags, "getall") and tags.getall("APIC"))
 
     @classmethod
@@ -749,6 +845,8 @@ class AudioMetadataHelper:
             changed = cls._write_id3_tags(audio.tags, values, overwrite)
         elif isinstance(audio.tags, MP4Tags):
             changed = cls._write_mp4_tags(audio.tags, values, overwrite)
+        elif isinstance(audio.tags, ASFTags):
+            changed = _write_asf_tags(audio.tags, values)
         else:
             changed = cls._write_text_tags(audio, values, overwrite, path)
         if changed:
@@ -896,19 +994,10 @@ class AudioMetadataHelper:
             cover_mime: str,
             overwrite: bool,
     ) -> None:
-        """为原生 ID3、FLAC、MP4/M4A 和 APE 写入内嵌封面，其它格式保留标签写入结果。"""
+        """按容器写入内嵌封面，链接隔离由外层写入事务统一负责。"""
         audio = cls._open_audio(path)
-        if isinstance(audio, MonkeysAudio):
-            if audio.tags is None:
-                audio.add_tags()
-            cover_key = "Cover Art (Front)"
-            if cover_key in audio.tags and not overwrite:
-                return
-            cover_filename = "cover.png" if cover_mime == "image/png" else "cover.jpg"
-            audio.tags[cover_key] = APEBinaryValue(
-                cover_filename.encode("ascii") + b"\x00" + cover_data
-            )
-            audio.save(path)
+        if isinstance(audio, (MonkeysAudio, ASF)):
+            _write_binary_cover(audio, path, cover_data, cover_mime, overwrite)
             return
         if isinstance(audio, FLAC):
             if audio.pictures and not overwrite:
@@ -957,7 +1046,7 @@ class AudioMetadataHelper:
 
     @staticmethod
     def _values(tags: Any, key: str) -> list[str]:
-        """从 Mutagen Easy 标签中提取非空字符串列表。"""
+        """从容器的统一文本字段或兼容标签中提取非空字符串列表。"""
         value = tags.get(key) if hasattr(tags, "get") else None
         if value is None or isinstance(value, APEBinaryValue):
             return []

docs/mcp-api.md

@@ -600,11 +600,17 @@ SSE 的 `candidate_items` 是站点原始返回数量，`match_counts` 记录身
 全部成功后原子替换目标；源文件字节与权限保持不变。实际发生写入后，目标成为普通文件，
 会独立占用磁盘空间。未发生标签变化且没有其它写入时保留原链接；跳过标签及封面可保持链接整理。
 写入失败或目标在处理期间改变时，丢弃临时副本，不覆盖原目标。LRC 与 Lyricsfile 旁挂也按目录项原子替换。
-WAV、DSF 和 MP3 使用原生 ID3 帧，MP4/M4A 使用标准 atom 及文本 freeform；无标签音频可直接补写，
+WAV、DSF、AIFF、DSDIFF 和 MP3 使用原生 ID3 帧，MP4/M4A 使用标准 atom 及文本 freeform；无标签音频可直接补写，
 错误扩展名按实际容器处理。Recording、Release、Release Group、Release Track ID 分别写入专用标签；
 仅有首发年份时只写原始年份标签，不能据此生成当前发行日期；已有同年的完整日期不因模型只携带年份而截断。
 APEv2 按[标准标签映射](https://picard-docs.musicbrainz.org/en/latest/appendices/tag_mapping.html)
 使用下划线形式的 MusicBrainz 字段和 `Originalyear`，不套用 ID3 的带空格描述；读取仍兼容旧字段别名。
+WMA/ASF 使用 `Title`、`Author`、`WM/AlbumTitle`、`WM/PartOfSet`、`MusicBrainz/*` 等原生属性，
+兼容旧版小写字段；标准 `WM/TrackNumber` 从 1 开始，旧 `WM/Track` 从 0 开始。
+`WM/Lyrics` 可读取纯文本歌词，`WM/Picture` 可补写封面，沿用相同的链接隔离与覆盖策略。
+WMA Lossless 只由实际 Codec List 类型确认，不从高码率或扩展名猜测，未提供的位深保持未知。
+独立乐团/演奏者字段 `WM/Orchestra`、`WM/Performer` 为自定义兼容属性。
+等价的组合/独立曲数标签不会触发音频复制；曲数、碟数和更精确的原有日期保持有效。
 
 音乐整理按每个文件的 `storage` 决定证据来源：远端仅使用名称、目录和原始种子线索，
 不读取本机同路径的标签、时长或 CUE。仅有 `.m4a` 扩展名时不声明 AAC/ALAC 或有损/无损。

skills/moviepilot-api/api/music.md

@@ -80,6 +80,14 @@ Music recognition, exploration, album and artist navigation, and recognition-cac
   the original-year tag (`ORIGINALDATE` or its native equivalent), never a
   fabricated current-release date. Preserve more precise existing dates in the
   same year. APEv2 MusicBrainz keys use underscores, not ID3's spaced descriptions.
+  WMA/ASF uses native Title/Author/WM and MusicBrainz properties, with legacy
+  lowercase aliases accepted. WM/TrackNumber is one-based; legacy WM/Track is
+  zero-based. WM/Lyrics supplies plain lyrics, and WM/Picture embeds covers under
+  the same link-isolation and overwrite policy. Only the actual codec type can
+  establish WMA Lossless; high bitrate and filename extensions cannot. Missing
+  bit depth stays unknown. WM/Orchestra and WM/Performer are custom compatibility
+  properties. Equivalent separate/combined track totals retain existing links.
+  AIFF and DSDIFF use the same native ID3 path as WAV and DSF.
 - Organization uses each file's `storage`: remote items use names and original
   torrent evidence, never local tags, duration, or CUE at an identical path.
   An `.m4a` suffix alone does not establish AAC/ALAC or lossless status.

tests/fixtures/audio/README.md

@@ -11,3 +11,9 @@
 移除了输入元数据，用于验证原生 MP4 atom/freeform 与 MP3 ID3 字段。
 测试直接通过 Mutagen 写入独立 MBID 和日期，CI 不依赖编码器。
 M4A 的 ORIGINALDATE 是可选自定义 freeform 兼容样本，不声明它是 MP4 标准日期字段。
+
+`silence.wma` 是同一静音 FLAC 使用 `ffmpeg -i silence.flac -map_metadata -1 -c:a wmav2 -b:a 128k silence.wma`
+转换后经 `mutagen.asf.ASF.clear()` / `save()` 清除标签的三秒 WMAv2 音频，44.1 kHz、双声道，
+用于标准 ASF 属性、旧曲序、封面、歌词和链接源文件保护测试。
+WMA Lossless 的 Codec List 分支使用受控流信息验证，不把此 WMAv2 载荷声称为无损编码。
+AIFF 和 DSDIFF 在测试中以标准库生成合法容器及静音数据，不新增二进制依赖或编码器要求。

tests/test_audio_asf.py

@@ -0,0 +1,156 @@
+"""真实ASF/WMA原生字段、旧编号、流声明和离线整理回归。"""
+
+import hashlib
+import os
+import shutil
+from pathlib import Path
+
+import pytest
+from mutagen.asf import ASF, ASFBoolAttribute, ASFByteArrayAttribute, ASFDWordAttribute
+
+from app.application.audio import AudioMetadataHelper
+from app.chain.transfer.music import _local_music_context, prepare_music_batch_context
+from app.domain.context import MusicInfo
+from app.schemas.types import MediaType
+from tests.test_music_source_protection import _COVER, _GROUP, _RECORDING, _RELEASE, _TRACK
+from tests.test_transfer_sync_extra_files import make_fileitem, make_transfer_chain
+
+
+@pytest.fixture
+def wma_path(tmp_path):
+    """复制无标签的真实三秒WMAv2音频，测试不需要编码器。"""
+    path = tmp_path / "03.wma"
+    shutil.copyfile(Path(__file__).parent / "fixtures/audio/silence.wma", path)
+    return path
+
+
+def _native_tags(path):
+    """独立按Windows Media/Picard原生字段构造输入，不借助被测写入器。"""
+    audio = ASF(path)
+    audio.tags.update({
+        "Title": ["晴天"], "Author": ["周杰伦"], "WM/AlbumTitle": ["叶惠美"], "WM/AlbumArtist": ["周杰伦"],
+        "WM/Year": ["2020-07-31"], "WM/OriginalReleaseTime": ["2003-07-31"],
+        "WM/TrackNumber": [ASFDWordAttribute(3)], "WM/PartOfSet": ["2/2"], "TOTALTRACKS": ["11"],
+        "WM/ISRC": ["USQX91300105"], "WM/IsCompilation": [ASFBoolAttribute(True)],
+        "MusicBrainz/Album Type": ["ep"], "MusicBrainz/Track Id": [_RECORDING],
+        "MusicBrainz/Album Id": [_RELEASE], "MusicBrainz/Release Group Id": [_GROUP],
+        "MusicBrainz/Release Track Id": [_TRACK], "WM/Composer": ["周杰伦"], "WM/Lyrics": ["歌词正文"],
+    })
+    audio.save()
+
+
+def test_native_asf_tags_reach_offline_batch_and_keep_distinct_identities(wma_path):
+    """标准标签贯通本地批次，不需要在线识别；发行和录音身份保持区分。"""
+    _native_tags(wma_path)
+    original = hashlib.sha256(wma_path.read_bytes()).digest()
+    meta = AudioMetadataHelper.read_tags(wma_path)
+    assert (meta.title, meta.artists, meta.album, meta.album_artist) == ("晴天", ["周杰伦"], "叶惠美", "周杰伦")
+    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, 11, 2, 2)
+    assert (meta.media_id, meta.musicbrainz_release_id, meta.musicbrainz_release_group_id, meta.musicbrainz_release_track_id) == (
+        _RECORDING, _RELEASE, _GROUP, _TRACK)
+    assert meta.album_type == "EP" and meta.secondary_types == ["Compilation"]
+    assert meta.composers == ["周杰伦"] and meta.field_sources["composers"] == "tag"
+    assert AudioMetadataHelper.read_lyrics(wma_path).plain_lyrics == "歌词正文"
+    owner = make_transfer_chain()
+    item = make_fileitem(str(wma_path))
+    batch = prepare_music_batch_context(owner, [(item, False)], MediaType.MUSIC)
+    _, info = _local_music_context(owner, item, wma_path, batch, AudioMetadataHelper.read(wma_path))
+    assert info.title == "晴天" and info.album == "叶惠美" and info.composers == ["周杰伦"]
+    assert hashlib.sha256(wma_path.read_bytes()).digest() == original
+
+
+@pytest.mark.parametrize("value,expected", [("0", 1), (ASFDWordAttribute(2), 3), ("-1", None), ("bad", None)])
+def test_asf_legacy_track_is_zero_based_and_never_negative(wma_path, value, expected):
+    """旧WM/Track接受字符串与DWORD，零是第一轨；坏值不能变成有效曲序。"""
+    audio = ASF(wma_path)
+    audio.tags["WM/Track"] = [value]
+    audio.save()
+    assert AudioMetadataHelper.read_tags(wma_path).track_number == expected
+
+
+def test_asf_standard_fields_override_old_aliases_and_correction_removes_stale_values(wma_path):
+    """原生字段优先于旧版小写字段，纠正时只清理同字段别名而保留未知用户数据。"""
+    _native_tags(wma_path)
+    audio = ASF(wma_path)
+    audio.tags.update({"artist": ["Old Artist"], "album": ["Old Album"], "WM/Track": [0],
+                       "orchestra": ["Old Orchestra"], "WM/Ensemble": ["Old Orchestra"],
+                       "performer:violin": ["Old Soloist"], "WM/Producer": ["Producer"], "USER/Binary": [b"\x01\x02"]})
+    audio.save()
+    meta = AudioMetadataHelper.read_tags(wma_path)
+    assert meta.artists == ["周杰伦"] and meta.album == "叶惠美" and meta.track_number == 3
+    meta.artists, meta.album, meta.track_number = ["New Artist"], "New Album", 4
+    meta.orchestras, meta.performers = ["New Orchestra"], {"violin": ["New Soloist"]}
+    assert AudioMetadataHelper.write(wma_path, meta)
+    native = ASF(wma_path)
+    assert [str(value) for value in native.tags["Author"]] == ["New Artist"]
+    assert [str(value) for value in native.tags["WM/TrackNumber"]] == ["4/11"]
+    assert not any(key in native.tags for key in ("artist", "album", "WM/Track", "orchestra", "WM/Ensemble", "performer:violin"))
+    assert str(native.tags["WM/Producer"][0]) == "Producer" and native.tags["USER/Binary"][0].value == b"\x01\x02"
+    actual = AudioMetadataHelper.read_tags(wma_path)
+    assert actual.orchestras == ["New Orchestra"] and actual.performers == {"violin": ["New Soloist"]}
+
+
+def test_asf_empty_and_binary_fields_do_not_fake_identity_or_tag_completeness(wma_path):
+    """无标签音频保留流信息；错误类型属性及只有发行曲目ID时不能制造录音身份。"""
+    meta = AudioMetadataHelper.read_tags(wma_path)
+    assert meta.duration == 3 and meta.sample_rate == 44100 and meta.bitrate == 128000
+    assert meta.audio_format == "WMA" and meta.audio_lossless is False
+    assert meta.field_sources["audio_format"] == meta.field_sources["audio_lossless"] == "stream"
+    assert meta.title is None and meta.artists == []
+    audio = ASF(wma_path)
+    audio.tags["WM/AlbumTitle"] = [ASFByteArrayAttribute(b"not text")]
+    audio.tags["MusicBrainz/Track Id"] = [ASFByteArrayAttribute(_RECORDING.encode())]
+    audio.tags["MusicBrainz/Release Track Id"] = [_TRACK]
+    audio.save()
+    meta = AudioMetadataHelper.read_tags(wma_path)
+    assert meta.album is None and meta.media_id is None and meta.musicbrainz_release_track_id == _TRACK
+
+
+@pytest.mark.parametrize("overwrite", [False, True])
+def test_asf_identical_tags_and_preserved_cover_keep_hardlink(wma_path, overwrite):
+    """完整年份精度与独立总曲数保持原样，缺失封面策略不触发无谓大文件复制。"""
+    _native_tags(wma_path)
+    audio = ASF(wma_path)
+    audio.tags["WM/Picture"] = [ASFByteArrayAttribute(b"existing-picture")]
+    audio.save()
+    meta = AudioMetadataHelper.read_tags(wma_path)
+    target = wma_path.with_name("library.wma")
+    os.link(wma_path, target)
+    before = target.read_bytes()
+    assert AudioMetadataHelper.write(target, meta, overwrite=overwrite, cover_data=_COVER, cover_overwrite=False)
+    assert os.path.samefile(wma_path, target) and target.read_bytes() == before
+
+
+@pytest.mark.parametrize("codec,expected_format,lossless", [
+    ("Windows Media Audio 9 Standard", "WMA", False),
+    ("Windows Media Audio 9 Professional", "WMA", False),
+    ("Windows Media Audio 9 Lossless", "WMA", True),
+    ("", None, None),
+])
+def test_asf_quality_uses_codec_identity_not_filename_or_bitrate(wma_path, monkeypatch, codec, expected_format, lossless):
+    """真实容器读取路径搭配受控Codec List投影；不声称静音WMAv2载荷变成了无损编码。"""
+    audio = ASF(wma_path)
+    audio.info.codec_type = codec
+    audio.info.codec_name = "Untrusted Lossless Label"
+    audio.info.codec_description = "Lossless 24 bit marketing text"
+    monkeypatch.setattr(AudioMetadataHelper, "_open_audio", lambda _path: audio)
+    meta = AudioMetadataHelper.read_tags(wma_path.with_suffix(".flac"))
+    assert meta.audio_format == expected_format and meta.audio_lossless is lossless
+    assert meta.bit_depth is None  # ASFInfo未提供实际位深，不能从描述伪造。
+
+
+def test_asf_failed_cover_write_preserves_both_library_and_seed(wma_path, monkeypatch):
+    """标签已写入临时副本但封面阶段失败时，原库目标和种子仍是原来的字节。"""
+    target = wma_path.with_name("library.wma")
+    os.link(wma_path, target)
+    before = target.read_bytes()
+
+    def fail_cover(**_kwargs):
+        """在外部封面写入边界模拟磁盘错误，保留前半段真实ASF写入。"""
+        raise OSError("cover failed")
+
+    monkeypatch.setattr(AudioMetadataHelper, "_write_cover", fail_cover)
+    assert not AudioMetadataHelper.write(target, MusicInfo(title="New"), cover_data=_COVER)
+    assert os.path.samefile(wma_path, target) and target.read_bytes() == before
+    assert not list(target.parent.glob("*.mp-audio-partial"))

tests/test_audio_containers.py

@@ -7,7 +7,7 @@ import wave
 from pathlib import Path
 
 import pytest
-from mutagen.dsf import DSF
+from mutagen import File as MutagenFile
 from mutagen.flac import FLAC
 from mutagen.id3 import TALB, TDRC, TIT2, TPE1, TPE2, TPOS, TRCK, TSRC, TXXX, UFID
 from mutagen.wave import WAVE
@@ -36,6 +36,29 @@ def _write_dsf(path: Path) -> Path:
     return path
 
 
+def _write_aiff(path: Path) -> Path:
+    """标准FORM/COMM/SSND块生成三秒44.1kHz PCM，采样率使用AIFF扩展浮点格式。"""
+    common = struct.pack(">hIh", 2, 44100 * 3, 16) + bytes.fromhex("400eac44000000000000")
+    sound = b"\0" * (8 + 44100 * 4 * 3)
+    chunks = b"COMM" + struct.pack(">I", len(common)) + common + b"SSND" + struct.pack(">I", len(sound)) + sound
+    path.write_bytes(b"FORM" + struct.pack(">I", len(chunks) + 4) + b"AIFF" + chunks)
+    return path
+
+
+def _write_dff(path: Path) -> Path:
+    """生成合法DSDIFF声道交织DSD数据及PROP，验证与DSF不同的原生ID3容器。"""
+    def chunk(name: bytes, data: bytes) -> bytes:
+        """DSDIFF块以64位大端长度计数，奇数字节补齐不计入内容长度。"""
+        return name + struct.pack(">Q", len(data)) + data + b"\0" * (len(data) % 2)
+
+    properties = b"SND " + chunk(b"FS  ", struct.pack(">I", 2822400))
+    properties += chunk(b"CHNL", struct.pack(">H", 2) + b"SLFTSRGT") + chunk(b"CMPR", b"DSD \x00")
+    content = b"DSD " + chunk(b"FVER", struct.pack(">I", 0x01050000)) + chunk(b"PROP", properties)
+    content += chunk(b"DSD ", b"\x69" * (2822400 * 3 * 2 // 8))
+    path.write_bytes(chunk(b"FRM8", content))
+    return path
+
+
 @pytest.fixture
 def flac_path(tmp_path):
     """复制无标签的三秒静音 FLAC，所有标签写入仅作用于测试临时文件。"""
@@ -61,11 +84,12 @@ def test_untagged_audio_keeps_stream_and_path_evidence(container, tmp_path, flac
     assert hashlib.sha256(path.read_bytes()).digest() == before
 
 
-@pytest.mark.parametrize("container", ["wav", "dsf"])
+@pytest.mark.parametrize("container", ["wav", "dsf", "aiff", "dff"])
 def test_raw_id3_container_reads_track_identity(container, tmp_path):
-    """WAV/DSF 原生 ID3 帧能提供曲目字段及 Recording 身份。"""
+    """WAV/DSF/AIFF/DSDIFF原生ID3帧能提供曲目字段及Recording身份。"""
     path = tmp_path / f"03.{container}"
-    audio = WAVE(_write_wave(path)) if container == "wav" else DSF(_write_dsf(path))
+    factory = {"wav": _write_wave, "dsf": _write_dsf, "aiff": _write_aiff, "dff": _write_dff}
+    audio = MutagenFile(factory[container](path))
     audio.add_tags()
     for frame in (
         TIT2(encoding=3, text=["晴天"]),

tests/test_music_classical_credits.py

@@ -11,6 +11,7 @@ from unittest.mock import AsyncMock, Mock
 
 import pytest
 from mutagen import File as MutagenFile
+from mutagen.asf import ASFTags
 from mutagen.id3 import ID3, TALB, TCOM, TIPL, TIT2, TMCL, TPE1, TPE3, TXXX
 from mutagen.mp4 import MP4Tags
 
@@ -33,7 +34,7 @@ from app.modules.musicbrainz import MusicBrainzModule
 from app.modules.musicbrainz.cache import MusicBrainzCache
 from app.modules.theaudiodb import TheAudioDbModule
 from app.schemas.types import MediaSource, MediaType
-from tests.test_audio_containers import _write_dsf, _write_wave
+from tests.test_audio_containers import _write_aiff, _write_dff, _write_dsf, _write_wave
 from tests.test_transfer_sync_extra_files import make_fileitem, make_transfer_chain
 
 CREDITS = {"composers": ["Ludwig van Beethoven"], "conductors": ["Herbert von Karajan"],
@@ -48,6 +49,10 @@ def _native_audio(tmp_path, container):
         _write_wave(path)
     elif container == "dsf":
         _write_dsf(path)
+    elif container == "aiff":
+        _write_aiff(path)
+    elif container == "dff":
+        _write_dff(path)
     else:
         shutil.copyfile(Path(__file__).parent / f"fixtures/audio/silence.{container}", path)
     audio = MutagenFile(path)
@@ -64,6 +69,10 @@ def _native_audio(tmp_path, container):
                            "\xa9wrt": CREDITS["composers"], "----:com.apple.iTunes:CONDUCTOR": [b"Herbert von Karajan"],
                            "----:com.apple.iTunes:ORCHESTRA": [b"Berliner Philharmoniker"],
                            "----:com.apple.iTunes:PERFORMER": [b"Anne-Sophie Mutter (violin)"]})
+    elif isinstance(audio.tags, ASFTags):
+        audio.tags.update({"Title": [TITLE], "Author": CREDITS["composers"], "WM/AlbumTitle": ["Violin Concerto"],
+                           "WM/Composer": CREDITS["composers"], "WM/Conductor": CREDITS["conductors"],
+                           "WM/Orchestra": CREDITS["orchestras"], "WM/Performer": ["Anne-Sophie Mutter (violin)"]})
     else:
         audio.tags.update({"title": [TITLE], "artist": CREDITS["composers"], "album": ["Violin Concerto"],
                            "composer": CREDITS["composers"], "conductor": CREDITS["conductors"],
@@ -72,7 +81,7 @@ def _native_audio(tmp_path, container):
     return path
 
 
-@pytest.mark.parametrize("container", ["flac", "mp3", "wav", "dsf", "m4a"])
+@pytest.mark.parametrize("container", ["flac", "mp3", "wav", "dsf", "m4a", "aiff", "dff", "wma"])
 def test_native_credits_survive_read_models_and_write_without_touching_seed(tmp_path, container):
     """原生角色到模型/API/pickle均不丢失，库内改写指挥不会修改硬链接做种源或主署名。"""
     source = _native_audio(tmp_path, container)

tests/test_music_source_protection.py

@@ -11,6 +11,7 @@ from unittest.mock import Mock
 import pytest
 from mutagen import File as MutagenFile
 from mutagen.apev2 import APEv2
+from mutagen.asf import ASF
 from mutagen.flac import FLAC
 from mutagen.id3 import ID3, TDOR, TDRC
 from mutagen.mp4 import MP4
@@ -23,7 +24,7 @@ from app.domain.meta.metamusic import MetaMusic
 from app.modules.filemanager.storages.local import LocalStorage
 from app.modules.filemanager.transhandler import TransHandler
 from app.schemas.file import FileItem
-from tests.test_audio_containers import _write_dsf, _write_wave
+from tests.test_audio_containers import _write_aiff, _write_dff, _write_dsf, _write_wave
 
 _COVER = base64.b64decode("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+aD1sAAAAASUVORK5CYII=")
 _RECORDING = "38035858-f990-4fbb-b3b2-f2f8b958eeba"
@@ -35,15 +36,14 @@ _TRACK = "b10bbbfc-cf9e-42e0-be17-e2c3e1d2600d"
 def _audio_file(tmp_path, container):
     """构造真实静音容器，编码来源与读取回归共用，不需要网络或ffmpeg。"""
     path = tmp_path / f"seed.{container}"
-    if container == "wav":
-        return _write_wave(path)
-    if container == "dsf":
-        return _write_dsf(path)
+    factories = {"wav": _write_wave, "dsf": _write_dsf, "aiff": _write_aiff, "dff": _write_dff}
+    if container in factories:
+        return factories[container](path)
     shutil.copyfile(Path(__file__).parent / f"fixtures/audio/silence.{container}", path)
     return path
 
 
-@pytest.mark.parametrize("container", ["wav", "dsf", "flac", "mp3", "m4a"])
+@pytest.mark.parametrize("container", ["wav", "dsf", "flac", "mp3", "m4a", "aiff", "dff", "wma"])
 @pytest.mark.parametrize("link_type", ["hardlink", "symlink"])
 @pytest.mark.parametrize("wrong_extension", [False, True])
 def test_native_container_writes_metadata_cover_and_release_ids_without_touching_seed(
@@ -76,6 +76,10 @@ def test_native_container_writes_metadata_cover_and_release_ids_without_touching
         assert native.pictures[0].data == _COVER
     elif isinstance(native, MP4):
         assert bytes(native.tags["covr"][0]) == _COVER
+    elif isinstance(native, ASF):
+        picture = native.tags["WM/Picture"][0].value
+        assert picture[:5] == bytes([3]) + len(_COVER).to_bytes(4, "little")
+        assert picture[5:] == "image/png\0\0".encode("utf-16-le") + _COVER
     else:
         assert isinstance(native.tags, ID3) and native.tags.getall("APIC")[0].data == _COVER