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 000000000..e69e37526 Binary files /dev/null and b/tests/fixtures/audio/silence.wma differ diff --git a/tests/test_audio_asf.py b/tests/test_audio_asf.py new file mode 100644 index 000000000..ebd2ad33d --- /dev/null +++ b/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")) diff --git a/tests/test_audio_containers.py b/tests/test_audio_containers.py index 3c1509fc5..bcb6c18a6 100644 --- a/tests/test_audio_containers.py +++ b/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=["晴天"]), 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