fix(music): retain untagged audio evidence and read native ID3 tags

app/application/audio.py

@@ -5,7 +5,7 @@ from uuid import UUID
 from mutagen import File as MutagenFile
 from mutagen.apev2 import APEBinaryValue
 from mutagen.flac import FLAC, Picture
-from mutagen.id3 import APIC, SYLT, USLT
+from mutagen.id3 import APIC, ID3, SYLT, USLT
 from mutagen.monkeysaudio import MonkeysAudio
 from mutagen.mp4 import MP4, MP4Cover
 
@@ -52,15 +52,18 @@ class AudioMetadataHelper:
         except Exception as err:
             logger.warning(f"读取音频标签失败：{path} - {err}")
             return None
-        if not audio:
+        # FileType 的真值取决于标签数量；无标签的有效音频仍包含整专对位所需的时长。
+        if audio is None:
             return None
 
-        tags = audio.tags or {}
+        tags = cls._readable_tags(audio.tags)
         track_number, total_tracks = cls._number_pair(
-            cls._first_of(tags, "tracknumber", "track")
+            cls._first_of(tags, "tracknumber", "track"),
+            cls._first_of(tags, "tracktotal", "totaltracks"),
         )
         disc_number, total_discs = cls._number_pair(
-            cls._first_of(tags, "discnumber", "disc")
+            cls._first_of(tags, "discnumber", "disc"),
+            cls._first_of(tags, "disctotal", "totaldiscs"),
         )
         musicbrainz_id = cls._normalize_musicbrainz_id(
             cls._first_of(
@@ -92,6 +95,35 @@ class AudioMetadataHelper:
             media_id=musicbrainz_id,
         )
 
+    @staticmethod
+    def _readable_tags(tags: Any) -> Any:
+        """统一 WAV/DSF 的原生 ID3 与其它容器的 Easy 标签，不修改文件标签。
+
+        Mutagen 的 easy=True 只适用于提供 Easy 包装的容器；WAV/DSF 仍返回
+        ID3 帧。Recording 的身份只取专用字段，不能误用 Release Track ID。
+        """
+        if not isinstance(tags, ID3):
+            return tags or {}
+        fields = {
+            "title": "TIT2", "artist": "TPE1", "album": "TALB",
+            "albumartist": "TPE2", "date": "TDRC", "originaldate": "TDOR",
+            "tracknumber": "TRCK", "discnumber": "TPOS", "isrc": "TSRC",
+            "subtitle": "TIT3",
+        }
+        values = {
+            key: [str(value) for frame in tags.getall(frame_id) for value in frame.text]
+            for key, frame_id in fields.items()
+        }
+        for frame in tags.getall("TXXX"):
+            key = str(frame.desc).casefold().replace(" ", "_")
+            if not values.get(key):
+                values[key] = [str(value) for value in frame.text]
+        for frame in tags.getall("UFID"):
+            if frame.owner == "http://musicbrainz.org":
+                values["musicbrainz_trackid"] = [frame.data.decode("ascii", errors="replace")]
+                break
+        return values
+
     @classmethod
     def read_lyrics(cls, path: Path) -> Optional[MusicLyrics]:
         """读取常见音频容器中的逐行同步歌词、纯文本歌词和 Lyricsfile 标签。"""
@@ -365,15 +397,23 @@ class AudioMetadataHelper:
             return None
 
     @staticmethod
-    def _number_pair(value: Optional[str]) -> tuple[Optional[int], Optional[int]]:
-        """解析 track/disc 标签中的当前编号和总数。"""
-        if not value:
-            return None, None
-        parts = str(value).split("/", 1)
-        current = AudioMetadataHelper._optional_int(parts[0])
-        total = AudioMetadataHelper._optional_int(parts[1]) if len(parts) > 1 else None
+    def _number_pair(
+            value: Optional[str],
+            total_value: Optional[str] = None,
+    ) -> tuple[Optional[int], Optional[int]]:
+        """读取组合或独立曲数/碟数，优先有效组合总数并排除非正编号。"""
+        parts = str(value or "").split("/", 1)
+        current = AudioMetadataHelper._positive_int(parts[0])
+        total = AudioMetadataHelper._positive_int(parts[1]) if len(parts) > 1 else None
+        total = total or AudioMetadataHelper._positive_int(total_value)
         return current, total
 
+    @staticmethod
+    def _positive_int(value: Any) -> Optional[int]:
+        """曲序、碟号和总数必须为正数，零及无效文本不构成位置证据。"""
+        number = AudioMetadataHelper._optional_int(value)
+        return number if number is not None and number > 0 else None
+
     @staticmethod
     def _year(value: Optional[str]) -> Optional[int]:
         """从完整或不完整日期标签中提取四位年份。"""

tests/fixtures/audio/README.md

@@ -0,0 +1,8 @@
+# 音频容器测试样本
+
+`silence.flac` 是本地生成的三秒静音音频：44.1 kHz、16 bit、双声道，移除了音频标签和填充区。
+它不含第三方录音内容，用于离线验证 Mutagen 对真实无标签 FLAC 的流参数读取。
+测试先复制到临时目录，再写入标签，不修改此原始样本，也不要求 CI 安装 ffmpeg。
+
+生成过程：标准库 `wave` 写入三秒零值 PCM，通过 ffmpeg 的 `flac` 编码器转换，再使用
+`mutagen.flac.FLAC.clear()` 和 `save(padding=lambda _: 0)` 清除标签及填充。

tests/test_audio_containers.py

@@ -0,0 +1,145 @@
+"""通过真实音频容器覆盖空标签、原生 ID3 和独立音轨总数的读取契约。"""
+
+import hashlib
+import shutil
+import struct
+import wave
+from pathlib import Path
+
+import pytest
+from mutagen.dsf import DSF
+from mutagen.flac import FLAC
+from mutagen.id3 import TALB, TDRC, TIT2, TPE1, TPE2, TPOS, TRCK, TSRC, TXXX, UFID
+from mutagen.wave import WAVE
+
+from app.application.audio import AudioMetadataHelper
+
+RECORDING_ID = "38035858-f990-4fbb-b3b2-f2f8b958eeba"
+
+
+def _write_wave(path: Path) -> Path:
+    """生成三秒双声道 PCM，避免测试依赖本机 ffmpeg。"""
+    with wave.open(str(path), "wb") as stream:
+        stream.setnchannels(2)
+        stream.setsampwidth(2)
+        stream.setframerate(44100)
+        stream.writeframes(b"\0" * 44100 * 4 * 3)
+    return path
+
+
+def _write_dsf(path: Path) -> Path:
+    """生成带单个双声道 DSD 数据块的合法 DSF，供实际 ID3 读写使用。"""
+    payload = b"\x69" * 8192
+    header = struct.pack("<4sQQQ", b"DSD ", 28, 92 + len(payload), 0)
+    fmt = struct.pack("<4sQIIIIIIQII", b"fmt ", 52, 1, 0, 2, 2, 2822400, 1, 32768, 4096, 0)
+    path.write_bytes(header + fmt + struct.pack("<4sQ", b"data", 12 + len(payload)) + payload)
+    return path
+
+
+@pytest.fixture
+def flac_path(tmp_path):
+    """复制无标签的三秒静音 FLAC，所有标签写入仅作用于测试临时文件。"""
+    path = tmp_path / "03.flac"
+    shutil.copyfile(Path(__file__).parent / "fixtures/audio/silence.flac", path)
+    return path
+
+
+@pytest.mark.parametrize("container", ["wav", "flac"])
+def test_untagged_audio_keeps_stream_and_path_evidence(container, tmp_path, flac_path):
+    """没有标签仍能读取时长和音质，并用路径补齐曲序而不改动文件。"""
+    path = _write_wave(tmp_path / "03.wav") if container == "wav" else flac_path
+    before = hashlib.sha256(path.read_bytes()).digest()
+
+    merged, tags, filename = AudioMetadataHelper.read_evidence(path)
+
+    assert tags is not None
+    assert tags.title is None
+    assert tags.duration == merged.duration == 3
+    assert merged.sample_rate == 44100
+    assert merged.bit_depth == 16
+    assert merged.track_number == filename.track_number == 3
+    assert hashlib.sha256(path.read_bytes()).digest() == before
+
+
+@pytest.mark.parametrize("container", ["wav", "dsf"])
+def test_raw_id3_container_reads_track_identity(container, tmp_path):
+    """WAV/DSF 原生 ID3 帧能提供曲目字段及 Recording 身份。"""
+    path = tmp_path / f"03.{container}"
+    audio = WAVE(_write_wave(path)) if container == "wav" else DSF(_write_dsf(path))
+    audio.add_tags()
+    for frame in (
+        TIT2(encoding=3, text=["晴天"]),
+        TPE1(encoding=3, text=["周杰伦", "嘉宾"]),
+        TALB(encoding=3, text=["叶惠美"]),
+        TPE2(encoding=3, text=["周杰伦"]),
+        TDRC(encoding=3, text=["2003-07-31"]),
+        TRCK(encoding=3, text=["3/11"]),
+        TPOS(encoding=3, text=["1/2"]),
+        TSRC(encoding=3, text=["USQX91300105"]),
+        TXXX(encoding=3, desc="VERSION", text=["Live"]),
+        UFID(owner="http://musicbrainz.org", data=RECORDING_ID.encode("ascii")),
+    ):
+        audio.tags.add(frame)
+    audio.save()
+    before = path.read_bytes()
+
+    meta = AudioMetadataHelper.read_tags(path)
+
+    assert meta.title == "晴天"
+    assert meta.artists == ["周杰伦", "嘉宾"]
+    assert meta.album == "叶惠美"
+    assert meta.album_artist == "周杰伦"
+    assert meta.year == 2003
+    assert (meta.track_number, meta.total_tracks) == (3, 11)
+    assert (meta.disc_number, meta.total_discs) == (1, 2)
+    assert meta.isrc == "USQX91300105"
+    assert meta.version == "Live"
+    assert meta.media_id == RECORDING_ID
+    assert meta.media_source == "musicbrainz"
+    assert path.read_bytes() == before
+
+
+@pytest.mark.parametrize("track_total,disc_total", [
+    ("TRACKTOTAL", "DISCTOTAL"), ("TOTALTRACKS", "TOTALDISCS"),
+])
+def test_flac_reads_separate_track_and_disc_totals(flac_path, track_total, disc_total):
+    """Vorbis 的独立总数字段与组合编号拥有同等语义。"""
+    audio = FLAC(flac_path)
+    audio.update({"title": ["晴天"], "tracknumber": ["3"], "discnumber": ["1"],
+                  track_total: ["11"], disc_total: ["2"]})
+    audio.save()
+
+    meta = AudioMetadataHelper.read_tags(flac_path)
+
+    assert (meta.track_number, meta.total_tracks) == (3, 11)
+    assert (meta.disc_number, meta.total_discs) == (1, 2)
+
+
+@pytest.mark.parametrize("position,expected", [("3/11", (3, 11)), ("3/0", (3, 12)),
+                                                ("0/0", (None, 12)), ("bad", (None, 12))])
+def test_combined_position_prefers_valid_total(flac_path, position, expected):
+    """有效组合总数优先，零或坏编号不能污染真实的独立总数。"""
+    audio = FLAC(flac_path)
+    audio.update({"tracknumber": [position], "tracktotal": ["12"]})
+    audio.save()
+
+    meta = AudioMetadataHelper.read_tags(flac_path)
+
+    assert (meta.track_number, meta.total_tracks) == expected
+
+
+def test_id3_release_track_id_is_not_recording_identity(tmp_path):
+    """发行曲目 ID 不能被当作 Recording ID 传入指纹和单曲识别。"""
+    path = _write_wave(tmp_path / "03.wav")
+    audio = WAVE(path)
+    audio.add_tags()
+    audio.tags.add(TIT2(encoding=3, text=["晴天"]))
+    audio.tags.add(TXXX(encoding=3, desc="MusicBrainz Release Track Id", text=[RECORDING_ID]))
+    audio.tags.add(UFID(owner="http://musicbrainz.org", data=b"\xffinvalid"))
+    audio.save()
+
+    meta = AudioMetadataHelper.read_tags(path)
+
+    assert meta.title == "晴天"
+    assert meta.media_id is None
+    assert meta.media_source is None