"""网易云音乐二维码登录模块 提供网易云音乐二维码登录功能，包括： - 二维码生成和显示 - 登录状态检查 - 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()