"""网易云音乐下载器（适配 ffmpeg 写入元数据） 提供功能： - 同步 / 异步下载（本地落盘 / 内存返回） - 批量并发下载 - 进度查询 - 基于 ffmpeg 的元数据 + 封面写入（MP3/FLAC/M4A/MP4/OGG/Opus） 设计要点： 1. ffmpeg 路径懒加载：每次写入前重新检测，支持服务启动后中途安装 2. 显式 -f 指定输出格式 + 临时文件保留原扩展名在最后，双保险 3. MP4（杜比全景声 EAC3）单独分支，启用 +faststart 4. 扩展名优先使用 API 返回的 type 字段，URL/Content-Type 兜底 """ from __future__ import annotations import contextlib import logging import os import re import shutil import subprocess import tempfile import threading import time from collections import deque from dataclasses import dataclass from enum import Enum from io import BytesIO from pathlib import Path from typing import Any, Dict, List, Optional, Tuple import requests from requests.adapters import HTTPAdapter from music_api import APIException, NeteaseAPI, load_cookies # P2-5：worker 线程的输出一律走 logger（坑 11——NSSM GBK 重定向下裸 print # 非可编码字符会 UnicodeEncodeError 崩掉 worker，尤其 ffmpeg stderr 原文回显） logger = logging.getLogger('music_downloader') # ============ HTTP 连接复用（提速核心） ============ # 常见浏览器 UA，避免直链 CDN 因空 UA 限速/拒绝 DEFAULT_UA = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" ) # 音频流式下载分块大小（256KB）：减少系统调用与磁盘 IO 次数，提升吞吐 DL_CHUNK_SIZE = 256 * 1024 # 线程本地 Session：同线程内复用 TCP/TLS 连接（keep-alive），批量并发显著提速 _SESSION_LOCAL = threading.local() # 第 97 条（P3）：下载内容嗅探的音频魔数（未知期望尺寸时唯一的内容防线） _AUDIO_MAGIC = (b'ID3', b'fLaC', b'OggS', b'RIFF') def _sniff_audio_header(path) -> bool: """读取文件头做基础音频魔数校验：MP3(ID3/裸帧)、FLAC、OGG/Opus、WAV(RIFF)、 M4A/MP4(ftyp)。未知期望尺寸的下载走"非空即通过"保留口径，本函数补上内容 防线——CDN 错误页/风控 JSON 不再能被存成音频交付。""" try: with open(path, 'rb') as f: head = f.read(16) except OSError: return False if not head: return False if any(head.startswith(m) for m in _AUDIO_MAGIC): return True if len(head) >= 8 and head[:4] == b'\x00\x00\x00' and head[4:8] == b'ftyp': return True return len(head) >= 2 and head[0] == 0xFF and (head[1] & 0xE0) == 0xE0 def get_http_session() -> requests.Session: """获取当前线程的共享 requests.Session（带连接池 + keep-alive）""" session: Optional[requests.Session] = getattr(_SESSION_LOCAL, "session", None) if session is None: session = requests.Session() adapter = HTTPAdapter(pool_connections=8, pool_maxsize=32, max_retries=0) session.mount("https://", adapter) session.mount("http://", adapter) session.headers.update({"User-Agent": DEFAULT_UA}) _SESSION_LOCAL.session = session return session # per-file 写锁：保证同一目标文件的 .part 临时文件同时只有一个写入者， # 防止并发下载同名歌曲（不同用户/任务）互相覆盖导致文件损坏 # F78：锁表带引用计数与回收——无人引用时从表中移除，防止长期运行无界增长 # A3：计数在注册表守卫锁内完成（"取出引用"即计数 +1），消除"取引用后、acquire 前 # 被另一个线程回收、第三个线程拿到新锁对象"的互斥破坏窗口 _FILE_LOCKS: Dict[str, "RefCountedFileLock"] = {} _FILE_LOCKS_GUARD = threading.Lock() class RefCountedFileLock: """带引用计数的文件锁：refs 表示"已持有或已登记即将持有"的数量，归零可回收""" def __init__(self): self._lock = threading.Lock() self.refs = 0 def acquire(self) -> None: self._lock.acquire() def release(self) -> None: self._lock.release() def locked(self) -> bool: return self._lock.locked() class _FileLockHandle: """兼容 `with get_file_lock(path):` 旧用法的句柄：语义与 file_lock_scope 一致""" def __init__(self, path: Path): self._path = path self._lock = None self._key = '' def __enter__(self) -> "RefCountedFileLock": key = str(self._path) with _FILE_LOCKS_GUARD: lock = _FILE_LOCKS.get(key) if lock is None: lock = RefCountedFileLock() _FILE_LOCKS[key] = lock lock.refs += 1 self._lock = lock self._key = key try: lock.acquire() except BaseException: # R5-17：acquire 被打断时回滚引用计数并摘除空表项，防锁表条目永久泄漏 with _FILE_LOCKS_GUARD: lock.refs -= 1 if lock.refs <= 0 and not lock.locked(): _FILE_LOCKS.pop(key, None) self._lock = None self._key = '' raise return lock def __exit__(self, exc_type, exc_val, exc_tb) -> None: if self._lock is None: return self._lock.release() with _FILE_LOCKS_GUARD: self._lock.refs -= 1 if self._lock.refs <= 0 and not self._lock.locked(): _FILE_LOCKS.pop(self._key, None) self._lock = None def get_file_lock(path: Path) -> "_FileLockHandle": """兼容入口：返回可用 with 语句的锁句柄（A3：引用计数在守卫锁内登记，无回收竞态）""" return _FileLockHandle(path) # 封面字节缓存：同一专辑封面只从网络拉取一次（ffmpeg 写元数据 与 ZIP 打包 共用） _COVER_CACHE: Dict[str, bytes] = {} _COVER_CACHE_ORDER: deque = deque(maxlen=128) _COVER_CACHE_LOCK = threading.Lock() def fetch_cover_bytes(pic_url: str) -> Optional[bytes]: """下载并缓存封面图片字节；失败返回 None（B3：字节预算 10MB，拒绝超大盘点阵/伪装文件）""" if not pic_url: return None with _COVER_CACHE_LOCK: cached = _COVER_CACHE.get(pic_url) if cached: return cached try: resp = get_http_session().get(pic_url, timeout=15, stream=True) try: resp.raise_for_status() try: declared = int(resp.headers.get('Content-Length') or 0) except ValueError: declared = 0 # 第 91 条（P3）：垃圾 Content-Length 不炸，按未知长度走流式 if declared > 10 * 1024 * 1024: return None # bytearray 就地追加（原 bytes += 是 O(n²) 拷贝，10MB 封面最坏多拷上百 MB） data = bytearray() for chunk in resp.iter_content(chunk_size=64 * 1024): data += chunk if len(data) > 10 * 1024 * 1024: return None if not data: return None data = bytes(data) finally: # 第 91 条（P3）：所有路径统一关闭响应（原 Content-Length 异常路径漏关） resp.close() except Exception: return None with _COVER_CACHE_LOCK: if pic_url not in _COVER_CACHE: # 简单 FIFO 淘汰，防止缓存无界增长 while len(_COVER_CACHE_ORDER) >= _COVER_CACHE_ORDER.maxlen: evict = _COVER_CACHE_ORDER.popleft() _COVER_CACHE.pop(evict, None) _COVER_CACHE[pic_url] = data _COVER_CACHE_ORDER.append(pic_url) return data # ============ 类型定义 ============ class QualityLevel(Enum): """音质等级枚举""" STANDARD = "standard" EXHIGH = "exhigh" LOSSLESS = "lossless" HIRES = "hires" SKY = "sky" JYEFFECT = "jyeffect" JYMASTER = "jymaster" DOLBY = "dolby" @dataclass class MusicInfo: """音乐信息数据类""" id: int name: str artists: str album: str pic_url: str duration: int track_number: int download_url: str file_type: str file_size: int quality: str lyric: str = "" tlyric: str = "" publish_year: Optional[int] = None @dataclass class DownloadResult: """下载结果数据类""" success: bool file_path: Optional[str] = None file_size: int = 0 error_message: str = "" music_info: Optional[MusicInfo] = None class DownloadException(Exception): """下载异常类""" pass class DownloadCanceled(DownloadException): """下载被用户取消""" pass class _RetryFromScratch(requests.RequestException): """内部信号：断点续传起点失效（HTTP 416 等），清除 .part 后从头重试""" pass class CountingSemaphore: """带活跃/等待计数的信号量（线程安全），用于实时展示排队情况 用法与 threading.BoundedSemaphore 一致（支持 with 语句）， 额外维护 active（正在执行）/ waiting（排队等待）两个计数， 供前端实时展示排队人数与预计等待时间。 """ def __init__(self, value: int = 1): self.capacity = max(1, int(value)) self._semaphore = threading.BoundedSemaphore(self.capacity) self._lock = threading.Lock() self.active = 0 self.waiting = 0 def acquire(self, blocking: bool = True) -> bool: if not blocking: if not self._semaphore.acquire(blocking=False): return False with self._lock: self.active += 1 return True with self._lock: self.waiting += 1 try: self._semaphore.acquire() except BaseException: with self._lock: self.waiting = max(0, self.waiting - 1) raise with self._lock: self.waiting = max(0, self.waiting - 1) self.active += 1 return True def acquire_cancellable(self, cancel_ev, poll: float = 1.0) -> bool: """可取消地等待名额（F91）：等待期间 cancel_ev 置位立即返回 False， 不再无期限阻塞线程与退款。返回 True 表示已获得名额。""" import time as _time with self._lock: self.waiting += 1 try: while True: if cancel_ev is not None and cancel_ev.is_set(): return False if self._semaphore.acquire(blocking=False): with self._lock: self.active += 1 return True _time.sleep(poll) finally: with self._lock: self.waiting = max(0, self.waiting - 1) def release(self) -> None: self._semaphore.release() with self._lock: self.active = max(0, self.active - 1) def __enter__(self) -> "CountingSemaphore": self.acquire() return self def __exit__(self, exc_type, exc_val, exc_tb) -> None: self.release() # ============ 工具映射表 ============ # API 返回的 type -> 文件扩展名 API_TYPE_TO_EXT = { "mp3": ".mp3", "flac": ".flac", "m4a": ".m4a", "mp4": ".mp4", "ogg": ".ogg", "opus": ".opus", } # 文件扩展名 -> ffmpeg 输出 muxer 名称 EXT_TO_FFMPEG_FORMAT = { ".mp3": "mp3", ".flac": "flac", ".m4a": "mp4", ".mp4": "mp4", ".ogg": "ogg", ".opus": "opus", } # 支持元数据写入的扩展名 METADATA_FORMATS = {".mp3", ".flac", ".m4a", ".mp4", ".ogg", ".opus"} # ============ 核心实现 ============ class MusicDownloader: """网易云音乐下载器 用法: downloader = MusicDownloader(download_dir="downloads") result = downloader.download_music_file(music_id, quality="lossless") """ def __init__( self, download_dir: str = "downloads", max_concurrent: int = 3, ffmpeg_path: Optional[str] = None, global_max_concurrent: int = 10, ): """ Args: download_dir: 下载目录 max_concurrent: 单个批量任务内的最大并发下载数 ffmpeg_path: ffmpeg 可执行文件路径，为 None 时启动时检测一次 global_max_concurrent: 整个服务同时下载的文件数（含 ffmpeg 元数据写入）。 所有请求共享该信号量，防止大量用户同时下载导致内存/CPU 耗尽。 """ self.download_dir = Path(download_dir) self.download_dir.mkdir(exist_ok=True) self.max_concurrent = max_concurrent self.api = NeteaseAPI() # 全局并发信号量：所有请求共用，限制同时进行的下载数 # （带计数，供前端实时展示排队情况） self._download_semaphore = CountingSemaphore(global_max_concurrent) # ffmpeg 元数据写入专用信号量（CPU 密集操作）： # 与下载分离限流 —— 元数据写满时只阻塞元数据，不占下载名额、不拖慢其它文件的传输 cpu_count = os.cpu_count() or 2 self._meta_semaphore = threading.BoundedSemaphore(max(1, min(4, cpu_count))) # ffmpeg 路径缓存：启动时检测一次，写入时还会再懒加载一次 self.ffmpeg_path = ffmpeg_path or shutil.which("ffmpeg") if not self.ffmpeg_path: logger.warning("未检测到 ffmpeg，写入元数据功能将不可用") # ----------- 工具方法 ----------- def _sanitize_filename(self, filename: str, max_bytes: int = 180) -> str: """清理文件名非法字符。 F79：换行/回车/制表等控制字符一律剔除（否则 send_file 头部 500）； F32：按 UTF-8 字节截断（Linux 文件系统 255 字节上限，中文按字符截 200 会超）， 且从 _build_file_path 传入时保证尾部 [ID 音质] 标记不被截掉。""" filename = re.sub(r'[\x00-\x1f\x7f]', '', filename) filename = re.sub(r'[<>:"/\\|?*]', '_', filename) filename = filename.strip(' .') # 按字节截断（在字符边界上回退，不产生半个字符） raw = filename.encode('utf-8') if len(raw) > max_bytes: cut = filename.encode('utf-8')[:max_bytes] # 回退到合法 UTF-8 边界 while cut and (cut[-1] & 0xC0) == 0x80: cut = cut[:-1] filename = cut.decode('utf-8', errors='ignore').rstrip(' .') return filename or "unknown" def _determine_file_extension( self, url: str, content_type: str = "", api_type: str = "", ) -> str: """确定文件扩展名 优先级: API 返回的 type > URL 后缀 > HTTP Content-Type > 默认 mp3 """ api_type = (api_type or "").lower() if api_type in API_TYPE_TO_EXT: return API_TYPE_TO_EXT[api_type] url_lower = url.lower() if ".flac" in url_lower: return ".flac" if ".mp3" in url_lower: return ".mp3" if ".m4a" in url_lower or ".mp4" in url_lower: return ".m4a" content_type = (content_type or "").lower() if "flac" in content_type: return ".flac" if "mpeg" in content_type or "mp3" in content_type: return ".mp3" if "mp4" in content_type or "m4a" in content_type: return ".m4a" return ".mp3" def _resolve_ffmpeg(self) -> Optional[str]: """懒加载解析 ffmpeg 路径 每次写入前调用一次：若启动时未检测到，但中途已安装，下一次就能用。 """ if self.ffmpeg_path: return self.ffmpeg_path path = shutil.which("ffmpeg") if path: self.ffmpeg_path = path # 回写缓存 return path @staticmethod def _unlink_quiet(path: Path) -> None: """安静地删除文件，忽略不存在 / 权限错误""" try: path.unlink() except OSError: pass # ----------- 信息获取 ----------- def get_music_info(self, music_id: int, quality: str = "standard", cookies: Optional[Dict[str, str]] = None) -> MusicInfo: """获取音乐详细信息（同步，URL/详情/歌词 3 个接口并行请求，加快单曲/批量启动）""" try: cookies = cookies if cookies is not None else load_cookies() # 3 个接口并行请求，单曲下载/批量预取都能明显提速 # F99：URL/详情拿到后立即 shutdown(cancel_futures)， # 已知失败不再被无关的歌词请求拖着等完才报错 from concurrent.futures import ThreadPoolExecutor _pool = ThreadPoolExecutor(max_workers=3) url_f = _pool.submit(self.api.get_song_url, music_id, quality, cookies) detail_f = _pool.submit(self.api.get_song_detail, music_id) lyric_f = _pool.submit(self.api.get_lyric, music_id, cookies) try: url_result = url_f.result() detail_result = detail_f.result() try: lyric_result = lyric_f.result() except Exception: lyric_result = None # 歌词接口抖动不阻断解析/下载 finally: try: _pool.shutdown(wait=False, cancel_futures=True) except TypeError: _pool.shutdown(wait=False) if not url_result.get("data") or not url_result["data"]: raise DownloadException(f"无法获取音乐ID {music_id} 的播放链接") # 第 91 条（P2-8）：上游 data 数组元素可能为 null（F94 同族） song_data = url_result["data"][0] or {} download_url = song_data.get("url", "") or "" if not download_url: raise DownloadException(f"音乐ID {music_id} 无可用的下载链接") if not detail_result.get("songs") or not detail_result["songs"]: raise DownloadException(f"无法获取音乐ID {music_id} 的详细信息") song_detail = detail_result["songs"][0] # F94：上游可选字段可能为 null（而非缺失），一律 (x or {}) / (x or []) 容错， # 可选展示信息异常不再让整首有效歌曲解析失败 lyric = ((lyric_result or {}).get("lrc") or {}).get("lyric", "") or "" tlyric = ((lyric_result or {}).get("tlyric") or {}).get("lyric", "") or "" artists = "/".join(filter(None, ((a or {}).get("name") for a in (song_detail.get("ar") or [])))) al = song_detail.get("al") or {} publish_time = al.get("publishTime", 0) or 0 publish_year = None if publish_time: import datetime try: publish_year = datetime.datetime.fromtimestamp(publish_time / 1000).year except (OSError, ValueError, OverflowError): publish_year = None return MusicInfo( id=music_id, name=song_detail.get("name") or "未知歌曲", artists=artists or "未知艺术家", album=al.get("name") or "未知专辑", pic_url=al.get("picUrl") or "", duration=(song_detail.get("dt") or 0) // 1000, track_number=song_detail.get("no", 0) or 0, download_url=download_url, file_type=(song_data.get("type") or "mp3").lower(), # 第 91 条（P2-8）：null 容错 file_size=song_data.get("size", 0) or 0, quality=quality, lyric=lyric, tlyric=tlyric, publish_year=publish_year, ) except APIException as e: raise DownloadException(f"API调用失败: {e}") except Exception as e: raise DownloadException(f"获取音乐信息时发生错误: {e}") # ----------- 文件路径 ----------- def _build_file_path(self, music_info: MusicInfo) -> Path: """根据 MusicInfo 生成下载文件路径 文件名附带歌曲 ID 与请求音质：不同歌曲同名（不同版本/重复收录）落盘互不冲突； 同一首歌不同音质也各自独立，避免低音质文件被当作高音质交付（调包）。 交付文件名由调用方另行决定，不受影响。 F32：先截断主名（按字节），再拼接 [ID 音质] 标记——标记永不丢失， 杜绝截断后两首不同 ID 的歌落到同一路径。 """ base = f"{music_info.artists} - {music_info.name}" tag = f" [{music_info.id} {music_info.quality}]" ext = self._determine_file_extension( music_info.download_url, api_type=music_info.file_type, ) # 第 97 条（P3）：按完整路径反算文件名可用长度——旧口径只按 Linux 255 字节 # 截断文件名，未启用 LongPathsEnabled 的 Windows 上超长歌名会在 open 时 # OSError 白白重试 3 轮。预算 = 240 - 目录前缀 - 扩展名 - ".part"等余量； # 中文名 UTF-8 字节数 ≥ 字符数，按字节截断天然保守。 # 目录过深时按比例压缩 tag 预算（对抗复审补：下限保底仍可能超 260 的 # 病理场景由部署侧控制 downloads 目录长度，见交接文档第 97 条备案） budget = 240 - len(str(self.download_dir)) - len(ext) - 8 if budget < 124: tag_budget = max(24, budget // 3) else: tag_budget = 60 safe_base = self._sanitize_filename(base, max_bytes=max(32, budget - tag_budget)) safe_tag = self._sanitize_filename(tag, max_bytes=tag_budget) return self.download_dir / f"{safe_base}{safe_tag}{ext}" # ----------- 下载（同步） ----------- def download_music_file( self, music_id: int, quality: str = "standard", write_meta: bool = True, meta_fields: Optional[List[str]] = None, cookies: Optional[Dict[str, str]] = None, retries: int = 2, cancel_event: Optional[threading.Event] = None, music_info: Optional[MusicInfo] = None, progress_cb=None, ) -> DownloadResult: """同步下载音乐到本地 Args: music_id: 音乐ID quality: 音质 write_meta: 是否写入元数据 meta_fields: 写入字段列表，None 表示全部 cookies: 用于获取直链/歌词的 Cookie，None 表示全局账号 retries: 网络/写入失败时的重试次数（默认 2 次） cancel_event: 取消事件，置位后立即中止下载（抛 DownloadCanceled） music_info: 已预取的歌曲信息（跳过重复 API 调用，用于批量加速） progress_cb: 可选进度回调 progress_cb(written_bytes, total_bytes)。 每写一个分块调用一次（含续传起点）；回调内部的异常会被吞掉， 绝不影响下载主流程。调用方自行限频。 说明: - 流式下载到 .part 临时文件，完整下载后再原子改名，半截文件不会被当作已下载 - 断点续传：失败保留 .part，重试/下次下载从已写字节续传（服务器支持 Range 时） - per-file 锁：同名歌曲并发下载互斥，防止 .part 被交叉写入损坏 - 连接复用：线程级 Session + keep-alive，避免每首歌重建 TCP/TLS 连接 - 网络传输与 ffmpeg 元数据写入分离限流，互不抢占对方资源 """ if music_info is None: music_info = self.get_music_info(music_id, quality, cookies) file_path = self._build_file_path(music_info) part_path0 = file_path.with_name(file_path.name + ".part") last_error = "" next_wait = 0.0 for attempt in range(retries + 1): # 第 91 条（P3）：重试退避睡眠在文件锁外执行——同名并发任务不再陪睡 if next_wait > 0: time.sleep(next_wait) next_wait = 0.0 # F117：重试轮次顶部的取消检查也要清理 .part（原实现在 try 外抛出会绕过清理） if cancel_event is not None and cancel_event.is_set(): self._unlink_quiet(part_path0) raise DownloadCanceled("下载已取消") # 同一目标文件的下载者串行化（不同文件之间仍然并行） with get_file_lock(file_path): try: # 其它线程可能刚好完成此文件，锁内复查 if file_path.exists(): # 缓存与本次期望不符（跨账号音质降级/上游变更都会让同 ID 同音质 # 的实际文件不同）：视为缓存 miss 删除重下，而不是把旧文件 # 当本次结果交付（原实现命中即返回、从不校验尺寸）。 # 第 91 条（P3）：期望未知（file_size=0）时至少拒绝 0 字节坏文件 st_size = file_path.stat().st_size cache_bad = ((music_info.file_size and st_size != music_info.file_size) or (not music_info.file_size and st_size == 0)) if cache_bad: self._unlink_quiet(file_path) else: # F31：缓存命中也尊重本次的元数据请求（重跑一次 ffmpeg，幂等） if write_meta: with self._meta_semaphore: self._write_metadata_with_ffmpeg(file_path, music_info, meta_fields) return DownloadResult( success=True, file_path=str(file_path), file_size=file_path.stat().st_size, music_info=music_info, ) part_path = file_path.with_name(file_path.name + ".part") resume_from = 0 if part_path.exists(): existing = part_path.stat().st_size if music_info.file_size > 0 and existing == music_info.file_size: # 上次已下完但未改名：直接转正（F30：转正后立即返回， # 不再发起网络请求——原实现离线时会把现成完整文件报为失败） # R5-15：改 >= 为 ==——超尺寸残片（上游多发货/坏分片）转正即交付坏文件 os.replace(part_path, file_path) if write_meta: with self._meta_semaphore: self._write_metadata_with_ffmpeg(file_path, music_info, meta_fields) return DownloadResult( success=True, file_path=str(file_path), file_size=file_path.stat().st_size, music_info=music_info, ) elif existing > 0: if music_info.file_size > 0 and existing > music_info.file_size: # R5-15：超尺寸残片没有续传价值（上游多发货/坏分片）， # 丢弃从头下——否则下轮被 >= 误转正交付坏文件 self._unlink_quiet(part_path) else: resume_from = existing # 断点续传起点 else: self._unlink_quiet(part_path) # 第 91 条（P2-7）：排队改用可取消等待（F91 的 acquire_cancellable）—— # 原阻塞式 with 使取消后仍要等全局 10 名额腾出，任务卡死、退款延迟 if not self._download_semaphore.acquire_cancellable(cancel_event, poll=0.5): raise DownloadCanceled("下载已取消") try: headers = {} mode = "wb" if resume_from > 0: headers["Range"] = f"bytes={resume_from}-" mode = "ab" if progress_cb is not None: try: progress_cb(resume_from, music_info.file_size or 0) except Exception: pass # 进度回调绝不影响下载主流程 resp: Optional[requests.Response] = None try: resp = get_http_session().get( music_info.download_url, stream=True, timeout=(10, 60), headers=headers, ) # 服务器忽略 Range 返回 200 → 从头重写而非追加 if resume_from > 0 and resp.status_code != 206: mode = "wb" # F29：206 必须从我们请求的起点开始；起点错位的分片拼起来是坏文件 if resume_from > 0 and resp.status_code == 206: cr = resp.headers.get('Content-Range') or '' m = re.match(r'bytes\s+(\d+)-', cr.strip()) if m and int(m.group(1)) != resume_from: self._unlink_quiet(part_path) raise _RetryFromScratch( f"Content-Range 起点错位（{m.group(1)} != {resume_from}），已从头重新下载") # R5-16：206 却解析不出起点（头被中间层剥离）——盲追加可能拼坏文件， # 与起点错位同口径处理：清残片从头下 if not m: self._unlink_quiet(part_path) raise _RetryFromScratch("206 响应缺少可用 Content-Range，已从头重新下载") resp.raise_for_status() # F20：下载体积硬上限（配置 500MB），超限即弃，防单文件拖垮磁盘 max_bytes = 500 * 1024 * 1024 written = resume_from if mode == 'ab' else 0 over_limit = False with open(part_path, mode) as f: for chunk in resp.iter_content(chunk_size=DL_CHUNK_SIZE): if cancel_event is not None and cancel_event.is_set(): raise DownloadCanceled("下载已取消") if chunk: f.write(chunk) written += len(chunk) if written > max_bytes: # C12：超限残片没有任何续传价值。第 91→93 条： # Windows 无法删除打开中的文件——置标志 break， # with 关闭后再清理（原实现在打开中 unlink 被静默吞掉） over_limit = True break if progress_cb is not None: try: progress_cb(written, music_info.file_size or 0) except Exception: pass # 进度回调绝不影响下载主流程 if over_limit: self._unlink_quiet(part_path) # 文件已关闭，清理可达 raise DownloadException( f"文件超过大小上限 {max_bytes // (1024*1024)}MB，已中止") except requests.HTTPError as e: # Range 起点无效（如 CDN 环节变化）：清除旧 part，从头重新下载 status = getattr(getattr(e, "response", None), "status_code", None) if status == 416: self._unlink_quiet(part_path) raise _RetryFromScratch("Range 起点失效，已从头重新下载") from e if status is not None and 400 <= status < 500 and status != 408: # 第 97 条（P3）：CDN 4xx（403 防盗链/429 风控等）是确定性 # 失败，对同一失效直链白重试只会加剧风控、拖慢失败返回 # （408 请求超时保留可重试） self._unlink_quiet(part_path) raise DownloadException( f"下载失败（HTTP {status}），下载链接可能已失效") from e raise finally: if resp is not None: resp.close() finally: # 第 91 条（P2-7）：与 acquire_cancellable 配对释放（原 with 语义） self._download_semaphore.release() # 完整性校验（F29：已知期望大小时必须精确相等；未知期望时至少非空） actual_size = part_path.stat().st_size if music_info.file_size > 0: complete = (actual_size == music_info.file_size) else: complete = (actual_size > 0) if not complete: # 保留 .part 以便断点续传；下次续传从现有字节继续 last_error = ( f"下载不完整: 期望 {music_info.file_size or '未知'} 字节, " f"实际 {actual_size} 字节" ) if attempt < retries: next_wait = 0.5 * (attempt + 1) # 锁外睡眠（第 91 条） continue # 第 97 条（P3）：转正前内容嗅探——未知期望尺寸（"非空即通过"口径） # 时 CDN 错误页/风控 JSON 曾可被存成音频交付 if music_info.file_size <= 0 and not _sniff_audio_header(part_path): self._unlink_quiet(part_path) raise DownloadException("下载内容校验失败（非音频数据）") os.replace(part_path, file_path) # 元数据写入移出下载信号量（CPU 操作不占网络名额）， # 但仍在文件锁内执行，保证其它请求看到的是已完成标记的最终文件 if write_meta: # 第 97 条（P3）：元数据段入口补取消检查点——已取消的任务不再 # 进入信号量排队/ffmpeg 等待（有界但可达 120s），收尾不再延迟 if cancel_event is not None and cancel_event.is_set(): raise DownloadCanceled("下载已取消") with self._meta_semaphore: self._write_metadata_with_ffmpeg(file_path, music_info, meta_fields) return DownloadResult( success=True, file_path=str(file_path), file_size=file_path.stat().st_size, music_info=music_info, ) except DownloadCanceled: # 用户明确放弃：清理半截临时文件 self._unlink_quiet(part_path) raise except DownloadException: # 第 94 条（P2-1）：确定性失败（如体积超上限）不重试——原落泛型 # handler 会删残片后从 0 重拉，3 轮白拉 ~1.5GB raise except _RetryFromScratch: # .part 已清理，直接进入下一轮从头下载 last_error = "续传起点失效，重新下载" if attempt < retries: next_wait = 0.5 * (attempt + 1) # 锁外睡眠（第 91 条） continue except (requests.RequestException, OSError) as e: # 网络错误：保留 .part 支持断点续传； # F117：但若任务已被取消，.part 没有任何续传价值，一并清理 if cancel_event is not None and cancel_event.is_set(): self._unlink_quiet(part_path) raise DownloadCanceled("下载已取消") from e last_error = f"下载请求失败: {e}" if attempt < retries: next_wait = 0.5 * (attempt + 1) # 锁外睡眠（第 91 条） continue except Exception as e: last_error = f"下载过程中发生错误: {e}" if attempt < retries: next_wait = 0.5 * (attempt + 1) # 锁外睡眠（第 91 条） continue # 最后一次尝试仍失败：若任务已取消按取消收尾（F117） if cancel_event is not None and cancel_event.is_set(): self._unlink_quiet(part_path0) raise DownloadCanceled("下载已取消") return DownloadResult(success=False, error_message=last_error or "下载失败") # ============ ffmpeg 元数据写入（核心） ============ # P3-42：download_music_to_memory / download_batch_async / get_download_progress # 三个方法已删除——全服务端与验证脚本零调用（CLI 遗留），且 to_memory 不接受 # cookies 参数、真被启用会有行为差异；如需内存下载/异步批量再按当前接口重写 def _write_metadata_with_ffmpeg( self, file_path: Path, music_info: MusicInfo, fields: Optional[List[str]] = None, ) -> bool: """使用 ffmpeg 写入音频元数据 处理: title/artist/album/track + 封面（如果是 mp3/flac/m4a/mp4） fields: 可选字段列表 ['title','artist','album','cover','lyrics','date']； None 表示全部写入。 Returns: 是否写入成功（失败不影响主流程） """ ffmpeg_path = self._resolve_ffmpeg() if not ffmpeg_path: logger.warning("跳过元数据写入: ffmpeg 不可用") return False file_ext = file_path.suffix.lower() if file_ext not in METADATA_FORMATS: logger.warning(f"跳过元数据写入: 不支持的格式 {file_ext}") return False if not file_path.exists(): logger.warning(f"跳过元数据写入: 文件不存在 {file_path}") return False # 临时输出文件：保留原扩展名在最后，便于 ffmpeg 识别 muxer # 例: 艾辰 - 错位时空.tagging.flac tmp_output = file_path.with_name( file_path.stem + ".tagging" + file_path.suffix ) # 临时封面文件 cover_tmp: Optional[str] = None cover_ok = False # F114：创建成功与否与"路径是否需要清理"分离管理 try: # 下载封面到临时文件（走共享缓存：与 ZIP 打包共用一次网络请求） include_cover = fields is None or "cover" in fields if include_cover and music_info.pic_url: cover_data = fetch_cover_bytes(music_info.pic_url) if cover_data: try: fd, cover_tmp = tempfile.mkstemp(suffix=".jpg", dir=str(self.download_dir)) os.close(fd) with open(cover_tmp, "wb") as cf: cf.write(cover_data) cover_ok = True except Exception as e: # cover_tmp 保持指向已创建的文件，finally 统一清理（F114： # 原实现置 None 会把已创建的零字节文件遗留在下载目录） logger.warning(f"封面写盘失败，跳过封面写入: {e}") else: logger.warning("封面下载失败，跳过封面写入") # 构造 ffmpeg 命令 # 顺序: global opts -> -i input1 -> -i input2 -> metadata -> mapping -> -f format -> output output_format = EXT_TO_FFMPEG_FORMAT.get(file_ext, "") cmd: List[str] = [ ffmpeg_path, "-y", "-loglevel", "error", "-i", str(file_path), ] use_cover = cover_ok # F35：OGG/Opus 的封面流 copy 不被支持（Unsupported codec id）， # 主动降级为"只写文本元数据"，不再让整个写入失败 if use_cover and file_ext in (".ogg", ".opus"): use_cover = False logger.info("OGG/Opus 不支持封面写入，仅写入文本元数据") if use_cover and cover_tmp: cmd += ["-i", cover_tmp] # 通用元数据（仅写入勾选的字段） meta = {} # F33：前端字段名为 name，与后端 title 统一映射 if fields is None or "title" in fields or "name" in fields: meta["title"] = music_info.name if fields is None or "artist" in fields: meta["artist"] = music_info.artists meta["album_artist"] = music_info.artists if fields is None or "album" in fields: meta["album"] = music_info.album if fields is None or "date" in fields: if music_info.publish_year: meta["date"] = str(music_info.publish_year) meta["year"] = str(music_info.publish_year) if fields is None or "lyrics" in fields: if music_info.lyric and music_info.lyric.strip(): # 去掉 LRC 时间标签，写入纯歌词 clean_lyric = re.sub( r"\[\d{1,2}:\d{1,2}([.:]\d+)?\]", "", music_info.lyric ).strip() if clean_lyric: meta["lyrics"] = clean_lyric if music_info.track_number > 0 and (fields is None or "track" in fields): # F33：只写当前曲号；没有真实专辑总数时不虚构 tracktotal meta["track"] = str(music_info.track_number) for k, v in meta.items(): if v: cmd += ["-metadata", f"{k}={v}"] # 各格式的 mapping / 编码选项 cmd += self._build_format_args(file_ext, has_cover=use_cover) # -f 必须放在所有 -i 之后（作为 output option）才生效 if output_format: cmd += ["-f", output_format] cmd.append(str(tmp_output)) # 执行 result = subprocess.run( cmd, capture_output=True, text=True, timeout=120, encoding="utf-8", errors="ignore", ) if result.returncode == 0 and tmp_output.exists(): # 第 97 条（P3）：os.replace 才是同卷原子替换——shutil.move 在目标已 # 存在时退化为 copy2+unlink，进程中断会留半截文件 os.replace(str(tmp_output), str(file_path)) return True # P2-5：ffmpeg stderr 原文走 logger（errors='ignore' 解码后的任意 Unicode # 在 GBK 重定向下裸 print 会崩 worker） logger.warning( f"ffmpeg 写入元数据失败: {(result.stderr or result.stdout or '').strip()}" ) return False except subprocess.TimeoutExpired: logger.warning("ffmpeg 执行超时") return False except FileNotFoundError: logger.warning(f"ffmpeg 可执行文件未找到: {ffmpeg_path}") return False except Exception as e: logger.warning(f"ffmpeg 写入元数据异常: {e}") return False finally: # 清理封面临时文件 if cover_tmp and os.path.exists(cover_tmp): try: os.unlink(cover_tmp) except OSError: pass # 清理可能残留的临时输出 if tmp_output.exists(): try: tmp_output.unlink() except OSError: pass def _build_format_args(self, file_ext: str, has_cover: bool) -> List[str]: """根据文件格式构造 mapping / 编码参数 F34：不写封面时用 -map 0 -c copy 保留全部既有流（含已有封面）， 原实现 -map 0:a 会把 MP3/M4A 已有封面静默删掉。""" if file_ext == ".mp3": # MP3: 封面通过 ID3v2 APIC 帧写入 if has_cover: return [ "-map", "0:a", "-map", "1:v", "-c:a", "copy", "-c:v", "copy", "-id3v2_version", "3", "-metadata:s:v", "title=Album cover", "-metadata:s:v", "comment=Cover (front)", ] return ["-map", "0", "-c", "copy", "-id3v2_version", "3"] if file_ext == ".flac": # FLAC: 原生支持封面 if has_cover: return [ "-map", "0", "-map", "1", "-c", "copy", "-disposition:v:0", "attached_pic", ] return ["-map", "0", "-c", "copy"] if file_ext == ".m4a": # M4A: AAC 音频 + mp4 容器 if has_cover: return [ "-map", "0:a", "-map", "1:v", "-c", "copy", "-disposition:v:0", "attached_pic", "-metadata:s:v", "title=Album cover", "-metadata:s:v", "comment=Cover (front)", ] # F34：保留既有封面流（原 -map 0:a 会删掉已有封面） return ["-map", "0", "-c", "copy"] if file_ext == ".mp4": # MP4 (杜比全景声 EAC3 / 其它 mp4 容器音频) # 杜比文件通常无视频流，封面作为 video stream 注入；+faststart 让 moov 在文件头 if has_cover: return [ "-map", "0", "-map", "1:v", "-c", "copy", "-disposition:v:0", "attached_pic", "-metadata:s:v", "title=Album cover", "-metadata:s:v", "comment=Cover (front)", "-movflags", "+faststart", ] return ["-map", "0", "-c", "copy", "-movflags", "+faststart"] if file_ext in (".ogg", ".opus"): if has_cover: return [ "-map", "0:a", "-map", "1:v", "-c:a", "copy", "-c:v", "copy", ] return ["-map", "0:a", "-c:a", "copy"] # 默认 if has_cover: return ["-map", "0:a", "-map", "1:v", "-c", "copy"] return ["-map", "0:a", "-c:a", "copy"] # ============ CLI 入口 ============ if __name__ == "__main__": import sys from dataclasses import asdict downloader = MusicDownloader() # 无参数：打印模块信息 if len(sys.argv) < 2: print("网易云音乐下载器") print(f"ffmpeg 路径: {downloader.ffmpeg_path or '未找到'}") print("支持的音质: standard / exhigh / lossless / hires / sky / jyeffect / jymaster / dolby") print("支持的功能: 同步/批量下载、ffmpeg 元数据写入") print() print("用法:") print(" python music_downloader.py [quality] # 仅打印信息") print(" python music_downloader.py [quality] --download # 打印并下载") print() print("示例:") print(" python music_downloader.py 123456") print(" python music_downloader.py 123456 lossless") print(" python music_downloader.py 123456 hires --download") sys.exit(0) # 解析参数 try: music_id = int(sys.argv[1]) except ValueError: print(f"错误: music_id 必须是整数，收到: {sys.argv[1]!r}") sys.exit(1) quality = sys.argv[2] if len(sys.argv) > 2 and not sys.argv[2].startswith("--") else "lossless" do_download = "--download" in sys.argv # 打印音乐信息 try: info = downloader.get_music_info(music_id, quality) except DownloadException as e: print(f"获取音乐信息失败: {e}") sys.exit(1) print() print("=" * 70) print(f"音乐信息 (ID={info.id}, quality={info.quality})") print("=" * 70) info_dict = asdict(info) for key, value in info_dict.items(): # 歌词太长，只显示行数 if key in ("lyric", "tlyric"): lines = [ln for ln in (value or "").splitlines() if ln.strip()] preview = " | ".join(lines[:2])[:80] if lines else "(空)" print(f" {key:<14}: {len(lines)} 行 预览: {preview}{'...' if len(lines) > 2 else ''}") continue # URL 类字段：完整展示（如果终端窄可换行） if key in ("download_url", "pic_url"): print(f" {key:<14}: {value}") continue print(f" {key:<14}: {value}") print("=" * 70) # 可选：触发下载 if do_download: print() print("开始下载...") result = downloader.download_music_file(music_id, quality) if result.success: print(f"下载完成: {result.file_path} ({result.file_size} 字节)") else: print(f"下载失败: {result.error_message}") sys.exit(1)