client

async def fetch(self, method: str, url: str, **kwargs: Any) -> httpx.Response:
    """发送底层 HTTP 请求.

    该方法提供并发控制、网络波动自动重试及网络异常转换.

    Args:
        method: HTTP 方法, 如 "GET" 或 "POST".
        url: 请求的 URL 地址.
        **kwargs: 传递给 httpx.AsyncClient.request 的附加参数.

    Returns:
        HTTP 响应对象.

    Raises:
        NetworkError: 网络请求在重试耗尽后仍然失败.
    """
    logger.debug("HTTP 请求开始: %s %s", method, url)

    await self._limiter.acquire()
    try:
        resp = await self._session.request(method, url, **kwargs)
        logger.debug("HTTP 请求完成: %s %s -> %s", method, url, resp.status_code)
        return resp
    except httpx.RequestError as exc:
        logger.debug("HTTP 请求重试耗尽: %s %s, error=%s", method, url, exc)
        raise NetworkError(f"Network error: {exc}", original_exc=exc) from exc
    finally:
        self._limiter.release()

def request_group(self, batch_size: int = 20, max_inflight_batches: int = 5) -> "RequestGroup":
    """创建并返回一个批量请求 (RequestGroup) 容器.

    适用于需合并多个相同协议 (JSON 或 JCE) 请求的场景.

    Args:
        batch_size: 单个批次的最大请求数量.
        max_inflight_batches: 允许同时发送的最多批次数量.

    Returns:
        批量请求对象.
    """
    from .request import RequestGroup

    return RequestGroup(self, batch_size=batch_size, max_inflight_batches=max_inflight_batches)

async def execute(self, request: "Request") -> Any:
    """执行单个请求描述符并解析返回结果.

    调用中间件进行请求预处理, 随后根据请求格式 (JCE/JSON) 分发调用底层发包方法,
    解析响应后自动组装成预期的 `response_model` 类型.

    Args:
        request: 请求描述符对象.

    Returns:
        解析后对应的响应对象模型.

    Raises:
        ApiError: 接口返回状态码异常或缺少预期字段.
    """
    data: RequestItem = {
        "module": request.module,
        "method": request.method,
        "param": request.param,
    }
    if request.is_jce:
        response = await self.request_jce(
            data=data,
            comm=request.comm,
            credential=request.credential,
        )
        item = response.data.get("req_0")
        if item is None:
            raise ApiError("缺少响应字段: req_0", code=-1, data=response)
        if item.code != 0:
            code, subcode = _extract_api_error_code(item)
            logger.debug(
                "JCE 请求返回错误: module=%s method=%s code=%s subcode=%s",
                request.module,
                request.method,
                code,
                subcode,
            )
            raise _build_api_error(
                code=code,
                subcode=subcode,
                data=item.data,
                context={"module": request.module, "method": request.method, "is_jce": True},
            )
        if item.data is None:
            raise ApiError("缺少响应数据: req_0.data", code=-1, data=item)
        if request.response_model is None:
            return item.data
        try:
            return self._build_result(item.data, request.response_model)
        except Exception as exc:
            raise ApiError("响应数据校验失败", code=-1, data=item.data, cause=exc) from exc

    response = await self.request_musicu(
        data=data,
        comm=request.comm,
        platform=request.platform,
        credential=request.credential,
        preserve_bool=request.preserve_bool,
    )
    item = response.get("req_0")
    if item is None:
        raise ApiError("缺少响应字段: req_0", code=-1, data=response)
    code, subcode = _extract_api_error_code(item)
    if code is not None and code != 0:
        logger.debug(
            "JSON 请求返回错误: module=%s method=%s code=%s subcode=%s",
            request.module,
            request.method,
            code,
            subcode,
        )
        raise _build_api_error(
            code=code,
            subcode=subcode,
            data=item.get("data"),
            context={"module": request.module, "method": request.method, "is_jce": False},
        )
    response_model = request.response_model
    raw = item.get("data", {})
    if not raw:
        raise ApiDataError("缺少响应数据: req_0.data", data=item)

    # dump_path = anyio.Path(f"responses/{request.module}_{request.method}.json")
    # await dump_path.parent.mkdir(parents=True, exist_ok=True)
    # await dump_path.write_text(json.dumps(raw).decode("utf-8"))
    if response_model is None:
        return raw
    try:
        return self._build_result(raw, response_model)
    except Exception as exc:
        raise ApiDataError("响应数据校验失败", data=raw) from exc

async def close(self) -> None:
    """关闭底层会话."""
    await self._session.aclose()

async def request(
    self,
    method: str,
    url: str,
    credential: Credential | None = None,
    platform: Platform | None = None,
    **kwargs: Any,
) -> httpx.Response:
    """发送带有凭证和 User-Agent 的 HTTP 请求.

    自动装配指定的客户端平台 User-Agent 及对应凭证的 Cookies.

    Args:
        method: HTTP 方法, 如 "GET" 或 "POST".
        url: 请求的 URL 地址.
        credential: 覆盖默认凭证, 可选.
        platform: 覆盖默认平台, 可选.
        **kwargs: 传递给 httpx 的其他参数.

    Returns:
        HTTP 响应对象.
    """
    auth_cookies = self._get_cookies(credential)
    if "cookies" in kwargs:
        auth_cookies.update(kwargs["cookies"])
    if auth_cookies:
        kwargs["cookies"] = auth_cookies

    headers = kwargs.get("headers", {})
    if "User-Agent" not in headers:
        headers["User-Agent"] = await self._get_user_agent(platform)
    kwargs["headers"] = headers

    logger.debug("发送请求: %s %s", method, url)
    return await self.fetch(method, url, **kwargs)

async def request_musicu(
    self,
    data: RequestItem | list[RequestItem],
    comm: dict[str, Any] | None = None,
    credential: Credential | None = None,
    url: str = "https://u.y.qq.com/cgi-bin/musicu.fcg",
    platform: Platform | None = None,
    *,
    preserve_bool: bool = False,
) -> dict[str, Any]:
    """发送标准 QQ 音乐请求 (Musicu/JSON) 并解析响应.

    Args:
        data: 请求项, 支持单个或批量.
        comm: 请求公共参数.
        credential: 请求凭证 (该方法底层未直接使用凭证参数, 供扩展).
        url: 请求的网关 URL, 默认为 musicu.fcg.
        platform: 请求发起的平台名称.
        preserve_bool: 是否保留 JSON 参数中的布尔字面量.

    Returns:
        解析后的 JSON 响应字典.

    Raises:
        HTTPError: HTTP 状态码不是 200.
        ApiError: JSON 解析错误或缺少关键字段.
    """
    requests = data if isinstance(data, list) else [data]
    logger.debug(
        "构建 JSON 批量请求: count=%s platform=%s preserve_bool=%s",
        len(requests),
        platform or self.platform,
        preserve_bool,
    )

    payload: dict[str, Any] = {
        "comm": await self._build_common_params(platform, credential or self.credential, comm),
    }
    for idx, req in enumerate(requests):
        payload[f"req_{idx}"] = {
            "module": req["module"],
            "method": req["method"],
            "param": req["param"] if preserve_bool else bool_to_int(req["param"]),
        }

    params: dict[str, Any] = {}

    if self.enable_sign:
        from ..algorithms.sign import sign_request

        if signature := sign_request(payload):
            params["sign"] = signature

    resp = await self.fetch(
        "POST",
        url,
        json=payload,
        params=params,
        headers={
            "Content-Type": "application/json",
            "User-Agent": await self._get_user_agent(Platform.ANDROID),
        },
    )

    if resp.status_code != 200:
        raise HTTPError(f"请求失败: {resp.text[:500]}", status_code=resp.status_code)

    try:
        return json.loads(resp.content)
    except Exception as exc:
        raise ApiError(f"JSON 解析失败: {exc!s}", code=-1, data=resp.text[:500], cause=exc) from exc

async def request_jce(
    self,
    data: RequestItem | list[RequestItem],
    credential: Credential | None = None,
    comm: dict[str, Any] | None = None,
    url: str = "http://u.y.qq.com/cgi-bin/musicw.fcg",
) -> JceResponse:
    """发送 Android 语义的 JCE 格式请求并解析响应.

    Args:
        data: JCE 请求项, 支持单个或批量.
        comm: 请求公共参数.
        credential: 请求凭证.
        url: JCE 网关 URL.

    Returns:
        解析后的 JCE 响应对象.

    Raises:
        HTTPError: HTTP 状态码不是 200.
        ApiError: JCE 解析失败.
    """
    requests = data if isinstance(data, list) else [data]
    logger.debug("构建 JCE 批量请求: count=%s", len(requests))

    def _ensure_jce_param(p: dict[str, Any] | dict[int, Any]) -> dict[int, Any]:
        if not all(isinstance(k, int) for k in p):
            raise TypeError("JCE param 必须是 dict[int, Any]")
        return {key: value for key, value in p.items() if isinstance(key, int)}

    payload = JceRequest(
        {
            k: str(v)
            for k, v in (
                await self._build_common_params(Platform.ANDROID, credential or self.credential, comm)
            ).items()
        },
        {
            f"req_{idx}": JceRequestItem(
                module=req["module"],
                method=req["method"],
                param=TarsDict(_ensure_jce_param(req["param"])),
            )
            for idx, req in enumerate(requests)
        },
    ).encode()

    headers = {
        "Content-Type": "application/x-www-form-urlencoded",
        "User-Agent": await self._get_user_agent(Platform.ANDROID),
        "x-sign-data-type": "jce",
    }

    resp = await self.fetch("POST", url, content=payload, headers=headers)

    if resp.status_code != 200:
        raise HTTPError(f"请求失败: {resp.text[:500]}", status_code=resp.status_code)

    try:
        return JceResponse.decode(resp.content)
    except Exception as exc:
        data_preview = resp.text[:500] if isinstance(resp.text, str) else str(resp.content[:500])
        raise ApiError(f"JCE 响应解析失败: {exc!s}", code=-1, data=data_preview, cause=exc) from exc