LoginApi

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

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

    Returns:
        bool: 是否已过期.
    """
    target = credential or self._client.credential
    try:
        await self._client.execute(
            self._build_request(
                module="music.UserInfo.userInfoServer",
                method="GetLoginUserInfo",
                param={},
                credential=target,
            ),
        )
        return False
    except LoginExpiredError:
        return True

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

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

    Returns:
        Credential: 刷新后的新凭证对象.
    """
    target = credential or self._client.credential
    try:
        data = await self._client.execute(
            self._build_request(
                module="music.login.LoginServer",
                method="Login",
                param={
                    "openid": target.openid,
                    "access_token": target.access_token,
                    "unionid": target.unionid,
                    "refresh_key": target.refresh_key,
                    "refresh_token": target.refresh_token,
                    "musickey": target.musickey,
                    "musicid": target.musicid,
                    "loginMode": 2,
                },
                comm={"tmeLoginType": target.login_type},
                credential=target,
            ),
        )
    except ApiError as exc:
        raise _raise_login_error("RefreshCredential", "刷新凭证失败", cause=exc) from exc

    return Credential.model_validate(data)

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 时包含) 的结果对象.
    """
    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(f"MQTT network error: {exc}", original_exc=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,
                        QRCodeLoginEvents.OTHER,
                    }:
                        return
        except ConnectionError as exc:
            raise NetworkError(f"MQTT network error: {exc}", original_exc=exc) from exc

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

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

    Returns:
        PhoneAuthCodeResult: 包含发送状态及附加信息的结果对象.
    """
    try:
        await self._client.execute(
            self._build_request(
                module="music.login.LoginServer",
                method="SendPhoneAuthCode",
                param={"tmeAppid": "qqmusic", "phoneNo": str(phone), "areaCode": str(country_code)},
                comm={"tmeLoginMethod": 3},
                platform=Platform.ANDROID,
            ),
        )
    except ApiError as exc:
        if exc.code == PhoneLoginEvents.CAPTCHA.value:
            return PhoneAuthCodeResult(event=PhoneLoginEvents.CAPTCHA)
        if exc.code == PhoneLoginEvents.FREQUENCY.value:
            return PhoneAuthCodeResult(event=PhoneLoginEvents.FREQUENCY)
        raise _raise_login_error("PhoneLogin", "发送验证码失败", code=exc.code, cause=exc) from exc

    return PhoneAuthCodeResult(event=PhoneLoginEvents.SEND)

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

    Args:
        phone: 手机号.
        auth_code: 验证码.
        country_code: 国家代码, 默认为 86.

    Returns:
        Credential: 登录成功后的凭证对象.
    """
    try:
        data = await self._client.execute(
            self._build_request(
                module="music.login.LoginServer",
                method="Login",
                param={
                    "code": str(auth_code),
                    "phoneNo": str(phone),
                    "areaCode": str(country_code),
                    "loginMode": 1,
                },
                comm={"tmeLoginMethod": 3, "tmeLoginType": 0},
                platform=Platform.ANDROID,
            ),
        )
    except ApiError as exc:
        if exc.code == 20274:
            raise _raise_login_error("PhoneLogin", "设备数量限制", code=exc.code, cause=exc) from exc
        if exc.code == 20271:
            raise _raise_login_error("PhoneLogin", "验证码错误或已鉴权", code=exc.code, cause=exc) from exc
        raise _raise_login_error("PhoneLogin", "鉴权失败", code=exc.code, cause=exc) from exc

    return Credential.model_validate(data)