From 2ebb3c7ad1471634f69ef682421f92ac0bc2da2d Mon Sep 17 00:00:00 2001 From: jxxghp Date: Wed, 30 Sep 2026 13:19:30 +0800 Subject: [PATCH] fix(music): read and write native ASF metadata safely --- app/application/audio.py | 123 +++++++++++++++++--- docs/mcp-api.md | 8 +- skills/moviepilot-api/api/music.md | 8 ++ tests/fixtures/audio/README.md | 6 + tests/fixtures/audio/silence.wma | Bin 0 -> 56058 bytes tests/test_audio_asf.py | 156 ++++++++++++++++++++++++++ tests/test_audio_containers.py | 32 +++++- tests/test_music_classical_credits.py | 13 ++- tests/test_music_source_protection.py | 16 ++- 9 files changed, 332 insertions(+), 30 deletions(-) create mode 100644 tests/fixtures/audio/silence.wma create mode 100644 tests/test_audio_asf.py diff --git a/app/application/audio.py b/app/application/audio.py index 3958624e9..c00411bef 100644 --- a/app/application/audio.py +++ b/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(" 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 [] diff --git a/docs/mcp-api.md b/docs/mcp-api.md index 6d60f320e..9c928a639 100644 --- a/docs/mcp-api.md +++ b/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 或有损/无损。 diff --git a/skills/moviepilot-api/api/music.md b/skills/moviepilot-api/api/music.md index 632e7a7d9..77e57ad7b 100644 --- a/skills/moviepilot-api/api/music.md +++ b/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. diff --git a/tests/fixtures/audio/README.md b/tests/fixtures/audio/README.md index dd7d7d4f3..6759d8885 100644 --- a/tests/fixtures/audio/README.md +++ b/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 在测试中以标准库生成合法容器及静音数据，不新增二进制依赖或编码器要求。 diff --git a/tests/fixtures/audio/silence.wma b/tests/fixtures/audio/silence.wma new file mode 100644 index 0000000000000000000000000000000000000000..e69e37526bb04e9644338396195f7a7f818aae79 GIT binary patch literal 56058 zcmeI5U2GIp6vyw`#kLEmtROAeVg*DLC@mo3SJ_Rrp%`h18ci&vDFL;K0b{8(K2<|B zN-Z%GH3m%!6|p5j8wpa3m=GTdk$zB2Oo$pVEdoinpDTe+aWweyLg^!{&EuiA9_>DmHiTA{wDMEAMAjx`?*r8|FAUrsH5 zc*|3EVdTE+n;_I)efGOg-lKFdn;G1j-?Y zxnMtM6F*6}^yrzs0gnG<;{|Gkj$S%-`Kv?qZM8d|?Cr??@!-kX=DB>gyL4Xeo!ujA zyR&7}>uR2PWPC}cXLz*m!f5*KUsc!O#sllem%LGU?#PjUea5rFyJg;QzPzq}D1G3H zYH!?-TeSY`6UTQo4W)PXYo$ItUmiS7mFkHieKQHFCQ7K>L7iE(Qu)L*HQ7APE~9)R zpU{(;help5s~S%4>(PBl*8aXZc<E&aXyM+l zR~@?kxQ2UY#^qUz%cy^d$W(yL?q0+Fq-#-k%QP8TMHC=DzG2t*%fNj^9t4CcKzyXi za^D2*g_?}CA__3|F>AK|Xwe5g`1nVPduO)8c41sb>_b4P0%U$4VYvSS+zT}sX+;#E zDQR?`~0yNEWKL+llx{SQS3NZCCYqtL2Gd{Tf zFz1Ds{lH26)r9ju!UkeqqylsUk4nRR6}T5^F_Q8qK+_%fjo{uR5JDmqpc#()PH->M zVkG5JfT@pJv-L-pJ`s`6|G1HRXSTyW$GG$uh>%DHXr|--XK*jlVkG5JfQlXWv-B=u z>~HdDgpfQ1sKjw!3-0Cli;VmVF!eEOw*GirUl4+if6U_Enfl{7j7z_Vh{{ud%yl>o z_aA_JdHy0JzXEiVTi>4n_kN8Km8Ssl)spRY4*%o>b+CxIbvyUY)E^BPmmUKV z5~%>);kfSr_aZGuQXU0pk>h><+5jrNqH1t>SNYy{qbi|e=OnN znfjwxU!o5C!#o}$Bu@dVaNO5`dwKpMBfkQ))N#KZ-1{{`RGtD<>A2qu?&bN5jQk2P z^)YL<{y1D@kAGBg?@awMf^q5h5K(yw&@#vU-{4-JzsSh104;aiSLjQxV}FxhCPalP zKr0;g8^FCVbCJ%33NZCCYqtK_nzHAA+{wK&^~Y-%mlN8E1;P}dm5%$P;9i)yNM}L? z=q|^79^6kT6BYz#fT@pJv-QWq$+rHun|o*Kj~vFO-$O*@ zDM0r)?pwgUJb#gqUje$;ao-K@{Td-EPXW5maX$?1<@t+@{0cDjF>AK|IIrsuKL7lF z?wzSW{=&HQdx)q!1*qC_zgYhzH1>zdvltn96rk0P`+9Ki5eOlX3Q)#z|0=i_X)%)W zD8SUmtm*nA-S9$c&lK+0i 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=["晴天"]), diff --git a/tests/test_music_classical_credits.py b/tests/test_music_classical_credits.py index c4dc5d81e..8fd5768d2 100644 --- a/tests/test_music_classical_credits.py +++ b/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) diff --git a/tests/test_music_source_protection.py b/tests/test_music_source_protection.py index 68afda68a..77999a20b 100644 --- a/tests/test_music_source_protection.py +++ b/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