client

async def close(self):
    """关闭客户端连接."""
    await self._session.close()

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

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

    Args:
        method: HTTP 方法.
        url: URL 地址.
        credential: 请求凭证.
        platform: 请求平台.
        lazy: 是否延迟发送请求.
        **kwargs: 其他参数.
    """
    cred = credential or self.credential
    user_cookies = kwargs.pop("cookies", {})
    cookies: dict[str, str] = {}
    if cred.musicid:
        cookies["uin"] = cred.str_musicid or str(cred.musicid)
        cookies["qqmusic_uin"] = cred.str_musicid or str(cred.musicid)
    if cred.musickey:
        cookies["qm_keyst"] = cred.musickey
        cookies["qqmusic_key"] = cred.musickey
    cookies.update(user_cookies)
    if cookies:
        kwargs["cookies"] = cookies

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

    try:
        resp = await self._session.request(
            method,
            url,
            **kwargs,
            proxies=self.proxies,
            hooks=self.hooks,
            cert=self.cert,
            verify=self.verify,
        )
        if not lazy:
            await self._session.gather(resp)
        return resp
    except RequestException as exc:
        raise NetworkError(str(exc)) from exc

async def request_api(
    self,
    data: list[RequestItem],
    comm: dict[str, Any] | None = None,
    credential: Credential | None = None,
    platform: Platform | None = None,
    *,
    override_comm: bool = False,
    is_jce: bool = False,
    lazy: bool = False,
    sign: bool = False,
) -> Response:
    """发送 API 请求."""
    target_platform = Platform.ANDROID if is_jce else platform or self.platform
    if target_platform == Platform.ANDROID:
        await self._ensure_session()
    device = await self._device_store.get_device()
    if override_comm:
        finalcomm = (comm or {}).copy()
    else:
        finalcomm = self._version_policy.build_comm(
            platform=target_platform,
            credential=credential or self.credential,
            device=device,
            qimei=cast("dict[str, str]", await self._qimei_manager.get_cached())
            if target_platform == Platform.ANDROID
            else None,
            guid=device.open_udid,
        )
        if comm:
            finalcomm.update(comm)

    user_agent = await self._get_user_agent(target_platform)

    try:
        if is_jce:
            for k, v in finalcomm.items():
                if not isinstance(v, str):
                    finalcomm[k] = str(v)
            content = JceRequest(
                finalcomm,
                {
                    f"req_{idx}": JceRequestItem(
                        module=req["module"],
                        method=req["method"],
                        param=TarsDict(cast("dict[int, Any]", req["param"])),
                    )
                    for idx, req in enumerate(data)
                },
            ).encode()
            resp = await self._session.post(
                "http://u.y.qq.com/cgi-bin/musicw.fcg",
                data=content,
                headers={"User-Agent": user_agent},
                proxies=self.proxies,
                hooks=self.hooks,
                cert=self.cert,
                verify=self.verify,
            )
            if not lazy:
                await self._session.gather(resp)
            return resp

        payload: dict[str, Any] = {
            "comm": finalcomm,
        }
        params: dict[str, str] = {}
        for idx, req in enumerate(data):
            payload[f"req_{idx}"] = {
                "module": req["module"],
                "method": req["method"],
                "param": req["param"] if req["preserve_bool"] else bool_to_int(req["param"]),
            }

        if sign:
            params["_"] = str(int(time.time() * 1000))
            params["sign"] = zzc_sign(json.dumps(payload))

        resp = await self._session.post(
            "https://u.y.qq.com/cgi-bin/musicu.fcg" if not sign else "https://u.y.qq.com/cgi-bin/musics.fcg",
            json=payload,
            params=params,
            headers={"User-Agent": user_agent},
            proxies=self.proxies,
            hooks=self.hooks,
            cert=self.cert,
            verify=self.verify,
        )
        if not lazy:
            await self._session.gather(resp)

        return resp
    except RequestException as exc:
        raise NetworkError(str(exc)) from exc

async def gather(
    self,
    requests: list[Request[Any]],
    *,
    batch_size: int = 20,
    return_exceptions: bool = False,
) -> list[Any]:
    """并发执行多个请求描述符并按输入顺序返回解析结果.

    可合并的请求会按协议、平台、公共参数和凭证分组, 每组按
    `batch_size` 拆分为批量请求发送。响应解析失败时, 默认抛出
    第一个异常; 当 `return_exceptions` 为 True 时, 异常会作为对应
    位置的结果返回。

    Args:
        requests: 待执行的请求描述符列表.
        batch_size: 每个批量请求包含的最大请求数.
        return_exceptions: 是否将单项解析异常作为结果返回.

    Returns:
        与 `requests` 顺序一致的解析结果列表.

    Raises:
        ValueError: 当 `batch_size` 小于等于 0, 响应为空, 响应缺少对应
            请求项, 或结果未能完整回填时抛出.
    """
    if batch_size <= 0:
        raise ValueError("batch_size 必须大于 0")

    if not requests:
        return []

    grouped_indices: dict[Any, list[int]] = defaultdict(list)
    for index, request in enumerate(requests):
        grouped_indices[request._group_key].append(index)

    batch_responses: list[tuple[list[int], Response]] = []

    for indices in grouped_indices.values():
        base_req = requests[indices[0]]

        for start in range(0, len(indices), batch_size):
            batch_indices = indices[start : start + batch_size]
            response_task = await self.request_api(
                data=[
                    {
                        "module": requests[i].module,
                        "method": requests[i].method,
                        "param": requests[i].param,
                        "preserve_bool": requests[i].preserve_bool,
                    }
                    for i in batch_indices
                ],
                comm=base_req.comm,
                override_comm=base_req.override_comm,
                credential=base_req.credential,
                platform=base_req.platform,
                is_jce=base_req.is_jce,
                lazy=True,
                sign=base_req.sign,
            )
            batch_responses.append((batch_indices, response_task))

    try:
        await self._session.gather(*(resp for _, resp in batch_responses))
    except RequestException as exc:
        raise NetworkError(str(exc)) from exc

    results: list[Any] = [_SENTINEL] * len(requests)

    for batch_indices, response in batch_responses:
        data = self._vaildate_resp(response, is_jce=requests[batch_indices[0]].is_jce)
        for batch_index, req_index in enumerate(batch_indices):
            request = requests[req_index]
            try:
                results[req_index] = self._parse_cgi_item(
                    data[f"req_{batch_index}"],
                    request,
                )
            except Exception as exc:
                if return_exceptions:
                    results[req_index] = exc
                else:
                    raise

    missing_indexes = [i for i, res in enumerate(results) if res is _SENTINEL]
    if missing_indexes:
        raise ApiDataError(f"缺少以下索引结果: {missing_indexes}")

    return results

async def execute(self, request: Request[RequestResultT]) -> RequestResultT:
    """执行单个请求描述符并解析响应结果.

    Args:
        request: 待执行的请求描述符.

    Returns:
        解析后的响应数据或响应模型.

    Raises:
        ValueError: 当响应为空、业务返回码非 0 或响应缺少 `req_0` 时抛出.
    """
    resp = await self.request_api(
        data=[
            {
                "module": request.module,
                "method": request.method,
                "param": request.param,
                "preserve_bool": request.preserve_bool,
            }
        ],
        comm=request.comm,
        override_comm=request.override_comm,
        credential=request.credential,
        platform=request.platform,
        is_jce=request.is_jce,
        sign=request.sign,
    )
    return self._parse_cgi_item(self._vaildate_resp(resp, is_jce=request.is_jce)["req_0"], request)