NeteaseUrl/music_api/qr.py

# -*- coding: utf-8 -*-
"""扫码登录管理器。

（自 music_api.py 拆出，行为与原实现一致。）
"""

import asyncio
import json
from random import randrange

from .client import (
    HTTPClient,
    _ensure_dict,
)
from .constants import (
    APIConstants,
)
from .crypto import (
    CryptoUtils,
)
from .errors import (
    APIException,
)


class QRLoginManager:
    """二维码登录管理器"""

    def __init__(self):
        self.http_client = HTTPClient()
        self.crypto_utils = CryptoUtils()

    async def generate_qr_key(self) -> str | None:
        """生成二维码的key

        Returns:
            成功返回unikey，失败返回None

        Raises:
            APIException: API调用失败时抛出
        """
        try:
            config = APIConstants.DEFAULT_CONFIG.copy()
            config["requestId"] = str(randrange(20000000, 30000000))

            payload = {
                'type': 1,
                'header': json.dumps(config)
            }

            params = self.crypto_utils.encrypt_params(APIConstants.QR_UNIKEY_API, payload)
            response = await self.http_client.post_request_full(APIConstants.QR_UNIKEY_API, params, {})

            result = json.loads(response.text)
            # 非 dict 响应（null/数组）不再 AttributeError 逃出捕获
            _ensure_dict(result, '生成二维码key响应异常（非标准 JSON 对象）')
            if result.get('code') == 200:
                return result.get('unikey')
            else:
                raise APIException(f"生成二维码key失败: {result.get('message', '未知错误')}")
        except (json.JSONDecodeError, KeyError) as e:
            raise APIException(f"解析二维码key响应失败: {e}") from e

    async def create_qr_login(self) -> str | None:
        """创建登录二维码并在控制台显示

        Returns:
            成功返回unikey，失败返回None
        """
        try:
            import qrcode

            unikey = await self.generate_qr_key()
            if not unikey:
                print("生成二维码key失败")
                return None

            # 创建二维码
            qr = qrcode.QRCode()
            qr.add_data(f'https://music.163.com/login?codekey={unikey}')
            qr.make(fit=True)

            # 在控制台显示二维码
            qr.print_ascii(tty=True)
            print("\n请使用网易云音乐APP扫描上方二维码登录")
            return unikey
        except ImportError:
            print("请安装qrcode库: pip install qrcode")
            return None
        except Exception as e:
            print(f"创建二维码失败: {e}")
            return None

    async def check_qr_login(self, unikey: str) -> tuple[int, dict[str, str]]:
        """检查二维码登录状态

        Args:
            unikey: 二维码key

        Returns:
            (登录状态码, cookie字典)

        Raises:
            APIException: API调用失败时抛出
        """
        try:
            config = APIConstants.DEFAULT_CONFIG.copy()
            config["requestId"] = str(randrange(20000000, 30000000))

            payload = {
                'key': unikey,
                'type': 1,
                'header': json.dumps(config)
            }

            params = self.crypto_utils.encrypt_params(APIConstants.QR_LOGIN_API, payload)
            response = await self.http_client.post_request_full(APIConstants.QR_LOGIN_API, params, {})

            result = json.loads(response.text)
            # 非 dict 响应不再 AttributeError 逃出捕获（同 generate_qr_key）
            _ensure_dict(result, '解析登录状态响应失败（非标准 JSON 对象）')
            cookie_dict = {}

            if result.get('code') == 803:
                # 登录成功，提取cookie。
                # 此前只提取 MUSIC_U——__csrf 从未入 Cookie 串，
                # weapi 写接口（发评论等）的 csrf_token 恒为空。改用 requests 的
                # cookie jar（requests 已把多条 Set-Cookie 合并解析，不受逗号合并
                # 的脆弱 split 影响）补齐 MUSIC_U 与 __csrf。
                for name in ('MUSIC_U', '__csrf'):
                    val = response.cookies.get(name)
                    if val:
                        cookie_dict[name] = val
                if not cookie_dict.get('MUSIC_U'):
                    # 兜底：jar 异常为空时回退 Set-Cookie 头解析 MUSIC_U
                    for cookie_str in response.headers.get('Set-Cookie', '').split(', '):
                        if 'MUSIC_U=' in cookie_str:
                            cookie_dict['MUSIC_U'] = cookie_str.split('MUSIC_U=')[1].split(';')[0]
                            break

            return result.get('code', -1), cookie_dict
        except (json.JSONDecodeError, KeyError) as e:
            raise APIException(f"解析登录状态响应失败: {e}") from e

    async def qr_login(self) -> str | None:
        """完整的二维码登录流程

        Returns:
            成功返回cookie字符串，失败返回None
        """
        try:
            unikey = await self.create_qr_login()
            if not unikey:
                return None

            while True:
                code, cookies = await self.check_qr_login(unikey)

                if code == 803:
                    print("\n登录成功！")
                    # CLI 序列化须与 qr_login 口径一致，带上 __csrf
                    # （此串正是写往全局 cookie.txt 的内容）
                    serialized = f"MUSIC_U={cookies['MUSIC_U']};os=pc;appver=8.9.70;"
                    if cookies.get('__csrf'):
                        serialized += f"__csrf={cookies['__csrf']};"
                    return serialized
                elif code == 801:
                    print("\r等待扫码...", end='')
                elif code == 802:
                    print("\r扫码成功，请在手机上确认登录...", end='')
                else:
                    print(f"\n登录失败，错误码：{code}")
                    return None

                await asyncio.sleep(2)
        except KeyboardInterrupt:
            print("\n用户取消登录")
            return None
        except Exception as e:
            print(f"\n登录过程中发生错误: {e}")
            return None