This repository has been archived on 2026-10-02. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files
NeteaseUrl/tests/test_tagging.py
T
snowy-yang ad8c30d932 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

189 lines
6.1 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- 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