refactor: 下载/标签/格式探测改用成熟库（aria2 + mutagen + filetype）

为什么：原先三处都在手写轮子——ffmpeg CLI 拼参数写标签、aiohttp 单连接手搓下载、自己比对魔数判格式。手写部分既难测又易错；改走成熟库后测试从 144 涨到 175，并且彻底去掉了 ffmpeg 这个外部二进制依赖。下载 → aria2c 常驻 JSON-RPC（aria2p 客户端），新增 _aria2.py - 单例后端：首次调用时拉起 aria2c，只监听 127.0.0.1 且强制随机 secret （无鉴权的 aria2 RPC 允许本机任意进程操纵队列、读写任意文件） - split=5 / max-connection-per-server=5 多连接 + 原生断点续传 - 异常语义与内置通道逐一对齐：取消/超限清残片抛 DownloadCanceled / DownloadException，任务失败/超时保留残片抛 RequestException 供续传 - 取不到 aria2c 时 get_backend() 返回 None，自动降级到内置 aiohttp 通道（本机无 aria2c 的降级路径已实测）标签 → mutagen 纯 Python 写入，新增 tagging.py - 取代 _write_metadata_with_ffmpeg + _build_format_args（226 行 shell 参数拼接） - MP3(ID3v2.3)/FLAC/M4A/OGG/Opus 分容器实现，无子进程、不重编码 - 就地改帧：TXXX 等未触及的标签原样保留（ffmpeg 整帧重写会丢） - 临时文件 + os.replace 原子替换，写坏不会留下半个文件 - 为什么不是 ffmpeg-python（原方案）：实测它用 **kwargs 建模选项，无法表达重复键（-metadata ×N），list 值还会被字符串化成 "['0', '1:v']"；-map 尚有内置的 stream 位置参数入口，但 -metadata 无解，且 overwrite_output=True 生成的 -y 被排到输出文件之后、可能失效格式探测 → filetype，改 _http.py - 删除手写的 _AUDIO_MAGIC 魔数表，改用 filetype.guess() 判 MIME （audio/* 放行，video/mp4 作为容器显式放行）配套 - Dockerfile：镜像不再装 ffmpeg，改装 aria2 - pyproject：新增 filetype / aria2p / mutagen；dev 加 httpx2（消除测试里的 StarletteDeprecationWarning，现 0 warning） - __main__.py / __init__.py / models.py：清理 ffmpeg 相关导出与文案测试：新增 31 个用例（test_aria2.py 13、test_tagging.py 21，含最小合法 FLAC 样本构造）；全套 175 passed / 0 warning
2026-09-26 13:54:44 +08:00
1 parent 183f5744d8
commit ad8c30d932
@@ -0,0 +1,188 @@
+# -*- coding: utf-8 -*-
+"""标签写入（mutagen）与内容嗅探（filetype）的单元测试。
+
+覆盖：
+- build_tags 的字段勾选、name/title 别名、LRC 时间标签剔除、曲号/年份缺省
+- write_tags 的失败路径（不支持的扩展名 / 文件不存在）
+- MP3 与 FLAC 的真实往返（写入后按标签读回）
+- _sniff_audio_header 对音频容器放行、对 CDN 错误页/JSON/空文件拒绝
+"""
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from music_downloader import MusicInfo
+from music_downloader._http import _sniff_audio_header
+from music_downloader.tagging import TagSet, build_tags, write_tags
+
+# 最小合法 FLAC：fLaC 魔数 + 一个 34 字节的 STREAMINFO（last-block 标志置位）。
+# STREAMINFO 不能全零——mutagen 会拒绝采样率为 0 的流；此处填 44100Hz/2ch/16bit：
+#   min_blocksize=4096 max_blocksize=4096 min_framesize=1000 max_framesize=2000
+#   随后 8 字节打包 (sample_rate 20bit | channels-1 3bit | bps-1 5bit | total 36bit)
+_STREAMINFO = bytes.fromhex(
+    '1000' '1000' '0003e8' '0007d0' '0ac442f00000ac44' + '00' * 16)
+assert len(_STREAMINFO) == 34
+_MINIMAL_FLAC = b'fLaC' + bytes([0x80, 0x00, 0x00, 0x22]) + _STREAMINFO
+
+# 最小 JPEG 头（仅用于标签写入，不参与解码）
+_JPEG = b'\xff\xd8\xff\xe0' + b'\x00' * 32
+
+
+def _info(**kw) -> MusicInfo:
+    base: dict[str, Any] = {
+        'id': 1, 'name': '歌名', 'artists': '歌手', 'album': '专辑', 'pic_url': '',
+        'duration': 200, 'track_number': 3,
+        'download_url': 'http://example.invalid/x', 'file_type': 'mp3',
+        'file_size': 0, 'quality': 'standard',
+        'lyric': '[00:12.34]第一行\n[00:15.00]第二行', 'tlyric': '',
+        'publish_year': 2024,
+    }
+    base.update(kw)
+    return MusicInfo(**base)
+
+
+# ---------- build_tags ----------
+
+def test_build_tags_all_fields():
+    tags = build_tags(_info(), None)
+    assert tags.title == '歌名'
+    assert tags.artist == '歌手'
+    assert tags.album_artist == '歌手'
+    assert tags.album == '专辑'
+    assert tags.date == '2024'
+    assert tags.track == 3
+    assert tags.lyrics == '第一行\n第二行'      # LRC 时间标签已剔除
+
+
+def test_build_tags_respects_selection():
+    tags = build_tags(_info(), ['title'])
+    assert tags.title == '歌名'
+    assert tags.artist is None
+    assert tags.album is None
+    assert tags.lyrics is None
+    assert tags.track is None
+
+
+def test_build_tags_accepts_name_alias():
+    assert build_tags(_info(), ['name']).title == '歌名'
+
+
+@pytest.mark.parametrize('kw, field', [
+    ({'track_number': 0}, 'track'),
+    ({'publish_year': None}, 'date'),
+    ({'lyric': ''}, 'lyrics'),
+    ({'lyric': '[00:01.00]'}, 'lyrics'),        # 只有时间标签 → 清洗后为空
+])
+def test_build_tags_omits_empty_values(kw, field):
+    assert getattr(build_tags(_info(**kw), None), field) is None
+
+
+# ---------- write_tags ----------
+
+def test_write_tags_rejects_unsupported_extension(tmp_path):
+    p = tmp_path / 'a.wav'
+    p.write_bytes(b'RIFF' + b'\x00' * 8)
+    assert write_tags(p, TagSet(title='x')) is False
+
+
+def test_write_tags_returns_false_when_missing(tmp_path):
+    assert write_tags(tmp_path / 'nope.mp3', TagSet(title='x')) is False
+
+
+def test_write_tags_mp3_roundtrip(tmp_path):
+    from mutagen.id3 import ID3
+
+    p = tmp_path / 'a.mp3'
+    ID3().save(str(p))          # 构造一个只含 ID3 标签区的合法 mp3
+    tags = TagSet(title='T', artist='A', album='B', album_artist='A2',
+                  date='2020', track=2, lyrics='L')
+    assert write_tags(p, tags, cover=_JPEG) is True
+
+    got = ID3(str(p))
+    assert got['TIT2'].text[0] == 'T'
+    assert got['TPE1'].text[0] == 'A'
+    assert got['TPE2'].text[0] == 'A2'
+    assert got['TALB'].text[0] == 'B'
+    assert got['TRCK'].text[0] == '2'
+    # USLT/APIC 的 HashKey 带语言/描述后缀，须用 getall 取
+    assert got.getall('USLT')[0].text == 'L'
+    assert got.getall('APIC')[0].mime == 'image/jpeg'
+
+
+def test_write_tags_mp3_keeps_untouched_frames(tmp_path):
+    from mutagen.id3 import ID3, TXXX
+
+    p = tmp_path / 'a.mp3'
+    audio = ID3()
+    audio['TXXX:keep'] = TXXX(encoding=3, desc='keep', text=['v'])
+    audio.save(str(p))
+
+    assert write_tags(p, TagSet(title='T')) is True
+    assert ID3(str(p))['TXXX:keep'].text[0] == 'v'   # 未触及的帧被保留
+
+
+def test_write_tags_flac_roundtrip(tmp_path):
+    from mutagen.flac import FLAC
+
+    p = tmp_path / 'a.flac'
+    p.write_bytes(_MINIMAL_FLAC)
+    assert write_tags(p, TagSet(title='T', artist='A', track=4), cover=_JPEG) is True
+
+    got = FLAC(str(p))
+    assert got['title'][0] == 'T'
+    assert got['artist'][0] == 'A'
+    assert got['tracknumber'][0] == '4'
+    assert len(got.pictures) == 1
+    assert got.pictures[0].mime == 'image/jpeg'
+
+
+def test_write_tags_failure_is_swallowed(tmp_path):
+    """写坏文件不应抛异常（主流程靠返回值判断）。"""
+    p = tmp_path / 'broken.flac'
+    p.write_bytes(b'fLaC')          # 截断的 FLAC，解析必然失败
+    assert write_tags(p, TagSet(title='T')) is False
+
+
+# ---------- 内容嗅探 ----------
+
+def test_sniff_accepts_id3(tmp_path):
+    p = tmp_path / 'x.mp3'
+    p.write_bytes(b'ID3\x03\x00\x00\x00' + b'\x00' * 64)
+    assert _sniff_audio_header(p) is True
+
+
+def test_sniff_accepts_flac(tmp_path):
+    p = tmp_path / 'x.flac'
+    p.write_bytes(_MINIMAL_FLAC)
+    assert _sniff_audio_header(p) is True
+
+
+def test_sniff_accepts_ogg(tmp_path):
+    p = tmp_path / 'x.ogg'
+    p.write_bytes(b'OggS' + b'\x00' * 64)
+    assert _sniff_audio_header(p) is True
+
+
+def test_sniff_rejects_risk_json(tmp_path):
+    """CDN 风控 JSON 不能被当成音频交付。"""
+    p = tmp_path / 'x.mp3'
+    p.write_bytes(b'{"code": 403, "msg": "risk"}')
+    assert _sniff_audio_header(p) is False
+
+
+def test_sniff_rejects_html_error_page(tmp_path):
+    p = tmp_path / 'x.mp3'
+    p.write_bytes(b'<!DOCTYPE html><html><body>403</body></html>')
+    assert _sniff_audio_header(p) is False
+
+
+def test_sniff_rejects_empty_file(tmp_path):
+    p = tmp_path / 'x.mp3'
+    p.write_bytes(b'')
+    assert _sniff_audio_header(p) is False
+
+
+def test_sniff_rejects_missing_file(tmp_path):
+    assert _sniff_audio_header(tmp_path / 'nope.mp3') is False