utils.network

网络请求

api_request

api_request(module: str, method: str, *, verify: bool = False, ignore_code: bool = False, process_bool: bool = True, catch_error_code: list[int] | None = None)

API请求装饰器.

用于将普通函数转换为 ApiRequest 实例的工厂函数.

参数	描述
`module`	模块名称 (e.g. `music.trackInfo.UniformRuleCtrl`). 类型： `str`
`method`	调用方法名称 (e.g. `CgiGetTrackInfo`). 类型： `str`
`verify`	是否验证凭证有效性.如果为 `True`,请求前会检查凭证是否过期,过期则抛出异常. 类型： `bool` 默认： `False`
`ignore_code`	是否忽略业务状态码检查.如果为 `True`,将跳过 `code != 0` 的验证. 类型： `bool` 默认： `False`
`process_bool`	是否转换布尔值.如果为 `True`,参数中的 `bool` 值会自动转换为 `int` (`0`/`1`). 类型： `bool` 默认： `True`
`catch_error_code`	视为成功的错误码列表.当响应 `code` 在此列表中时,不会抛出异常. 类型： `list[int] \| None` 默认： `None`

返回	描述
	一个装饰器,将函数转换为返回 `ApiRequest` 的可调用对象.

源代码位于： qqmusic_api/utils/network.py

def api_request(
    module: str,
    method: str,
    *,
    verify: bool = False,
    ignore_code: bool = False,
    process_bool: bool = True,
    catch_error_code: list[int] | None = None,
):
    """API请求装饰器.

    用于将普通函数转换为 `ApiRequest` 实例的工厂函数.

    Args:
        module: 模块名称 (e.g. `music.trackInfo.UniformRuleCtrl`).
        method: 调用方法名称 (e.g. `CgiGetTrackInfo`).
        verify: 是否验证凭证有效性.如果为 `True`,请求前会检查凭证是否过期,过期则抛出异常.
        ignore_code: 是否忽略业务状态码检查.如果为 `True`,将跳过 `code != 0` 的验证.
        process_bool: 是否转换布尔值.如果为 `True`,参数中的 `bool` 值会自动转换为 `int` (`0`/`1`).
        catch_error_code: 视为成功的错误码列表.当响应 `code` 在此列表中时,不会抛出异常.

    Returns:
        一个装饰器,将函数转换为返回 `ApiRequest` 的可调用对象.
    """

    def decorator(
        api_func: Callable[_P, Coroutine[None, None, tuple[dict[Any, Any], Callable[[dict[str, Any]], _R]]]],
    ):
        return ApiRequest[_P, _R](
            module,
            method,
            api_func=api_func,
            verify=verify,
            ignore_code=ignore_code,
            process_bool=process_bool,
            catch_error_code=catch_error_code,
        )

    return decorator

BaseRequest

BaseRequest(common: dict[str, Any] | None = None, credential: Credential | None = None, verify: bool = False, ignore_code: bool = False)

Bases: ABC

请求基类

源代码位于： qqmusic_api/utils/network.py

def __init__(
    self,
    common: dict[str, Any] | None = None,
    credential: Credential | None = None,
    verify: bool = False,
    ignore_code: bool = False,
) -> None:
    self._common = common or {}
    self._credential = credential
    self.verify = verify
    self.ignore_code = ignore_code

session `property`

session: Session

获取请求会话

credential `property` `writable`

credential: Credential

获取请求凭证

common `property` `writable`

common: dict[str, Any]

公共参数

build_cookies `staticmethod`

build_cookies(credential: Credential) -> Cookies | None

根据凭证生成 cookies

源代码位于： qqmusic_api/utils/network.py

@staticmethod
def build_cookies(credential: Credential) -> httpx.Cookies | None:
    """根据凭证生成 cookies"""
    if credential.has_musicid() and credential.has_musickey():
        cookies = httpx.Cookies()
        cookies.set("uin", str(credential.musicid), domain=".qq.com")
        cookies.set("qqmusic_key", credential.musickey, domain=".qq.com")
        cookies.set("qm_keyst", credential.musickey, domain=".qq.com")
        cookies.set("tmeLoginType", str(credential.login_type), domain=".qq.com")
        return cookies
    return None

build_request_data `abstractmethod`

build_request_data() -> dict[str, Any]

构建请求体数据

源代码位于： qqmusic_api/utils/network.py

@abstractmethod
def build_request_data(self) -> dict[str, Any]:
    """构建请求体数据"""
    pass

build_request

build_request() -> dict[str, Any]

统一构建请求参数

源代码位于： qqmusic_api/utils/network.py

def build_request(self) -> dict[str, Any]:
    """统一构建请求参数"""
    data = self.build_request_data()
    config = self.session.api_config
    request_params = {
        "url": config["enc_endpoint" if config["enable_sign"] else "endpoint"],
        "json": data,
    }
    if config["enable_sign"]:
        request_params["params"] = {"sign": sign(data)}
    return request_params

request `async`

request() -> Response

执行请求

源代码位于： qqmusic_api/utils/network.py

async def request(self) -> httpx.Response:
    """执行请求"""
    if self.verify:
        self.credential.raise_for_invalid()
    request_data = self.build_request()
    logger.debug(f"发送请求: {request_data}")
    self._set_cookies(self.credential)
    resp = await self.session.post(**request_data)
    if not self.ignore_code:
        resp.raise_for_status()
    return resp

ApiRequest

ApiRequest(module: str, method: str, api_func: Callable[_P, Coroutine[None, None, tuple[dict[Any, Any], Callable[[dict[str, Any]], _R]]]] | None = None, *, params: dict[str, Any] | None = None, common: dict[str, Any] | None = None, credential: Credential | None = None, verify: bool = False, ignore_code: bool = False, process_bool: bool = True, catch_error_code: list[int] | None = None)

Bases: BaseRequest, Generic[_P, _R]

API 请求处理器.

封装单个 API 请求的构建、发送和处理逻辑.

初始化 ApiRequest.

参数	描述
`module`	模块名. 类型： `str`
`method`	方法名. 类型： `str`
`api_func`	被装饰的原始 API 函数类型： `Callable[_P, Coroutine[None, None, tuple[dict[Any, Any], Callable[[dict[str, Any]], _R]]]] \| None` 默认： `None`
`params`	请求参数字典. 类型： `dict[str, Any] \| None` 默认： `None`
`common`	公共参数字典. 类型： `dict[str, Any] \| None` 默认： `None`
`credential`	请求凭证. 类型： `Credential \| None` 默认： `None`
`verify`	是否验证凭证. 类型： `bool` 默认： `False`
`ignore_code`	是否忽略错误码. 类型： `bool` 默认： `False`
`process_bool`	是否处理布尔值. 类型： `bool` 默认： `True`
`catch_error_code`	捕获的错误码列表. 类型： `list[int] \| None` 默认： `None`

源代码位于： qqmusic_api/utils/network.py

def __init__(
    self,
    module: str,
    method: str,
    api_func: Callable[_P, Coroutine[None, None, tuple[dict[Any, Any], Callable[[dict[str, Any]], _R]]]]
    | None = None,
    *,
    params: dict[str, Any] | None = None,
    common: dict[str, Any] | None = None,
    credential: Credential | None = None,
    verify: bool = False,
    ignore_code: bool = False,
    process_bool: bool = True,
    catch_error_code: list[int] | None = None,
) -> None:
    """初始化 ApiRequest.

    Args:
        module: 模块名.
        method: 方法名.
        api_func: 被装饰的原始 API 函数
        params: 请求参数字典.
        common: 公共参数字典.
        credential: 请求凭证.
        verify: 是否验证凭证.
        ignore_code: 是否忽略错误码.
        process_bool: 是否处理布尔值.
        catch_error_code: 捕获的错误码列表.
    """
    super().__init__(common, credential, verify, ignore_code)
    self.module = module
    self.method = method
    self.params = params or {}
    self.api_func = api_func
    self.process_bool = process_bool
    self.processor: Callable[[dict[str, Any]], Any] = NO_PROCESSOR
    self.catch_error_code = catch_error_code or []

build_request_data

build_request_data() -> dict[str, Any]

构建请求体数据

源代码位于： qqmusic_api/utils/network.py

@override
def build_request_data(self) -> dict[str, Any]:
    return {"comm": self.common, f"{self.module}.{self.method}": self.data}

data `property`

data: dict[str, Any]

API 请求数据

RequestItem

Bases: Generic[TReq, TProc], TypedDict

请求 Item (支持泛型)

RequestGroup

RequestGroup(common: dict[str, Any] | None = None, credential: Credential | None = None, limit: int = 20)

Bases: BaseRequest

合并多个 API 请求,支持组级公共参数和重复模块方法处理

初始化 RequestGroup.

参数	描述
`common`	组级公共参数,将合并到每个子请求中. 类型： `dict[str, Any] \| None` 默认： `None`
`credential`	组级凭证,将应用于所有子请求. 类型： `Credential \| None` 默认： `None`
`limit`	单次请求的最大子请求数量.超过此数量会自动分批发送. 类型： `int` 默认： `20`

源代码位于： qqmusic_api/utils/network.py

def __init__(
    self,
    common: dict[str, Any] | None = None,
    credential: Credential | None = None,
    limit: int = 20,
):
    """初始化 RequestGroup.

    Args:
        common: 组级公共参数,将合并到每个子请求中.
        credential: 组级凭证,将应用于所有子请求.
        limit: 单次请求的最大子请求数量.超过此数量会自动分批发送.
    """
    super().__init__(common, credential)
    self._requests: list[RequestItem] = []
    self.limit = limit
    self._key_counter = defaultdict(int)
    self._results = []

add_request

add_request(request: ApiRequest[_P, _R], *args: args, **kwargs: kwargs) -> None

添加请求到组.

参数	描述
`request`	`ApiRequest` 实例或被 `@api_request` 装饰的函数. 类型： `ApiRequest[_P, _R]`
`*args`	传递给 API 函数的位置参数. 类型： `args` 默认： `()`
`**kwargs`	传递给 API 函数的关键字参数. 类型： `kwargs` 默认： `{}`

源代码位于： qqmusic_api/utils/network.py

def add_request(self, request: ApiRequest[_P, _R], *args: _P.args, **kwargs: _P.kwargs) -> None:
    """添加请求到组.

    Args:
        request: `ApiRequest` 实例或被 `@api_request` 装饰的函数.
        *args: 传递给 API 函数的位置参数.
        **kwargs: 传递给 API 函数的关键字参数.
    """
    base_key = f"{request.module}.{request.method}"
    self._key_counter[base_key] += 1
    count = self._key_counter[base_key]
    unique_key = f"{base_key}.{count}" if count > 1 else base_key
    request = request.copy()
    request.credential = self.credential

    self._requests.append(
        RequestItem(
            id=len(self._requests),
            key=unique_key,
            request=request,
            args=args,
            kwargs=kwargs,
            processor=request.processor,
        )
    )

build_request_data

build_request_data()

构建请求

源代码位于： qqmusic_api/utils/network.py

@override
def build_request_data(self):
    """构建请求"""
    merged_data = {req["key"]: req["request"].data for req in self._requests}
    data = {"comm": self.common}
    data.update(merged_data)
    return data

execute `async`

execute() -> list[Any]

执行合并请求。

如果请求数量超过 limit,会自动分批执行并合并结果。

返回	描述
`list[Any]`	list[Any]: 请求结果列表

源代码位于： qqmusic_api/utils/network.py

async def execute(self) -> list[Any]:
    """执行合并请求。

    如果请求数量超过 `limit`,会自动分批执行并合并结果。

    Returns:
        list[Any]: 请求结果列表
    """
    if not self._requests:
        return []

    # 未设置 limit 或请求数未超过 limit 时直接处理
    if self.limit <= 0 or len(self._requests) <= self.limit:
        return await self._execute()

    # 分批次处理
    batches = [self._requests[i : i + self.limit] for i in range(0, len(self._requests), self.limit)]
    all_results = []

    for batch in batches:
        # 创建新 RequestGroup 处理当前批次
        batch_group = RequestGroup(
            common=self.common.copy(),
            credential=self.credential,
        )

        # 添加当前批次的请求
        for req_item in batch:
            batch_group.add_request(req_item["request"], *req_item["args"], **req_item["kwargs"])

        # 执行并收集结果
        batch_results = await batch_group._execute()
        all_results.extend(batch_results)

    return all_results

utils.network

api_request

BaseRequest

session property

credential property writable

common property writable

build_cookies staticmethod

build_request_data abstractmethod

build_request

request async

ApiRequest

build_request_data

data property

RequestItem

RequestGroup

add_request

build_request_data

execute async

session `property`

credential `property` `writable`

common `property` `writable`

build_cookies `staticmethod`

build_request_data `abstractmethod`

request `async`

data `property`

execute `async`