refactor(async): music_api 异步化 + 全项目 lint 收敛

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 条。
2026-09-26 00:32:45 +08:00
1 parent 4962542e78
commit ce7ea66d6a
@@ -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)
 # ============ 辅助 ============
@@ -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
@@ -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 typing import Any
-from urllib.parse import quote, urlsplit, parse_qs
+from urllib.parse import parse_qs, quote, urlsplit
-from flask import Flask, request, send_file, render_template, Response, jsonify, stream_with_context, make_response, g
+
 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,
+    VALID_QUALITY_LEVELS,
-    MAX_ACTIVE_TASKS_PER_USER, MAX_TOTAL_TASKS, MAX_TERMINAL_TASKS,
+    Settings,
    BATCH_ZIP_CONCURRENCY, VALID_QUALITY_LEVELS,
 )
 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, BATCH_DURATIONS_LOCK,
+    BATCH_DURATIONS_LOCK,
-    DOWNLOAD_TASKS, DOWNLOAD_TASKS_LOCK,
+    DEVICE_LAST_SEEN,
-    DEVICE_LAST_SEEN, DEVICE_LAST_SEEN_LOCK,
+    DEVICE_LAST_SEEN_LOCK,
-    REQUEST_STATS, STATS_LOCK as _STATS_LOCK, REQ_LOG,
+    DOWNLOAD_TASKS,
-    GLOBAL_MUSIC_U_CACHE as _GLOBAL_MUSIC_U_CACHE,
+    DOWNLOAD_TASKS_LOCK,
-    VIP_CACHE as _VIP_CACHE, VIP_USER_CACHE as _VIP_USER_CACHE,
+    REQ_LOG,
-    SERVER_NICK_CACHE as _SERVER_NICK_CACHE, SERVER_NICK_LOCK as _SERVER_NICK_LOCK,
+    REQUEST_STATS,
    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,
 )
 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
@@ -96,7 +122,7 @@ 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,
@@ -108,7 +134,7 @@ class APIResponse:
        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,
@@ -166,7 +192,7 @@ class MusicAPIService:
        return logger
-    def _get_cookies(self) -> Dict[str, str]:
+    def _get_cookies(self) -> dict[str, str]:
        """直接从 cookie.txt 读取并解析为 dict"""
        try:
            cookies = load_cookies()
@@ -240,14 +266,14 @@ class MusicAPIService:
        }
        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_done: list[str] = []
-            finished_retry: List[str] = []
+            finished_retry: list[str] = []
-            expired: List[tuple] = []
+            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）——上游黑洞时并发请求不再各自串行
@@ -1669,7 +1694,7 @@ def search_music_api():
        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)
@@ -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,
+def _podcast_resolve_audio(program: dict[str, Any], quality: str,
-                           cookies: Dict[str, str]) -> Dict[str, Any]:
+                           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 已迁至 core.config（顶部 import 同名共享，此处不再重复定义）
 VALID_QUALITY_LEVELS = ['standard', 'exhigh', 'lossless', 'hires', 'sky', 'jyeffect', 'jymaster', 'dolby']
@app.route('/global/playlists', methods=['GET', 'POST'])
@@ -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
+    zip_tmp_path: str | None = None
    zipped_files: List[str] = []
    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
+    file_path_str: str | None = None
-    zip_tmp_path: Optional[str] = None
+    zip_tmp_path: str | None = None
-    task_dir: Optional[Path] = 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
+    file_path_str: str | None = None
-    zip_tmp_path: Optional[str] = None
+    zip_tmp_path: str | None = None
-    task_dir_path: Optional[str] = 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:
@@ -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("  ├─ GET  /health        - 健康检查")
-        _safe_console_print(f"  ├─ POST /search        - 搜索音乐（支持歌曲/歌单/专辑/歌手）")
+        _safe_console_print("  ├─ POST /search        - 搜索音乐（支持歌曲/歌单/专辑/歌手）")
-        _safe_console_print(f"  ├─ POST /song          - 获取歌曲信息")
+        _safe_console_print("  ├─ POST /song          - 获取歌曲信息")
-        _safe_console_print(f"  ├─ POST /playlist      - 获取歌单详情")
+        _safe_console_print("  ├─ POST /playlist      - 获取歌单详情")
-        _safe_console_print(f"  ├─ POST /album         - 获取专辑详情")
+        _safe_console_print("  ├─ POST /album         - 获取专辑详情")
-        _safe_console_print(f"  ├─ POST /toplist       - 获取官方榜单列表")
+        _safe_console_print("  ├─ POST /toplist       - 获取官方榜单列表")
-        _safe_console_print(f"  ├─ POST /artist_songs  - 获取歌手信息与热门歌曲")
+        _safe_console_print("  ├─ POST /artist_songs  - 获取歌手信息与热门歌曲")
-        _safe_console_print(f"  ├─ POST /comments      - 获取歌曲评论")
+        _safe_console_print("  ├─ POST /comments      - 获取歌曲评论")
-        _safe_console_print(f"  ├─ POST /download      - 下载音乐")
+        _safe_console_print("  ├─ POST /download      - 下载音乐")
-        _safe_console_print(f"  ├─ GET/POST /global/playlists - 全局账号歌单列表")
+        _safe_console_print("  ├─ GET/POST /global/playlists - 全局账号歌单列表")
-        _safe_console_print(f"  ├─ GET/POST /global/playlist  - 全局账号歌单详情")
+        _safe_console_print("  ├─ GET/POST /global/playlist  - 全局账号歌单详情")
-        _safe_console_print(f"  ├─ GET/POST /api/playlist/play - 公开歌单播放API")
+        _safe_console_print("  ├─ GET/POST /api/playlist/play - 公开歌单播放API")
-        _safe_console_print(f"  └─ GET  /api/info      - 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")
@@ -10,19 +10,22 @@
 - 二维码登录
 """
 import asyncio
 import json
 import os
 import urllib.parse
 import time
-from random import randrange
+import urllib.parse
 from typing import Dict, List, Optional, Tuple, Any
 from hashlib import md5
 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):
    """音质等级枚举"""
@@ -95,7 +98,7 @@ class CryptoUtils:
        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")
@@ -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
@@ -177,7 +180,7 @@ 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}")
@@ -263,7 +266,7 @@ class NeteaseAPI:
        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:
@@ -292,7 +295,7 @@ class NeteaseAPI:
                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 失真
@@ -305,7 +308,7 @@ class NeteaseAPI:
        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:
@@ -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()
@@ -341,7 +344,7 @@ class NeteaseAPI:
        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:
@@ -372,7 +375,7 @@ class NeteaseAPI:
                '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()
@@ -387,8 +390,8 @@ class NeteaseAPI:
        except json.JSONDecodeError as e:
            raise APIException(f"解析歌词响应失败: {e}")
-    def search_music(self, keywords: str, cookies: Dict[str, str], limit: int = 10,
+    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]]:
+                     search_type: int = 1, offset: int = 0) -> list[dict[str, Any]]:
        """搜索音乐
        Args:
@@ -412,7 +415,7 @@ class NeteaseAPI:
                '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()
@@ -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——上游单条残缺（下架/屏蔽）不再拖死整页搜索）
@@ -489,7 +492,7 @@ class NeteaseAPI:
        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:
@@ -509,7 +512,7 @@ class NeteaseAPI:
                '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()
@@ -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()
@@ -584,7 +587,7 @@ class NeteaseAPI:
        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:
@@ -604,7 +607,7 @@ class NeteaseAPI:
                '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()
@@ -640,7 +643,7 @@ class NeteaseAPI:
        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],
+    async def detect_vip_by_probe(self, cookies: dict[str, str],
-                            probe_song_id: int = 347230) -> Dict[str, Any]:
+                            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],
+    async def probe_levels(self, cookies: dict[str, str],
-                     probe_song_id: int = 347230) -> List[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],
+    async def subscribe_playlist(self, playlist_id, cookies: dict[str, str],
-                           action: int = 1) -> Dict[str, Any]:
+                           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],
+    async def subscribe_album(self, album_id, cookies: dict[str, str],
-                        action: int = 1) -> Dict[str, Any]:
+                        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,
+    async def post_comment(self, thread_id: str, content: str, action: int = 1,
-                     comment_id: Optional[int] = None,
+                     comment_id: int | None = None,
-                     cookies: Dict[str, str] = None) -> Dict[str, Any]:
+                     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()
@@ -1114,7 +1117,7 @@ class NeteaseAPI:
        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:
@@ -1138,7 +1141,7 @@ class QRLoginManager:
        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:
@@ -1157,7 +1160,7 @@ class QRLoginManager:
            }
            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 逃出捕获
@@ -1169,7 +1172,7 @@ class QRLoginManager:
        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:
@@ -1178,7 +1181,7 @@ class QRLoginManager:
        try:
            import qrcode
-            unikey = self.generate_qr_key()
+            unikey = await self.generate_qr_key()
            if not unikey:
                print("生成二维码key失败")
                return None
@@ -1199,7 +1202,7 @@ class QRLoginManager:
            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:
@@ -1222,7 +1225,7 @@ class QRLoginManager:
            }
            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）
@@ -1250,19 +1253,19 @@ class QRLoginManager:
        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登录成功！")
@@ -1280,7 +1283,7 @@ class QRLoginManager:
                    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,
+def search_music(keywords: str, cookies: dict[str, str], limit: int = 10,
-                 search_type: int = 1, offset: int = 0) -> List[Dict[str, Any]]:
+                 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,
+                 comment_id: int | None = None,
-                 cookies: Dict[str, str] = None) -> Dict[str, Any]:
+                 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()
@@ -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)，
+            # F99：URL/详情拿到后立即取消未完成的歌词请求，已知失败不再被拖着等完
-            # 已知失败不再被无关的歌词请求拖着等完才报错
+            # 异步化：改用 asyncio.gather 并发（api 方法已异步），失败语义与旧线程池一致
-            from concurrent.futures import ThreadPoolExecutor
+            import asyncio
-            _pool = ThreadPoolExecutor(max_workers=3)
+
-            url_f = _pool.submit(self.api.get_song_url, music_id, quality, cookies)
+            async def _gather_info():
-            detail_f = _pool.submit(self.api.get_song_detail, music_id)
+                url_t = self.api.get_song_url(music_id, quality, cookies)
-            lyric_f = _pool.submit(self.api.get_lyric, music_id, cookies)
+                detail_t = self.api.get_song_detail(music_id)
-            try:
+                lyric_t = self.api.get_lyric(music_id, cookies)
-                url_result = url_f.result()
+                return await asyncio.gather(url_t, detail_t, lyric_t, return_exceptions=True)
-                detail_result = detail_f.result()
+
-                try:
+            url_result, detail_result, lyric_result = run_sync(_gather_info())
-                    lyric_result = lyric_f.result()
+            if isinstance(url_result, BaseException):
-                except Exception:
+                raise url_result
-                    lyric_result = None  # 歌词接口抖动不阻断解析/下载
+            if isinstance(detail_result, BaseException):
-            finally:
+                raise detail_result
-                try:
+            if isinstance(lyric_result, BaseException):
-                    _pool.shutdown(wait=False, cancel_futures=True)
+                lyric_result = None  # 歌词接口抖动不阻断解析/下载
                except TypeError:
                    _pool.shutdown(wait=False)
            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,
+        meta_fields: list[str] | None = None,
-        cookies: Optional[Dict[str, str]] = None,
+        cookies: dict[str, str] | None = None,
        retries: int = 2,
-        cancel_event: Optional[threading.Event] = None,
+        cancel_event: threading.Event | None = None,
-        music_info: Optional[MusicInfo] = 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 已有封面静默删掉。"""
@@ -19,8 +19,7 @@
 import json
 import re
 import time
-from random import randrange
+from typing import Any
 from typing import Dict, List, Optional, 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,
+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]:
+                       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,
+def resolve_episode_url(song_id: Any, cookies: dict[str, str] | None = None,
-                        level: str = 'auto') -> Optional[Dict[str, Any]]:
+                        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,
+def parse_podcast_ref(raw: str, cookies: dict[str, str] | None = None,
-                      page_size: int = 30) -> Dict[str, Any]:
+                      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,
+def list_radio_all_program_ids(radio_id: Any, cookies: dict[str, str] | None = None,
-                               max_count: int = 200) -> List[Dict[str, Any]]:
+                               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
@@ -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]
@@ -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()