utils.network

网络请求

api_request

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

API请求

Source code in qqmusic_api/utils/network.py

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

    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,
            cacheable=cacheable,
            cache_ttl=cache_ttl,
            exclude_params=exclude_params,
            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

请求基类

Source code in 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
    self.cache = self.session._cache

session property

session: Session

获取请求会话

credential property writable

credential: Credential

获取请求凭证

common property

common: dict[str, Any]

公共参数

commom

commom(value: dict[str, Any])

设置公共参数

Source code in qqmusic_api/utils/network.py

@common.setter
def commom(self, value: dict[str, Any]):
    """设置公共参数"""
    self._common = value

build_request_data abstractmethod

build_request_data() -> dict[str, Any]

构建请求体数据

Source code in qqmusic_api/utils/network.py

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

build_request

build_request() -> dict[str, Any]

统一构建请求参数

Source code in 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

执行请求

Source code in 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, cache_ttl: int | None = None, cacheable: bool = True, exclude_params: list[str] = [], catch_error_code: list[int] = [])

Bases: BaseRequest, Generic[_P, _R]

API 请求处理器

Source code in 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,
    cache_ttl: int | None = None,
    cacheable: bool = True,
    exclude_params: list[str] = [],
    catch_error_code: list[int] = [],
) -> None:
    super().__init__(common, credential, verify, ignore_code)
    self.module = module
    self.method = method
    self.params = params or {}
    self.api_func = api_func
    self.proceduce_bool = process_bool
    self.processor: Callable[[dict[str, Any]], Any] = NO_PROCESSOR
    self.cacheable = cacheable
    self.cache_ttl = cache_ttl
    self.exclude_params = exclude_params
    self.catch_error_code = catch_error_code

copy

copy() -> ApiRequest[_P, _R]

创建当前 ApiRequest 实例的副本

Source code in qqmusic_api/utils/network.py

def copy(self) -> "ApiRequest[_P, _R]":
    """创建当前 ApiRequest 实例的副本"""
    req = ApiRequest[_P, _R](
        module=self.module,
        method=self.method,
        api_func=self.api_func,
        params=self.params.copy(),
        common=self._common.copy(),
        credential=self.credential,
        verify=self.verify,
        ignore_code=self.ignore_code,
        process_bool=self.proceduce_bool,
        cacheable=self.cacheable,
        cache_ttl=self.cache_ttl,
        exclude_params=self.exclude_params.copy(),
        catch_error_code=self.catch_error_code,
    )
    req.processor = self.processor
    return req

build_request_data

build_request_data() -> dict[str, Any]

构建请求体数据

Source code in 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: TypedDict

请求 Item

RequestGroup

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

Bases: BaseRequest

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

Source code in qqmusic_api/utils/network.py

def __init__(
    self,
    common: dict[str, Any] | None = None,
    credential: Credential | None = None,
    limit: int = 30,
):
    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

添加请求,自动生成唯一键

Source code in qqmusic_api/utils/network.py

def add_request(self, request: ApiRequest[_P, _R], *args: _P.args, **kwargs: _P.kwargs) -> None:
    """添加请求,自动生成唯一键"""
    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) - 1,
            key=unique_key,
            request=request,
            args=args,
            kwargs=kwargs,
            processor=request.processor,
        )
    )

build_request_data

build_request_data()

构建请求

Source code in 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]

执行合并请求

Source code in qqmusic_api/utils/network.py

async def execute(self) -> 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