def set_token(self, user_token: str, refresh_token: str) -> None:
    """设置统一认证 Token。

    Args:
        user_token: `userToken`。对豫见郑大 APP 抓包获取，或账密登录后访问 [`user_token`][zzupy.aio.app.auth.CASClient.user_token] 获取
        refresh_token: `refreshToken`。对豫见郑大 APP 抓包获取，或账密登录后访问 [`refresh_token`][zzupy.aio.app.auth.CASClient.refresh_token] 获取
    """
    self._user_token = user_token
    self._refresh_token = refresh_token

def set_device(self, device_id: str) -> None:
    """设置认证请求使用的设备标识。

    Args:
        device_id: 登录和 MFA 检测请求中的 `deviceId`。
    """
    self._device_id = device_id
    self.mfa.reset()

async def login(self, force_login: bool = False) -> None:
    """登录统一认证。

    成功后，[`userToken`][zzupy.aio.app.auth.CASClient.user_token] 和 [`refreshToken`][zzupy.aio.app.auth.CASClient.refresh_token] 会被存储在实例中.

    若 [`user_token`][zzupy.aio.app.auth.CASClient.user_token] 和 [`refresh_token`][zzupy.aio.app.auth.CASClient.refresh_token] 已通过 [`set_token`][zzupy.aio.app.auth.CASClient.set_token] 设置且有效，则会跳过账密登录。

    Args:
        force_login: 强制使用账密登录

    Raises:
        MFAError: 如果当前登录需要 MFA 但尚未完成验证。
        LoginError: 如果登录失败。
        ParsingError: 如果服务器响应无法解析。
        NetworkError: 如果出现网络错误。
    """
    if not force_login:
        if self._user_token is None or self._refresh_token is None:
            logger.debug("userToken 或 refreshToken 不存在，使用账密登录")
        elif self._validate_jwt(True):
            logger.debug("userToken 和 refreshToken 已设置且有效，跳过账密登录")
            self._logged_in = True
            return
    else:
        logger.info("强制使用账密登录")

    if self._public_key is None:
        self._public_key = await self._get_public_key()

    assert self._public_key is not None

    if not self.mfa.state:
        await self.mfa.is_required()
    if self.mfa.required and not self.mfa.verified:
        raise MFAError("当前登录需要完成 MFA 验证")

    encrypted_account = self._encrypt_and_encode(self._account, self._public_key)
    encrypted_password = self._encrypt_and_encode(self._password, self._public_key)

    headers = {"User-Agent": f"{self.APP_VERSION}()"}
    params = {
        "username": encrypted_account,
        "password": encrypted_password,
        "appId": self.APP_ID,
        "osType": self.OS_TYPE,
        "geo": "",
        "deviceId": self._device_id,
        "clientId": "",
        "mfaState": self.mfa.state,
    }

    try:
        logger.debug("正在向 {} 发送登录请求...", self.LOGIN_URL)
        response = await self._client.post(
            self.LOGIN_URL, params=params, headers=headers
        )
        response.raise_for_status()

        log_http_response_body(
            self.LOGIN_URL,
            response.text,
            content_type=response.headers.get("content-type"),
            level="DEBUG",
        )

        data = response.json()

        if data.get("code") != 0:
            error_message = data.get("message", "未知错误")
            logger.error("登录请求失败: {}", error_message)
            raise LoginError(f"登录失败: {error_message}")

        token_data = data["data"]
        self._user_token = token_data["idToken"]
        self._refresh_token = token_data["refreshToken"]
        self._validate_jwt()
        self._logged_in = True

        logger.info("统一认证登录成功")

    except httpx2.HTTPStatusError as exc:
        logger.error("登录请求返回失败状态码: {}", exc.response.status_code)
        raise LoginError.from_http_status(
            exc,
            "服务器返回错误状态",
            context={"url": self.LOGIN_URL},
        ) from exc
    except (json.JSONDecodeError, KeyError, TypeError) as exc:
        logger.error("从 /passwordLogin 响应中提取 token 失败: {}", exc)
        raise ParsingError.from_exception(
            exc,
            "服务器响应格式不正确",
            context={"url": self.LOGIN_URL},
        ) from exc
    except httpx2.RequestError as exc:
        logger.error("登录网络请求失败: {}", exc)
        raise NetworkError.from_exception(
            exc,
            "网络连接异常",
            context={"url": self.LOGIN_URL},
        ) from exc

@require_auth
async def get_user_info(self) -> PersonalInfo:
    """获取当前用户的聚合个人信息。

    返回学号、姓名、身份类型、学院、邮箱未读数、一卡通余额和科研信息数量。

    Returns:
        当前用户的个人信息

    Raises:
        NotLoggedInError: 如果当前认证状态缺少 userToken。
        OperationError: 如果服务端返回失败结果。
        ParsingError: 如果响应解析失败。
        NetworkError: 如果网络请求失败。
    """
    headers = {"X-Id-Token": self._require_user_token()}
    url = f"{self.PERSONAL_INFO_URL}"
    try:
        response = await self._client.get(url, headers=headers)
        response.raise_for_status()
        log_http_response_body(
            url,
            response.text,
            content_type=response.headers.get("content-type"),
        )

        response_data = response.json()

    except httpx2.HTTPStatusError as exc:
        logger.error("{}请求返回失败状态码: {}", url, exc.response.status_code)
        raise OperationError.from_http_status(
            exc,
            "服务器返回错误状态",
            context={"url": url},
        ) from exc
    except (json.JSONDecodeError, TypeError) as exc:
        logger.error("从 {} 响应中提取数据失败: {}", url, exc)
        raise ParsingError.from_exception(
            exc,
            "服务器响应格式不正确",
            context={"url": url},
        ) from exc
    except httpx2.RequestError as exc:
        logger.error("{} 请求失败: {}", url, exc)
        raise NetworkError.from_exception(
            exc,
            "网络连接异常",
            context={"url": url},
        ) from exc

    if response_data["code"] != 0:
        logger.error("服务器返回消息 {}", response_data["message"])
        raise OperationError(f"服务器返回消息 {response_data['message']}")

    try:
        personal_info_data = PersonalInfoModel.model_validate(response_data)
    except ValidationError as exc:
        logger.error("从 {} 响应中解析数据失败: {}", url, exc)
        raise ParsingError.from_exception(
            exc,
            "服务器响应格式不正确",
            context={"url": url},
        ) from exc

    url = f"{self.PERSONAL_INFO_CARD_URL}"
    try:
        response = await self._client.get(url, headers=headers)
        response.raise_for_status()
        log_http_response_body(
            url,
            response.text,
            content_type=response.headers.get("content-type"),
        )

        response_data = response.json()

    except httpx2.HTTPStatusError as exc:
        logger.error("{}请求返回失败状态码: {}", url, exc.response.status_code)
        raise OperationError.from_http_status(
            exc,
            "服务器返回错误状态",
            context={"url": url},
        ) from exc
    except (json.JSONDecodeError, TypeError) as exc:
        logger.error("从 {} 响应中提取数据失败: {}", url, exc)
        raise ParsingError.from_exception(
            exc,
            "服务器响应格式不正确",
            context={"url": url},
        ) from exc
    except httpx2.RequestError as exc:
        logger.error("{} 请求失败: {}", url, exc)
        raise NetworkError.from_exception(
            exc,
            "网络连接异常",
            context={"url": url},
        ) from exc

    if response_data["code"] != 0:
        logger.error("服务器返回消息 {}", response_data["message"])
        raise OperationError(f"服务器返回消息 {response_data['message']}")

    try:
        personal_info_card_data = PersonalInfoCardModel.model_validate(
            response_data
        )
    except ValidationError as exc:
        logger.error("从 {} 响应中解析数据失败: {}", url, exc)
        raise ParsingError.from_exception(
            exc,
            "服务器响应格式不正确",
            context={"url": url},
        ) from exc

    return PersonalInfo(
        uid=personal_info_data.data.attributes.user_uid,
        name=personal_info_data.data.attributes.user_name,
        student_type=personal_info_data.data.attributes.identity_type_name,
        student_type_id=personal_info_data.data.attributes.identity_type_id,
        college=personal_info_data.data.attributes.organization_name,
        college_id=personal_info_data.data.attributes.organization_id,
        unread_email_count=int(personal_info_card_data.data[0].amount),
        balance=float(personal_info_card_data.data[1].amount),
        research_count=int(personal_info_card_data.data[2].amount),
    )

@require_auth
def logout(self) -> None:
    """登出账户，清除 Cookie 但保留连接池"""
    self._client.cookies.clear()
    self._client.headers.clear()
    self._user_token = None
    self._refresh_token = None
    self.mfa.reset()
    if self._refresh_task is not None:
        self._refresh_task.cancel()
        self._refresh_task = None
    self._logged_in = False

async def close(self) -> None:
    """清除 Cookie 和连接池"""
    if self._logged_in:
        self.logout()
    await self._client.aclose()

async def login(self) -> None:
    """登录到校园卡系统

    Raises:
        ParsingError: 如果响应解析失败。
        NetworkError: 如果网络请求失败。
    """
    logger.debug("开始登录校园卡系统")
    await self._get_tid()
    await self._get_tokens()
    self._logged_in = True
    self._schedule_token_refresh()
    logger.info("校园卡系统登录成功")

@require_auth
async def get_default_room(self) -> str:
    """获取账户默认房间

    Returns:
        默认的房间

    Raises:
        ParsingError: 如果响应解析失败。
        NetworkError: 如果网络请求失败。
    """

    if self._default_room is not None:
        return self._default_room

    headers = {"Authorization": self._require_access_token()}
    data = {"utilityType": "electric"}

    try:
        logger.debug("正在向 {} 发送请求获取默认房间...", self.CONFIG_URL)
        response = await self._client.post(
            self.CONFIG_URL,
            headers=headers,
            json=data,
        )
        response.raise_for_status()

        log_http_response_body(
            self.CONFIG_URL,
            response.text,
            content_type=response.headers.get("content-type"),
            level="DEBUG",
        )

        response_data = response.json()

        if "resultData" not in response_data:
            logger.error("响应中缺少 resultData")
            raise ParsingError("服务器响应格式不正确")

        room = response_data["resultData"]["location"]["room"]
        logger.info("获取默认房间成功: {}", room)
        self._default_room = room
        return room

    except httpx2.HTTPStatusError as exc:
        logger.error("获取默认房间请求返回失败状态码: {}", exc.response.status_code)
        raise NetworkError.from_http_status(
            exc,
            "服务器返回错误状态",
            context={"url": self.CONFIG_URL},
        ) from exc
    except (json.JSONDecodeError, KeyError, TypeError) as exc:
        logger.error("从 /utilities/config 响应中提取房间信息失败: {}", exc)
        raise ParsingError.from_exception(
            exc,
            "服务器响应格式不正确",
            context={"url": self.CONFIG_URL},
        ) from exc
    except httpx2.RequestError as exc:
        logger.error("获取默认房间网络请求失败: {}", exc)
        raise NetworkError.from_exception(
            exc,
            "网络连接异常",
            context={"url": self.CONFIG_URL},
        ) from exc

@require_auth
async def recharge_energy(self, payment_password: str, amt: int, room: str) -> None:
    """为 room 充值电费

    Args:
        payment_password: 支付密码
        amt: 充值金额
        room: 房间 ID 。格式应为 'areaid-buildingid--unitid-roomid'，可通过
            [`get_room_dict()`][zzupy.aio.app.ecard.ECardClient.get_room_dict] 获取

    Raises:
        InvalidArgumentError: 如果金额或房间参数不合法。
        OperationError: 如果充值失败。
        ParsingError: 如果响应解析失败。
        NetworkError: 如果网络请求失败。
    """

    if amt <= 0:
        logger.error("充值金额必须大于0")
        raise InvalidArgumentError("充值金额必须大于 0", context={"amt": amt})

    logger.debug("准备为房间 {} 充值 {} 元", room, amt)

    headers = {"Authorization": self._require_access_token()}

    try:
        # 获取加密信息
        logger.debug("正在向 {} 发送请求获取加密信息...", self.ENCRYPT_URL)
        response = await self._client.post(
            self.ENCRYPT_URL,
            headers=headers,
        )
        response.raise_for_status()

        response_data = response.json()

        if "resultData" not in response_data:
            logger.error("响应中缺少 resultData")
            raise ParsingError("服务器响应格式不正确")

        pay_id = response_data["resultData"]["id"]
        encrypted_public_key = response_data["resultData"]["publicKey"]

        logger.debug("开始解密公钥")
        # 解密被加密的公钥
        public_key = sm4_decrypt_ecb(
            base64.b64decode(encrypted_public_key),
            self.SM4_KEY,
        )

        # 解析房间信息
        try:
            area, building = room.split("--")[0].split("-")
            level = room.split("--")[1].split("-")[0]
        except (IndexError, ValueError) as exc:
            logger.error("房间格式不正确: {}", room)
            raise InvalidArgumentError(
                f"房间格式不正确: {room}",
                context={"room": room},
            ) from exc

        # 构建请求体
        json_data = {
            "utilityType": "electric",
            "payCode": "06",
            "password": payment_password,
            "amt": str(amt),
            "timestamp": int(round(time.time() * 1000)),
            "bigArea": "",
            "area": area,
            "building": building,
            "unit": "",
            "level": level,
            "room": room,
            "subArea": "",
            "customfield": {},
        }
        json_string = json.dumps(json_data, separators=(",", ":"))

        logger.debug("开始加密支付信息")
        # 加密 params
        sm2 = gmalg.SM2(pk=bytes.fromhex(public_key))
        encrypted_params = sm2.encrypt(json_string.encode())
        data = {"id": pay_id, "params": (encrypted_params.hex())[2:]}

        logger.debug("正在向 {} 发送充值请求...", self.PAY_URL)
        response = await self._client.post(
            self.PAY_URL,
            headers=headers,
            json=data,
        )
        response.raise_for_status()

        log_http_response_body(
            self.PAY_URL,
            response.text,
            content_type=response.headers.get("content-type"),
            level="DEBUG",
        )

        response_data = response.json()
        if response_data.get("success") is False:
            error_msg = response_data.get("message", "充值失败")
            logger.error("充值失败: {}", error_msg)
            raise OperationError(error_msg)

        logger.info("成功为房间 {} 充值 {} 元", room, amt)

    except httpx2.HTTPStatusError as exc:
        logger.error("充值请求返回失败状态码: {}", exc.response.status_code)
        raise NetworkError.from_http_status(
            exc,
            "服务器返回错误状态",
            context={"url": self.PAY_URL, "room": room},
        ) from exc
    except (json.JSONDecodeError, KeyError, TypeError) as exc:
        logger.error("从充值响应中提取数据失败: {}", exc)
        raise ParsingError.from_exception(
            exc,
            "服务器响应格式不正确",
            context={"url": self.PAY_URL, "room": room},
        ) from exc
    except httpx2.RequestError as exc:
        logger.error("充值网络请求失败: {}", exc)
        raise NetworkError.from_exception(
            exc,
            "网络连接异常",
            context={"url": self.PAY_URL, "room": room},
        ) from exc

@require_auth
async def get_balance(self) -> float:
    """获取校园卡余额

    Returns:
        校园卡余额

    Raises:
        ParsingError: 如果响应解析失败。
        NetworkError: 如果网络请求失败。
    """
    headers = {"X-Id-Token": self._require_user_token()}

    try:
        logger.debug("正在向 {} 发送请求获取校园卡余额...", self.BALANCE_URL)
        response = await self._client.get(
            self.BALANCE_URL,
            headers=headers,
        )
        response.raise_for_status()

        log_http_response_body(
            self.BALANCE_URL,
            response.text,
            content_type=response.headers.get("content-type"),
            level="DEBUG",
        )

        response_data = response.json()

        if "data" not in response_data or len(response_data["data"]) < 2:
            logger.error("响应数据格式不正确")
            raise ParsingError("服务器响应格式不正确")

        balance = float(response_data["data"][1]["amount"])
        logger.info("获取校园卡余额成功: {} 元", balance)
        return balance

    except httpx2.HTTPStatusError as exc:
        logger.error("获取余额请求返回失败状态码: {}", exc.response.status_code)
        raise NetworkError.from_http_status(
            exc,
            "服务器返回错误状态",
            context={"url": self.BALANCE_URL},
        ) from exc
    except (
        json.JSONDecodeError,
        KeyError,
        IndexError,
        TypeError,
        ValueError,
    ) as exc:
        logger.error("从余额响应中提取数据失败: {}", exc)
        raise ParsingError.from_exception(
            exc,
            "服务器响应格式不正确",
            context={"url": self.BALANCE_URL},
        ) from exc
    except httpx2.RequestError as exc:
        logger.error("获取余额网络请求失败: {}", exc)
        raise NetworkError.from_exception(
            exc,
            "网络连接异常",
            context={"url": self.BALANCE_URL},
        ) from exc

@require_auth
async def get_room_dict(self, room_id: str) -> dict:
    """获取房间的字典

    Args:
        room_id: 已知房间 ID 。例如: '', '99', '99-12', '99-12--33'

    Returns:
        对应的字典

    Raises:
        InvalidArgumentError: 如果参数格式不正确。
        ParsingError: 如果响应解析失败。
        NetworkError: 如果网络请求失败。
    """

    logger.debug("正在获取房间列表，room_id: {}", room_id)

    # 解析房间ID
    num = room_id.count("-")
    if num == 0 and room_id == "":
        area = building = level = ""
        location_type = "bigArea"
    elif num == 0 and room_id != "":
        building = level = ""
        area = room_id
        location_type = "building"
    elif num == 1:
        area, building = room_id.split("-")
        level = ""
        location_type = "unit"
    elif num == 3:
        try:
            area, building = room_id.split("--")[0].split("-")
            level = room_id.split("--")[1]
            location_type = "room"
        except (IndexError, ValueError) as exc:
            logger.error("房间ID格式不正确: {}", room_id)
            raise InvalidArgumentError(
                f"房间ID格式不正确: {room_id}",
                context={"room_id": room_id},
            ) from exc
    else:
        logger.error("房间ID格式不合法: {}", room_id)
        raise InvalidArgumentError(
            f"房间ID格式不合法: {room_id}",
            context={"room_id": room_id},
        )

    headers = {"Authorization": self._require_access_token()}
    data = {
        "utilityType": "electric",
        "locationType": location_type,
        "bigArea": "",
        "area": area,
        "building": building,
        "unit": "",
        "level": level,
        "room": "",
        "subArea": "",
    }

    try:
        logger.debug("正在向 {} 发送请求获取房间列表...", self.LOCATION_URL)
        response = await self._client.post(
            self.LOCATION_URL,
            headers=headers,
            json=data,
        )
        response.raise_for_status()

        log_http_response_body(
            self.LOCATION_URL,
            response.text,
            content_type=response.headers.get("content-type"),
            level="DEBUG",
        )

        response_data = response.json()

        if "resultData" not in response_data:
            logger.error("响应中缺少 resultData")
            raise ParsingError("服务器响应格式不正确")

        location_list = response_data["resultData"].get("locationList", [])

        room_dict = {}
        for location in location_list:
            room_dict[location["id"]] = location["name"]

        logger.info("成功获取房间列表，共 {} 个房间", len(room_dict))
        return room_dict

    except httpx2.HTTPStatusError as exc:
        logger.error("获取房间列表请求返回失败状态码: {}", exc.response.status_code)
        raise NetworkError.from_http_status(
            exc,
            "服务器返回错误状态",
            context={"url": self.LOCATION_URL, "room_id": room_id},
        ) from exc
    except (json.JSONDecodeError, KeyError, TypeError) as exc:
        logger.error("从房间列表响应中提取数据失败: {}", exc)
        raise ParsingError.from_exception(
            exc,
            "服务器响应格式不正确",
            context={"url": self.LOCATION_URL, "room_id": room_id},
        ) from exc
    except httpx2.RequestError as exc:
        logger.error("获取房间列表网络请求失败: {}", exc)
        raise NetworkError.from_exception(
            exc,
            "网络连接异常",
            context={"url": self.LOCATION_URL, "room_id": room_id},
        ) from exc

@require_auth
async def get_remaining_energy(self, room: str | None = None) -> float:
    """获取剩余电量

    Args:
        room: 房间 ID 。格式应为 'areaid-buildingid--unitid-roomid'，可通过
            [`get_room_dict()`][zzupy.aio.app.ecard.ECardClient.get_room_dict] 获取

    Returns:
        剩余电量

    Raises:
        InvalidArgumentError: 如果房间参数不合法。
        ParsingError: 如果响应解析失败。
        NetworkError: 如果网络请求失败。
    """
    room = await self.get_default_room() if room is None else room
    logger.debug("正在获取房间 {} 的剩余电量", room)

    # 解析房间信息
    try:
        area, building = room.split("--")[0].split("-")
        level = room.split("--")[1].split("-")[0]
    except (IndexError, ValueError) as exc:
        logger.error("房间格式不正确: {}", room)
        raise InvalidArgumentError(
            f"房间格式不正确: {room}",
            context={"room": room},
        ) from exc

    headers = {"Authorization": self._require_access_token()}
    data = {
        "utilityType": "electric",
        "bigArea": "",
        "area": area,
        "building": building,
        "unit": "",
        "level": level,
        "room": room,
        "subArea": "",
    }

    try:
        logger.debug("正在向 {} 发送请求获取剩余电量...", self.ACCOUNT_URL)
        response = await self._client.post(
            self.ACCOUNT_URL,
            headers=headers,
            json=data,
        )
        response.raise_for_status()

        log_http_response_body(
            self.ACCOUNT_URL,
            response.text,
            content_type=response.headers.get("content-type"),
            level="DEBUG",
        )

        response_data = response.json()
        account_data = ECardAccountModel.model_validate(response_data)
        remaining_energy = account_data.remaining_energy

        if remaining_energy is None:
            raise ParsingError("服务器响应数据不完整，无法找到剩余电量 quantity")

        logger.info("房间 {} 剩余电量: {} 度", room, remaining_energy)
        return remaining_energy

    except httpx2.HTTPStatusError as exc:
        logger.error("获取剩余电量请求返回失败状态码: {}", exc.response.status_code)
        raise NetworkError.from_http_status(
            exc,
            "服务器返回错误状态",
            context={"url": self.ACCOUNT_URL, "room": room},
        ) from exc
    except (
        json.JSONDecodeError,
        KeyError,
        IndexError,
        TypeError,
        ValueError,
        ValidationError,
    ) as exc:
        logger.error("从剩余电量响应中提取数据失败: {}", exc)
        raise ParsingError.from_exception(
            exc,
            "服务器响应格式不正确",
            context={"url": self.ACCOUNT_URL, "room": room},
        ) from exc
    except httpx2.RequestError as exc:
        logger.error("获取剩余电量网络请求失败: {}", exc)
        raise NetworkError.from_exception(
            exc,
            "网络连接异常",
            context={"url": self.ACCOUNT_URL, "room": room},
        ) from exc

@require_auth
def logout(self) -> None:
    """登出账户，清除 Cookie 但保留连接池"""
    logger.debug("正在登出校园卡系统")
    self._cancel_token_refresh()
    self._access_token = None
    self._refresh_token = None
    self._tid = None
    self._client.cookies.clear()
    self._client.headers.clear()
    self._logged_in = False
    logger.info("已登出校园卡系统")

async def close(self) -> None:
    """清除 Cookie 和连接池"""
    logger.debug("正在关闭校园卡客户端")
    self._cancel_token_refresh()
    if self._logged_in:
        self.logout()
    await self._client.aclose()
    logger.info("校园卡客户端已关闭")