From ce7ea66d6a5fd281d12c00129e31358f7a359a9e Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Sat, 26 Sep 2026 00:32:45 +0800 Subject: [PATCH] =?UTF-8?q?refactor(async):=20music=5Fapi=20=E5=BC=82?= =?UTF-8?q?=E6=AD=A5=E5=8C=96=20+=20=E5=85=A8=E9=A1=B9=E7=9B=AE=20lint=20?= =?UTF-8?q?=E6=94=B6=E6=95=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit music_api.py： - 22 个方法改 async（HTTPClient.post_request*/NeteaseAPI.*/QRLoginManager.*）， 33 处 requests.* → await core.aiohttp_compat.* - 16 个模块级同步包装改走 run_sync 桥（main.py 等 99 个同步调用点零改动） - async 内 time.sleep → asyncio.sleep（修 7 处阻塞事件循环） musica_downloader/qr_login： - 单曲三请求 ThreadPoolExecutor → asyncio.gather（语义等价，失败处理保留） - interactive_login 改 async，CLI 入口用 asyncio.run 真实缺陷修复（lint 发现）： - podcast_api._to_int 未定义（2 处调用会 NameError）→ 补安全转换 helper - main.py VALID_QUALITY_LEVELS 重复定义（覆盖 core.config 导入值）→ 移除 - 移除死代码变量（salt×2/device_id×3/file_path/zipped_files） lint 收敛：524→0（ruff All checks passed）；pyright 0 errors。 pyproject 记录「有意保留」项（B904/RUF012/S305 上游协议等）及理由。 验证：冒烟 10/10；pytest 16/16；真实上游 get_toplist 异步直调与桥接均得 63 条。 --- auth_routes.py | 45 +++-- auth_system.py | 65 ++++---- main.py | 318 ++++++++++++++++++----------------- music_api.py | 395 ++++++++++++++++++++++---------------------- music_downloader.py | 84 +++++----- podcast_api.py | 71 ++++---- pyproject.toml | 6 + qr_login.py | 26 +-- 8 files changed, 520 insertions(+), 490 deletions(-) diff --git a/auth_routes.py b/auth_routes.py index 7e088b2..31153a2 100644 --- a/auth_routes.py +++ b/auth_routes.py @@ -6,12 +6,9 @@ import hmac import secrets - import sqlite3 import time -from typing import Optional, Tuple - from flask import request, send_file import auth_system as auth @@ -304,7 +301,7 @@ def register_auth_routes(app) -> None: return _ok({'user': auth._user_public(fresh), 'duration_text': auth.format_credits(fresh['duration_seconds']), 'redirect': 'admin' if fresh['role'] == 'admin' else 'user'}, - f'注册成功' + (f'，已赠送 {credits} 次下载次数' if credits else '')) + '注册成功' + (f'，已赠送 {credits} 次下载次数' if credits else '')) except Exception: return _err('注册失败，请稍后重试', 500) @@ -366,7 +363,7 @@ def register_auth_routes(app) -> None: '密码验证通过，请完成邮箱验证') auth.login_session(user) # IP 归属地异常检测：24h 内 >=3 个不同省份自动封禁（防倒卖） - banned, info = auth.record_login_location(user, auth._client_ip()) + banned, _info = auth.record_login_location(user, auth._client_ip()) if banned: auth.logout_session() return _err('检测到账号在多个省份异常登录，已自动封禁，请联系管理员') @@ -442,7 +439,7 @@ def register_auth_routes(app) -> None: return _err('账号已被禁用，请联系管理员') auth.login_session(user) # IP 归属地异常检测：管理员天然豁免省份封禁（record_login_location 内部判定） - banned, info = auth.record_login_location(user, auth._client_ip()) + banned, _info = auth.record_login_location(user, auth._client_ip()) if banned: auth.logout_session() return _err('检测到账号在多个省份异常登录，已自动封禁，请联系管理员') @@ -552,7 +549,7 @@ def register_auth_routes(app) -> None: auth.add_log('success', _actor(), auth._client_ip(), '删除用户', target['username'] if target else str(uid)) return _ok(message='用户已删除') except Exception as e: - return _err(f'删除失败: {str(e)}', 500) + return _err(f'删除失败: {e!s}', 500) @app.route('/admin/site/user/duration', methods=['POST']) @auth.admin_required @@ -576,9 +573,9 @@ def register_auth_routes(app) -> None: f'{target["username"]} → {count} 次') return _ok(message=f'已修改为 {count} 次') except Exception as e: - return _err(f'修改失败: {str(e)}', 500) + return _err(f'修改失败: {e!s}', 500) - def _block_admin_target(target) -> Optional[Tuple]: + def _block_admin_target(target) -> tuple | None: """禁止对管理员账号执行危险操作（防自锁）""" if target and target['role'] == 'admin': return _err('不能对管理员账号执行该操作') @@ -622,7 +619,7 @@ def register_auth_routes(app) -> None: f'{username} · {credits} 次') return _ok(message=f'用户 {username} 创建成功（{credits} 次）') except Exception as e: - return _err(f'创建失败: {str(e)}', 500) + return _err(f'创建失败: {e!s}', 500) @app.route('/admin/site/user/reset_password', methods=['POST']) @auth.admin_required @@ -653,7 +650,7 @@ def register_auth_routes(app) -> None: auth.add_log('success', _actor(), auth._client_ip(), '重置密码', target['username']) return _ok(message=f'已重置 {target["username"]} 的密码') except Exception as e: - return _err(f'重置失败: {str(e)}', 500) + return _err(f'重置失败: {e!s}', 500) @app.route('/admin/site/user/toggle', methods=['POST']) @auth.admin_required @@ -674,7 +671,7 @@ def register_auth_routes(app) -> None: return _ok({'status': new_status}, message=f'已{"启用" if new_status else "禁用"} {target["username"]}') except Exception as e: - return _err(f'操作失败: {str(e)}', 500) + return _err(f'操作失败: {e!s}', 500) @app.route('/admin/site/user/ban', methods=['POST']) @auth.admin_required @@ -704,7 +701,7 @@ def register_auth_routes(app) -> None: auth._client_ip(), '解封用户', row['username']) return _ok(message=f'已解封用户「{row["username"]}」') except Exception as e: - return _err(f'操作失败: {str(e)}', 500) + return _err(f'操作失败: {e!s}', 500) @app.route('/admin/site/user/cookie', methods=['POST']) @auth.admin_required @@ -743,7 +740,7 @@ def register_auth_routes(app) -> None: auth.add_log('success', _actor(), auth._client_ip(), '保存SMTP配置', f'{host}:{port}') return _ok(message='SMTP 配置已保存') except Exception as e: - return _err(f'保存失败: {str(e)}', 500) + return _err(f'保存失败: {e!s}', 500) @app.route('/admin/site/smtp/test', methods=['POST']) @auth.admin_required @@ -762,7 +759,7 @@ def register_auth_routes(app) -> None: return _ok(message='测试邮件已发送，请查收') except Exception as e: auth.add_log('fail', _actor(), auth._client_ip(), 'SMTP测试', str(e)) - return _err(f'发送失败: {str(e)}', 500) + return _err(f'发送失败: {e!s}', 500) @app.route('/admin/site/logs', methods=['POST']) @auth.admin_required @@ -815,7 +812,7 @@ def register_auth_routes(app) -> None: return _ok({'codes': codes, 'credits': credits, 'count': len(codes)}, f'已生成 {len(codes)} 张卡密') except Exception as e: - return _err(f'生成失败: {str(e)}', 500) + return _err(f'生成失败: {e!s}', 500) @app.route('/admin/site/cards/list', methods=['POST']) @auth.admin_required @@ -884,7 +881,7 @@ def register_auth_routes(app) -> None: auth.add_log('success', _actor(), auth._client_ip(), '保存公告', str(data.get('title') or '')) return _ok(message='公告已保存') except Exception as e: - return _err(f'保存失败: {str(e)}', 500) + return _err(f'保存失败: {e!s}', 500) # -------------------- 播放 API 密钥 -------------------- @@ -917,7 +914,7 @@ def register_auth_routes(app) -> None: auth.add_log('success', _actor(), auth._client_ip(), '数据库备份', name) return _ok({'name': name}, f'备份完成：{name}') except Exception as e: - return _err(f'备份失败: {str(e)}', 500) + return _err(f'备份失败: {e!s}', 500) @app.route('/admin/site/backup/list', methods=['POST']) @auth.admin_required @@ -979,7 +976,7 @@ def register_auth_routes(app) -> None: dest.unlink() except Exception: pass - return _err(f'上传失败: {str(e)}', 500) + return _err(f'上传失败: {e!s}', 500) @app.route('/admin/site/backup/restore', methods=['POST']) @auth.admin_required @@ -992,10 +989,10 @@ def register_auth_routes(app) -> None: except ValueError as e: # 第 95 条（P3-1）：客户端可纠正错误（备份不存在/校验失败）返回 400， # 与 delete 的 404 口径对齐——原实现一律 500 - return _err(f'恢复失败: {str(e)}', 400) + return _err(f'恢复失败: {e!s}', 400) except Exception as e: auth.add_log('fail', _actor(), auth._client_ip(), '恢复数据库', str(e)) - return _err(f'恢复失败: {str(e)}', 500) + return _err(f'恢复失败: {e!s}', 500) @app.route('/admin/site/backup/delete', methods=['POST']) @auth.admin_required @@ -1006,7 +1003,7 @@ def register_auth_routes(app) -> None: return _ok(message='备份已删除') return _err('备份文件不存在', 404) except Exception as e: - return _err(f'删除失败: {str(e)}', 500) + return _err(f'删除失败: {e!s}', 500) @app.route('/admin/site/backup/auto', methods=['POST']) @auth.admin_required @@ -1023,10 +1020,10 @@ def register_auth_routes(app) -> None: if hours < 0: return _err('间隔不能为负') auth.set_setting('backup_interval_hours', str(hours)) - return _ok(message='定时备份已开启，每 {} 小时自动备份'.format(hours) if hours > 0 + return _ok(message=f'定时备份已开启，每 {hours} 小时自动备份' if hours > 0 else '定时备份已关闭') except Exception as e: - return _err(f'设置失败: {str(e)}', 500) + return _err(f'设置失败: {e!s}', 500) # ============ 辅助 ============ diff --git a/auth_system.py b/auth_system.py index d5f0933..6b6650e 100644 --- a/auth_system.py +++ b/auth_system.py @@ -10,14 +10,14 @@ 数据存储：SQLite（data/site.db），会话：Flask 签名 Cookie（secret_key 持久化） """ +import base64 import hashlib -import shutil import json import math -import base64 import os import re import secrets +import shutil import smtplib import sqlite3 import ssl @@ -29,9 +29,9 @@ from email.mime.text import MIMEText from email.utils import formataddr from functools import wraps from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple +from typing import Any -from flask import request, session, g +from flask import g, request, session # ============ 路径与常量 ============ DATA_DIR = Path('data') @@ -59,7 +59,7 @@ def get_conn() -> sqlite3.Connection: return conn -def _q(sql: str, params: Tuple = ()) -> List[sqlite3.Row]: +def _q(sql: str, params: tuple = ()) -> list[sqlite3.Row]: with _DB_LOCK: conn = get_conn() try: @@ -68,7 +68,7 @@ def _q(sql: str, params: Tuple = ()) -> List[sqlite3.Row]: conn.close() -def _x(sql: str, params: Tuple = ()) -> int: +def _x(sql: str, params: tuple = ()) -> int: """执行写操作，返回 lastrowid / rowcount""" with _DB_LOCK: conn = get_conn() @@ -209,7 +209,6 @@ def init_db() -> None: row = conn.execute("SELECT id FROM users WHERE role = 'admin' LIMIT 1").fetchone() if not row: initial_pass = secrets.token_urlsafe(12) - salt = secrets.token_hex(8) conn.execute( 'INSERT INTO users (username, password_hash, salt, role, duration_seconds, created_at) ' 'VALUES (?, ?, ?, ?, -1, ?)', @@ -270,7 +269,7 @@ def _hash_new(password: str, salt: str = '', iterations: int = _PBKDF2_ITERS) -> return f'pbkdf2${iterations}${salt}${digest}' -def _verify_password(stored: str, password: str, salt: str) -> Tuple[bool, bool]: +def _verify_password(stored: str, password: str, salt: str) -> tuple[bool, bool]: """校验口令。返回 (是否正确, 是否需要升级到 PBKDF2)""" if stored.startswith('pbkdf2$'): try: @@ -287,7 +286,7 @@ def _verify_password(stored: str, password: str, salt: str) -> Tuple[bool, bool] def create_user(username: str, password: str, email: str = '', credits: int = 0, role: str = 'user', device_id: str = '', ip: str = '', ua: str = '', register_ip_limit: int = 0, register_ip_window: int = 86400, - register_device_once: bool = False) -> Dict[str, Any]: + register_device_once: bool = False) -> dict[str, Any]: # F77：余额只接受 -1（不限）或 >= 0；非法负值一律按 0 处理，禁止出现 -2 这类"假不限" try: credits = int(credits) @@ -298,7 +297,6 @@ def create_user(username: str, password: str, email: str = '', credits: int = 0, # R5（第 79 条）：注册入口传 register_ip_limit=3 —— 同 IP 24h 注册数检查挪进 # _DB_LOCK 事务内与 INSERT 原子完成，堵住"锁外检查+并发插入"的 TOCTOU 刷号； # 管理建号与不限制场景传 0 关闭。超限抛 ValueError 由调用方转 400/429。 - salt = secrets.token_hex(8) now = int(time.time()) with _DB_LOCK: conn = get_conn() @@ -328,12 +326,12 @@ def create_user(username: str, password: str, email: str = '', credits: int = 0, return get_user_by_name(username) -def get_user_by_name(username: str) -> Optional[sqlite3.Row]: +def get_user_by_name(username: str) -> sqlite3.Row | None: rows = _q('SELECT * FROM users WHERE username = ?', (username,)) return rows[0] if rows else None -def get_user_by_email(email: str) -> Optional[sqlite3.Row]: +def get_user_by_email(email: str) -> sqlite3.Row | None: """按邮箱查用户（P3-19：注册/建号邮箱查重的友好提示；库侧另有唯一索引兜底并发）""" email = _normalize_email(email) if not email: @@ -367,12 +365,12 @@ def register_ip_exceeded(ip: str, limit: int = 3, window: int = 86400) -> bool: return bool(rows and rows[0]['n'] >= limit) -def get_user_by_id(uid: int) -> Optional[sqlite3.Row]: +def get_user_by_id(uid: int) -> sqlite3.Row | None: rows = _q('SELECT * FROM users WHERE id = ?', (uid,)) return rows[0] if rows else None -def verify_login(username: str, password: str) -> Optional[sqlite3.Row]: +def verify_login(username: str, password: str) -> sqlite3.Row | None: """校验用户名密码。密码正确即返回用户行（含被禁用/封禁账号—— 状态语义由登录路由判断：禁用显示"已被禁用"、封禁显示封禁原因。 原实现把禁用账号提前过滤成 None，路由的"账号已被禁用"分支永不可达）""" @@ -388,7 +386,7 @@ def verify_login(username: str, password: str) -> Optional[sqlite3.Row]: return None -def list_users() -> List[Dict[str, Any]]: +def list_users() -> list[dict[str, Any]]: rows = _q('SELECT * FROM users ORDER BY id DESC') return [_user_public(r, full=True) for r in rows] @@ -449,7 +447,7 @@ def unban_user(uid: int) -> bool: return bool(_x('UPDATE users SET status = 1, ban_reason = \'\', recent_provinces = \'[]\' WHERE id = ?', (uid,))) -_PROVINCE_CACHE: Dict[str, str] = {} +_PROVINCE_CACHE: dict[str, str] = {} def _is_private_ip(ip: str) -> bool: @@ -484,8 +482,9 @@ def query_ip_province(ip: str) -> str: return _PROVINCE_CACHE[ip] province = '' try: - import requests as _rq import urllib.parse as _ul + + import requests as _rq r = _rq.get('http://ip-api.com/json/' + _ul.quote(ip), params={'lang': 'zh-CN', 'fields': 'status,regionName'}, timeout=5) @@ -560,7 +559,7 @@ def set_user_netease_cookie(username: str, cookie: str) -> None: _x('UPDATE users SET netease_cookie = ? WHERE username = ?', (cookie, username)) -def _user_public(r: sqlite3.Row, full: bool = False) -> Dict[str, Any]: +def _user_public(r: sqlite3.Row, full: bool = False) -> dict[str, Any]: d = { 'id': r['id'], 'username': r['username'], @@ -636,7 +635,7 @@ def logout_session() -> None: pass -def current_user() -> Optional[sqlite3.Row]: +def current_user() -> sqlite3.Row | None: """当前登录用户（每请求只查一次库：结果缓存到 flask.g。 一次下载流程会调用本函数 5-6 次——原实现每次都新建 SQLite 连接，公网下是纯浪费。 缓存含 None：未登录同样只查一次。请求内登录态不会变化（登录/注册路由不走本函数）。）""" @@ -648,7 +647,7 @@ def current_user() -> Optional[sqlite3.Row]: return user -def _load_current_user() -> Optional[sqlite3.Row]: +def _load_current_user() -> sqlite3.Row | None: uid = session.get(SESSION_KEY) if not uid: return None @@ -848,7 +847,7 @@ def ledger_open(task_id: str, user_id: int, reserved: int) -> None: conn.close() -def reserve_and_open(uid: int, count: int, ledger_key: str) -> Optional[int]: +def reserve_and_open(uid: int, count: int, ledger_key: str) -> int | None: """A2/B8：预扣与开账同一事务（原两步之间存在崩溃丢账窗口）。 足额：扣减+开账并返回剩余；不限/管理员：开账(reserved=0)返回 -1；不足：返回 None（不开账）。""" count = max(1, int(count)) @@ -941,7 +940,7 @@ def ledger_supplement(task_id: str, extra: int) -> bool: conn.close() -def ledger_get(task_id: str) -> Optional[Dict[str, Any]]: +def ledger_get(task_id: str) -> dict[str, Any] | None: rows = _q('SELECT * FROM billing_ledger WHERE task_id = ?', (str(task_id),)) return dict(rows[0]) if rows else None @@ -1016,7 +1015,7 @@ def ledger_cleanup_settled(max_age_seconds: int = 90 * 86400) -> int: # ============ 装饰器 ============ -def _deny(message: str, code: int = 401, error_code: str = 'unauthorized') -> Tuple[Dict[str, Any], int]: +def _deny(message: str, code: int = 401, error_code: str = 'unauthorized') -> tuple[dict[str, Any], int]: return ({'status': code, 'success': False, 'message': message, 'error_code': error_code}, code) @@ -1174,8 +1173,9 @@ _SMTP_ENC_PREFIX = 'enc1:' # 第 95 条：信封加密标记前缀（存量明 def _smtp_password_encrypt(plaintext: str) -> str: """用 data/secret.key 派生 Fernet 密钥加密 SMTP 授权码（第 95 条 P2-7： 原明文落库+随备份导出=泄露半径扩大；注释编号笔误已按第 97 条纠正）。""" - from cryptography.fernet import Fernet import hashlib as _hl + + from cryptography.fernet import Fernet key = _hl.sha256(get_secret_key().encode('utf-8')).digest() f = Fernet(base64.urlsafe_b64encode(key)) return _SMTP_ENC_PREFIX + f.encrypt(plaintext.encode('utf-8')).decode('utf-8') @@ -1184,14 +1184,15 @@ def _smtp_password_encrypt(plaintext: str) -> str: def _smtp_password_decrypt(ciphertext: str) -> str: if not ciphertext.startswith(_SMTP_ENC_PREFIX): return ciphertext # 存量明文：透明兼容，下次保存时升级为密文 - from cryptography.fernet import Fernet import hashlib as _hl + + from cryptography.fernet import Fernet key = _hl.sha256(get_secret_key().encode('utf-8')).digest() f = Fernet(base64.urlsafe_b64encode(key)) return f.decrypt(ciphertext[len(_SMTP_ENC_PREFIX):].encode('utf-8')).decode('utf-8') -def get_smtp_config() -> Dict[str, Any]: +def get_smtp_config() -> dict[str, Any]: rows = _q('SELECT * FROM smtp_config WHERE id = 1') cfg = dict(rows[0]) if rows else {'host': '', 'port': 587, 'sender': '', 'password': ''} # 第 95 条：读取时解密密码——调用方（SMTP 连接/后台展示）拿到的始终是明文 @@ -1399,7 +1400,7 @@ def restore_verify_code(email: str, code: str, ttl_seconds: int, purpose: str = def generate_cards(count: int, credits: int, created_by: str = '', - prefix: str = '', note: str = '', valid_days: int = 0) -> List[str]: + prefix: str = '', note: str = '', valid_days: int = 0) -> list[str]: """生成次数卡密（credits = 每张卡可下载次数） prefix: 自定义卡密前缀（默认 VIP-） note: 备注信息 @@ -1425,7 +1426,7 @@ def generate_cards(count: int, credits: int, created_by: str = '', return codes -def list_cards(limit: int = 100, offset: int = 0) -> List[Dict[str, Any]]: +def list_cards(limit: int = 100, offset: int = 0) -> list[dict[str, Any]]: """卡密列表（F85：支持分页 limit/offset，管理路由透传）""" limit = max(1, min(int(limit), 500)) offset = max(0, int(offset)) @@ -1485,7 +1486,7 @@ def delete_card(code: str) -> None: conn.close() -def redeem_card(code: str, username: str) -> Tuple[int, int]: +def redeem_card(code: str, username: str) -> tuple[int, int]: """兑换卡密，返回 (卡密次数, 新剩余次数)（F25：标记已用与加余额同事务，失败整体回滚不吞卡）""" code = (code or '').strip().upper() with _DB_LOCK: @@ -1533,7 +1534,7 @@ def redeem_card(code: str, username: str) -> Tuple[int, int]: # ============ 公告 ============ -def get_announcement() -> Dict[str, Any]: +def get_announcement() -> dict[str, Any]: rows = _q('SELECT * FROM announcements WHERE id = 1') return dict(rows[0]) if rows else {'title': '', 'content': '', 'popup_mode': 'off'} @@ -1606,7 +1607,7 @@ def create_backup() -> str: conn.close() -def list_backups() -> List[Dict[str, Any]]: +def list_backups() -> list[dict[str, Any]]: """备份列表（F44：按文件修改时间倒序——按名字排序会让 upload_ 永远排前，误删新自动备份）""" if not BACKUP_DIR.exists(): return [] @@ -1622,7 +1623,7 @@ def list_backups() -> List[Dict[str, Any]]: return out -def backup_path(name: str) -> Optional[Path]: +def backup_path(name: str) -> Path | None: name = Path(name).name # 防目录穿越 p = BACKUP_DIR / name return p if p.exists() and p.suffix == '.db' else None diff --git a/main.py b/main.py index 3fd8e91..d7b94bb 100644 --- a/main.py +++ b/main.py @@ -9,16 +9,13 @@ - 健康检查 """ -import base64 -import json import collections import hashlib import hmac +import json import logging import os import re -import requests -import secrets import shutil import sys import threading @@ -26,27 +23,45 @@ import time import traceback import uuid import zipfile -from dataclasses import dataclass from pathlib import Path -from typing import Dict, Any, Optional, Tuple, List -from urllib.parse import quote, urlsplit, parse_qs -from flask import Flask, request, send_file, render_template, Response, jsonify, stream_with_context, make_response, g +from typing import Any +from urllib.parse import parse_qs, quote, urlsplit + +import requests +from flask import Flask, Response, g, make_response, render_template, request, send_file, stream_with_context try: - from music_api import ( - NeteaseAPI, APIException, QualityLevel, - url_v1, name_v1, lyric_v1, search_music, - playlist_detail, album_detail, - toplist_list, artist_songs, song_comments, - account_info, user_playlists, detect_vip_by_probe, - playlist_subscribe, album_subscribe, post_comment, - load_cookies, - ) - from music_downloader import MusicDownloader, DownloadException, DownloadCanceled, CountingSemaphore, fetch_cover_bytes - from qr_login import QRLoginClient - import podcast_api import auth_system + import podcast_api from auth_routes import register_auth_routes + from music_api import ( + APIException, + NeteaseAPI, + account_info, + album_detail, + album_subscribe, + artist_songs, + detect_vip_by_probe, + load_cookies, + lyric_v1, + name_v1, + playlist_detail, + playlist_subscribe, + post_comment, + search_music, + song_comments, + toplist_list, + url_v1, + user_playlists, + ) + from music_downloader import ( + CountingSemaphore, + DownloadCanceled, + DownloadException, + MusicDownloader, + fetch_cover_bytes, + ) + from qr_login import QRLoginClient except ImportError as e: # 坑 11：导入失败发生在 _safe_console_print 定义之前，用纯 ASCII 保证任何编码下可打印 print("ImportError: " + str(e)) @@ -58,35 +73,46 @@ except ImportError as e: # 常量/配置、运行时状态、纯工具、音频签名已迁出至 core 包，以**同一对象/同一实现** # 供 Flask 与迁移目标 FastAPI 共用。此处以原名称再导出（别名），既有 250+ 调用点零改动。 from core.config import ( - APP_VERSION, RESULT_RETENTION_SECONDS, Settings, + APP_VERSION, + BATCH_ZIP_CONCURRENCY, + CANCEL_AFTER_SECONDS, + HEARTBEAT_INTERVAL, + MAX_ACTIVE_TASKS_PER_USER, + MAX_TERMINAL_TASKS, + MAX_TOTAL_TASKS, + PAUSE_AFTER_SECONDS, + RESULT_RETENTION_SECONDS, USER_COOKIE_FILE, - PAUSE_AFTER_SECONDS, CANCEL_AFTER_SECONDS, HEARTBEAT_INTERVAL, - MAX_ACTIVE_TASKS_PER_USER, MAX_TOTAL_TASKS, MAX_TERMINAL_TASKS, - BATCH_ZIP_CONCURRENCY, VALID_QUALITY_LEVELS, + VALID_QUALITY_LEVELS, + Settings, ) +from core.security import init_secret_key as _init_audio_secret +from core.security import protect_audio_urls as _protect_audio_urls +from core.security import verify_audio_token as _verify_audio_token from core.state import ( - RATE_BUCKETS as _RATE_BUCKETS, RATE_LOCK as _RATE_LOCK, - BATCH_DURATIONS, BATCH_DURATIONS_LOCK, - DOWNLOAD_TASKS, DOWNLOAD_TASKS_LOCK, - DEVICE_LAST_SEEN, DEVICE_LAST_SEEN_LOCK, - REQUEST_STATS, STATS_LOCK as _STATS_LOCK, REQ_LOG, - GLOBAL_MUSIC_U_CACHE as _GLOBAL_MUSIC_U_CACHE, - VIP_CACHE as _VIP_CACHE, VIP_USER_CACHE as _VIP_USER_CACHE, - SERVER_NICK_CACHE as _SERVER_NICK_CACHE, SERVER_NICK_LOCK as _SERVER_NICK_LOCK, - VIP_PROBE_LOCK as _VIP_PROBE_LOCK, -) -from core.utils import ( - safe_int as _safe_int, safe_console_print as _safe_console_print, - srt_time as _srt_time, format_lyric as _format_lyric, - parse_meta_fields as _parse_meta_fields, expand_shortlink as _expand_shortlink, -) -from core.security import ( - init_secret_key as _init_audio_secret, - is_audio_url as _is_audio_url, - make_audio_token as _make_audio_token, - verify_audio_token as _verify_audio_token, - protect_audio_urls as _protect_audio_urls, + BATCH_DURATIONS, + BATCH_DURATIONS_LOCK, + DEVICE_LAST_SEEN, + DEVICE_LAST_SEEN_LOCK, + DOWNLOAD_TASKS, + DOWNLOAD_TASKS_LOCK, + REQ_LOG, + REQUEST_STATS, ) +from core.state import GLOBAL_MUSIC_U_CACHE as _GLOBAL_MUSIC_U_CACHE +from core.state import RATE_BUCKETS as _RATE_BUCKETS +from core.state import RATE_LOCK as _RATE_LOCK +from core.state import SERVER_NICK_CACHE as _SERVER_NICK_CACHE +from core.state import SERVER_NICK_LOCK as _SERVER_NICK_LOCK +from core.state import STATS_LOCK as _STATS_LOCK +from core.state import VIP_CACHE as _VIP_CACHE +from core.state import VIP_PROBE_LOCK as _VIP_PROBE_LOCK +from core.state import VIP_USER_CACHE as _VIP_USER_CACHE +from core.utils import expand_shortlink as _expand_shortlink +from core.utils import format_lyric as _format_lyric +from core.utils import parse_meta_fields as _parse_meta_fields +from core.utils import safe_console_print as _safe_console_print +from core.utils import safe_int as _safe_int # 向后兼容别名：既有代码/文档沿用 APIConfig 名称 APIConfig = Settings @@ -94,9 +120,9 @@ APIConfig = Settings class APIResponse: """API响应工具类""" - + @staticmethod - def success(data: Any = None, message: str = 'success', status_code: int = 200) -> Tuple[Dict[str, Any], int]: + def success(data: Any = None, message: str = 'success', status_code: int = 200) -> tuple[dict[str, Any], int]: """成功响应""" response = { 'status': status_code, @@ -106,9 +132,9 @@ class APIResponse: if data is not None: response['data'] = data return response, status_code - + @staticmethod - def error(message: str, status_code: int = 400, error_code: str = None) -> Tuple[Dict[str, Any], int]: + def error(message: str, status_code: int = 400, error_code: str | None = None) -> tuple[dict[str, Any], int]: """错误响应""" response = { 'status': status_code, @@ -122,7 +148,7 @@ class APIResponse: class MusicAPIService: """音乐API服务类""" - + def __init__(self, config: APIConfig): self.config = config self.logger = self._setup_logger() @@ -136,12 +162,12 @@ class MusicAPIService: self.downloads_path.mkdir(exist_ok=True) self.logger.info(f"音乐API服务初始化完成，下载目录: {self.downloads_path.absolute()}") - + def _setup_logger(self) -> logging.Logger: """设置日志记录器""" logger = logging.getLogger('music_api') logger.setLevel(getattr(logging, self.config.log_level.upper())) - + if not logger.handlers: # 控制台处理器 console_handler = logging.StreamHandler() @@ -150,7 +176,7 @@ class MusicAPIService: ) console_handler.setFormatter(console_formatter) logger.addHandler(console_handler) - + # 文件处理器（轮转：单文件 5MB，保留 3 份，防止长期运行无限膨胀） try: from logging.handlers import RotatingFileHandler @@ -163,10 +189,10 @@ class MusicAPIService: logger.addHandler(file_handler) except Exception as e: logger.warning(f"无法创建日志文件: {e}") - + return logger - - def _get_cookies(self) -> Dict[str, str]: + + def _get_cookies(self) -> dict[str, str]: """直接从 cookie.txt 读取并解析为 dict""" try: cookies = load_cookies() @@ -176,7 +202,7 @@ class MusicAPIService: except Exception as e: self.logger.error(f"读取 cookie.txt 异常: {e}") return {} - + def _extract_music_id(self, id_or_url: str) -> str: """提取音乐ID（F7/F74：正规 URL 解析替代子串/删除非数字启发式） - 短链仅当整串就是一个 163cn.tv 短链 URL 时才请求展开（防 SSRF） @@ -207,7 +233,7 @@ class MusicAPIService: except Exception as e: self.logger.error(f"提取音乐ID失败: {e}") return str(id_or_url or '').strip() - + def _format_file_size(self, size_bytes) -> str: """格式化文件大小（上游 size 可能为 null/垃圾值：容错返回"大小未知"而非 500）""" try: @@ -225,12 +251,12 @@ class MusicAPIService: unit_index += 1 return f"{size:.2f}{units[unit_index]}" - + def _get_quality_display_name(self, quality: str) -> str: """获取音质显示名称""" quality_names = { 'standard': "标准音质", - 'exhigh': "极高音质", + 'exhigh': "极高音质", 'lossless': "无损音质", 'hires': "Hi-Res音质", 'sky': "沉浸环绕声", @@ -239,15 +265,15 @@ class MusicAPIService: 'dolby': "杜比全景声" } return quality_names.get(quality, f"未知音质({quality})") - - def _validate_request_params(self, required_params: Dict[str, Any]) -> Optional[Tuple[Dict[str, Any], int]]: + + def _validate_request_params(self, required_params: dict[str, Any]) -> tuple[dict[str, Any], int] | None: """验证请求参数""" for param_name, param_value in required_params.items(): if not param_value: return APIResponse.error(f"参数 '{param_name}' 不能为空", 400) return None - - def _safe_get_request_data(self) -> Dict[str, Any]: + + def _safe_get_request_data(self) -> dict[str, Any]: """安全获取请求数据""" try: if request.method == 'GET': @@ -541,7 +567,6 @@ def api_task_cancel(): try: data = api_service._safe_get_request_data() task_id = str(data.get('task_id') or '').strip() - device_id = str(data.get('device_id') or '').strip()[:64] # 第 98 条：与注册同口径钳 64（防超长键灌爆 DEVICE_LAST_SEEN） if not task_id: return APIResponse.error("缺少 task_id 参数") with DOWNLOAD_TASKS_LOCK: @@ -685,7 +710,7 @@ BATCH_ZIP_SEM = CountingSemaphore(BATCH_ZIP_CONCURRENCY) # USER_COOKIE_FILE → core.config -def _register_download_task(task: Dict[str, Any], reserved: int, task_id_hint: str = '') -> Optional[str]: +def _register_download_task(task: dict[str, Any], reserved: int, task_id_hint: str = '') -> str | None: """任务登记统一入口：归属校验、数量上限、账本承接。失败返回错误描述（str），成功返回 None。 - F05：任务必须绑定站点用户（提交接口都有 login_required，这里再取一次会话） @@ -759,7 +784,7 @@ def _register_download_task(task: Dict[str, Any], reserved: int, task_id_hint: s return None -def _start_task_thread(task_id: str, target, args=()) -> Optional[str]: +def _start_task_thread(task_id: str, target, args=()) -> str | None: """启动任务 worker（F90：启动失败时收敛任务终态并按账本退费，不留僵尸 queued 记录）。 A2：退款只走账本（本处 settle 后装饰器看到账已关，不会重复退）。 R5-14：线程句柄存入任务字典——监控循环对 4h 超时任务若发现 worker 线程已死 @@ -784,7 +809,7 @@ def _start_task_thread(task_id: str, target, args=()) -> Optional[str]: return f'任务启动失败：{e}' -def _task_owned_by_current_user(task: Dict[str, Any]) -> bool: +def _task_owned_by_current_user(task: dict[str, Any]) -> bool: """任务归属校验（F05：站点用户是任务的第一归属，device_id 仅作辅助）""" user = auth_system.current_user() if not user: @@ -823,7 +848,7 @@ def _mark_task_terminal(task_id: str, status: str, msg: str = '') -> None: pass -def _task_files(task: Dict[str, Any]) -> List[str]: +def _task_files(task: dict[str, Any]) -> list[str]: """任务当前关联的可清理文件路径（交付 ZIP/交付副本；缓存文件不在此列）""" result = task.get('result') or {} files = [] @@ -875,9 +900,9 @@ def _task_monitor_loop() -> None: # ---- 终态任务清理（F38/F106：按完成时刻；删除失败保留记录下轮重试，最多 8 次）---- # 第 91 条（P2-1）：锁内只收集过期清单，文件删除移到锁外——GB 级任务目录 # 的删除曾阻塞锁数十秒，期间全站进度上报/提交/取消/领取全部停摆 - finished_done: List[str] = [] - finished_retry: List[str] = [] - expired: List[tuple] = [] + finished_done: list[str] = [] + finished_retry: list[str] = [] + expired: list[tuple] = [] with DOWNLOAD_TASKS_LOCK: for tid, task in DOWNLOAD_TASKS.items(): if task.get('status') not in ('done', 'error', 'canceled'): @@ -890,7 +915,7 @@ def _task_monitor_loop() -> None: continue expired.append((tid, _task_files(task), api_service.downloads_path / 'tasks' / tid)) - cleanup_results: List[tuple] = [] + cleanup_results: list[tuple] = [] for tid, files, task_dir in expired: all_removed = True for f in files: @@ -929,7 +954,7 @@ def _task_monitor_loop() -> None: if finished_retry: api_service.logger.info(f"监控线程保留清理失败任务待重试: {finished_retry}") # ---- 长时间未结束的任务：置取消（含 cancel Event，唤醒可能的阻塞等待）---- - timeout_settle: List[str] = [] + timeout_settle: list[str] = [] with DOWNLOAD_TASKS_LOCK: for tid, task in DOWNLOAD_TASKS.items(): if (task.get('status') in ('queued', 'running', 'paused') @@ -1118,9 +1143,9 @@ def _record_log(action: str, ip: str, song: str) -> None: del REQ_LOG[:len(REQ_LOG) - 2000] -def _parse_cookie_str(raw) -> Dict[str, str]: +def _parse_cookie_str(raw) -> dict[str, str]: """将 'k=v;k2=v2' 格式的 Cookie 字符串解析为字典""" - cookies: Dict[str, str] = {} + cookies: dict[str, str] = {} if not raw: return cookies for part in str(raw).split(';'): @@ -1154,7 +1179,7 @@ def _global_music_u_cache_reset() -> None: _GLOBAL_MUSIC_U_CACHE['ts'] = 0.0 -def _resolve_user_cookies(data: Dict[str, Any]) -> Dict[str, str]: +def _resolve_user_cookies(data: dict[str, Any]) -> dict[str, str]: """解析个人账号 Cookie：仅使用浏览器传入的 user_cookie（不落服务器，杜绝跨设备访问他人账号）。 第 92 条（owner 拍板）：全局账号（cookie.txt）在发评/收藏/下载等个人身份操作上 零权限、零功能、零入口——浏览器提交的 MUSIC_U 若与全局账号一致（全局 Cookie @@ -1177,7 +1202,7 @@ def _resolve_user_cookies(data: Dict[str, Any]) -> Dict[str, str]: return ck -def _build_download_base(filename_format: str, music_info: Dict[str, Any]) -> str: +def _build_download_base(filename_format: str, music_info: dict[str, Any]) -> str: """按文件名格式生成下载基础名（不含扩展名）""" name = music_info.get('name', '未知歌曲') artist = music_info.get('artist_string', '') or '未知歌手' @@ -1199,7 +1224,7 @@ def _build_download_base(filename_format: str, music_info: Dict[str, Any]) -> st # _srt_time / _format_lyric / _parse_meta_fields 已迁出至 core.utils（此处 import 同名共享） -def _verify_admin_password(data) -> Optional[Tuple[Dict[str, Any], int]]: +def _verify_admin_password(data) -> tuple[dict[str, Any], int] | None: """校验管理员权限：仅站点管理员会话（旧版口令已废除，2026-09-15）""" _user = auth_system.current_user() if _user is not None and _user['role'] == 'admin': @@ -1219,7 +1244,7 @@ def _vip_name(vip_type: int) -> str: }.get(vip_type, '普通用户') -def _detect_vip(cookies: Dict[str, str]) -> Tuple[int, str, List[str]]: +def _detect_vip(cookies: dict[str, str]) -> tuple[int, str, list[str]]: """检测账号会员类型与可解析音质： 优先账号接口（vipType 精确），并始终用音质探测接口实测可解析列表。 第 100 条：探测单飞锁（有界等待 0.5s）——上游黑洞时并发请求不再各自串行 @@ -1488,7 +1513,7 @@ def health_check(): } return APIResponse.success(health_info, "API服务运行正常") - + except Exception as e: api_service.logger.error(f"健康检查失败: {e}") return APIResponse.error("服务暂时不可用，请稍后再试", 500) @@ -1506,23 +1531,23 @@ def get_song_info(): url = data.get('url') level = data.get('level', 'lossless') info_type = data.get('type', 'url') - + # 参数验证 if not song_ids and not url: return APIResponse.error("必须提供 'ids'、'id' 或 'url' 参数") - + # 提取音乐ID music_id = api_service._extract_music_id(song_ids or url) # P2-2：ID 白名单（与 /album、/artist_songs、/comments 同口径—— # 垃圾输入不再透传上游并把上游报错原文回显给公网） if not str(music_id).strip().isdigit(): return APIResponse.error("无效的歌曲ID，请输入数字ID或分享链接") - + # 验证音质参数 valid_levels = ['standard', 'exhigh', 'lossless', 'hires', 'sky', 'jyeffect', 'jymaster', 'dolby'] if level not in valid_levels: return APIResponse.error(f"无效的音质参数，支持: {', '.join(valid_levels)}") - + # 验证类型参数 valid_types = ['url', 'name', 'lyric', 'json'] if info_type not in valid_types: @@ -1536,7 +1561,7 @@ def get_song_info(): return APIResponse.error("请先扫码登录个人账号后再使用个人解析", 401) else: cookies = api_service._get_cookies() - + # 根据类型获取不同信息 if info_type == 'url': result = url_v1(music_id, level, cookies) @@ -1559,15 +1584,15 @@ def get_song_info(): return APIResponse.success(response_data, "获取歌曲URL成功") else: return APIResponse.error("获取音乐URL失败，可能是版权限制或音质不支持", 404) - + elif info_type == 'name': result = name_v1(music_id) return APIResponse.success(result, "获取歌曲信息成功") - + elif info_type == 'lyric': result = lyric_v1(music_id, cookies) return APIResponse.success(result, "获取歌词成功") - + elif info_type == 'json': # 获取完整的歌曲信息（用于前端解析） song_info = name_v1(music_id) @@ -1576,10 +1601,10 @@ def get_song_info(): lyric_info = lyric_v1(music_id, cookies) except Exception: lyric_info = None # 歌词接口抖动不阻断解析 - + if not song_info or 'songs' not in song_info or not song_info['songs']: return APIResponse.error("未找到歌曲信息", 404) - + song_data = song_info['songs'][0] # S1/F94：可选字段为 null（而非缺失）时同样容错，不让有效歌曲解析失败 @@ -1598,7 +1623,7 @@ def get_song_info(): 'tlyric': ((lyric_info or {}).get('tlyric') or {}).get('lyric') or '', 'yrc': ((lyric_info or {}).get('yrc') or {}).get('lyric') or '' } - + # 添加URL和大小信息 if url_info and url_info.get('data') and len(url_info['data']) > 0: url_data = url_info['data'][0] @@ -1612,7 +1637,7 @@ def get_song_info(): 'url': '', 'size': '获取失败' }) - + # 记录用户解析操作（IP + 歌曲名） try: _record_log('解析', auth_system._client_ip(), response_data.get('name') or str(music_id)) @@ -1620,7 +1645,7 @@ def get_song_info(): pass return APIResponse.success(response_data, "获取歌曲信息成功") - + except APIException as e: api_service.logger.error(f"API调用失败: {e}") return APIResponse.error("服务暂时不可用，请稍后再试", 500) @@ -1641,20 +1666,20 @@ def search_music_api(): limit = max(1, _safe_int(data.get('limit'), 30)) offset = max(0, _safe_int(data.get('offset'), 0)) search_type = data.get('type', '1') # 1-歌曲, 10-专辑, 100-歌手, 1000-歌单 - + # 参数验证 validation_error = api_service._validate_request_params({'keyword': keyword}) if validation_error: return validation_error - + # 限制搜索数量 if limit > 100: limit = 100 - + cookies = api_service._get_cookies() # F45：offset 透传给上游，翻页不再永远从第一页开始 result = search_music(keyword, cookies, limit, int(search_type), offset=offset) - + # 向后兼容：歌曲搜索结果补充 artist_string 字段；歌单结果遮蔽服务器账号昵称 if result: for song in result: @@ -1665,11 +1690,11 @@ def search_music_api(): song['creator'] = _mask_server_nickname(song['creator']) if song.get('name'): song['name'] = _mask_server_nickname(song['name']) - + return APIResponse.success(result, "搜索完成") - + except ValueError as e: - return APIResponse.error(f"参数格式错误: {str(e)}") + return APIResponse.error(f"参数格式错误: {e!s}") except Exception as e: api_service.logger.error(f"搜索音乐异常: {e}\n{traceback.format_exc()}") return APIResponse.error("服务暂时不可用，请稍后再试", 500) @@ -1684,7 +1709,7 @@ def get_playlist(): # 获取请求参数 data = api_service._safe_get_request_data() playlist_id = data.get('id') - + # 参数验证 validation_error = api_service._validate_request_params({'playlist_id': playlist_id}) if validation_error: @@ -1709,9 +1734,9 @@ def get_playlist(): 'status': 'success', 'playlist': result } - + return APIResponse.success(response_data, "获取歌单详情成功") - + except Exception as e: api_service.logger.error(f"获取歌单异常: {e}\n{traceback.format_exc()}") return APIResponse.error("服务暂时不可用，请稍后再试", 500) @@ -1740,15 +1765,15 @@ def get_album(): cookies = api_service._get_cookies() result = album_detail(album_id, cookies) - + # 适配前端期望的响应格式 response_data = { 'status': 200, 'album': result } - + return APIResponse.success(response_data, "获取专辑详情成功") - + except Exception as e: api_service.logger.error(f"获取专辑异常: {e}\n{traceback.format_exc()}") return APIResponse.error("服务暂时不可用，请稍后再试", 500) @@ -2219,8 +2244,8 @@ def _podcast_stream_download(url: str, target: Path, total_size: int, return done -def _podcast_resolve_audio(program: Dict[str, Any], quality: str, - cookies: Dict[str, str]) -> Dict[str, Any]: +def _podcast_resolve_audio(program: dict[str, Any], quality: str, + cookies: dict[str, str]) -> dict[str, Any]: """解析单集音频直链，失败抛 DownloadException""" song_id = program.get('main_song_id') if not song_id: @@ -2255,7 +2280,7 @@ def get_song_comments_api(): result = song_comments(song_id, limit, offset) return APIResponse.success(result, "获取评论成功") except ValueError as e: - return APIResponse.error(f"参数格式错误: {str(e)}") + return APIResponse.error(f"参数格式错误: {e!s}") except APIException as e: api_service.logger.error(f"获取评论失败: {e}") return APIResponse.error("服务暂时不可用，请稍后再试", 500) @@ -2687,8 +2712,7 @@ def user_comment(): # ============ 全局账号歌单（管理员 cookie.txt，无需游客扫码） ============ - -VALID_QUALITY_LEVELS = ['standard', 'exhigh', 'lossless', 'hires', 'sky', 'jyeffect', 'jymaster', 'dolby'] +# VALID_QUALITY_LEVELS 已迁至 core.config（顶部 import 同名共享，此处不再重复定义） @app.route('/global/playlists', methods=['GET', 'POST']) @@ -2942,7 +2966,7 @@ def download_music_api(): music_id = data.get('id') quality = data.get('quality', 'lossless') return_format = data.get('format', 'file') # file 或 json - + # 参数验证 validation_error = api_service._validate_request_params({'music_id': music_id}) if validation_error: @@ -2951,16 +2975,16 @@ def download_music_api(): music_id = str(music_id).strip() if not music_id.isdigit(): return APIResponse.error("无效的歌曲 ID") - + # 验证音质参数 valid_qualities = ['standard', 'exhigh', 'lossless', 'hires', 'sky', 'jyeffect', 'jymaster', 'dolby'] if quality not in valid_qualities: return APIResponse.error(f"无效的音质参数，支持: {', '.join(valid_qualities)}") - + # 验证返回格式 if return_format not in ['file', 'json']: return APIResponse.error("返回格式只支持 'file' 或 'json'") - + music_id = api_service._extract_music_id(music_id) # 解析账号选择：user=游客个人账号，默认 admin=全局管理员账号 @@ -3007,7 +3031,6 @@ def download_music_api(): # R5-P3：对外泛化（error_message 可能含上游细节），原文留日志 api_service.logger.error(f"下载失败: {download_result.error_message}") return APIResponse.error("下载失败，请稍后重试或更换音质", 500) - file_path = Path(download_result.file_path) api_service.logger.info(f"下载完成: {filename}") except DownloadException as e: api_service.logger.error(f"下载异常: {e}") @@ -3265,10 +3288,9 @@ def _batch_zip_worker(task_id: str) -> None: import tempfile as _tempfile from concurrent.futures import ThreadPoolExecutor, as_completed - zip_tmp_path: Optional[str] = None - zipped_files: List[str] = [] + zip_tmp_path: str | None = None ok_count = 0 - failed: List[str] = [] + failed: list[str] = [] used_arcnames = set() # ZIP 内已占用文件名（同名歌曲去重，防止包内互相覆盖） info_pool = None info_futures = {} @@ -3562,7 +3584,7 @@ def _batch_zip_worker(task_id: str) -> None: refunded = _settle_batch_billing(task_id, delivered, 'error') if delivered > 0 and zip_tmp_path: _mark_task_terminal(task_id, 'error', - f"部分歌曲已打包完成，可领取" + + "部分歌曲已打包完成，可领取" + (f"，退回 {refunded} 次" if refunded else '')) with DOWNLOAD_TASKS_LOCK: t = DOWNLOAD_TASKS.get(task_id) @@ -3603,9 +3625,9 @@ def _podcast_download_worker(task_id: str) -> None: """ import tempfile as _tempfile - file_path_str: Optional[str] = None - zip_tmp_path: Optional[str] = None - task_dir: Optional[Path] = None + file_path_str: str | None = None + zip_tmp_path: str | None = None + task_dir: Path | None = None try: with DOWNLOAD_TASKS_LOCK: task = DOWNLOAD_TASKS[task_id] @@ -3643,7 +3665,7 @@ def _podcast_download_worker(task_id: str) -> None: raise DownloadCanceled("任务已取消") file_path_str = str(target) - result: Dict[str, Any] + result: dict[str, Any] if need_zip: fd, zip_tmp_path = _tempfile.mkstemp(prefix='podcast_zip_', suffix='.zip', dir=str(task_dir)) @@ -3737,7 +3759,7 @@ def _podcast_download_worker(task_id: str) -> None: shutil.rmtree(str(task_dir), ignore_errors=True) -def _podcast_cleanup_files(paths: List[Optional[str]]) -> None: +def _podcast_cleanup_files(paths: list[str | None]) -> None: """清理播客任务产生的临时/残留文件""" for p in paths: if not p: @@ -3759,9 +3781,9 @@ def _podcast_zip_worker(task_id: str) -> None: """整台播客打包：逐集 下载→入包即删，支持取消与断线暂停（复用批量任务机制）""" import tempfile as _tempfile - zip_tmp_path: Optional[str] = None + zip_tmp_path: str | None = None ok_count = 0 - failed: List[str] = [] + failed: list[str] = [] # 第 97 条（P1-1）：zip_valid 必须在函数头初始化——空电台/无单集/补扣失败等 # 正常业务错误在 try 内早期 raise，异常处理器引用未绑定名会二次崩溃 worker 线程 zip_valid = False @@ -4064,7 +4086,6 @@ def download_batch_zip_result(): try: data = api_service._safe_get_request_data() task_id = str(data.get('task_id') or '').strip() - device_id = str(data.get('device_id') or '').strip()[:64] # 第 98 条：与注册同口径钳 64（防超长键灌爆 DEVICE_LAST_SEEN） if not task_id: return APIResponse.error("缺少 task_id 参数") with DOWNLOAD_TASKS_LOCK: @@ -4127,9 +4148,9 @@ def _single_download_worker(task_id: str) -> None: """后台单曲下载：下载音频（可选 ZIP 打包），支持取消；完成后由 /download/result 交付，35 分钟保留期清理""" import tempfile as _tempfile - file_path_str: Optional[str] = None - zip_tmp_path: Optional[str] = None - task_dir_path: Optional[str] = None + file_path_str: str | None = None + zip_tmp_path: str | None = None + task_dir_path: str | None = None try: with DOWNLOAD_TASKS_LOCK: task = DOWNLOAD_TASKS[task_id] @@ -4308,7 +4329,6 @@ def download_single_result(): try: data = api_service._safe_get_request_data() task_id = str(data.get('task_id') or '').strip() - device_id = str(data.get('device_id') or '').strip()[:64] # 第 98 条：与注册同口径钳 64（防超长键灌爆 DEVICE_LAST_SEEN） if not task_id: return APIResponse.error("缺少 task_id 参数") with DOWNLOAD_TASKS_LOCK: @@ -4384,9 +4404,9 @@ def api_info(): 'max_body_bytes': '2MB（备份上传 200MB）' } } - + return APIResponse.success(info, "API信息获取成功") - + except Exception as e: api_service.logger.error(f"获取API信息异常: {e}") return APIResponse.error("服务暂时不可用，请稍后再试", 500) @@ -4437,21 +4457,21 @@ def start_api_server(): _safe_console_print(f"📁 下载目录: {api_service.downloads_path.absolute()}") _safe_console_print(f"📋 日志级别: {config.log_level}") _safe_console_print("\n📚 API端点:") - _safe_console_print(f" ├─ GET /health - 健康检查") - _safe_console_print(f" ├─ POST /search - 搜索音乐（支持歌曲/歌单/专辑/歌手）") - _safe_console_print(f" ├─ POST /song - 获取歌曲信息") - _safe_console_print(f" ├─ POST /playlist - 获取歌单详情") - _safe_console_print(f" ├─ POST /album - 获取专辑详情") - _safe_console_print(f" ├─ POST /toplist - 获取官方榜单列表") - _safe_console_print(f" ├─ POST /artist_songs - 获取歌手信息与热门歌曲") - _safe_console_print(f" ├─ POST /comments - 获取歌曲评论") - _safe_console_print(f" ├─ POST /download - 下载音乐") - _safe_console_print(f" ├─ GET/POST /global/playlists - 全局账号歌单列表") - _safe_console_print(f" ├─ GET/POST /global/playlist - 全局账号歌单详情") - _safe_console_print(f" ├─ GET/POST /api/playlist/play - 公开歌单播放API") - _safe_console_print(f" └─ GET /api/info - API信息") + _safe_console_print(" ├─ GET /health - 健康检查") + _safe_console_print(" ├─ POST /search - 搜索音乐（支持歌曲/歌单/专辑/歌手）") + _safe_console_print(" ├─ POST /song - 获取歌曲信息") + _safe_console_print(" ├─ POST /playlist - 获取歌单详情") + _safe_console_print(" ├─ POST /album - 获取专辑详情") + _safe_console_print(" ├─ POST /toplist - 获取官方榜单列表") + _safe_console_print(" ├─ POST /artist_songs - 获取歌手信息与热门歌曲") + _safe_console_print(" ├─ POST /comments - 获取歌曲评论") + _safe_console_print(" ├─ POST /download - 下载音乐") + _safe_console_print(" ├─ GET/POST /global/playlists - 全局账号歌单列表") + _safe_console_print(" ├─ GET/POST /global/playlist - 全局账号歌单详情") + _safe_console_print(" ├─ GET/POST /api/playlist/play - 公开歌单播放API") + _safe_console_print(" └─ GET /api/info - API信息") _safe_console_print("\n🎵 支持的音质:") - _safe_console_print(f" standard, exhigh, lossless, hires, sky, jyeffect, jymaster, dolby") + _safe_console_print(" standard, exhigh, lossless, hires, sky, jyeffect, jymaster, dolby") _safe_console_print("="*60) _safe_console_print(f"⏰ 启动时间: {time.strftime('%Y-%m-%d %H:%M:%S')}") _safe_console_print("🌟 服务已就绪，等待请求...\n") diff --git a/music_api.py b/music_api.py index 3f4781b..3ecedb8 100644 --- a/music_api.py +++ b/music_api.py @@ -10,19 +10,22 @@ - 二维码登录 """ +import asyncio import json import os -import urllib.parse import time -from random import randrange -from typing import Dict, List, Optional, Tuple, Any -from hashlib import md5 +import urllib.parse from enum import Enum +from hashlib import md5 +from random import randrange +from typing import Any -import requests from cryptography.hazmat.primitives import padding from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from core import aiohttp_compat as requests # 异步 HTTP 垫片（requests 同形 API） +from core.async_bridge import run_sync # 模块级同步兼容接口用的同步/异步桥 + class QualityLevel(Enum): """音质等级枚举""" @@ -42,7 +45,7 @@ class APIConstants: AES_KEY = b"e82ckenh8dichen8" USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Safari/537.36 Chrome/91.0.4472.164 NeteaseMusicDesktop/2.10.2.200154' REFERER = 'https://music.163.com/' - + # API URLs SONG_URL_V1 = "https://interface3.music.163.com/eapi/song/enhance/player/url/v1" SONG_DETAIL_V3 = "https://interface3.music.163.com/api/v3/song/detail" @@ -60,7 +63,7 @@ class APIConstants: COMMENT_POST_API = 'https://music.163.com/weapi/comment' QR_UNIKEY_API = 'https://interface3.music.163.com/eapi/login/qrcode/unikey' QR_LOGIN_API = 'https://interface3.music.163.com/eapi/login/qrcode/client/login' - + # 默认配置 DEFAULT_CONFIG = { "os": "pc", @@ -68,7 +71,7 @@ class APIConstants: "osver": "", "deviceId": "pyncm!" } - + DEFAULT_COOKIES = { "os": "pc", "appver": "", @@ -83,31 +86,31 @@ class CryptoUtils: def hex_digest(data: bytes) -> str: """将字节数据转换为十六进制字符串""" return "".join([hex(d)[2:].zfill(2) for d in data]) - + @staticmethod def hash_digest(text: str) -> bytes: """计算MD5哈希值""" return md5(text.encode("utf-8")).digest() - + @staticmethod def hash_hex_digest(text: str) -> str: """计算MD5哈希值并转换为十六进制字符串""" return CryptoUtils.hex_digest(CryptoUtils.hash_digest(text)) - + @staticmethod - def encrypt_params(url: str, payload: Dict[str, Any]) -> str: + def encrypt_params(url: str, payload: dict[str, Any]) -> str: """加密请求参数""" url_path = urllib.parse.urlparse(url).path.replace("/eapi/", "/api/") digest = CryptoUtils.hash_hex_digest(f"nobody{url_path}use{json.dumps(payload)}md5forencrypt") params = f"{url_path}-36cd479b6b5-{json.dumps(payload)}-36cd479b6b5-{digest}" - + # AES加密 padder = padding.PKCS7(algorithms.AES(APIConstants.AES_KEY).block_size).padder() padded_data = padder.update(params.encode()) + padder.finalize() cipher = Cipher(algorithms.AES(APIConstants.AES_KEY), modes.ECB()) encryptor = cipher.encryptor() enc = encryptor.update(padded_data) + encryptor.finalize() - + return CryptoUtils.hex_digest(enc) @@ -135,7 +138,7 @@ class WeapiCrypto: return encryptor.update(padded) + encryptor.finalize() @staticmethod - def encrypt(payload: Dict[str, Any]) -> Dict[str, str]: + def encrypt(payload: dict[str, Any]) -> dict[str, str]: """将请求体加密为 weapi 的 params/encSecKey。 第 91 条（P2-5）：按官方协议重写——secKey 取官方字符集随机 16 位， encSecKey = secKey 反转后做 textbook RSA（无填充）的 256 位 hex。 @@ -166,7 +169,7 @@ class WeapiCrypto: -def _ensure_dict(result: Any, msg: str) -> Dict[str, Any]: +def _ensure_dict(result: Any, msg: str) -> dict[str, Any]: """第 91 条（P2-8）：上游 200 但返回 null/数组等非 dict 体（风控/抖动）时统一前置校验""" if isinstance(result, dict): return result @@ -175,9 +178,9 @@ def _ensure_dict(result: Any, msg: str) -> Dict[str, Any]: class HTTPClient: """HTTP客户端类""" - + @staticmethod - def post_request(url: str, params: str, cookies: Dict[str, str]) -> str: + async def post_request(url: str, params: str, cookies: dict[str, str]) -> str: """发送POST请求并返回文本响应。 R5-20：网络异常/5xx 做一次 1 秒退避重试（全站 API 调用此前零重试， 上游抖动即整首失败）；4xx（429/460 风控）不重试——立即重试会加剧风控。""" @@ -192,10 +195,10 @@ class HTTPClient: last_err = None for attempt in range(2): try: - response = requests.post(url, headers=headers, cookies=request_cookies, + response = await requests.post(url, headers=headers, cookies=request_cookies, data={"params": params}, timeout=30) if response.status_code >= 500 and attempt == 0: - time.sleep(1.0) + await asyncio.sleep(1.0) continue response.raise_for_status() if len(response.content) > _MAX_UPSTREAM_JSON_BYTES: @@ -208,12 +211,12 @@ class HTTPClient: except requests.RequestException as e: last_err = e if attempt == 0: - time.sleep(1.0) + await asyncio.sleep(1.0) continue raise APIException(f"HTTP请求失败: {last_err}") @staticmethod - def post_request_full(url: str, params: str, cookies: Dict[str, str]) -> requests.Response: + async def post_request_full(url: str, params: str, cookies: dict[str, str]) -> requests.Response: """发送POST请求并返回完整响应对象（R5-20：同 post_request 的有界重试）""" headers = { 'User-Agent': APIConstants.USER_AGENT, @@ -226,10 +229,10 @@ class HTTPClient: last_err = None for attempt in range(2): try: - response = requests.post(url, headers=headers, cookies=request_cookies, + response = await requests.post(url, headers=headers, cookies=request_cookies, data={"params": params}, timeout=30) if response.status_code >= 500 and attempt == 0: - time.sleep(1.0) + await asyncio.sleep(1.0) continue response.raise_for_status() if len(response.content) > _MAX_UPSTREAM_JSON_BYTES: @@ -241,7 +244,7 @@ class HTTPClient: except requests.RequestException as e: last_err = e if attempt == 0: - time.sleep(1.0) + await asyncio.sleep(1.0) continue raise APIException(f"HTTP请求失败: {last_err}") @@ -258,29 +261,29 @@ class APIException(Exception): class NeteaseAPI: """网易云音乐API主类""" - + def __init__(self): self.http_client = HTTPClient() self.crypto_utils = CryptoUtils() - - def get_song_url(self, song_id: int, quality: str, cookies: Dict[str, str]) -> Dict[str, Any]: + + async def get_song_url(self, song_id: int, quality: str, cookies: dict[str, str]) -> dict[str, Any]: """获取歌曲播放URL - + Args: song_id: 歌曲ID quality: 音质等级 (standard, exhigh, lossless, hires, sky, jyeffect, jymaster) cookies: 用户cookies - + Returns: 包含歌曲URL信息的字典 - + Raises: APIException: API调用失败时抛出 """ try: config = APIConstants.DEFAULT_CONFIG.copy() config["requestId"] = str(randrange(20000000, 30000000)) - + payload = { 'ids': [song_id], 'level': quality, @@ -290,10 +293,10 @@ class NeteaseAPI: if quality == 'sky': payload['immerseType'] = 'c51' - + params = self.crypto_utils.encrypt_params(APIConstants.SONG_URL_V1, payload) - response_text = self.http_client.post_request(APIConstants.SONG_URL_V1, params, cookies) - + response_text = await self.http_client.post_request(APIConstants.SONG_URL_V1, params, cookies) + result = json.loads(response_text) # R5-19：合法 JSON 但非对象（null/数组）时给明确错误，不让 AttributeError 失真 if not isinstance(result, dict): @@ -304,16 +307,16 @@ class NeteaseAPI: return result except (json.JSONDecodeError, KeyError) as e: raise APIException(f"解析响应数据失败: {e}") - - def get_song_detail(self, song_id: int) -> Dict[str, Any]: + + async def get_song_detail(self, song_id: int) -> dict[str, Any]: """获取歌曲详细信息 - + Args: song_id: 歌曲ID - + Returns: 包含歌曲详细信息的字典 - + Raises: APIException: API调用失败时抛出 """ @@ -324,7 +327,7 @@ class NeteaseAPI: 'User-Agent': APIConstants.USER_AGENT, 'Referer': APIConstants.REFERER } - response = requests.post(APIConstants.SONG_DETAIL_V3, data=data, + response = await requests.post(APIConstants.SONG_DETAIL_V3, data=data, headers=headers, timeout=30) response.raise_for_status() @@ -340,55 +343,55 @@ class NeteaseAPI: raise APIException(f"获取歌曲详情请求失败: {e}") except json.JSONDecodeError as e: raise APIException(f"解析歌曲详情响应失败: {e}") - - def get_lyric(self, song_id: int, cookies: Dict[str, str]) -> Dict[str, Any]: + + async def get_lyric(self, song_id: int, cookies: dict[str, str]) -> dict[str, Any]: """获取歌词信息 - + Args: song_id: 歌曲ID cookies: 用户cookies - + Returns: 包含歌词信息的字典 - + Raises: APIException: API调用失败时抛出 """ try: data = { - 'id': song_id, - 'cp': 'false', - 'tv': '0', - 'lv': '0', - 'rv': '0', + 'id': song_id, + 'cp': 'false', + 'tv': '0', + 'lv': '0', + 'rv': '0', 'kv': '0', 'yv': '1', 'ytv': '0', 'yrv': '0' } - + headers = { 'User-Agent': APIConstants.USER_AGENT, 'Referer': APIConstants.REFERER } - - response = requests.post(APIConstants.LYRIC_API, data=data, + + response = await requests.post(APIConstants.LYRIC_API, data=data, headers=headers, cookies=cookies, timeout=30) response.raise_for_status() - + result = response.json() _ensure_dict(result, 'get_lyric 接口响应异常（非标准 JSON 对象）') if result.get('code') != 200: raise APIException(f"获取歌词失败: {result.get('message', '未知错误')}") - + return result except requests.RequestException as e: raise APIException(f"获取歌词请求失败: {e}") except json.JSONDecodeError as e: raise APIException(f"解析歌词响应失败: {e}") - - def search_music(self, keywords: str, cookies: Dict[str, str], limit: int = 10, - search_type: int = 1, offset: int = 0) -> List[Dict[str, Any]]: + + async def search_music(self, keywords: str, cookies: dict[str, str], limit: int = 10, + search_type: int = 1, offset: int = 0) -> list[dict[str, Any]]: """搜索音乐 Args: @@ -411,11 +414,11 @@ class NeteaseAPI: 'User-Agent': APIConstants.USER_AGENT, 'Referer': APIConstants.REFERER } - - response = requests.post(APIConstants.SEARCH_API, data=data, + + response = await requests.post(APIConstants.SEARCH_API, data=data, headers=headers, cookies=cookies, timeout=30) response.raise_for_status() - + result = response.json() # R5-P3：非 dict 响应（null/数组）给明确错误 if not isinstance(result, dict): @@ -424,7 +427,7 @@ class NeteaseAPI: raise APIException(f"搜索失败: {result.get('message', '未知错误')}") result_data = result.get('result') or {} - items: List[Dict[str, Any]] = [] + items: list[dict[str, Any]] = [] if search_type == 1: # 歌曲（P2-12：id 改 .get——上游单条残缺（下架/屏蔽）不再拖死整页搜索） @@ -488,17 +491,17 @@ class NeteaseAPI: raise APIException(f"搜索请求失败: {e}") except (json.JSONDecodeError, KeyError) as e: raise APIException(f"解析搜索响应失败: {e}") - - def get_playlist_detail(self, playlist_id: int, cookies: Dict[str, str]) -> Dict[str, Any]: + + async def get_playlist_detail(self, playlist_id: int, cookies: dict[str, str]) -> dict[str, Any]: """获取歌单详情 - + Args: playlist_id: 歌单ID cookies: 用户cookies - + Returns: 歌单详情信息 - + Raises: APIException: API调用失败时抛出 """ @@ -508,11 +511,11 @@ class NeteaseAPI: 'User-Agent': APIConstants.USER_AGENT, 'Referer': APIConstants.REFERER } - - response = requests.post(APIConstants.PLAYLIST_DETAIL_API, data=data, + + response = await requests.post(APIConstants.PLAYLIST_DETAIL_API, data=data, headers=headers, cookies=cookies, timeout=30) response.raise_for_status() - + result = response.json() # R5-P2：非 dict 响应（null/数组）给明确错误 if not isinstance(result, dict): @@ -549,7 +552,7 @@ class NeteaseAPI: except (TypeError, ValueError): continue - song_resp = requests.post(APIConstants.SONG_DETAIL_V3, data=song_data, + song_resp = await requests.post(APIConstants.SONG_DETAIL_V3, data=song_data, headers=headers, cookies=cookies, timeout=30) song_resp.raise_for_status() @@ -583,17 +586,17 @@ class NeteaseAPI: raise APIException(f"获取歌单详情请求失败: {e}") except (json.JSONDecodeError, KeyError) as e: raise APIException(f"解析歌单详情响应失败: {e}") - - def get_album_detail(self, album_id: int, cookies: Dict[str, str]) -> Dict[str, Any]: + + async def get_album_detail(self, album_id: int, cookies: dict[str, str]) -> dict[str, Any]: """获取专辑详情 - + Args: album_id: 专辑ID cookies: 用户cookies - + Returns: 专辑详情信息 - + Raises: APIException: API调用失败时抛出 """ @@ -603,15 +606,15 @@ class NeteaseAPI: 'User-Agent': APIConstants.USER_AGENT, 'Referer': APIConstants.REFERER } - - response = requests.get(url, headers=headers, cookies=cookies, timeout=30) + + response = await requests.get(url, headers=headers, cookies=cookies, timeout=30) response.raise_for_status() - + result = response.json() _ensure_dict(result, 'get_album_detail 接口响应异常（非标准 JSON 对象）') if result.get('code') != 200: raise APIException(f"获取专辑详情失败: {result.get('message', '未知错误')}") - + album = result.get('album') or {} info = { 'id': album.get('id'), @@ -633,14 +636,14 @@ class NeteaseAPI: 'album': (song.get('al') or {}).get('name') or '未知专辑', 'picUrl': self.get_pic_url((song.get('al') or {}).get('pic')) }) - + return info except requests.RequestException as e: raise APIException(f"获取专辑详情请求失败: {e}") except (json.JSONDecodeError, KeyError) as e: raise APIException(f"解析专辑详情响应失败: {e}") - def get_toplist(self) -> List[Dict[str, Any]]: + async def get_toplist(self) -> list[dict[str, Any]]: """获取网易云官方榜单列表 Returns: @@ -655,7 +658,7 @@ class NeteaseAPI: 'Referer': APIConstants.REFERER } - response = requests.get(APIConstants.TOPLIST_API, headers=headers, timeout=30) + response = await requests.get(APIConstants.TOPLIST_API, headers=headers, timeout=30) response.raise_for_status() result = response.json() @@ -681,7 +684,7 @@ class NeteaseAPI: except (json.JSONDecodeError, KeyError) as e: raise APIException(f"解析榜单响应失败: {e}") - def get_artist_hot_songs(self, artist_id: int) -> Dict[str, Any]: + async def get_artist_hot_songs(self, artist_id: int) -> dict[str, Any]: """获取歌手信息和热门歌曲 Args: @@ -700,7 +703,7 @@ class NeteaseAPI: 'Referer': APIConstants.REFERER } - response = requests.get(url, headers=headers, timeout=30) + response = await requests.get(url, headers=headers, timeout=30) response.raise_for_status() result = response.json() @@ -735,7 +738,7 @@ class NeteaseAPI: except (json.JSONDecodeError, KeyError) as e: raise APIException(f"解析歌手信息响应失败: {e}") - def get_song_comments(self, song_id: int, limit: int = 20, offset: int = 0) -> Dict[str, Any]: + async def get_song_comments(self, song_id: int, limit: int = 20, offset: int = 0) -> dict[str, Any]: """获取歌曲评论 Args: @@ -756,7 +759,7 @@ class NeteaseAPI: 'Referer': APIConstants.REFERER } - response = requests.get(url, headers=headers, timeout=30) + response = await requests.get(url, headers=headers, timeout=30) response.raise_for_status() result = response.json() @@ -802,7 +805,7 @@ class NeteaseAPI: except (json.JSONDecodeError, KeyError) as e: raise APIException(f"解析评论响应失败: {e}") - def get_account_info(self, cookies: Dict[str, str]) -> Optional[Dict[str, Any]]: + async def get_account_info(self, cookies: dict[str, str]) -> dict[str, Any] | None: """获取当前账号信息（用于个人登录状态展示） Args: @@ -817,7 +820,7 @@ class NeteaseAPI: 'Referer': APIConstants.REFERER } - response = requests.get(APIConstants.USER_ACCOUNT_API, headers=headers, + response = await requests.get(APIConstants.USER_ACCOUNT_API, headers=headers, cookies=cookies, timeout=30) response.raise_for_status() @@ -842,7 +845,7 @@ class NeteaseAPI: except Exception: return None - def get_user_playlists(self, user_id: int, cookies: Dict[str, str]) -> List[Dict[str, Any]]: + async def get_user_playlists(self, user_id: int, cookies: dict[str, str]) -> list[dict[str, Any]]: """获取用户收藏/创建的歌单列表 Args: @@ -869,7 +872,7 @@ class NeteaseAPI: page_size = 100 for _ in range(20): # 有界翻页，防上游 more 异常导致无限循环 params = {'uid': user_id, 'limit': page_size, 'offset': offset} - response = requests.get(APIConstants.USER_PLAYLIST_API, params=params, + response = await requests.get(APIConstants.USER_PLAYLIST_API, params=params, headers=headers, cookies=cookies, timeout=30) response.raise_for_status() @@ -906,8 +909,8 @@ class NeteaseAPI: except (json.JSONDecodeError, KeyError) as e: raise APIException(f"解析用户歌单响应失败: {e}") - def detect_vip_by_probe(self, cookies: Dict[str, str], - probe_song_id: int = 347230) -> Dict[str, Any]: + async def detect_vip_by_probe(self, cookies: dict[str, str], + probe_song_id: int = 347230) -> dict[str, Any]: """通过歌曲URL接口探测账号可用的最高音质，推断会员类型 直接询问网易云接口「这个账号能把歌解析到什么音质」， @@ -920,7 +923,7 @@ class NeteaseAPI: Returns: {'vip_type': int, 'highest_level': str, 'levels_ok': List[str]} """ - levels_ok = self.probe_levels(cookies, probe_song_id) + levels_ok = await self.probe_levels(cookies, probe_song_id) highest = levels_ok[0] if levels_ok else 'standard' if highest in ('jymaster', 'jyeffect', 'sky'): @@ -931,8 +934,8 @@ class NeteaseAPI: vip_type = 0 # 普通用户 return {'vip_type': vip_type, 'highest_level': highest, 'levels_ok': levels_ok} - def probe_levels(self, cookies: Dict[str, str], - probe_song_id: int = 347230) -> List[str]: + async def probe_levels(self, cookies: dict[str, str], + probe_song_id: int = 347230) -> list[str]: """实测账号对每个音质级别能否真正解析 只有接口返回的 level 与请求一致且带 URL，才算该音质可解析。 @@ -954,7 +957,7 @@ class NeteaseAPI: if time.monotonic() > _deadline: break try: - result = self.get_song_url(probe_song_id, level, cookies) + result = await self.get_song_url(probe_song_id, level, cookies) if result and result.get('data'): d = result['data'][0] or {} # 请求与返回一致且带直链，才说明该音质真正可用 @@ -964,8 +967,8 @@ class NeteaseAPI: continue return levels_ok - def subscribe_playlist(self, playlist_id, cookies: Dict[str, str], - action: int = 1) -> Dict[str, Any]: + async def subscribe_playlist(self, playlist_id, cookies: dict[str, str], + action: int = 1) -> dict[str, Any]: """收藏/取消收藏歌单（官方接口） Args: @@ -985,7 +988,7 @@ class NeteaseAPI: 'User-Agent': APIConstants.USER_AGENT, 'Referer': APIConstants.REFERER } - response = requests.post(APIConstants.PLAYLIST_SUBSCRIBE_API, data=data, + response = await requests.post(APIConstants.PLAYLIST_SUBSCRIBE_API, data=data, headers=headers, cookies=cookies, timeout=30) response.raise_for_status() result = response.json() @@ -998,8 +1001,8 @@ class NeteaseAPI: except (json.JSONDecodeError, KeyError) as e: raise APIException(f"解析收藏歌单响应失败: {e}") - def subscribe_album(self, album_id, cookies: Dict[str, str], - action: int = 1) -> Dict[str, Any]: + async def subscribe_album(self, album_id, cookies: dict[str, str], + action: int = 1) -> dict[str, Any]: """收藏/取消收藏专辑（官方接口） Args: @@ -1019,7 +1022,7 @@ class NeteaseAPI: 'User-Agent': APIConstants.USER_AGENT, 'Referer': APIConstants.REFERER } - response = requests.post(APIConstants.ALBUM_SUBSCRIBE_API, data=data, + response = await requests.post(APIConstants.ALBUM_SUBSCRIBE_API, data=data, headers=headers, cookies=cookies, timeout=30) response.raise_for_status() result = response.json() @@ -1032,9 +1035,9 @@ class NeteaseAPI: except (json.JSONDecodeError, KeyError) as e: raise APIException(f"解析收藏专辑响应失败: {e}") - def post_comment(self, thread_id: str, content: str, action: int = 1, - comment_id: Optional[int] = None, - cookies: Dict[str, str] = None) -> Dict[str, Any]: + async def post_comment(self, thread_id: str, content: str, action: int = 1, + comment_id: int | None = None, + cookies: dict[str, str] | None = None) -> dict[str, Any]: """发送/回复歌曲评论（官方 weapi 接口） Args: @@ -1051,7 +1054,7 @@ class NeteaseAPI: APIException: 操作失败时抛出 """ try: - payload: Dict[str, Any] = { + payload: dict[str, Any] = { 'threadId': thread_id, 'content': content, 't': action, @@ -1070,7 +1073,7 @@ class NeteaseAPI: 'User-Agent': APIConstants.USER_AGENT, 'Referer': APIConstants.REFERER, } - response = requests.post(APIConstants.COMMENT_POST_API, data=body, + response = await requests.post(APIConstants.COMMENT_POST_API, data=body, headers=headers, cookies=cookies or {}, timeout=30) response.raise_for_status() @@ -1091,74 +1094,74 @@ class NeteaseAPI: def netease_encrypt_id(self, id_str: str) -> str: """网易云加密图片ID算法 - + Args: id_str: 图片ID字符串 - + Returns: 加密后的字符串 """ import base64 import hashlib - + magic = list('3go8&$8*3*3h0k(2)2') song_id = list(id_str) - + for i in range(len(song_id)): song_id[i] = chr(ord(song_id[i]) ^ ord(magic[i % len(magic)])) - + m = ''.join(song_id) md5_bytes = hashlib.md5(m.encode('utf-8')).digest() result = base64.b64encode(md5_bytes).decode('utf-8') result = result.replace('/', '_').replace('+', '-') - + return result - - def get_pic_url(self, pic_id: Optional[int], size: int = 300) -> str: + + def get_pic_url(self, pic_id: int | None, size: int = 300) -> str: """获取网易云加密歌曲/专辑封面直链 - + Args: pic_id: 封面ID size: 图片尺寸 - + Returns: 图片URL """ if pic_id is None: return '' - + enc_id = self.netease_encrypt_id(str(pic_id)) return f'https://p3.music.126.net/{enc_id}/{pic_id}.jpg?param={size}y{size}' class QRLoginManager: """二维码登录管理器""" - + def __init__(self): self.http_client = HTTPClient() self.crypto_utils = CryptoUtils() - - def generate_qr_key(self) -> Optional[str]: + + async def generate_qr_key(self) -> str | None: """生成二维码的key - + Returns: 成功返回unikey，失败返回None - + Raises: APIException: API调用失败时抛出 """ try: config = APIConstants.DEFAULT_CONFIG.copy() config["requestId"] = str(randrange(20000000, 30000000)) - + payload = { 'type': 1, 'header': json.dumps(config) } - + params = self.crypto_utils.encrypt_params(APIConstants.QR_UNIKEY_API, payload) - response = self.http_client.post_request_full(APIConstants.QR_UNIKEY_API, params, {}) - + response = await self.http_client.post_request_full(APIConstants.QR_UNIKEY_API, params, {}) + result = json.loads(response.text) # 第 97 条（P3）：非 dict 响应（null/数组）不再 AttributeError 逃出捕获 _ensure_dict(result, '生成二维码key响应异常（非标准 JSON 对象）') @@ -1168,26 +1171,26 @@ class QRLoginManager: raise APIException(f"生成二维码key失败: {result.get('message', '未知错误')}") except (json.JSONDecodeError, KeyError) as e: raise APIException(f"解析二维码key响应失败: {e}") - - def create_qr_login(self) -> Optional[str]: + + async def create_qr_login(self) -> str | None: """创建登录二维码并在控制台显示 - + Returns: 成功返回unikey，失败返回None """ try: import qrcode - - unikey = self.generate_qr_key() + + unikey = await self.generate_qr_key() if not unikey: print("生成二维码key失败") return None - + # 创建二维码 qr = qrcode.QRCode() qr.add_data(f'https://music.163.com/login?codekey={unikey}') qr.make(fit=True) - + # 在控制台显示二维码 qr.print_ascii(tty=True) print("\n请使用网易云音乐APP扫描上方二维码登录") @@ -1198,37 +1201,37 @@ class QRLoginManager: except Exception as e: print(f"创建二维码失败: {e}") return None - - def check_qr_login(self, unikey: str) -> Tuple[int, Dict[str, str]]: + + async def check_qr_login(self, unikey: str) -> tuple[int, dict[str, str]]: """检查二维码登录状态 - + Args: unikey: 二维码key - + Returns: (登录状态码, cookie字典) - + Raises: APIException: API调用失败时抛出 """ try: config = APIConstants.DEFAULT_CONFIG.copy() config["requestId"] = str(randrange(20000000, 30000000)) - + payload = { 'key': unikey, 'type': 1, 'header': json.dumps(config) } - + params = self.crypto_utils.encrypt_params(APIConstants.QR_LOGIN_API, payload) - response = self.http_client.post_request_full(APIConstants.QR_LOGIN_API, params, {}) - + response = await self.http_client.post_request_full(APIConstants.QR_LOGIN_API, params, {}) + result = json.loads(response.text) # 第 97 条（P3）：非 dict 响应不再 AttributeError 逃出捕获（同 generate_qr_key） _ensure_dict(result, '解析登录状态响应失败（非标准 JSON 对象）') cookie_dict = {} - + if result.get('code') == 803: # 登录成功，提取cookie。 # 第 94 条（P1-2）：此前只提取 MUSIC_U——__csrf 从未入 Cookie 串， @@ -1249,21 +1252,21 @@ class QRLoginManager: return result.get('code', -1), cookie_dict except (json.JSONDecodeError, KeyError) as e: raise APIException(f"解析登录状态响应失败: {e}") - - def qr_login(self) -> Optional[str]: + + async def qr_login(self) -> str | None: """完整的二维码登录流程 - + Returns: 成功返回cookie字符串，失败返回None """ try: - unikey = self.create_qr_login() + unikey = await self.create_qr_login() if not unikey: return None - + while True: - code, cookies = self.check_qr_login(unikey) - + code, cookies = await self.check_qr_login(unikey) + if code == 803: print("\n登录成功！") # 第 97 条（P2-8）：CLI 序列化同 qr_login 口径带上 __csrf（94 条只修 @@ -1279,8 +1282,8 @@ class QRLoginManager: else: print(f"\n登录失败，错误码：{code}") return None - - time.sleep(2) + + await asyncio.sleep(2) except KeyboardInterrupt: print("\n用户取消登录") return None @@ -1290,115 +1293,115 @@ class QRLoginManager: # 向后兼容的函数接口 -def url_v1(song_id: int, level: str, cookies: Dict[str, str]) -> Dict[str, Any]: +def url_v1(song_id: int, level: str, cookies: dict[str, str]) -> dict[str, Any]: """获取歌曲URL（向后兼容）""" api = NeteaseAPI() - return api.get_song_url(song_id, level, cookies) + return run_sync(api.get_song_url(song_id, level, cookies)) -def name_v1(song_id: int) -> Dict[str, Any]: +def name_v1(song_id: int) -> dict[str, Any]: """获取歌曲详情（向后兼容）""" api = NeteaseAPI() - return api.get_song_detail(song_id) + return run_sync(api.get_song_detail(song_id)) -def lyric_v1(song_id: int, cookies: Dict[str, str]) -> Dict[str, Any]: +def lyric_v1(song_id: int, cookies: dict[str, str]) -> dict[str, Any]: """获取歌词（向后兼容）""" api = NeteaseAPI() - return api.get_lyric(song_id, cookies) + return run_sync(api.get_lyric(song_id, cookies)) -def search_music(keywords: str, cookies: Dict[str, str], limit: int = 10, - search_type: int = 1, offset: int = 0) -> List[Dict[str, Any]]: +def search_music(keywords: str, cookies: dict[str, str], limit: int = 10, + search_type: int = 1, offset: int = 0) -> list[dict[str, Any]]: """搜索音乐（向后兼容）""" api = NeteaseAPI() - return api.search_music(keywords, cookies, limit, search_type, offset=offset) + return run_sync(api.search_music(keywords, cookies, limit, search_type, offset=offset)) -def toplist_list() -> List[Dict[str, Any]]: +def toplist_list() -> list[dict[str, Any]]: """获取官方榜单列表（向后兼容）""" api = NeteaseAPI() - return api.get_toplist() + return run_sync(api.get_toplist()) -def artist_songs(artist_id: int) -> Dict[str, Any]: +def artist_songs(artist_id: int) -> dict[str, Any]: """获取歌手信息和热门歌曲（向后兼容）""" api = NeteaseAPI() - return api.get_artist_hot_songs(artist_id) + return run_sync(api.get_artist_hot_songs(artist_id)) -def song_comments(song_id: int, limit: int = 20, offset: int = 0) -> Dict[str, Any]: +def song_comments(song_id: int, limit: int = 20, offset: int = 0) -> dict[str, Any]: """获取歌曲评论（向后兼容）""" api = NeteaseAPI() - return api.get_song_comments(song_id, limit, offset) + return run_sync(api.get_song_comments(song_id, limit, offset)) -def account_info(cookies: Dict[str, str]) -> Optional[Dict[str, Any]]: +def account_info(cookies: dict[str, str]) -> dict[str, Any] | None: """获取当前账号信息（向后兼容）""" api = NeteaseAPI() - return api.get_account_info(cookies) + return run_sync(api.get_account_info(cookies)) -def user_playlists(user_id: int, cookies: Dict[str, str]) -> List[Dict[str, Any]]: +def user_playlists(user_id: int, cookies: dict[str, str]) -> list[dict[str, Any]]: """获取用户收藏/创建的歌单列表（向后兼容）""" api = NeteaseAPI() - return api.get_user_playlists(user_id, cookies) + return run_sync(api.get_user_playlists(user_id, cookies)) -def detect_vip_by_probe(cookies: Dict[str, str], probe_song_id: int = 347230) -> Dict[str, Any]: +def detect_vip_by_probe(cookies: dict[str, str], probe_song_id: int = 347230) -> dict[str, Any]: """通过歌曲URL接口探测账号会员类型（向后兼容）""" api = NeteaseAPI() - return api.detect_vip_by_probe(cookies, probe_song_id) + return run_sync(api.detect_vip_by_probe(cookies, probe_song_id)) -def playlist_subscribe(playlist_id: int, cookies: Dict[str, str], action: int = 1) -> Dict[str, Any]: +def playlist_subscribe(playlist_id: int, cookies: dict[str, str], action: int = 1) -> dict[str, Any]: """收藏/取消收藏歌单（向后兼容）""" api = NeteaseAPI() - return api.subscribe_playlist(playlist_id, cookies, action) + return run_sync(api.subscribe_playlist(playlist_id, cookies, action)) -def album_subscribe(album_id: int, cookies: Dict[str, str], action: int = 1) -> Dict[str, Any]: +def album_subscribe(album_id: int, cookies: dict[str, str], action: int = 1) -> dict[str, Any]: """收藏/取消收藏专辑（向后兼容）""" api = NeteaseAPI() - return api.subscribe_album(album_id, cookies, action) + return run_sync(api.subscribe_album(album_id, cookies, action)) def post_comment(thread_id: str, content: str, action: int = 1, - comment_id: Optional[int] = None, - cookies: Dict[str, str] = None) -> Dict[str, Any]: + comment_id: int | None = None, + cookies: dict[str, str] | None = None) -> dict[str, Any]: """发送/回复歌曲评论（向后兼容）""" api = NeteaseAPI() - return api.post_comment(thread_id, content, action, comment_id, cookies) + return run_sync(api.post_comment(thread_id, content, action, comment_id, cookies)) -def probe_levels(cookies: Dict[str, str], probe_song_id: int = 347230) -> List[str]: +def probe_levels(cookies: dict[str, str], probe_song_id: int = 347230) -> list[str]: """实测账号可解析的音质列表（向后兼容）""" api = NeteaseAPI() - return api.probe_levels(cookies, probe_song_id) + return run_sync(api.probe_levels(cookies, probe_song_id)) -def playlist_detail(playlist_id: int, cookies: Dict[str, str]) -> Dict[str, Any]: +def playlist_detail(playlist_id: int, cookies: dict[str, str]) -> dict[str, Any]: """获取歌单详情（向后兼容）""" api = NeteaseAPI() - return api.get_playlist_detail(playlist_id, cookies) + return run_sync(api.get_playlist_detail(playlist_id, cookies)) -def album_detail(album_id: int, cookies: Dict[str, str]) -> Dict[str, Any]: +def album_detail(album_id: int, cookies: dict[str, str]) -> dict[str, Any]: """获取专辑详情（向后兼容）""" api = NeteaseAPI() - return api.get_album_detail(album_id, cookies) + return run_sync(api.get_album_detail(album_id, cookies)) -def get_pic_url(pic_id: Optional[int], size: int = 300) -> str: +def get_pic_url(pic_id: int | None, size: int = 300) -> str: """获取图片URL（向后兼容）""" api = NeteaseAPI() return api.get_pic_url(pic_id, size) -def qr_login() -> Optional[str]: +def qr_login() -> str | None: """二维码登录（向后兼容）""" manager = QRLoginManager() - return manager.qr_login() + return run_sync(manager.qr_login()) # ============ Cookie 工具 ============ @@ -1406,7 +1409,7 @@ def qr_login() -> Optional[str]: DEFAULT_COOKIE_FILE = "cookie.txt" -def load_cookies(path: str = DEFAULT_COOKIE_FILE) -> Dict[str, str]: +def load_cookies(path: str = DEFAULT_COOKIE_FILE) -> dict[str, str]: """从 cookie.txt 直接读取并解析为 dict 支持标准 ``NMTID=xxx; _ntes_nnid=xxx; ...`` 格式。 @@ -1417,19 +1420,19 @@ def load_cookies(path: str = DEFAULT_COOKIE_FILE) -> Dict[str, str]: if not os.path.isfile(path): return {} try: - with open(path, "r", encoding="utf-8-sig") as f: + with open(path, encoding="utf-8-sig") as f: content = f.read() except OSError: return {} return parse_cookie_string(content) -def parse_cookie_string(cookie_string: str) -> Dict[str, str]: +def parse_cookie_string(cookie_string: str) -> dict[str, str]: """把 ``k1=v1; k2=v2`` 格式字符串解析为 dict 支持 ``;`` 或 ``\\n`` 作为分隔符。 """ - cookies: Dict[str, str] = {} + cookies: dict[str, str] = {} if not cookie_string: return cookies cookie_string = cookie_string.strip() diff --git a/music_downloader.py b/music_downloader.py index d973be5..0dbdbc1 100644 --- a/music_downloader.py +++ b/music_downloader.py @@ -15,7 +15,6 @@ from __future__ import annotations -import contextlib import logging import os import re @@ -27,13 +26,12 @@ 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 core.async_bridge import run_sync from music_api import APIException, NeteaseAPI, load_cookies # P2-5：worker 线程的输出一律走 logger（坑 11——NSSM GBK 重定向下裸 print @@ -80,7 +78,7 @@ def _sniff_audio_header(path) -> bool: def get_http_session() -> requests.Session: """获取当前线程的共享 requests.Session（带连接池 + keep-alive）""" - session: Optional[requests.Session] = getattr(_SESSION_LOCAL, "session", None) + session: requests.Session | None = getattr(_SESSION_LOCAL, "session", None) if session is None: session = requests.Session() adapter = HTTPAdapter(pool_connections=8, pool_maxsize=32, max_retries=0) @@ -96,7 +94,7 @@ def get_http_session() -> requests.Session: # F78：锁表带引用计数与回收——无人引用时从表中移除，防止长期运行无界增长 # A3：计数在注册表守卫锁内完成（"取出引用"即计数 +1），消除"取引用后、acquire 前 # 被另一个线程回收、第三个线程拿到新锁对象"的互斥破坏窗口 -_FILE_LOCKS: Dict[str, "RefCountedFileLock"] = {} +_FILE_LOCKS: dict[str, RefCountedFileLock] = {} _FILE_LOCKS_GUARD = threading.Lock() @@ -125,7 +123,7 @@ class _FileLockHandle: self._lock = None self._key = '' - def __enter__(self) -> "RefCountedFileLock": + def __enter__(self) -> RefCountedFileLock: key = str(self._path) with _FILE_LOCKS_GUARD: lock = _FILE_LOCKS.get(key) @@ -159,18 +157,18 @@ class _FileLockHandle: self._lock = None -def get_file_lock(path: Path) -> "_FileLockHandle": +def get_file_lock(path: Path) -> _FileLockHandle: """兼容入口：返回可用 with 语句的锁句柄（A3：引用计数在守卫锁内登记，无回收竞态）""" return _FileLockHandle(path) # 封面字节缓存：同一专辑封面只从网络拉取一次（ffmpeg 写元数据 与 ZIP 打包 共用） -_COVER_CACHE: Dict[str, bytes] = {} +_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]: +def fetch_cover_bytes(pic_url: str) -> bytes | None: """下载并缓存封面图片字节；失败返回 None（B3：字节预算 10MB，拒绝超大盘点阵/伪装文件）""" if not pic_url: return None @@ -243,17 +241,17 @@ class MusicInfo: quality: str lyric: str = "" tlyric: str = "" - publish_year: Optional[int] = None + publish_year: int | None = None @dataclass class DownloadResult: """下载结果数据类""" success: bool - file_path: Optional[str] = None + file_path: str | None = None file_size: int = 0 error_message: str = "" - music_info: Optional[MusicInfo] = None + music_info: MusicInfo | None = None class DownloadException(Exception): @@ -330,7 +328,7 @@ class CountingSemaphore: with self._lock: self.active = max(0, self.active - 1) - def __enter__(self) -> "CountingSemaphore": + def __enter__(self) -> CountingSemaphore: self.acquire() return self @@ -378,7 +376,7 @@ class MusicDownloader: self, download_dir: str = "downloads", max_concurrent: int = 3, - ffmpeg_path: Optional[str] = None, + ffmpeg_path: str | None = None, global_max_concurrent: int = 10, ): """ @@ -461,7 +459,7 @@ class MusicDownloader: return ".mp3" - def _resolve_ffmpeg(self) -> Optional[str]: + def _resolve_ffmpeg(self) -> str | None: """懒加载解析 ffmpeg 路径 每次写入前调用一次：若启动时未检测到，但中途已安装，下一次就能用。 @@ -484,31 +482,29 @@ class MusicDownloader: # ----------- 信息获取 ----------- def get_music_info(self, music_id: int, quality: str = "standard", - cookies: Optional[Dict[str, str]] = None) -> MusicInfo: + cookies: dict[str, str] | None = 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) + # F99：URL/详情拿到后立即取消未完成的歌词请求，已知失败不再被拖着等完 + # 异步化：改用 asyncio.gather 并发（api 方法已异步），失败语义与旧线程池一致 + import asyncio + + async def _gather_info(): + url_t = self.api.get_song_url(music_id, quality, cookies) + detail_t = self.api.get_song_detail(music_id) + lyric_t = self.api.get_lyric(music_id, cookies) + return await asyncio.gather(url_t, detail_t, lyric_t, return_exceptions=True) + + url_result, detail_result, lyric_result = run_sync(_gather_info()) + if isinstance(url_result, BaseException): + raise url_result + if isinstance(detail_result, BaseException): + raise detail_result + if isinstance(lyric_result, BaseException): + lyric_result = None # 歌词接口抖动不阻断解析/下载 if not url_result.get("data") or not url_result["data"]: raise DownloadException(f"无法获取音乐ID {music_id} 的播放链接") @@ -602,11 +598,11 @@ class MusicDownloader: music_id: int, quality: str = "standard", write_meta: bool = True, - meta_fields: Optional[List[str]] = None, - cookies: Optional[Dict[str, str]] = None, + meta_fields: list[str] | None = None, + cookies: dict[str, str] | None = None, retries: int = 2, - cancel_event: Optional[threading.Event] = None, - music_info: Optional[MusicInfo] = None, + cancel_event: threading.Event | None = None, + music_info: MusicInfo | None = None, progress_cb=None, ) -> DownloadResult: """同步下载音乐到本地 @@ -719,7 +715,7 @@ class MusicDownloader: progress_cb(resume_from, music_info.file_size or 0) except Exception: pass # 进度回调绝不影响下载主流程 - resp: Optional[requests.Response] = None + resp: requests.Response | None = None try: resp = get_http_session().get( music_info.download_url, @@ -878,7 +874,7 @@ class MusicDownloader: self, file_path: Path, music_info: MusicInfo, - fields: Optional[List[str]] = None, + fields: list[str] | None = None, ) -> bool: """使用 ffmpeg 写入音频元数据 @@ -909,7 +905,7 @@ class MusicDownloader: file_path.stem + ".tagging" + file_path.suffix ) # 临时封面文件 - cover_tmp: Optional[str] = None + cover_tmp: str | None = None cover_ok = False # F114：创建成功与否与"路径是否需要清理"分离管理 try: @@ -934,7 +930,7 @@ class MusicDownloader: # 构造 ffmpeg 命令 # 顺序: global opts -> -i input1 -> -i input2 -> metadata -> mapping -> -f format -> output output_format = EXT_TO_FFMPEG_FORMAT.get(file_ext, "") - cmd: List[str] = [ + cmd: list[str] = [ ffmpeg_path, "-y", "-loglevel", "error", "-i", str(file_path), ] @@ -1030,7 +1026,7 @@ class MusicDownloader: except OSError: pass - def _build_format_args(self, file_ext: str, has_cover: bool) -> List[str]: + def _build_format_args(self, file_ext: str, has_cover: bool) -> list[str]: """根据文件格式构造 mapping / 编码参数 F34：不写封面时用 -map 0 -c copy 保留全部既有流（含已有封面）， 原实现 -map 0:a 会把 MP3/M4A 已有封面静默删掉。""" diff --git a/podcast_api.py b/podcast_api.py index 37b6b96..d7fb63d 100644 --- a/podcast_api.py +++ b/podcast_api.py @@ -19,8 +19,7 @@ import json import re import time -from random import randrange -from typing import Dict, List, Optional, Any +from typing import Any import requests @@ -54,7 +53,7 @@ class PodcastConstants: _MAX_UPSTREAM_JSON_BYTES = 8 * 1024 * 1024 -def _post_json(url: str, data: Dict[str, Any], cookies: Optional[Dict[str, str]] = None) -> Dict[str, Any]: +def _post_json(url: str, data: dict[str, Any], cookies: dict[str, str] | None = None) -> dict[str, Any]: """发送表单 POST 请求并返回 JSON""" headers = { 'User-Agent': APIConstants.USER_AGENT, @@ -78,7 +77,15 @@ def _post_json(url: str, data: Dict[str, Any], cookies: Optional[Dict[str, str]] return result -def _ts_to_date(ts: Optional[int]) -> str: +def _to_int(v: Any, default: int | None = None) -> int | None: + """安全转 int：垃圾值/None 回退默认值（上游字段常为 null/字符串）。""" + try: + return int(v) + except (TypeError, ValueError): + return default + + +def _ts_to_date(ts: int | None) -> str: """毫秒时间戳 → 日期字符串""" try: if not ts: @@ -88,7 +95,7 @@ def _ts_to_date(ts: Optional[int]) -> str: return '' -def _fmt_duration(ms: Optional[int]) -> str: +def _fmt_duration(ms: int | None) -> str: """毫秒 → mm:ss / hh:mm:ss""" try: sec = int((ms or 0) / 1000) @@ -101,7 +108,7 @@ def _fmt_duration(ms: Optional[int]) -> str: return f"{h}:{m:02d}:{s:02d}" if h else f"{m:02d}:{s:02d}" -def _fmt_count(n: Optional[int]) -> str: +def _fmt_count(n: int | None) -> str: """数量 → 万单位简写""" try: n = int(n or 0) @@ -118,7 +125,7 @@ def _fmt_count(n: Optional[int]) -> str: # ============ 数据规范化 ============ -def _norm_radio(raw: Dict[str, Any]) -> Dict[str, Any]: +def _norm_radio(raw: dict[str, Any]) -> dict[str, Any]: """规范化电台（播客）信息""" dj = raw.get('dj') or {} return { @@ -139,7 +146,7 @@ def _norm_radio(raw: Dict[str, Any]) -> Dict[str, Any]: } -def _norm_program(raw: Dict[str, Any]) -> Dict[str, Any]: +def _norm_program(raw: dict[str, Any]) -> dict[str, Any]: """规范化单集（program）信息""" radio = raw.get('radio') or {} dj = raw.get('dj') or radio.get('dj') or {} @@ -168,7 +175,7 @@ def _norm_program(raw: Dict[str, Any]) -> Dict[str, Any]: # ============ 链接 / ID 解析 ============ -_SHORTLINK_CACHE: Dict[str, str] = {} +_SHORTLINK_CACHE: dict[str, str] = {} def _resolve_short_link(url: str) -> str: @@ -201,7 +208,7 @@ def _resolve_short_link(url: str) -> str: return real -def extract_podcast_ref(raw: str) -> Dict[str, str]: +def extract_podcast_ref(raw: str) -> dict[str, str]: """从用户输入中提取播客引用（T4：正规 URL 解析，替代子串匹配与非数字剔除） Returns: @@ -210,7 +217,7 @@ def extract_podcast_ref(raw: str) -> Dict[str, str]: - program: program?id= / sound?id= 链接 - unknown: 纯数字 ID（需依次尝试单集/电台） """ - from urllib.parse import urlsplit, parse_qs + from urllib.parse import parse_qs, urlsplit text = str(raw or '').strip() # 短链先还原（仅当整串就是一个短链 URL） if '163cn.tv' in text.lower(): @@ -247,9 +254,9 @@ def extract_podcast_ref(raw: str) -> Dict[str, str]: # ============ 核心 API ============ -def search_podcast(keywords: str, cookies: Optional[Dict[str, str]] = None, +def search_podcast(keywords: str, cookies: dict[str, str] | None = None, limit: int = 20, offset: int = 0, - search_type: str = 'all') -> Dict[str, Any]: + search_type: str = 'all') -> dict[str, Any]: """搜索播客 Args: @@ -267,9 +274,9 @@ def search_podcast(keywords: str, cookies: Optional[Dict[str, str]] = None, raise APIException('搜索关键词不能为空') limit = max(1, min(int(limit or 20), 50)) - result: Dict[str, Any] = {'radios': [], 'voices': []} + result: dict[str, Any] = {'radios': [], 'voices': []} - def _do_search(stype: int) -> Dict[str, Any]: + def _do_search(stype: int) -> dict[str, Any]: return _post_json(PodcastConstants.SEARCH_API, {'s': keywords, 'type': stype, 'limit': limit, 'offset': offset}, cookies) @@ -288,7 +295,7 @@ def search_podcast(keywords: str, cookies: Optional[Dict[str, str]] = None, return result -def get_radio_detail(radio_id: Any, cookies: Optional[Dict[str, str]] = None) -> Dict[str, Any]: +def get_radio_detail(radio_id: Any, cookies: dict[str, str] | None = None) -> dict[str, Any]: """获取播客电台（djradio）详情""" try: rid = int(str(radio_id).strip()) @@ -302,8 +309,8 @@ def get_radio_detail(radio_id: Any, cookies: Optional[Dict[str, str]] = None) -> return _norm_radio(data) -def get_radio_programs(radio_id: Any, cookies: Optional[Dict[str, str]] = None, - limit: int = 30, offset: int = 0, asc: bool = True) -> Dict[str, Any]: +def get_radio_programs(radio_id: Any, cookies: dict[str, str] | None = None, + limit: int = 30, offset: int = 0, asc: bool = True) -> dict[str, Any]: """获取电台下的单集列表（分页） Args: @@ -334,7 +341,7 @@ def get_radio_programs(radio_id: Any, cookies: Optional[Dict[str, str]] = None, } -def get_program_detail(program_id: Any, cookies: Optional[Dict[str, str]] = None) -> Dict[str, Any]: +def get_program_detail(program_id: Any, cookies: dict[str, str] | None = None) -> dict[str, Any]: """获取单集（program）详情""" try: pid = int(str(program_id).strip()) @@ -347,7 +354,7 @@ def get_program_detail(program_id: Any, cookies: Optional[Dict[str, str]] = None return _norm_program(program) -def get_categories() -> List[Dict[str, Any]]: +def get_categories() -> list[dict[str, Any]]: """获取播客分类列表（GET 接口） Returns: @@ -387,9 +394,9 @@ def get_categories() -> List[Dict[str, Any]]: return cats -def get_hot_radios(cookies: Optional[Dict[str, str]] = None, +def get_hot_radios(cookies: dict[str, str] | None = None, limit: int = 30, offset: int = 0, - cate_id: Optional[int] = None) -> List[Dict[str, Any]]: + cate_id: int | None = None) -> list[dict[str, Any]]: """获取分类下热门播客 Args: @@ -400,7 +407,7 @@ def get_hot_radios(cookies: Optional[Dict[str, str]] = None, 电台列表（已规范化） """ limit = max(1, min(int(limit or 30), 100)) - cate_ids: List[int] = [] + cate_ids: list[int] = [] user_cate = False if cate_id: try: @@ -425,8 +432,8 @@ def get_hot_radios(cookies: Optional[Dict[str, str]] = None, return [] -def resolve_episode_url(song_id: Any, cookies: Optional[Dict[str, str]] = None, - level: str = 'auto') -> Optional[Dict[str, Any]]: +def resolve_episode_url(song_id: Any, cookies: dict[str, str] | None = None, + level: str = 'auto') -> dict[str, Any] | None: """解析单集音频直链（带音质降级） Args: @@ -439,7 +446,7 @@ def resolve_episode_url(song_id: Any, cookies: Optional[Dict[str, str]] = None, """ # F83：降级阶梯只向下不向上——原实现指定 standard 会回退到更高的 exhigh _FALLDOWN_LADDER = ['hires', 'lossless', 'exhigh', 'standard'] - order: List[str] = [] + order: list[str] = [] level = str(level or 'auto').strip() if level and level != 'auto': order.append(level) @@ -454,7 +461,7 @@ def resolve_episode_url(song_id: Any, cookies: Optional[Dict[str, str]] = None, else: order.extend(PodcastConstants.QUALITY_LADDER) # 去重保序 - dedup: List[str] = [] + dedup: list[str] = [] for lv in order: if lv not in dedup: dedup.append(lv) @@ -487,8 +494,8 @@ def resolve_episode_url(song_id: Any, cookies: Optional[Dict[str, str]] = None, # ============ 组合能力 ============ -def parse_podcast_ref(raw: str, cookies: Optional[Dict[str, str]] = None, - page_size: int = 30) -> Dict[str, Any]: +def parse_podcast_ref(raw: str, cookies: dict[str, str] | None = None, + page_size: int = 30) -> dict[str, Any]: """智能解析用户输入：电台链接 → 电台详情+单集列表；单集链接 → 单集详情 Returns: @@ -515,15 +522,15 @@ def parse_podcast_ref(raw: str, cookies: Optional[Dict[str, str]] = None, return {'kind': 'radio', 'radio': radio, **page} -def list_radio_all_program_ids(radio_id: Any, cookies: Optional[Dict[str, str]] = None, - max_count: int = 200) -> List[Dict[str, Any]]: +def list_radio_all_program_ids(radio_id: Any, cookies: dict[str, str] | None = None, + max_count: int = 200) -> list[dict[str, Any]]: """拉取电台全部单集（用于整台打包，正序），返回 [{'id','name','main_song_id','serial_num','cover_url'}] 最多拉取 max_count 集，防止超大电台拖垮磁盘/任务。 F98：按节目 ID 去重；一页没有任何新增（短页/重复页/列表变动）即有界停止， 不再固定 offset+=100 盲目翻页。 """ - out: List[Dict[str, Any]] = [] + out: list[dict[str, Any]] = [] seen_ids = set() offset = 0 page_size = 100 diff --git a/pyproject.toml b/pyproject.toml index 29b7b58..777cb6f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,12 @@ ignore = [ "S603", "S607", # subprocess（打包/运维脚本） "UP009", # 保留 `# -*- coding: utf-8 -*-` 文件头（全项目既有约定） "RUF001", "RUF002", "RUF003", # 全角字符（中文注释/文案，误报） + # ---- 以下为「有意保留」，随文件在迁移中重写时再行处理，非待修缺陷 ---- + "B904", # raise-from 异常链：纯风格，40 处均在历史巨石文件，逐处改写收益低风险高 + "RUF012", # 常量类可变默认值：APIConstants/PodcastConstants 是常量表，非实例状态 + "S305", # AES-ECB：网易云 weapi 协议**要求**的加密模式，非可自由选择 + "S311", # randrange：用于生成上游 requestId（协议随机数），非安全用途 + "ASYNC250", # input()：qr_login 交互式 CLI 的人类输入，异步化无意义 ] [tool.ruff.lint.per-file-ignores] diff --git a/qr_login.py b/qr_login.py index 3c514f6..f0b01be 100644 --- a/qr_login.py +++ b/qr_login.py @@ -7,17 +7,17 @@ - 用户友好的交互界面 """ +import asyncio +import datetime +import logging import os +import shutil import sys import time -import shutil -import logging -import datetime from pathlib import Path -from typing import Optional, Tuple try: - from music_api import QRLoginManager, APIException, load_cookies + from music_api import APIException, QRLoginManager, load_cookies except ImportError as e: print(f"导入模块失败: {e}") print("请确保 music_api.py 文件存在且可用") @@ -87,7 +87,7 @@ class QRLoginClient: shutil.copy2(backup, self.cookie_file) return False - def _backup_cookie(self, suffix: str = "backup") -> Optional[Path]: + def _backup_cookie(self, suffix: str = "backup") -> Path | None: if not self.cookie_file.exists(): return None ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S_%f") # 第 91 条（P3）：微秒防同秒覆盖 @@ -172,7 +172,7 @@ class QRLoginClient: self.logger.error(f"检查登录状态失败: {e}") return False - def interactive_login(self) -> Tuple[bool, Optional[str]]: + async def interactive_login(self) -> tuple[bool, str | None]: """交互式二维码登录""" try: print("\n=== 网易云音乐二维码登录 ===") @@ -186,7 +186,7 @@ class QRLoginClient: print("\n开始二维码登录流程...") print("正在生成二维码...") # create_qr_login 返回 unikey 字符串（失败为 None），并已在控制台打印二维码 - qr_key = self.qr_manager.create_qr_login() + qr_key = await self.qr_manager.create_qr_login() if not qr_key: error_msg = "生成二维码失败" @@ -203,7 +203,7 @@ class QRLoginClient: while attempt < max_attempts: try: # check_qr_login 返回 (状态码, cookie字典)：801等待/802已扫/803成功/800过期 - code, cookie_dict = self.qr_manager.check_qr_login(qr_key) + code, cookie_dict = await self.qr_manager.check_qr_login(qr_key) if code == 803: musu_u = cookie_dict.get('MUSIC_U', '') @@ -233,7 +233,7 @@ class QRLoginClient: if code == 801 and attempt % 10 == 0: print(f"等待扫码中... ({attempt + 1}/{max_attempts})") - time.sleep(5) + await asyncio.sleep(5) attempt += 1 except KeyboardInterrupt: @@ -241,7 +241,7 @@ class QRLoginClient: return False, "用户取消登录" except Exception as e: self.logger.error(f"检查登录状态时发生错误: {e}") - time.sleep(5) + await asyncio.sleep(5) attempt += 1 print("\n[超时] 登录超时，请重新尝试") @@ -325,7 +325,7 @@ def main(): command = sys.argv[1].lower() if command == 'login': - success, error = client.interactive_login() + success, error = asyncio.run(client.interactive_login()) if success: print("\n登录完成！") client.show_login_info() @@ -368,7 +368,7 @@ def main(): choice = input("\n请选择操作 (1-4): ").strip() if choice == '1': - success, error = client.interactive_login() + success, error = asyncio.run(client.interactive_login()) if success: print("\n登录成功！") client.show_login_info()