LoginApi

async def check_expired(self, credential: Credential | None = None) -> bool:
    """检查登录凭证是否已过期.

    Args:
        credential: 待检查的凭证. 若为 None 则检查当前客户端已存储的凭证.

    Returns:
        bool: 是否已过期.
    """
    target = credential or self._client.credential
    if self._client.platform == Platform.WEB:
        resp = await self._client.request(
            "GET",
            "https://c6.y.qq.com/rsc/fcgi-bin/fcg_get_profile_homepage.fcg",
            credential=target,
            params={
                "g_tk": hash33(target.musickey, 5381),
                "format": "json",
                "inCharset": "utf-8",
                "outCharset": "utf-8",
                "notice": 0,
                "cid": 205360838,
                "needNewCode": 0,
                "loginUin": target.musicid,
                "hostUin": 0,
                "userid": target.musicid,
                "reqfrom": "1",
            },
        )
        return resp.json().get("code") != 0

    data = await self._build_request(
        module="music.UserInfo.userInfoServer",
        method="GetLoginUserInfo",
        param={},
        credential=target,
        allow_error_codes=(1000, 104401, 104400),
    )
    return data.get("code", 0) != 0

async def refresh_credential(self, credential: Credential | None = None) -> Credential:
    """尝试刷新登录凭证.

    Args:
        credential: 待刷新的凭证.

    Raises:
        CredentialRefreshError: 凭证刷新失败.

    Returns:
        Credential: 刷新后的新凭证对象.
    """
    target = credential or self._client.credential
    self._require_login(target)
    match target.login_type:
        case 1:
            param = {
                "openid": target.openid,
                "refresh_token": target.refresh_token,
                "str_musicid": target.str_musicid or str(target.musicid),
                "musickey": target.musickey,
                "unionid": target.unionid,
                "refresh_key": target.refresh_key,
                "loginMode": 2,
            }

        case 2:
            param = {
                "openid": target.openid,
                "access_token": target.access_token,
                "refresh_token": target.refresh_token,
                "expired_in": target.expired_at,
                "musicid": target.musicid,
                "musickey": target.musickey,
                "refresh_key": target.refresh_key,
                "loginMode": 2,
            }

        case _:
            param = {
                "openid": target.openid,
                "access_token": target.access_token,
                "refresh_token": target.refresh_token,
                "expired_in": target.expired_at,
                "str_musicid": target.str_musicid or str(target.musicid),
                "musicid": target.musicid,
                "musickey": target.musickey,
                "unionid": target.unionid,
                "refresh_key": target.refresh_key,
                "loginMode": 2,
            }
    data = await self._build_request(
        module="music.login.LoginServer",
        method="Login",
        param=param,
        comm={"tmeLoginType": target.login_type},
        credential=target,
        allow_error_codes=_ERROR_CODE,
    )
    try:
        return Credential.model_validate(self._validate_result(data))
    except LoginError as exc:
        raise CredentialRefreshError(message=exc.message, code=exc.code, data=exc.data) from exc

async def get_qrcode(self, login_type: QRLoginType) -> QR:
    """获取指定类型的登录二维码.

    Args:
        login_type: 登录类型 (QQ/微信/手机客户端).

    Returns:
        QR: 包含二维码二进制数据及标识符的对象.
    """
    if login_type == QRLoginType.WX:
        return await self._get_wx_qr()
    if login_type == QRLoginType.MOBILE:
        return await self._get_mobile_qr()
    return await self._get_qq_qr()

async def check_qrcode(self, qrcode: QR) -> QRLoginResult:
    """检查二维码登录状态.

    Args:
        qrcode: 待检查的二维码对象.

    Returns:
        QRLoginResult: 包含当前状态和凭证 (仅在 DONE 时包含) 的结果对象.

    Raises:
        LoginAuthExpiredError: 登录鉴权参数无效或已过期 (code=1000/104401/104400).
        LoginDeviceLimitError: 登录设备数量超限 (code=20279).
        LoginAccountRestrictedError: 账号受限或已被封禁 (code=20277/20278/20450).
        LoginRateLimitError: 操作过于频繁 (code=104604).
        LoginError: 其他登录业务异常.
    """
    if qrcode.qr_type == QRLoginType.WX:
        return await self._check_wx_qr(qrcode)
    return await self._check_qq_qr(qrcode)

async def checking_mobile_qrcode(self, qrcode: QR, deadline: float | None = None) -> QRLoginStream:
    """检查手机登录二维码状态 (单次 MQTT 连接生命周期).

    建立 MQTT 订阅并持续产出服务端推送的登录状态事件.
    当收到终端事件 (DONE/REFUSE/TIMEOUT/OTHER) 时会主动结束迭代.

    Args:
        qrcode: 待检查的二维码对象.
        deadline: 基于 `anyio.current_time()` 的最长等待截止时间. 为 None 时不额外限制超时.

    Yields:
        QRLoginResult: 包含当前状态和凭证的结果对象.

    Raises:
        NetworkError: MQTT 建连、订阅或消息监听过程中发生网络错误.
    """
    client_id = f"{int(time() * 1000)}{random.randint(1000, 9999)}"

    def get_timeout_left() -> float | None:
        """返回当前 deadline 剩余秒数."""
        if deadline is None:
            return None
        return deadline - anyio.current_time()

    async def await_before_deadline(operation: Callable[[], Any]) -> Any:
        """在 deadline 之前完成单次异步操作."""
        timeout_left = get_timeout_left()
        if timeout_left is None:
            return await operation()
        if timeout_left <= 0:
            raise TimeoutError
        with anyio.fail_after(timeout_left):
            return await operation()

    async with MqttClient(
        client_id=client_id,
        host="mu.y.qq.com",
        port=443,
        path="/ws/handshake",
        keep_alive=45,
    ) as client:
        try:
            await await_before_deadline(lambda: self._connect_mobile_mqtt(client, qrcode.identifier))
            topic = f"management.qrcode_login/{qrcode.identifier}"
            await await_before_deadline(
                lambda: client.subscribe(
                    topic,
                    properties={PropertyId.USER_PROPERTY: [("authorization", "tmelogin"), ("pubsub", "unicast")]},
                ),
            )
        except TimeoutError:
            yield QRLoginResult(event=QRCodeLoginEvents.TIMEOUT)
            return
        except ConnectionError as exc:
            raise NetworkError(str(exc)) from exc

        yield QRLoginResult(event=QRCodeLoginEvents.SCAN)

        try:
            async with aclosing(client.messages()) as messages:
                while True:
                    try:
                        message = await await_before_deadline(lambda: anext(messages))
                    except StopAsyncIteration:
                        return
                    except TimeoutError:
                        yield QRLoginResult(event=QRCodeLoginEvents.TIMEOUT)
                        return

                    message_type = message.properties.get("type")
                    message_payload = message.json
                    try:
                        event_item = await await_before_deadline(
                            lambda message_type=message_type, message_payload=message_payload: (
                                self._handle_mobile_message(
                                    qrcode.identifier,
                                    message_type,
                                    message_payload,
                                )
                            ),
                        )
                    except TimeoutError:
                        yield QRLoginResult(event=QRCodeLoginEvents.TIMEOUT)
                        return
                    if event_item is None:
                        continue

                    yield event_item

                    if event_item.event in {
                        QRCodeLoginEvents.DONE,
                        QRCodeLoginEvents.REFUSE,
                        QRCodeLoginEvents.TIMEOUT,
                    }:
                        return
        except ConnectionError as exc:
            raise NetworkError(str(exc)) from exc

async def send_authcode(
    self,
    phone: int | str,
    country_code: int = 86,
) -> PhoneAuthCodeResult:
    """发送手机验证码.

    固定使用 Android 平台.

    Args:
        phone: 手机号 (int) 或加密手机号 (str).
        country_code: 国家代码, 默认为 86 (中国).

    Returns:
        PhoneAuthCodeResult: 包含发送状态及附加信息的结果对象.
    """
    param: dict[str, str] = {"tmeAppid": "qqmusic", "areaCode": str(country_code)}
    if isinstance(phone, str):
        param["encryptedPhoneNo"] = phone
    else:
        param["phoneNo"] = str(phone)

    resp = await self._build_request(
        module="music.login.LoginServer",
        method="SendPhoneAuthCode",
        param=param,
        comm={"tmeLoginMethod": 3},
        platform=Platform.ANDROID,
        allow_error_codes="all",
    )
    code = resp.get("code", 0)
    data = resp.get("data", {})
    match code:
        case PhoneLoginEvents.CAPTCHA.value:
            return PhoneAuthCodeResult(event=PhoneLoginEvents.CAPTCHA, info=data.get("securityURL"))
        case PhoneLoginEvents.FREQUENCY.value:
            return PhoneAuthCodeResult(event=PhoneLoginEvents.FREQUENCY)
        case PhoneLoginEvents.SEND.value:
            return PhoneAuthCodeResult(event=PhoneLoginEvents.SEND)
        case _:
            raise LoginError("发送验证码失败", code=code, data=data)

async def phone_authorize(
    self,
    phone: int | str,
    auth_code: str,
) -> Credential:
    """使用手机验证码鉴权.

    固定使用 Android 平台.

    Args:
        phone: 手机号 (int) 或加密手机号 (str).
        auth_code: 验证码.

    Returns:
        Credential: 登录成功后的凭证对象.

    Raises:
        LoginAuthExpiredError: 登录鉴权参数无效或已过期 (code=1000/104401/104400).
        LoginDeviceLimitError: 登录设备数量超限 (code=20279).
        LoginAccountRestrictedError: 账号受限或已被封禁 (code=20277/20278/20450).
        LoginRateLimitError: 操作过于频繁 (code=104604).
        LoginError: 其他登录业务异常.
    """
    param: dict[str, str | int] = {"code": auth_code, "loginMode": 1}
    if isinstance(phone, str):
        param["encryptedPhoneNo"] = phone
    else:
        param["phoneNo"] = str(phone)

    data = await self._build_request(
        module="music.login.LoginServer",
        method="Login",
        param=param,
        comm={"tmeLoginMethod": 3, "tmeLoginType": 0},
        platform=Platform.ANDROID,
        allow_error_codes=_ERROR_CODE,
    )

    return Credential.model_validate(self._validate_result(data))