Archived
snowy-yang
e8cc7cb34a
为什么：文件 IO 这条线上 to_thread 用得过宽。凡是「单文件 exists / unlink / stat」 这类微秒级元数据操作，线程调度开销（几十微秒起）比操作本身还大，套 to_thread 只是 给同步调用加了层异步外壳，不产生任何并发收益；真正需要卸载的是整库复制、ZIP 整段 拷贝、SQLite 探测这类活。本批按这个口径把两端都摆正，并顺手修掉一处真实缺陷。 改动： - web/routers/admin_site.py 备份上传：原 `content = await f.read()` 会把整库一次性 读进内存（端点对外无大小上限，等于给了上传方一个内存放大器），落盘后还要把文件 重新打开读一次头。改为 _save_upload_stream 按 1 MiB 逐块搬运——内存占用恒为一块， 文件头在写第一块时就地截获（省掉那趟回读）。仍属阻塞 IO，故保留在 to_thread 里。 - web/routers/admin_site.py backup_list：撤掉仅为一次 stat 而拆出的嵌套函数 + to_thread，直接在事件循环里做完。 - web/routers/admin.py：save_cookie / logout / _get_cookie_info 四处 to_thread 改 直接调用——它们读写 1KB 级的 cookie.txt，全程不碰网络。 - music_api/qr_client.py：logout() 的 print 改 logger。该方法既被 CLI 也被 admin 端点调用，在 web 语境里 print 会把内容直接喷到服务端 stdout，绕过 logging 配置、 无法被收集，也污染容器日志。 - auth_system/backup.py：三处单文件 unlink（占位文件回滚、删备份、裁剪 pre_restore_） 改直接同步；把只为配合 to_thread 而拆出的 _backup_path_sync 并回 backup_path。 保留在线程里的：整库 copy2 / os.replace、glob + N 次 stat、SQLite 探测、以及持 跨线程锁的 _reserve_backup_path（内联会让事件循环等一把别的线程可能持有的锁）。 - web/routers/delivery.py：两处 os.path.exists 改直接同步（带 ASYNC240 行内豁免， 与 core/db.py、auth_system/db.py 同一写法），顺带删掉随之未使用的 asyncio 导入。 - pyproject.toml / uv.lock：移除 aiofiles。它全仓零引用，且底层同样是 to_thread—— 对 1KB 的 cookie 文件不会带来任何真异步收益。 验证： - uv run ruff check . → All checks passed! - uv run pyright → 0 errors, 0 warnings, 0 informations - uv run pytest → 392 项，0 失败 / 0 error / 0 skipped （含 test_admin_backup_upload 四条：接受真实 SQLite / 缺字段拒 / 后缀拒 / 后缀对但内容非库必须拒且不留半成品） - data/site.db sha256 前后一致（a19f5dbf…6487），无 -wal/-shm 残留
398 lines
15 KiB
Python
398 lines
15 KiB
Python
"""网易云音乐二维码登录模块
|
||
|
||
提供网易云音乐二维码登录功能，包括：
|
||
- 二维码生成和显示
|
||
- 登录状态检查
|
||
- Cookie 获取和保存（直接读写 cookie.txt）
|
||
- 用户友好的交互界面
|
||
"""
|
||
|
||
import asyncio
|
||
import datetime
|
||
import logging
|
||
import os
|
||
import shutil
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
from .cookies import load_cookies
|
||
from .errors import APIException
|
||
from .qr import QRLoginManager
|
||
|
||
# 判断 cookie 是否"有效"：包含任一关键字段即可
|
||
IMPORTANT_COOKIE_KEYS = ("MUSIC_U", "__csrf", "NMTID")
|
||
|
||
|
||
class QRLoginClient:
|
||
"""二维码登录客户端"""
|
||
|
||
def __init__(self, cookie_file: str = "cookie.txt"):
|
||
"""
|
||
Args:
|
||
cookie_file: Cookie 保存文件路径
|
||
"""
|
||
self.cookie_file = Path(cookie_file)
|
||
self.qr_manager = QRLoginManager()
|
||
self.logger = logging.getLogger(__name__)
|
||
|
||
if not self.logger.handlers:
|
||
handler = logging.StreamHandler()
|
||
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
|
||
handler.setFormatter(formatter)
|
||
self.logger.addHandler(handler)
|
||
self.logger.setLevel(logging.INFO)
|
||
|
||
# ----------- Cookie 文件直接操作 -----------
|
||
|
||
def _read_cookie(self) -> str:
|
||
if not self.cookie_file.exists():
|
||
return ""
|
||
try:
|
||
return self.cookie_file.read_text(encoding="utf-8").strip()
|
||
except OSError as e:
|
||
self.logger.error(f"读取 Cookie 失败: {e}")
|
||
return ""
|
||
|
||
def _write_cookie(self, cookie: str) -> bool:
|
||
"""临时文件 + 原子替换写入；失败自动恢复备份，杜绝半截文件毁掉在用 Cookie。
|
||
临时文件名带随机后缀：同进程多线程并发写同一 Cookie 文件不再互相覆盖 .tmp"""
|
||
if not cookie or not cookie.strip():
|
||
self.logger.error("Cookie 内容不能为空")
|
||
return False
|
||
backup = self._backup_cookie(suffix="prewrite")
|
||
tmp = None
|
||
try:
|
||
import tempfile
|
||
fd, tmpname = tempfile.mkstemp(
|
||
prefix=self.cookie_file.name + '.tmp-', dir=str(self.cookie_file.parent))
|
||
os.close(fd)
|
||
tmp = Path(tmpname)
|
||
tmp.write_text(cookie.strip() + "\n", encoding="utf-8")
|
||
tmp.replace(self.cookie_file)
|
||
return True
|
||
except OSError as e:
|
||
self.logger.error(f"写入 Cookie 失败: {e}")
|
||
# 写失败：恢复备份，保证在用文件仍是旧的有效内容
|
||
try:
|
||
if tmp is not None:
|
||
tmp.unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
if backup is not None and backup.exists() and not self.cookie_file.exists():
|
||
shutil.copy2(backup, self.cookie_file)
|
||
return False
|
||
|
||
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") # 微秒防同秒覆盖
|
||
backup_path = self.cookie_file.with_name(
|
||
f"{self.cookie_file.stem}.{suffix}.{ts}{self.cookie_file.suffix}"
|
||
)
|
||
try:
|
||
shutil.copy2(self.cookie_file, backup_path)
|
||
except OSError as e:
|
||
self.logger.warning(f"备份 Cookie 失败: {e}")
|
||
return None
|
||
# 备份保留上限：同一 suffix 只留最近 3 份——备份含真实 MUSIC_U，
|
||
# 无限累积既占磁盘也会跟着部署包外泄（_mkpkg ignore 已同步排除）
|
||
try:
|
||
siblings = sorted(
|
||
self.cookie_file.parent.glob(
|
||
f"{self.cookie_file.stem}.{suffix}.*{self.cookie_file.suffix}"),
|
||
key=lambda p: p.stat().st_mtime)
|
||
for old in siblings[:-3]:
|
||
old.unlink(missing_ok=True)
|
||
# 顺手清理崩溃遗留的原子写临时文件（cookie.txt.tmp-xxxx，超过 1 小时的必是残骸）
|
||
now = time.time()
|
||
for stale in self.cookie_file.parent.glob(f"{self.cookie_file.name}.tmp-*"):
|
||
try:
|
||
if now - stale.stat().st_mtime > 3600:
|
||
stale.unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
except OSError:
|
||
pass
|
||
return backup_path
|
||
|
||
def _clear_cookie(self) -> bool:
|
||
if not self.cookie_file.exists():
|
||
return True
|
||
try:
|
||
self.cookie_file.unlink()
|
||
return True
|
||
except OSError as e:
|
||
self.logger.error(f"清除 Cookie 失败: {e}")
|
||
return False
|
||
|
||
def _is_cookie_valid(self) -> bool:
|
||
# 登录态凭据是 MUSIC_U 且必须非空——
|
||
# 只有 __csrf/NMTID 这类匿名 cookie 也存在时不能算"有效登录"
|
||
cookies = load_cookies(str(self.cookie_file))
|
||
return bool((cookies.get('MUSIC_U') or '').strip())
|
||
|
||
def _get_cookie_info(self) -> dict:
|
||
cookies = load_cookies(str(self.cookie_file))
|
||
present = [k for k in IMPORTANT_COOKIE_KEYS
|
||
if k == 'MUSIC_U' or (cookies.get(k) or '').strip()]
|
||
valid = bool((cookies.get('MUSIC_U') or '').strip()) # 以 MUSIC_U 为准
|
||
info = {
|
||
"file_path": str(self.cookie_file.absolute()),
|
||
"file_exists": self.cookie_file.exists(),
|
||
"cookie_count": len(cookies),
|
||
"is_valid": valid,
|
||
"important_cookies_present": present,
|
||
"missing_important_cookies": [k for k in IMPORTANT_COOKIE_KEYS
|
||
if not (cookies.get(k) or '').strip()],
|
||
}
|
||
if self.cookie_file.exists():
|
||
try:
|
||
mtime = self.cookie_file.stat().st_mtime
|
||
info["last_modified"] = datetime.datetime.fromtimestamp(mtime).strftime("%Y-%m-%d %H:%M:%S")
|
||
except OSError:
|
||
pass
|
||
return info
|
||
|
||
# ----------- 业务方法 -----------
|
||
|
||
def check_existing_login(self) -> bool:
|
||
"""检查是否已有有效登录"""
|
||
try:
|
||
if self._is_cookie_valid():
|
||
self.logger.info("检测到有效的登录 Cookie")
|
||
return True
|
||
self.logger.info("未检测到有效的登录 Cookie")
|
||
return False
|
||
except Exception as e:
|
||
self.logger.error(f"检查登录状态失败: {e}")
|
||
return False
|
||
|
||
async def interactive_login(self) -> tuple[bool, str | None]:
|
||
"""交互式二维码登录"""
|
||
try:
|
||
print("\n=== 网易云音乐二维码登录 ===")
|
||
|
||
if self.check_existing_login():
|
||
choice = input("检测到已有有效登录，是否重新登录？(y/N): ").strip().lower()
|
||
if choice not in ('y', 'yes', '是'):
|
||
print("使用现有登录状态")
|
||
return True, None
|
||
|
||
print("\n开始二维码登录流程...")
|
||
print("正在生成二维码...")
|
||
# create_qr_login 返回 unikey 字符串（失败为 None），并已在控制台打印二维码
|
||
qr_key = await self.qr_manager.create_qr_login()
|
||
|
||
if not qr_key:
|
||
error_msg = "生成二维码失败"
|
||
self.logger.error(error_msg)
|
||
return False, error_msg
|
||
|
||
print("\n二维码已生成！")
|
||
print("请使用网易云音乐手机 APP 扫描二维码进行登录")
|
||
print("二维码有效期: 3 分钟")
|
||
print("\n等待扫码中...")
|
||
|
||
max_attempts = 60 # 最多 5 分钟
|
||
attempt = 0
|
||
while attempt < max_attempts:
|
||
try:
|
||
# check_qr_login 返回 (状态码, cookie字典)：801等待/802已扫/803成功/800过期
|
||
code, cookie_dict = await self.qr_manager.check_qr_login(qr_key)
|
||
|
||
if code == 803:
|
||
musu_u = cookie_dict.get('MUSIC_U', '')
|
||
if not musu_u:
|
||
return False, "登录成功但未获取到 Cookie"
|
||
# 序列化带上 __csrf——历史上只修了 web 的两条路径，
|
||
# 漏了 CLI 换号：会导致全站 __csrf 恒空、weapi 发评论必败
|
||
serialized = f"MUSIC_U={musu_u};os=pc;appver=8.9.70;"
|
||
csrf = cookie_dict.get('__csrf', '')
|
||
if csrf:
|
||
serialized += f"__csrf={csrf};"
|
||
if self.save_cookie(serialized):
|
||
# 注：不用 emoji——中文 Windows 控制台默认 GBK 编码不了它，
|
||
# print 抛 UnicodeEncodeError 会把"保存成功"误报成"未知错误"
|
||
print("\n[成功] 登录成功！Cookie 已保存")
|
||
return True, None
|
||
return False, "登录成功但 Cookie 保存失败"
|
||
|
||
if code == 802:
|
||
print("二维码已扫描，请在手机上确认登录")
|
||
elif code == 800:
|
||
print("\n[过期] 二维码已过期，请重新尝试")
|
||
return False, "二维码已过期"
|
||
elif code != 801:
|
||
print(f"\n登录状态异常（{code}）")
|
||
|
||
if code == 801 and attempt % 10 == 0:
|
||
print(f"等待扫码中... ({attempt + 1}/{max_attempts})")
|
||
|
||
await asyncio.sleep(5)
|
||
attempt += 1
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n用户取消登录")
|
||
return False, "用户取消登录"
|
||
except Exception as e:
|
||
self.logger.error(f"检查登录状态时发生错误: {e}")
|
||
await asyncio.sleep(5)
|
||
attempt += 1
|
||
|
||
print("\n[超时] 登录超时，请重新尝试")
|
||
return False, "登录超时"
|
||
|
||
except APIException as e:
|
||
self.logger.error(f"API 调用失败: {e}")
|
||
return False, f"API 调用失败: {e}"
|
||
except Exception as e:
|
||
self.logger.error(f"登录过程中发生未知错误: {e}")
|
||
return False, f"登录过程中发生未知错误: {e}"
|
||
|
||
def save_cookie(self, cookie: str) -> bool:
|
||
"""保存 Cookie 到文件"""
|
||
try:
|
||
if self.cookie_file.exists():
|
||
backup = self._backup_cookie()
|
||
if backup:
|
||
self.logger.info(f"已备份现有 Cookie 到: {backup}")
|
||
|
||
if not self._write_cookie(cookie):
|
||
return False
|
||
|
||
if self._is_cookie_valid():
|
||
self.logger.info("Cookie 保存并验证成功")
|
||
return True
|
||
|
||
self.logger.warning("Cookie 保存成功但验证失败（缺少关键字段）")
|
||
return False
|
||
except Exception as e:
|
||
self.logger.error(f"保存 Cookie 时发生错误: {e}")
|
||
return False
|
||
|
||
def show_login_info(self) -> None:
|
||
"""显示登录信息"""
|
||
try:
|
||
info = self._get_cookie_info()
|
||
print("\n=== 登录状态信息 ===")
|
||
print(f"Cookie 文件: {info['file_path']}")
|
||
print(f"文件存在: {'是' if info['file_exists'] else '否'}")
|
||
print(f"Cookie 数量: {info['cookie_count']}")
|
||
print(f"登录状态: {'有效' if info['is_valid'] else '无效'}")
|
||
if info.get("last_modified"):
|
||
print(f"最后更新: {info['last_modified']}")
|
||
if info["is_valid"]:
|
||
present = info.get("important_cookies_present", [])
|
||
print(f"重要 Cookie: {', '.join(present)}")
|
||
else:
|
||
missing = info.get("missing_important_cookies", [])
|
||
if missing:
|
||
print(f"缺少 Cookie: {', '.join(missing)}")
|
||
except Exception as e:
|
||
print(f"获取登录信息失败: {e}")
|
||
|
||
def logout(self) -> bool:
|
||
"""登出（清除 Cookie）。
|
||
|
||
输出走 logger 而不是 print：本方法既被 CLI（main）调用，也被 admin 端点
|
||
（/api/v1/admin/cookie/delete）调用——在 web 语境里 print 会把内容直接喷到
|
||
服务端 stdout，绕过 logging 配置、无法被收集，也会污染容器日志。
|
||
"""
|
||
try:
|
||
if self.cookie_file.exists():
|
||
backup = self._backup_cookie("logout")
|
||
if backup:
|
||
self.logger.info(f"Cookie 已备份到: {backup}")
|
||
if self._clear_cookie():
|
||
self.logger.info("已成功登出")
|
||
return True
|
||
self.logger.error("登出失败")
|
||
return False
|
||
except Exception as e:
|
||
self.logger.error(f"登出时发生错误: {e}")
|
||
return False
|
||
|
||
|
||
def main():
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||
)
|
||
|
||
client = QRLoginClient()
|
||
|
||
if len(sys.argv) > 1:
|
||
command = sys.argv[1].lower()
|
||
|
||
if command == 'login':
|
||
success, error = asyncio.run(client.interactive_login())
|
||
if success:
|
||
print("\n登录完成！")
|
||
client.show_login_info()
|
||
sys.exit(0)
|
||
print(f"\n登录失败: {error}")
|
||
sys.exit(1)
|
||
|
||
elif command in ('status', 'info'):
|
||
client.show_login_info()
|
||
sys.exit(0)
|
||
|
||
elif command == 'logout':
|
||
sys.exit(0 if client.logout() else 1)
|
||
|
||
elif command in ('help', '-h', '--help'):
|
||
print("网易云音乐二维码登录工具")
|
||
print("\n用法: python -m music_api.qr_client [命令]")
|
||
print("\n命令:")
|
||
print(" login - 执行二维码登录")
|
||
print(" status - 显示登录状态")
|
||
print(" logout - 登出（清除 Cookie）")
|
||
print(" help - 显示此帮助信息")
|
||
print("\n如果不提供命令，将进入交互模式")
|
||
sys.exit(0)
|
||
|
||
else:
|
||
print(f"未知命令: {command}")
|
||
print("使用 'python -m music_api.qr_client help' 查看帮助")
|
||
sys.exit(1)
|
||
|
||
# 交互模式
|
||
try:
|
||
while True:
|
||
print("\n=== 网易云音乐登录工具 ===")
|
||
print("1. 二维码登录")
|
||
print("2. 查看登录状态")
|
||
print("3. 登出")
|
||
print("4. 退出")
|
||
|
||
choice = input("\n请选择操作 (1-4): ").strip()
|
||
|
||
if choice == '1':
|
||
success, error = asyncio.run(client.interactive_login())
|
||
if success:
|
||
print("\n登录成功！")
|
||
client.show_login_info()
|
||
else:
|
||
print(f"\n登录失败: {error}")
|
||
elif choice == '2':
|
||
client.show_login_info()
|
||
elif choice == '3':
|
||
client.logout()
|
||
elif choice == '4':
|
||
print("再见！")
|
||
break
|
||
else:
|
||
print("无效选择，请重试")
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n\n程序已退出")
|
||
sys.exit(0)
|
||
except Exception as e:
|
||
print(f"\n程序运行出错: {e}")
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|