From 3b3383cde596a8828196ea91f62be8f701c305c2 Mon Sep 17 00:00:00 2001 From: Suxiaoqinx <3074193836@qq.com> Date: Sun, 21 Jun 2026 04:03:28 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E6=9D=9C=E6=AF=94?= =?UTF-8?q?=E5=85=A8=E6=99=AF=E5=A3=B0=E6=94=AF=E6=8C=81=EF=BC=8C=E9=87=8D?= =?UTF-8?q?=E6=9E=84Cookie=E7=AE=A1=E7=90=86=E4=B8=8E=E9=A1=B9=E7=9B=AE?= =?UTF-8?q?=E9=85=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 添加前端页面的杜比全景声音质选择选项，更新后端音质参数验证列表 - 适配杜比全景声的编码格式，将dolby音质的encodeType设为mp4 - 重构Cookie管理模块：移除独立的cookie_manager.py，将Cookie读写、解析逻辑迁移至music_api.py，简化qr_login工具实现 - 更新Dockerfile，添加ffmpeg依赖以支持音频转码处理 - 更新requirements.txt依赖配置 - 更新使用文档，调整项目文件结构说明 - 优化项目依赖，移除冗余的cookie_manager模块依赖 --- Dockerfile | 1 + cookie.txt | 2 +- cookie_manager.py | 469 --------------------- main.py | 40 +- music_api.py | 55 ++- music_downloader.py | 984 ++++++++++++++++++++++++------------------- qr_login.py | 394 +++++++++-------- requirements.txt | Bin 948 -> 430 bytes templates/index.html | 1 + 使用文档.md | 5 +- 10 files changed, 823 insertions(+), 1128 deletions(-) delete mode 100644 cookie_manager.py diff --git a/Dockerfile b/Dockerfile index a5c7f04..d72206f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,5 +1,6 @@ FROM python:3.9.22-alpine3.21 WORKDIR /app +RUN apk add --no-cache ffmpeg COPY requirements.txt requirements.txt RUN pip3 config set global.index-url https://mirrors.aliyun.com/pypi/simple/ && \ pip3 install --no-cache-dir -r requirements.txt diff --git a/cookie.txt b/cookie.txt index 51aec91..a475f38 100644 --- a/cookie.txt +++ b/cookie.txt @@ -1 +1 @@ -MUSIC_U=1eb9ce22024bb666e99b6743b2222f29ef64a9e88fda0fd5754714b900a5d70d993166e004087dd3b95085f6a85b059f5e9aba41e3f2646e3cebdbec0317df58c119e5;os=pc;appver=8.9.75; +MUSIC_U=00C09F150B1C965F6D668A94D0FBC54690D76D5A2043282656875EAE3F0B99B667F736421093E1B4AF4E3DF93C69F7D40C582783F8A1CCA17D1B701944878BBC29AAE08CDF8017734DF9D12275011772CDA37A08D560D2910FD3367F28E5FE7F69BBD339DDF371212A28E7FDCBD48758B2F13D9D4A96BFED34C4458755B59342895B8C92E0A28984445BC2A64F35B6EA550B280CD20F87BC10636369C0F3CB71BC68B3A55542334B314BF90FA460B4032B2F298CAA4522E445D665B8A498FD4A0CC9FCD3E8AD0A775609B56C7772E957000419B67B74480CFCCE61E78AAEB8E851E57B8F878C4E76095F6ACBB3247788DBF594265E30E2E668102EB9160299921775D563E2EC7A40D29E58482B6AC2A728980F76632571D5AABCA86169E79EA9BBCC054E5A2CE26FBC3723086EC5D3F9BC081BBB35A17B796283E8712AA64F3B2CA952F76A0C1CA2CD8F6A524B05142175CF2542D2214341A0B328566888F79130A4074D82D2E60C551A3582275ABBDEE9059C5240E5CD2C1703ACABA4D15DAF466D0E8A8604CA4A11A4496D7630478FB0E662E835A6D2A6B3EB2596191E3E5CCF;appver=8.9.75; \ No newline at end of file diff --git a/cookie_manager.py b/cookie_manager.py deleted file mode 100644 index 9275b00..0000000 --- a/cookie_manager.py +++ /dev/null @@ -1,469 +0,0 @@ -"""Cookie管理器模块 - -提供网易云音乐Cookie管理功能，包括： -- Cookie文件读取和写入 -- Cookie格式验证和解析 -- Cookie有效性检查 -- 自动过期处理 -""" - -import os -import json -import time -from typing import Dict, Optional, List, Tuple, Any -from pathlib import Path -from dataclasses import dataclass -from datetime import datetime, timedelta -import logging - - -@dataclass -class CookieInfo: - """Cookie信息数据类""" - name: str - value: str - domain: str = "" - path: str = "/" - expires: Optional[int] = None - secure: bool = False - http_only: bool = False - - -class CookieException(Exception): - """Cookie相关异常类""" - pass - - -class CookieManager: - """Cookie管理器主类""" - - def __init__(self, cookie_file: str = "cookie.txt"): - """ - 初始化Cookie管理器 - - Args: - cookie_file: Cookie文件路径 - """ - self.cookie_file = Path(cookie_file) - self.logger = logging.getLogger(__name__) - - # 网易云音乐相关的重要Cookie字段 - self.important_cookies = { - 'MUSIC_U', # 用户标识 - 'MUSIC_A', # 用户认证 - '__csrf', # CSRF令牌 - 'NMTID', # 设备标识 - 'WEVNSM', # 会话管理 - 'WNMCID', # 客户端标识 - } - - # 确保cookie文件存在 - self._ensure_cookie_file_exists() - - def _ensure_cookie_file_exists(self) -> None: - """确保Cookie文件存在""" - if not self.cookie_file.exists(): - self.cookie_file.touch() - self.logger.info(f"创建Cookie文件: {self.cookie_file}") - - def read_cookie(self) -> str: - """读取Cookie文件内容 - - Returns: - Cookie字符串内容 - - Raises: - CookieException: 读取失败时抛出 - """ - try: - if not self.cookie_file.exists(): - self.logger.warning(f"Cookie文件不存在: {self.cookie_file}") - return "" - - content = self.cookie_file.read_text(encoding='utf-8').strip() - - if not content: - self.logger.warning("Cookie文件为空") - return "" - - self.logger.debug(f"成功读取Cookie文件，长度: {len(content)}") - return content - - except UnicodeDecodeError as e: - raise CookieException(f"Cookie文件编码错误: {e}") - except PermissionError as e: - raise CookieException(f"没有权限读取Cookie文件: {e}") - except Exception as e: - raise CookieException(f"读取Cookie文件失败: {e}") - - def write_cookie(self, cookie_content: str) -> bool: - """写入Cookie到文件 - - Args: - cookie_content: Cookie内容字符串 - - Returns: - 是否写入成功 - - Raises: - CookieException: 写入失败时抛出 - """ - try: - if not cookie_content or not cookie_content.strip(): - raise CookieException("Cookie内容不能为空") - - # 验证Cookie格式 - if not self.validate_cookie_format(cookie_content): - raise CookieException("Cookie格式无效") - - # 写入文件 - self.cookie_file.write_text(cookie_content.strip(), encoding='utf-8') - - self.logger.info(f"成功写入Cookie到文件: {self.cookie_file}") - return True - - except PermissionError as e: - raise CookieException(f"没有权限写入Cookie文件: {e}") - except Exception as e: - raise CookieException(f"写入Cookie文件失败: {e}") - - def parse_cookies(self) -> Dict[str, str]: - """解析Cookie字符串为字典 - - Returns: - Cookie字典 - - Raises: - CookieException: 解析失败时抛出 - """ - try: - cookie_content = self.read_cookie() - if not cookie_content: - return {} - - return self.parse_cookie_string(cookie_content) - - except Exception as e: - raise CookieException(f"解析Cookie失败: {e}") - - def parse_cookie_string(self, cookie_string: str) -> Dict[str, str]: - """解析Cookie字符串 - - Args: - cookie_string: Cookie字符串 - - Returns: - Cookie字典 - """ - if not cookie_string or not cookie_string.strip(): - return {} - - cookies = {} - - try: - # 处理多种Cookie格式 - cookie_string = cookie_string.strip() - - # 分割Cookie项 - cookie_pairs = [] - if ';' in cookie_string: - cookie_pairs = cookie_string.split(';') - elif '\n' in cookie_string: - cookie_pairs = cookie_string.split('\n') - else: - cookie_pairs = [cookie_string] - - for pair in cookie_pairs: - pair = pair.strip() - if not pair or '=' not in pair: - continue - - # 分割键值对 - key, value = pair.split('=', 1) - key = key.strip() - value = value.strip() - - if key and value: - cookies[key] = value - - self.logger.debug(f"解析得到 {len(cookies)} 个Cookie项") - return cookies - - except Exception as e: - self.logger.error(f"解析Cookie字符串失败: {e}") - return {} - - def validate_cookie_format(self, cookie_string: str) -> bool: - """验证Cookie格式是否有效 - - Args: - cookie_string: Cookie字符串 - - Returns: - 是否格式有效 - """ - if not cookie_string or not cookie_string.strip(): - return False - - try: - # 尝试解析Cookie - cookies = self.parse_cookie_string(cookie_string) - - # 检查是否至少包含一个有效的Cookie - if not cookies: - return False - - # 检查Cookie名称是否合法 - for name, value in cookies.items(): - if not name or not isinstance(name, str): - return False - if not isinstance(value, str): - return False - # 检查是否包含非法字符 - if any(char in name for char in [' ', '\t', '\n', '\r', ';', ',']): - return False - - return True - - except Exception: - return False - - def is_cookie_valid(self) -> bool: - """检查Cookie是否有效 - - Returns: - Cookie是否有效 - """ - try: - cookies = self.parse_cookies() - - if not cookies: - self.logger.warning("Cookie为空") - return False - - # 检查重要Cookie是否存在 - missing_cookies = self.important_cookies - set(cookies.keys()) - if missing_cookies: - self.logger.warning(f"缺少重要Cookie: {missing_cookies}") - return False - - # 检查MUSIC_U是否有效（基本验证） - music_u = cookies.get('MUSIC_U', '') - if not music_u or len(music_u) < 10: - self.logger.warning("MUSIC_U Cookie无效") - return False - - self.logger.debug("Cookie验证通过") - return True - - except Exception as e: - self.logger.error(f"Cookie验证失败: {e}") - return False - - def get_cookie_info(self) -> Dict[str, Any]: - """获取Cookie详细信息 - - Returns: - 包含Cookie信息的字典 - """ - try: - cookies = self.parse_cookies() - - info = { - 'file_path': str(self.cookie_file), - 'file_exists': self.cookie_file.exists(), - 'file_size': self.cookie_file.stat().st_size if self.cookie_file.exists() else 0, - 'cookie_count': len(cookies), - 'is_valid': self.is_cookie_valid(), - 'important_cookies_present': list(self.important_cookies & set(cookies.keys())), - 'missing_important_cookies': list(self.important_cookies - set(cookies.keys())), - 'all_cookie_names': list(cookies.keys()) - } - - # 添加文件修改时间 - if self.cookie_file.exists(): - mtime = self.cookie_file.stat().st_mtime - info['last_modified'] = datetime.fromtimestamp(mtime).isoformat() - - return info - - except Exception as e: - return { - 'error': str(e), - 'file_path': str(self.cookie_file), - 'file_exists': False, - 'is_valid': False - } - - def backup_cookie(self, backup_suffix: str = None) -> str: - """备份Cookie文件 - - Args: - backup_suffix: 备份文件后缀，默认使用时间戳 - - Returns: - 备份文件路径 - - Raises: - CookieException: 备份失败时抛出 - """ - try: - if not self.cookie_file.exists(): - raise CookieException("Cookie文件不存在，无法备份") - - if backup_suffix is None: - backup_suffix = datetime.now().strftime("%Y%m%d_%H%M%S") - - backup_path = self.cookie_file.with_suffix(f".{backup_suffix}.bak") - - # 复制文件内容 - content = self.cookie_file.read_text(encoding='utf-8') - backup_path.write_text(content, encoding='utf-8') - - self.logger.info(f"Cookie备份成功: {backup_path}") - return str(backup_path) - - except Exception as e: - raise CookieException(f"备份Cookie文件失败: {e}") - - def restore_cookie(self, backup_path: str) -> bool: - """从备份恢复Cookie - - Args: - backup_path: 备份文件路径 - - Returns: - 是否恢复成功 - - Raises: - CookieException: 恢复失败时抛出 - """ - try: - backup_file = Path(backup_path) - if not backup_file.exists(): - raise CookieException(f"备份文件不存在: {backup_path}") - - # 读取备份内容 - backup_content = backup_file.read_text(encoding='utf-8') - - # 验证备份内容 - if not self.validate_cookie_format(backup_content): - raise CookieException("备份文件中的Cookie格式无效") - - # 写入当前Cookie文件 - self.write_cookie(backup_content) - - self.logger.info(f"从备份恢复Cookie成功: {backup_path}") - return True - - except Exception as e: - raise CookieException(f"恢复Cookie失败: {e}") - - def clear_cookie(self) -> bool: - """清空Cookie文件 - - Returns: - 是否清空成功 - """ - try: - if self.cookie_file.exists(): - self.cookie_file.write_text("", encoding='utf-8') - self.logger.info("Cookie文件已清空") - return True - - except Exception as e: - self.logger.error(f"清空Cookie文件失败: {e}") - return False - - def update_cookie(self, new_cookies: Dict[str, str]) -> bool: - """更新Cookie - - Args: - new_cookies: 新的Cookie字典 - - Returns: - 是否更新成功 - """ - try: - if not new_cookies: - raise CookieException("新Cookie不能为空") - - # 读取现有Cookie - existing_cookies = self.parse_cookies() - - # 合并Cookie - existing_cookies.update(new_cookies) - - # 转换为Cookie字符串 - cookie_string = '; '.join(f"{k}={v}" for k, v in existing_cookies.items()) - - # 写入文件 - return self.write_cookie(cookie_string) - - except Exception as e: - self.logger.error(f"更新Cookie失败: {e}") - return False - - def get_cookie_for_request(self) -> Dict[str, str]: - """获取用于HTTP请求的Cookie字典 - - Returns: - 适用于requests库的Cookie字典 - """ - try: - cookies = self.parse_cookies() - - # 过滤掉空值 - filtered_cookies = {k: v for k, v in cookies.items() if k and v} - - return filtered_cookies - - except Exception as e: - self.logger.error(f"获取请求Cookie失败: {e}") - return {} - - def format_cookie_string(self, cookies: Dict[str, str]) -> str: - """将Cookie字典格式化为字符串 - - Args: - cookies: Cookie字典 - - Returns: - Cookie字符串 - """ - if not cookies: - return "" - - return '; '.join(f"{k}={v}" for k, v in cookies.items() if k and v) - - def __str__(self) -> str: - """字符串表示""" - info = self.get_cookie_info() - return f"CookieManager(file={info['file_path']}, valid={info['is_valid']}, count={info['cookie_count']})" - - def __repr__(self) -> str: - """详细字符串表示""" - return self.__str__() - - -if __name__ == "__main__": - # 测试代码 - import sys - - # 配置日志 - logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') - - manager = CookieManager() - - print("Cookie管理器模块") - print("支持的功能:") - print("- Cookie文件读写") - print("- Cookie格式验证") - print("- Cookie有效性检查") - print("- Cookie备份和恢复") - print("- Cookie信息查看") - - # 显示当前Cookie信息 - info = manager.get_cookie_info() - print(f"\n当前Cookie状态: {manager}") - print(f"详细信息: {info}") diff --git a/main.py b/main.py index 96e6f0c..024901f 100644 --- a/main.py +++ b/main.py @@ -21,10 +21,10 @@ from flask import Flask, request, send_file, render_template, Response try: from music_api import ( NeteaseAPI, APIException, QualityLevel, - url_v1, name_v1, lyric_v1, search_music, - playlist_detail, album_detail + url_v1, name_v1, lyric_v1, search_music, + playlist_detail, album_detail, + load_cookies, ) - from cookie_manager import CookieManager, CookieException from music_downloader import MusicDownloader, DownloadException, AudioFormat except ImportError as e: print(f"导入模块失败: {e}") @@ -79,14 +79,13 @@ class MusicAPIService: def __init__(self, config: APIConfig): self.config = config self.logger = self._setup_logger() - self.cookie_manager = CookieManager() self.netease_api = NeteaseAPI() self.downloader = MusicDownloader() - + # 创建下载目录 self.downloads_path = Path(config.downloads_dir) self.downloads_path.mkdir(exist_ok=True) - + self.logger.info(f"音乐API服务初始化完成，下载目录: {self.downloads_path.absolute()}") def _setup_logger(self) -> logging.Logger: @@ -117,15 +116,14 @@ class MusicAPIService: return logger def _get_cookies(self) -> Dict[str, str]: - """获取Cookie""" + """直接从 cookie.txt 读取并解析为 dict""" try: - cookie_str = self.cookie_manager.read_cookie() - return self.cookie_manager.parse_cookie_string(cookie_str) - except CookieException as e: - self.logger.warning(f"获取Cookie失败: {e}") - return {} + cookies = load_cookies() + if not cookies: + self.logger.warning("cookie.txt 为空或不存在，部分接口（VIP 音质）将不可用") + return cookies except Exception as e: - self.logger.error(f"Cookie处理异常: {e}") + self.logger.error(f"读取 cookie.txt 异常: {e}") return {} def _extract_music_id(self, id_or_url: str) -> str: @@ -260,17 +258,19 @@ def index() -> str: def health_check(): """健康检查API""" try: - # 检查Cookie状态 - cookie_status = api_service.cookie_manager.is_cookie_valid() - + # 检查 Cookie 状态：直接读 cookie.txt + cookies = load_cookies() + cookie_status = 'valid' if cookies else 'invalid' + health_info = { 'service': 'running', 'timestamp': int(time.time()) if 'time' in sys.modules else None, - 'cookie_status': 'valid' if cookie_status else 'invalid', + 'cookie_status': cookie_status, + 'cookie_count': len(cookies), 'downloads_dir': str(api_service.downloads_path.absolute()), 'version': '2.0.0' } - + return APIResponse.success(health_info, "API服务运行正常") except Exception as e: @@ -298,7 +298,7 @@ def get_song_info(): music_id = api_service._extract_music_id(song_ids or url) # 验证音质参数 - valid_levels = ['standard', 'exhigh', 'lossless', 'hires', 'sky', 'jyeffect', 'jymaster'] + valid_levels = ['standard', 'exhigh', 'lossless', 'hires', 'sky', 'jyeffect', 'jymaster', 'dolby'] if level not in valid_levels: return APIResponse.error(f"无效的音质参数，支持: {', '.join(valid_levels)}") @@ -500,7 +500,7 @@ def download_music_api(): return validation_error # 验证音质参数 - valid_qualities = ['standard', 'exhigh', 'lossless', 'hires', 'sky', 'jyeffect', 'jymaster'] + valid_qualities = ['standard', 'exhigh', 'lossless', 'hires', 'sky', 'jyeffect', 'jymaster', 'dolby'] if quality not in valid_qualities: return APIResponse.error(f"无效的音质参数，支持: {', '.join(valid_qualities)}") diff --git a/music_api.py b/music_api.py index 20cef52..2ac1c30 100644 --- a/music_api.py +++ b/music_api.py @@ -10,6 +10,7 @@ """ import json +import os import urllib.parse import time from random import randrange @@ -177,10 +178,10 @@ class NeteaseAPI: payload = { 'ids': [song_id], 'level': quality, - 'encodeType': 'flac', + 'encodeType': 'mp4' if quality == 'dolby' else 'flac', 'header': json.dumps(config), } - + if quality == 'sky': payload['immerseType'] = 'c51' @@ -661,6 +662,56 @@ def qr_login() -> Optional[str]: return manager.qr_login() +# ============ Cookie 工具 ============ + +DEFAULT_COOKIE_FILE = "cookie.txt" + + +def load_cookies(path: str = DEFAULT_COOKIE_FILE) -> Dict[str, str]: + """从 cookie.txt 直接读取并解析为 dict + + 支持标准 ``NMTID=xxx; _ntes_nnid=xxx; ...`` 格式。 + 文件不存在或解析失败返回空 dict（不抛异常，调用方决定如何处理）。 + """ + if not os.path.isfile(path): + return {} + try: + with open(path, "r", encoding="utf-8") as f: + content = f.read().strip() + except OSError: + return {} + return parse_cookie_string(content) + + +def parse_cookie_string(cookie_string: str) -> Dict[str, str]: + """把 ``k1=v1; k2=v2`` 格式字符串解析为 dict + + 支持 ``;`` 或 ``\\n`` 作为分隔符。 + """ + cookies: Dict[str, str] = {} + if not cookie_string: + return cookies + cookie_string = cookie_string.strip() + if not cookie_string: + return cookies + for sep in (";", "\n"): + if sep in cookie_string: + pairs = cookie_string.split(sep) + break + else: + pairs = [cookie_string] + for pair in pairs: + pair = pair.strip() + if not pair or "=" not in pair: + continue + key, value = pair.split("=", 1) + key = key.strip() + value = value.strip() + if key and value: + cookies[key] = value + return cookies + + if __name__ == "__main__": # 测试代码 print("网易云音乐API模块") diff --git a/music_downloader.py b/music_downloader.py index 96fea69..fde896d 100644 --- a/music_downloader.py +++ b/music_downloader.py @@ -1,52 +1,60 @@ -"""音乐下载器模块 +"""网易云音乐下载器（适配 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 asyncio import os import re -import asyncio -import aiohttp -import aiofiles -from io import BytesIO -from typing import Dict, List, Optional, Tuple, Any, Union -from pathlib import Path +import shutil +import subprocess +import tempfile 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 aiofiles +import aiohttp import requests -from mutagen.flac import FLAC -from mutagen.mp3 import MP3 -from mutagen.id3 import ID3, TIT2, TPE1, TALB, TDRC, TRCK, APIC -from mutagen.mp4 import MP4 -from music_api import NeteaseAPI, APIException -from cookie_manager import CookieManager +from music_api import APIException, NeteaseAPI, load_cookies +# ============ 类型定义 ============ + class AudioFormat(Enum): """音频格式枚举""" MP3 = "mp3" FLAC = "flac" M4A = "m4a" + MP4 = "mp4" UNKNOWN = "unknown" class QualityLevel(Enum): """音质等级枚举""" - STANDARD = "standard" # 标准 - EXHIGH = "exhigh" # 极高 - LOSSLESS = "lossless" # 无损 - HIRES = "hires" # Hi-Res - SKY = "sky" # 沉浸环绕声 - JYEFFECT = "jyeffect" # 高清环绕声 - JYMASTER = "jymaster" # 超清母带 - DOLBY = "dolby" # 杜比全景声 + STANDARD = "standard" + EXHIGH = "exhigh" + LOSSLESS = "lossless" + HIRES = "hires" + SKY = "sky" + JYEFFECT = "jyeffect" + JYMASTER = "jymaster" + DOLBY = "dolby" @dataclass @@ -82,507 +90,625 @@ class DownloadException(Exception): pass +# ============ 工具映射表 ============ + +# 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: - """音乐下载器主类""" - - def __init__(self, download_dir: str = "downloads", max_concurrent: int = 3): + """网易云音乐下载器 + + 用法: + 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, + ): """ - 初始化音乐下载器 - Args: download_dir: 下载目录 max_concurrent: 最大并发下载数 + ffmpeg_path: ffmpeg 可执行文件路径，为 None 时启动时检测一次 """ self.download_dir = Path(download_dir) self.download_dir.mkdir(exist_ok=True) self.max_concurrent = max_concurrent - - # 初始化依赖 - self.cookie_manager = CookieManager() + self.api = NeteaseAPI() - - # 支持的文件格式 - self.supported_formats = { - 'mp3': AudioFormat.MP3, - 'flac': AudioFormat.FLAC, - 'm4a': AudioFormat.M4A - } - + + # ffmpeg 路径缓存：启动时检测一次，写入时还会再懒加载一次 + self.ffmpeg_path = ffmpeg_path or shutil.which("ffmpeg") + if not self.ffmpeg_path: + print("警告: 未检测到 ffmpeg，写入元数据功能将不可用") + + # ----------- 工具方法 ----------- + def _sanitize_filename(self, filename: str) -> str: - """清理文件名，移除非法字符 - - Args: - filename: 原始文件名 - - Returns: - 清理后的安全文件名 - """ - # 移除或替换非法字符 - illegal_chars = r'[<>:"/\\|?*]' - filename = re.sub(illegal_chars, '_', filename) - - # 移除前后空格和点 + """清理文件名非法字符""" + filename = re.sub(r'[<>:"/\\|?*]', '_', filename) filename = filename.strip(' .') - - # 限制长度 if len(filename) > 200: filename = filename[:200] - return filename or "unknown" - - def _determine_file_extension(self, url: str, content_type: str = "") -> str: - """根据URL和Content-Type确定文件扩展名 - - Args: - url: 下载URL - content_type: HTTP Content-Type头 - - Returns: - 文件扩展名 + + def _determine_file_extension( + self, + url: str, + content_type: str = "", + api_type: str = "", + ) -> str: + """确定文件扩展名 + + 优先级: API 返回的 type > URL 后缀 > HTTP Content-Type > 默认 mp3 """ - # 首先尝试从URL获取 - if '.flac' in url.lower(): - return '.flac' - elif '.mp3' in url.lower(): - return '.mp3' - elif '.m4a' in url.lower(): - return '.m4a' - - # 从Content-Type获取 - content_type = content_type.lower() - if 'flac' in content_type: - return '.flac' - elif 'mpeg' in content_type or 'mp3' in content_type: - return '.mp3' - elif 'mp4' in content_type or 'm4a' in content_type: - return '.m4a' - - return '.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 + + # ----------- 信息获取 ----------- + def get_music_info(self, music_id: int, quality: str = "standard") -> MusicInfo: - """获取音乐详细信息 - - Args: - music_id: 音乐ID - quality: 音质等级 - - Returns: - 音乐信息对象 - - Raises: - DownloadException: 获取信息失败时抛出 - """ + """获取音乐详细信息（同步）""" try: - # 获取cookies - cookies = self.cookie_manager.parse_cookies() - - # 获取音乐URL信息 + cookies = load_cookies() + url_result = self.api.get_song_url(music_id, quality, cookies) - if not url_result.get('data') or not url_result['data']: + print(url_result) + if not url_result.get("data") or not url_result["data"]: raise DownloadException(f"无法获取音乐ID {music_id} 的播放链接") - - song_data = url_result['data'][0] - download_url = song_data.get('url', '') + + song_data = url_result["data"][0] + download_url = song_data.get("url", "") if not download_url: raise DownloadException(f"音乐ID {music_id} 无可用的下载链接") - - # 获取音乐详情 + detail_result = self.api.get_song_detail(music_id) - if not detail_result.get('songs') or not detail_result['songs']: + if not detail_result.get("songs") or not detail_result["songs"]: raise DownloadException(f"无法获取音乐ID {music_id} 的详细信息") - - song_detail = detail_result['songs'][0] - - # 获取歌词 + + song_detail = detail_result["songs"][0] + lyric_result = self.api.get_lyric(music_id, cookies) - lyric = lyric_result.get('lrc', {}).get('lyric', '') if lyric_result else '' - tlyric = lyric_result.get('tlyric', {}).get('lyric', '') if lyric_result else '' - - # 构建艺术家字符串 - artists = '/'.join(artist['name'] for artist in song_detail.get('ar', [])) - - # 创建MusicInfo对象 - music_info = MusicInfo( + lyric = lyric_result.get("lrc", {}).get("lyric", "") if lyric_result else "" + tlyric = lyric_result.get("tlyric", {}).get("lyric", "") if lyric_result else "" + + artists = "/".join(a["name"] for a in song_detail.get("ar", [])) + + return MusicInfo( id=music_id, - name=song_detail.get('name', '未知歌曲'), - artists=artists or '未知艺术家', - album=song_detail.get('al', {}).get('name', '未知专辑'), - pic_url=song_detail.get('al', {}).get('picUrl', ''), - duration=song_detail.get('dt', 0) // 1000, # 转换为秒 - track_number=song_detail.get('no', 0), + name=song_detail.get("name", "未知歌曲"), + artists=artists or "未知艺术家", + album=song_detail.get("al", {}).get("name", "未知专辑"), + pic_url=song_detail.get("al", {}).get("picUrl", ""), + duration=song_detail.get("dt", 0) // 1000, + track_number=song_detail.get("no", 0), download_url=download_url, - file_type=song_data.get('type', 'mp3').lower(), - file_size=song_data.get('size', 0), + file_type=song_data.get("type", "mp3").lower(), + file_size=song_data.get("size", 0), quality=quality, lyric=lyric, - tlyric=tlyric + tlyric=tlyric, ) - - return music_info - + except APIException as e: raise DownloadException(f"API调用失败: {e}") except Exception as e: raise DownloadException(f"获取音乐信息时发生错误: {e}") - - def download_music_file(self, music_id: int, quality: str = "standard") -> DownloadResult: - """下载音乐文件到本地 - - Args: - music_id: 音乐ID - quality: 音质等级 - - Returns: - 下载结果对象 - """ + + # ----------- 文件路径 ----------- + + def _build_file_path(self, music_info: MusicInfo) -> Path: + """根据 MusicInfo 生成下载文件路径""" + filename = f"{music_info.artists} - {music_info.name}" + safe = self._sanitize_filename(filename) + ext = self._determine_file_extension( + music_info.download_url, + api_type=music_info.file_type, + ) + return self.download_dir / f"{safe}{ext}" + + # ----------- 下载（同步） ----------- + + def download_music_file( + self, + music_id: int, + quality: str = "standard", + ) -> DownloadResult: + """同步下载音乐到本地""" try: - # 获取音乐信息 music_info = self.get_music_info(music_id, quality) - - # 生成文件名 - filename = f"{music_info.artists} - {music_info.name}" - safe_filename = self._sanitize_filename(filename) - - # 确定文件扩展名 - file_ext = self._determine_file_extension(music_info.download_url) - file_path = self.download_dir / f"{safe_filename}{file_ext}" - - # 检查文件是否已存在 + file_path = self._build_file_path(music_info) + + # 已存在则直接返回 if file_path.exists(): return DownloadResult( success=True, file_path=str(file_path), file_size=file_path.stat().st_size, - music_info=music_info + music_info=music_info, ) - - # 下载文件 + + # 流式下载 response = requests.get(music_info.download_url, stream=True, timeout=30) response.raise_for_status() - - # 写入文件 - with open(file_path, 'wb') as f: + with open(file_path, "wb") as f: for chunk in response.iter_content(chunk_size=8192): if chunk: f.write(chunk) - - # 写入音乐标签 - self._write_music_tags(file_path, music_info) - + + # 写入元数据（失败不影响主流程） + self._write_metadata_with_ffmpeg(file_path, music_info) + return DownloadResult( success=True, file_path=str(file_path), file_size=file_path.stat().st_size, - music_info=music_info + music_info=music_info, ) - + except DownloadException: raise except requests.RequestException as e: - return DownloadResult( - success=False, - error_message=f"下载请求失败: {e}" - ) + return DownloadResult(success=False, error_message=f"下载请求失败: {e}") except Exception as e: - return DownloadResult( - success=False, - error_message=f"下载过程中发生错误: {e}" - ) - - async def download_music_file_async(self, music_id: int, quality: str = "standard") -> DownloadResult: - """异步下载音乐文件到本地 - - Args: - music_id: 音乐ID - quality: 音质等级 - - Returns: - 下载结果对象 - """ + return DownloadResult(success=False, error_message=f"下载过程中发生错误: {e}") + + # ----------- 下载（异步） ----------- + + async def download_music_file_async( + self, + music_id: int, + quality: str = "standard", + ) -> DownloadResult: + """异步下载音乐到本地""" try: - # 获取音乐信息（同步操作） music_info = self.get_music_info(music_id, quality) - - # 生成文件名 - filename = f"{music_info.artists} - {music_info.name}" - safe_filename = self._sanitize_filename(filename) - - # 确定文件扩展名 - file_ext = self._determine_file_extension(music_info.download_url) - file_path = self.download_dir / f"{safe_filename}{file_ext}" - - # 检查文件是否已存在 + file_path = self._build_file_path(music_info) + if file_path.exists(): return DownloadResult( success=True, file_path=str(file_path), file_size=file_path.stat().st_size, - music_info=music_info + music_info=music_info, ) - - # 异步下载文件 + async with aiohttp.ClientSession() as session: - async with session.get(music_info.download_url) as response: - response.raise_for_status() - - async with aiofiles.open(file_path, 'wb') as f: - async for chunk in response.content.iter_chunked(8192): + async with session.get(music_info.download_url) as resp: + resp.raise_for_status() + async with aiofiles.open(file_path, "wb") as f: + async for chunk in resp.content.iter_chunked(8192): await f.write(chunk) - - # 写入音乐标签 - self._write_music_tags(file_path, music_info) - + + self._write_metadata_with_ffmpeg(file_path, music_info) + return DownloadResult( success=True, file_path=str(file_path), file_size=file_path.stat().st_size, - music_info=music_info + music_info=music_info, ) - + except DownloadException: raise except aiohttp.ClientError as e: - return DownloadResult( - success=False, - error_message=f"异步下载请求失败: {e}" - ) + return DownloadResult(success=False, error_message=f"异步下载请求失败: {e}") except Exception as e: - return DownloadResult( - success=False, - error_message=f"异步下载过程中发生错误: {e}" - ) - - def download_music_to_memory(self, music_id: int, quality: str = "standard") -> Tuple[bool, BytesIO, MusicInfo]: - """下载音乐到内存 - - Args: - music_id: 音乐ID - quality: 音质等级 - - Returns: - (是否成功, 音乐数据流, 音乐信息) - - Raises: - DownloadException: 下载失败时抛出 - """ + return DownloadResult(success=False, error_message=f"异步下载过程中发生错误: {e}") + + # ----------- 下载到内存 ----------- + + def download_music_to_memory( + self, + music_id: int, + quality: str = "standard", + ) -> Tuple[bool, BytesIO, MusicInfo]: + """下载音乐到内存""" try: - # 获取音乐信息 music_info = self.get_music_info(music_id, quality) - - # 下载到内存 - response = requests.get(music_info.download_url, timeout=30) - response.raise_for_status() - - # 创建BytesIO对象 - audio_data = BytesIO(response.content) - - return True, audio_data, music_info - + resp = requests.get(music_info.download_url, timeout=30) + resp.raise_for_status() + return True, BytesIO(resp.content), music_info + except DownloadException: raise except requests.RequestException as e: raise DownloadException(f"下载到内存失败: {e}") except Exception as e: raise DownloadException(f"内存下载过程中发生错误: {e}") - - async def download_batch_async(self, music_ids: List[int], quality: str = "standard") -> List[DownloadResult]: - """批量异步下载音乐 - - Args: - music_ids: 音乐ID列表 - quality: 音质等级 - - Returns: - 下载结果列表 - """ + + # ----------- 批量下载 ----------- + + async def download_batch_async( + self, + music_ids: List[int], + quality: str = "standard", + ) -> List[DownloadResult]: + """批量并发下载""" semaphore = asyncio.Semaphore(self.max_concurrent) - - async def download_with_semaphore(music_id: int) -> DownloadResult: + + async def _one(music_id: int) -> DownloadResult: async with semaphore: return await self.download_music_file_async(music_id, quality) - - tasks = [download_with_semaphore(music_id) for music_id in music_ids] + + tasks = [_one(mid) for mid in music_ids] results = await asyncio.gather(*tasks, return_exceptions=True) - - # 处理异常结果 - processed_results = [] - for i, result in enumerate(results): - if isinstance(result, Exception): - processed_results.append(DownloadResult( + + processed: List[DownloadResult] = [] + for i, r in enumerate(results): + if isinstance(r, Exception): + processed.append(DownloadResult( success=False, - error_message=f"下载音乐ID {music_ids[i]} 时发生异常: {result}" + error_message=f"下载音乐ID {music_ids[i]} 时发生异常: {r}", )) else: - processed_results.append(result) - - return processed_results - - def _write_music_tags(self, file_path: Path, music_info: MusicInfo) -> None: - """写入音乐标签信息 - - Args: - file_path: 音乐文件路径 - music_info: 音乐信息 - """ - try: - file_ext = file_path.suffix.lower() - - if file_ext == '.mp3': - self._write_mp3_tags(file_path, music_info) - elif file_ext == '.flac': - self._write_flac_tags(file_path, music_info) - elif file_ext == '.m4a': - self._write_m4a_tags(file_path, music_info) - - except Exception as e: - print(f"写入音乐标签失败: {e}") - - def _write_mp3_tags(self, file_path: Path, music_info: MusicInfo) -> None: - """写入MP3标签""" - try: - audio = MP3(str(file_path), ID3=ID3) - - # 添加ID3标签 - audio.tags.add(TIT2(encoding=3, text=music_info.name)) - audio.tags.add(TPE1(encoding=3, text=music_info.artists)) - audio.tags.add(TALB(encoding=3, text=music_info.album)) - - if music_info.track_number > 0: - audio.tags.add(TRCK(encoding=3, text=str(music_info.track_number))) - - # 下载并添加封面 - if music_info.pic_url: - try: - pic_response = requests.get(music_info.pic_url, timeout=10) - pic_response.raise_for_status() - audio.tags.add(APIC( - encoding=3, - mime='image/jpeg', - type=3, - desc='Cover', - data=pic_response.content - )) - except: - pass # 封面下载失败不影响主流程 - - audio.save() - except Exception as e: - print(f"写入MP3标签失败: {e}") - - def _write_flac_tags(self, file_path: Path, music_info: MusicInfo) -> None: - """写入FLAC标签""" - try: - audio = FLAC(str(file_path)) - - audio['TITLE'] = music_info.name - audio['ARTIST'] = music_info.artists - audio['ALBUM'] = music_info.album - - if music_info.track_number > 0: - audio['TRACKNUMBER'] = str(music_info.track_number) - - # 下载并添加封面 - if music_info.pic_url: - try: - pic_response = requests.get(music_info.pic_url, timeout=10) - pic_response.raise_for_status() - - from mutagen.flac import Picture - picture = Picture() - picture.type = 3 # Cover (front) - picture.mime = 'image/jpeg' - picture.desc = 'Cover' - picture.data = pic_response.content - audio.add_picture(picture) - except: - pass # 封面下载失败不影响主流程 - - audio.save() - except Exception as e: - print(f"写入FLAC标签失败: {e}") - - def _write_m4a_tags(self, file_path: Path, music_info: MusicInfo) -> None: - """写入M4A标签""" - try: - audio = MP4(str(file_path)) - - audio['\xa9nam'] = music_info.name - audio['\xa9ART'] = music_info.artists - audio['\xa9alb'] = music_info.album - - if music_info.track_number > 0: - audio['trkn'] = [(music_info.track_number, 0)] - - # 下载并添加封面 - if music_info.pic_url: - try: - pic_response = requests.get(music_info.pic_url, timeout=10) - pic_response.raise_for_status() - audio['covr'] = [pic_response.content] - except: - pass # 封面下载失败不影响主流程 - - audio.save() - except Exception as e: - print(f"写入M4A标签失败: {e}") - - def get_download_progress(self, music_id: int, quality: str = "standard") -> Dict[str, Any]: - """获取下载进度信息 - - Args: - music_id: 音乐ID - quality: 音质等级 - - Returns: - 包含进度信息的字典 - """ + processed.append(r) + return processed + + # ----------- 进度查询 ----------- + + def get_download_progress( + self, + music_id: int, + quality: str = "standard", + ) -> Dict[str, Any]: + """查询下载进度""" try: music_info = self.get_music_info(music_id, quality) - - filename = f"{music_info.artists} - {music_info.name}" - safe_filename = self._sanitize_filename(filename) - file_ext = self._determine_file_extension(music_info.download_url) - file_path = self.download_dir / f"{safe_filename}{file_ext}" - + file_path = self._build_file_path(music_info) + if file_path.exists(): current_size = file_path.stat().st_size - progress = (current_size / music_info.file_size * 100) if music_info.file_size > 0 else 0 - + progress = ( + current_size / music_info.file_size * 100 + if music_info.file_size > 0 + else 0 + ) return { - 'music_id': music_id, - 'filename': safe_filename + file_ext, - 'total_size': music_info.file_size, - 'current_size': current_size, - 'progress': min(progress, 100), - 'completed': current_size >= music_info.file_size + "music_id": music_id, + "filename": file_path.name, + "total_size": music_info.file_size, + "current_size": current_size, + "progress": min(progress, 100), + "completed": current_size >= music_info.file_size, } - else: - return { - 'music_id': music_id, - 'filename': safe_filename + file_ext, - 'total_size': music_info.file_size, - 'current_size': 0, - 'progress': 0, - 'completed': False - } - - except Exception as e: + return { - 'music_id': music_id, - 'error': str(e), - 'progress': 0, - 'completed': False + "music_id": music_id, + "filename": file_path.name, + "total_size": music_info.file_size, + "current_size": 0, + "progress": 0, + "completed": False, } + except Exception as e: + return { + "music_id": music_id, + "error": str(e), + "progress": 0, + "completed": False, + } + + # ============ ffmpeg 元数据写入（核心） ============ + + def _write_metadata_with_ffmpeg( + self, + file_path: Path, + music_info: MusicInfo, + ) -> bool: + """使用 ffmpeg 写入音频元数据 + + 处理: title/artist/album/track + 封面（如果是 mp3/flac/m4a/mp4） + + Returns: + 是否写入成功（失败不影响主流程） + """ + ffmpeg_path = self._resolve_ffmpeg() + if not ffmpeg_path: + print("跳过元数据写入: ffmpeg 不可用") + return False + + file_ext = file_path.suffix.lower() + if file_ext not in METADATA_FORMATS: + print(f"跳过元数据写入: 不支持的格式 {file_ext}") + return False + + if not file_path.exists(): + print(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 + + try: + # 下载封面到临时文件 + if music_info.pic_url: + try: + pic_resp = requests.get(music_info.pic_url, timeout=10) + pic_resp.raise_for_status() + fd, cover_tmp = tempfile.mkstemp(suffix=".jpg", dir=str(self.download_dir)) + os.close(fd) + with open(cover_tmp, "wb") as cf: + cf.write(pic_resp.content) + except Exception as e: + print(f"封面下载失败，跳过封面写入: {e}") + cover_tmp = None + + # 构造 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), + ] + if cover_tmp: + cmd += ["-i", cover_tmp] + + # 通用元数据 + meta = { + "title": music_info.name, + "artist": music_info.artists, + "album": music_info.album, + "album_artist": music_info.artists, + "comment": "Downloaded from Netease Cloud Music", + } + if music_info.track_number > 0: + meta["track"] = str(music_info.track_number) + meta["tracktotal"] = 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=bool(cover_tmp)) + + # -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(): + # 原子替换原文件 + shutil.move(str(tmp_output), str(file_path)) + return True + + print( + f"ffmpeg 写入元数据失败: " + f"{(result.stderr or result.stdout or '').strip()}" + ) + return False + + except subprocess.TimeoutExpired: + print("ffmpeg 执行超时") + return False + except FileNotFoundError: + print(f"ffmpeg 可执行文件未找到: {ffmpeg_path}") + return False + except Exception as e: + print(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 / 编码参数""" + 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:a", "-c:a", "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)", + ] + return ["-map", "0:a", "-c:a", "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() - print("音乐下载器模块") - print("支持的功能:") - print("- 同步下载") - print("- 异步下载") - print("- 批量下载") - print("- 内存下载") - print("- 音乐标签写入") - print("- 下载进度跟踪") + + # 无参数：打印模块信息 + 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) diff --git a/qr_login.py b/qr_login.py index e108178..86bbbc0 100644 --- a/qr_login.py +++ b/qr_login.py @@ -3,153 +3,193 @@ 提供网易云音乐二维码登录功能，包括： - 二维码生成和显示 - 登录状态检查 -- Cookie获取和保存 +- Cookie 获取和保存（直接读写 cookie.txt） - 用户友好的交互界面 """ +import os import sys import time +import shutil import logging -from typing import Optional, Dict, Any, Tuple +import datetime from pathlib import Path +from typing import Optional, Tuple try: - from music_api import QRLoginManager, APIException - from cookie_manager import CookieManager, CookieException + from music_api import QRLoginManager, APIException, load_cookies except ImportError as e: print(f"导入模块失败: {e}") - print("请确保 music_api.py 和 cookie_manager.py 文件存在且可用") + print("请确保 music_api.py 文件存在且可用") sys.exit(1) +# 判断 cookie 是否"有效"：包含任一关键字段即可 +IMPORTANT_COOKIE_KEYS = ("MUSIC_U", "__csrf", "NMTID") + + class QRLoginClient: """二维码登录客户端""" - + def __init__(self, cookie_file: str = "cookie.txt"): """ - 初始化二维码登录客户端 - Args: - cookie_file: Cookie保存文件路径 + cookie_file: Cookie 保存文件路径 """ - self.cookie_manager = CookieManager(cookie_file) + self.cookie_file = Path(cookie_file) self.qr_manager = QRLoginManager() self.logger = logging.getLogger(__name__) - - # 配置日志 + if not self.logger.handlers: handler = logging.StreamHandler() formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') handler.setFormatter(formatter) self.logger.addHandler(handler) self.logger.setLevel(logging.INFO) - - def check_existing_login(self) -> bool: - """检查是否已有有效登录 - - Returns: - 是否已登录 - """ + + # ----------- Cookie 文件直接操作 ----------- + + def _read_cookie(self) -> str: + if not self.cookie_file.exists(): + return "" try: - if self.cookie_manager.is_cookie_valid(): - self.logger.info("检测到有效的登录Cookie") + return self.cookie_file.read_text(encoding="utf-8").strip() + except OSError as e: + self.logger.error(f"读取 Cookie 失败: {e}") + return "" + + def _write_cookie(self, cookie: str) -> bool: + if not cookie or not cookie.strip(): + self.logger.error("Cookie 内容不能为空") + return False + try: + self.cookie_file.write_text(cookie.strip() + "\n", encoding="utf-8") + return True + except OSError as e: + self.logger.error(f"写入 Cookie 失败: {e}") + return False + + def _backup_cookie(self, suffix: str = "backup") -> Optional[Path]: + if not self.cookie_file.exists(): + return None + ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") + backup_path = self.cookie_file.with_name( + f"{self.cookie_file.stem}.{suffix}.{ts}{self.cookie_file.suffix}" + ) + try: + shutil.copy2(self.cookie_file, backup_path) + return backup_path + except OSError as e: + self.logger.warning(f"备份 Cookie 失败: {e}") + return None + + def _clear_cookie(self) -> bool: + if not self.cookie_file.exists(): + return True + try: + self.cookie_file.unlink() + return True + except OSError as e: + self.logger.error(f"清除 Cookie 失败: {e}") + return False + + def _is_cookie_valid(self) -> bool: + cookies = load_cookies(str(self.cookie_file)) + return any(k in cookies for k in IMPORTANT_COOKIE_KEYS) + + def _get_cookie_info(self) -> dict: + cookies = load_cookies(str(self.cookie_file)) + present = [k for k in IMPORTANT_COOKIE_KEYS if k in cookies] + info = { + "file_path": str(self.cookie_file.absolute()), + "file_exists": self.cookie_file.exists(), + "cookie_count": len(cookies), + "is_valid": bool(present), + "important_cookies_present": present, + "missing_important_cookies": [k for k in IMPORTANT_COOKIE_KEYS if k not in cookies], + } + if self.cookie_file.exists(): + try: + mtime = self.cookie_file.stat().st_mtime + info["last_modified"] = datetime.datetime.fromtimestamp(mtime).strftime("%Y-%m-%d %H:%M:%S") + except OSError: + pass + return info + + # ----------- 业务方法 ----------- + + def check_existing_login(self) -> bool: + """检查是否已有有效登录""" + try: + if self._is_cookie_valid(): + self.logger.info("检测到有效的登录 Cookie") return True - else: - self.logger.info("未检测到有效的登录Cookie") - return False + self.logger.info("未检测到有效的登录 Cookie") + return False except Exception as e: self.logger.error(f"检查登录状态失败: {e}") return False - + def interactive_login(self) -> Tuple[bool, Optional[str]]: - """交互式二维码登录 - - Returns: - (登录是否成功, 错误信息) - """ + """交互式二维码登录""" try: print("\n=== 网易云音乐二维码登录 ===") - - # 检查现有登录状态 + if self.check_existing_login(): choice = input("检测到已有有效登录，是否重新登录？(y/N): ").strip().lower() - if choice not in ['y', 'yes', '是']: + if choice not in ('y', 'yes', '是'): print("使用现有登录状态") return True, None - + print("\n开始二维码登录流程...") - - # 生成二维码 print("正在生成二维码...") qr_result = self.qr_manager.create_qr_login() - - if not qr_result['success']: + + if not qr_result.get('success'): error_msg = f"生成二维码失败: {qr_result.get('message', '未知错误')}" self.logger.error(error_msg) return False, error_msg - + qr_key = qr_result['qr_key'] - print(f"\n二维码已生成！") - print(f"请使用网易云音乐手机APP扫描二维码进行登录") - print(f"二维码有效期: 3分钟") + print("\n二维码已生成！") + print("请使用网易云音乐手机 APP 扫描二维码进行登录") + print("二维码有效期: 3 分钟") print("\n等待扫码中...") - - # 轮询检查登录状态 - max_attempts = 60 # 最多等待5分钟 + + max_attempts = 60 # 最多 5 分钟 attempt = 0 - while attempt < max_attempts: try: - # 检查登录状态 status_result = self.qr_manager.check_qr_login(qr_key) - - if status_result['success']: - if status_result['status'] == 'success': - # 登录成功 + + if status_result.get('success'): + status = status_result.get('status') + if status == 'success': cookie = status_result.get('cookie', '') - if cookie: - # 保存Cookie - success = self.save_cookie(cookie) - if success: - print("\n✅ 登录成功！Cookie已保存") - return True, None - else: - error_msg = "登录成功但Cookie保存失败" - self.logger.error(error_msg) - return False, error_msg - else: - error_msg = "登录成功但未获取到Cookie" - self.logger.error(error_msg) - return False, error_msg - - elif status_result['status'] == 'waiting': - # 等待扫码 - if attempt % 10 == 0: # 每10次显示一次提示 - print(f"等待扫码中... ({attempt + 1}/{max_attempts})") - - elif status_result['status'] == 'scanned': - # 已扫码，等待确认 + if not cookie: + return False, "登录成功但未获取到 Cookie" + if self.save_cookie(cookie): + print("\n✅ 登录成功！Cookie 已保存") + return True, None + return False, "登录成功但 Cookie 保存失败" + + if status == 'waiting' and attempt % 10 == 0: + print(f"等待扫码中... ({attempt + 1}/{max_attempts})") + elif status == 'scanned': print("二维码已扫描，请在手机上确认登录") - - elif status_result['status'] == 'expired': - # 二维码过期 - error_msg = "二维码已过期，请重新尝试" - print(f"\n❌ {error_msg}") - return False, error_msg - - elif status_result['status'] == 'error': - # 登录错误 - error_msg = f"登录失败: {status_result.get('message', '未知错误')}" - print(f"\n❌ {error_msg}") - return False, error_msg - + elif status == 'expired': + print("\n❌ 二维码已过期，请重新尝试") + return False, "二维码已过期" + elif status == 'error': + msg = status_result.get('message', '未知错误') + print(f"\n❌ 登录失败: {msg}") + return False, f"登录失败: {msg}" else: self.logger.warning(f"检查登录状态失败: {status_result.get('message', '未知错误')}") - - # 等待5秒后重试 + time.sleep(5) attempt += 1 - + except KeyboardInterrupt: print("\n用户取消登录") return False, "用户取消登录" @@ -157,169 +197,119 @@ class QRLoginClient: self.logger.error(f"检查登录状态时发生错误: {e}") time.sleep(5) attempt += 1 - - # 超时 - error_msg = "登录超时，请重新尝试" - print(f"\n❌ {error_msg}") - return False, error_msg - + + print("\n❌ 登录超时，请重新尝试") + return False, "登录超时" + except APIException as e: - error_msg = f"API调用失败: {e}" - self.logger.error(error_msg) - return False, error_msg + self.logger.error(f"API 调用失败: {e}") + return False, f"API 调用失败: {e}" except Exception as e: - error_msg = f"登录过程中发生未知错误: {e}" - self.logger.error(error_msg) - return False, error_msg - + self.logger.error(f"登录过程中发生未知错误: {e}") + return False, f"登录过程中发生未知错误: {e}" + def save_cookie(self, cookie: str) -> bool: - """保存Cookie到文件 - - Args: - cookie: Cookie字符串 - - Returns: - 是否保存成功 - """ + """保存 Cookie 到文件""" try: - # 备份现有Cookie（如果存在） - if self.cookie_manager.cookie_file.exists(): - try: - backup_path = self.cookie_manager.backup_cookie() - self.logger.info(f"已备份现有Cookie到: {backup_path}") - except Exception as e: - self.logger.warning(f"备份Cookie失败: {e}") - - # 保存新Cookie - success = self.cookie_manager.write_cookie(cookie) - - if success: - # 验证保存的Cookie - if self.cookie_manager.is_cookie_valid(): - self.logger.info("Cookie保存并验证成功") - return True - else: - self.logger.warning("Cookie保存成功但验证失败") - return False - else: - self.logger.error("Cookie保存失败") + if self.cookie_file.exists(): + backup = self._backup_cookie() + if backup: + self.logger.info(f"已备份现有 Cookie 到: {backup}") + + if not self._write_cookie(cookie): return False - - except CookieException as e: - self.logger.error(f"Cookie操作失败: {e}") + + if self._is_cookie_valid(): + self.logger.info("Cookie 保存并验证成功") + return True + + self.logger.warning("Cookie 保存成功但验证失败（缺少关键字段）") return False except Exception as e: - self.logger.error(f"保存Cookie时发生错误: {e}") + self.logger.error(f"保存 Cookie 时发生错误: {e}") return False - + def show_login_info(self) -> None: """显示登录信息""" try: - info = self.cookie_manager.get_cookie_info() - + info = self._get_cookie_info() print("\n=== 登录状态信息 ===") - print(f"Cookie文件: {info['file_path']}") + print(f"Cookie 文件: {info['file_path']}") print(f"文件存在: {'是' if info['file_exists'] else '否'}") - print(f"Cookie数量: {info['cookie_count']}") + print(f"Cookie 数量: {info['cookie_count']}") print(f"登录状态: {'有效' if info['is_valid'] else '无效'}") - - if info.get('last_modified'): + if info.get("last_modified"): print(f"最后更新: {info['last_modified']}") - - if info['is_valid']: - present_cookies = info.get('important_cookies_present', []) - print(f"重要Cookie: {', '.join(present_cookies)}") + if info["is_valid"]: + present = info.get("important_cookies_present", []) + print(f"重要 Cookie: {', '.join(present)}") else: - missing_cookies = info.get('missing_important_cookies', []) - if missing_cookies: - print(f"缺少Cookie: {', '.join(missing_cookies)}") - + missing = info.get("missing_important_cookies", []) + if missing: + print(f"缺少 Cookie: {', '.join(missing)}") except Exception as e: print(f"获取登录信息失败: {e}") - + def logout(self) -> bool: - """登出（清除Cookie） - - Returns: - 是否登出成功 - """ + """登出（清除 Cookie）""" try: - # 备份Cookie - if self.cookie_manager.cookie_file.exists(): - try: - backup_path = self.cookie_manager.backup_cookie("logout") - print(f"Cookie已备份到: {backup_path}") - except Exception as e: - self.logger.warning(f"备份Cookie失败: {e}") - - # 清除Cookie - success = self.cookie_manager.clear_cookie() - - if success: + if self.cookie_file.exists(): + backup = self._backup_cookie("logout") + if backup: + print(f"Cookie 已备份到: {backup}") + if self._clear_cookie(): print("已成功登出") return True - else: - print("登出失败") - return False - + print("登出失败") + return False except Exception as e: print(f"登出时发生错误: {e}") return False def main(): - """主函数""" - # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) - + client = QRLoginClient() - - # 解析命令行参数 + if len(sys.argv) > 1: command = sys.argv[1].lower() - + if command == 'login': - # 执行登录 success, error = client.interactive_login() if success: print("\n登录完成！") client.show_login_info() sys.exit(0) - else: - print(f"\n登录失败: {error}") - sys.exit(1) - - elif command == 'status' or command == 'info': - # 显示登录状态 + print(f"\n登录失败: {error}") + sys.exit(1) + + elif command in ('status', 'info'): client.show_login_info() sys.exit(0) - + elif command == 'logout': - # 登出 - success = client.logout() - sys.exit(0 if success else 1) - - elif command == 'help' or command == '-h' or command == '--help': - # 显示帮助 + sys.exit(0 if client.logout() else 1) + + elif command in ('help', '-h', '--help'): print("网易云音乐二维码登录工具") - print("\n用法:") - print(" python qr_login.py [命令]") + print("\n用法: python qr_login.py [命令]") print("\n命令:") print(" login - 执行二维码登录") print(" status - 显示登录状态") - print(" logout - 登出（清除Cookie）") + print(" logout - 登出（清除 Cookie）") print(" help - 显示此帮助信息") print("\n如果不提供命令，将进入交互模式") sys.exit(0) - + else: print(f"未知命令: {command}") print("使用 'python qr_login.py help' 查看帮助") sys.exit(1) - + # 交互模式 try: while True: @@ -328,9 +318,9 @@ def main(): print("2. 查看登录状态") print("3. 登出") print("4. 退出") - + choice = input("\n请选择操作 (1-4): ").strip() - + if choice == '1': success, error = client.interactive_login() if success: @@ -338,20 +328,16 @@ def main(): client.show_login_info() else: print(f"\n登录失败: {error}") - elif choice == '2': client.show_login_info() - elif choice == '3': client.logout() - elif choice == '4': print("再见！") break - else: print("无效选择，请重试") - + except KeyboardInterrupt: print("\n\n程序已退出") sys.exit(0) diff --git a/requirements.txt b/requirements.txt index 4fdb48e1d1f82d9c550a7077917fa04a2a36af3e..81ebd83430a09fce14df8e41d32515a835e90bf0 100644 GIT binary patch literal 430 zcmX9)J$J((4BYuI)ic=n=)l?`?{(_XS;5#=O@IhIIKRHa*(BYaPA8-^QSilDGk}Ip zA*BU#-u>?H7(xT!4EmVYnyu|Y8_=gVC^Sb5)@m4N6ZF#E1I5*I13#s8o-0ix za3_Q-^vX@G)u1o3-7Uz)*N>R^fq`C4*vnR-1#+vkR}fR>Qb>#sw6+uQtQ_yTrkOD> zyR}0H9T@c$u!usA5c$OE^sY229-Hwhi(yu7>a-1U2bV4f&pXHlU*7Gl?aOcbMvLX-P&N| e5a`~wTb?s{T5G;5o>Vijl~-L?^wv`$%(=%he=o04 zk1&5&?O4hatJo8>5W10LNkm8yUBg~d`;D5;SycTv#tYx?$&tTsHpQJ#9$0hNQ4mdW z)z9qS-||tIT#J8;jyt#-zV@!Rw4aeznxr}*N>_m{af&H?2V#S2q2!LEjOkFl8uWDZ zw}r2vGsThGGaSq`qGGBZKUu0&EA1LaF{Jyf^q3?u?H~#OyN?aEod*T)Pq(+?!r8-Yq zfux)s1gWV3nsBhQ*=z{w?IppYMdL*CmS*jpcJ^>3`%2u^3t($ zK!xs!x+5P^dC57ErCU*QKjx##Y`7b7^4~#FPjr4x_(}cH>5CI|c74@Vy沉浸环绕声 +
diff --git a/使用文档.md b/使用文档.md index 9e58a23..220a216 100644 --- a/使用文档.md +++ b/使用文档.md @@ -417,10 +417,9 @@ tail -f music_api.log ``` Netease_url/ ├── main.py # 主程序入口 -├── music_api.py # 音乐API核心模块 +├── music_api.py # 音乐API核心模块（含 load_cookies 工具） ├── music_downloader.py # 音乐下载模块 -├── cookie_manager.py # Cookie管理模块 -├── qr_login.py # 二维码登录模块 +├── qr_login.py # 二维码登录模块（直接读写 cookie.txt） ├── templates/ │ └── index.html # Web界面模板 ├── downloads/ # 下载文件目录