API 编写指南

# qqmusic_api/modules/foo.py
from ._base import ApiModule
from ..core.endpoint import cgi_endpoint, CgiRequestData


class FooApi(ApiModule):
    """Foo 相关 API."""

    @cgi_endpoint(
        key="foo.get_something",
        module="music.foo.Svc",
        method="GetSomething",
    )
    def get_something(self, id: int) -> CgiRequestData:
        """获取某项数据."""
        return CgiRequestData(param={"id": id})

# qqmusic_api/core/client.py
from functools import cached_property


class Client:
    @cached_property
    def foo(self) -> "FooApi":
        from ..modules.foo import FooApi

        return FooApi(self)

from ..core.endpoint import cgi_endpoint, CgiRequestData


@cgi_endpoint(
    key="song.get_detail",  # 端点唯一标识
    module="music.songDetail",  # 接口所属模块
    method="GetDetail",  # 方法名
)
def get_detail(self, song_id: int) -> CgiRequestData:
    """获取歌曲详情."""
    return CgiRequestData(
        param={"songid": song_id},  # 业务参数
    )

def get_detail(self, song_id: int):
    """获取歌曲详情."""
    return self._build_cgi(
        module="music.songDetail",  # 接口所属模块
        method="GetDetail",  # 方法名
        param={"songid": song_id},  # 业务参数
    )

from ..core.endpoint import http_endpoint, HttpRequestData


@http_endpoint(
    key="search.quick_search",
    method="GET",
    url="https://c.y.qq.com/splcloud/fcgi-bin/smartbox_new.fcg",
)
def quick_search(self, keyword: str) -> HttpRequestData:
    """快速搜索 (直接返回解析后的 JSON 数据)."""
    return HttpRequestData(
        params={"key": keyword},
    )

async def quick_search(self, keyword: str) -> dict[str, Any]:
    """快速搜索 (直接返回解析后的 JSON 数据)."""
    resp = await self._build_http(
        "GET",
        "https://c.y.qq.com/splcloud/fcgi-bin/smartbox_new.fcg",
        params={"key": keyword},
    )
    return resp["data"]

# GET 请求
req = self._build_http("GET", "https://example.com/api", params={"key": "value"})

# POST JSON
req = self._build_http("POST", "https://example.com/api", json={"key": "value"})

# 覆盖凭证
req = self._build_http("GET", "https://example.com/api", credential=my_credential)

# 返回 RawPayload 而非解析 JSON
req = self._build_http("GET", "https://example.com/api", raw=True)

from pydantic import Field

from .request import Response


class MyResponse(Response):
    """我的响应模型."""

    name: str
    count: int

class SonglistMeta(Response):
    """歌单元数据示例."""

    id: int = Field(json_schema_extra={"jsonpath": "$.result.tid"})
    dirid: int = Field(json_schema_extra={"jsonpath": "$.result.dirId"})
    name: str = Field(json_schema_extra={"jsonpath": "$.result.dirName"})

class CommentListResponse(Response):
    """评论列表响应."""

    comments: list[Comment] = Field(
        default_factory=list,
        json_schema_extra={"jsonpath": "$.commentlist[*]"},
    )

class Singer(Response):
    """歌手信息."""

    id: int = Field(
        default=-1,
        validation_alias=AliasChoices("id", "singerID", "singerId", "SingerID"),
    )
    mid: str = Field(
        default="",
        validation_alias=AliasChoices("mid", "singerMid", "singerMID"),
    )

@cgi_endpoint(
    key="user.get_vip_info",
    module="VipLogin.VipLoginInter",
    method="vip_login_base",
    response_model=UserVipInfoResponse,
    require_login=True,
)
def get_vip_info(self, *, credential: Credential | None = None) -> CgiRequestData:
    """获取 VIP 信息."""
    return CgiRequestData(credential=credential)

from ..core.pagination import OffsetStrategy


@cgi_endpoint(
    key="songlist.get_detail",
    module="music.srfDissInfo.DissInfo",
    method="CgiGetDiss",
    response_model=GetSonglistDetailResponse,
    item_type=Song,
)
def get_detail(self, songlist_id: int, num: int = 10, page: int = 1) -> CgiRequestData:
    """获取歌单详情."""
    return CgiRequestData(
        param={
            "disstid": songlist_id,
            "song_begin": num * (page - 1),
            "song_num": num,
        },
        pager_strategy=OffsetStrategy[GetSonglistDetailResponse](
            offset_key="song_begin",
            page_size_key="song_num",
            has_more_extractor=lambda response: bool(response.hasmore),
            total_extractor=lambda response: response.total,
            count_extractor=lambda response: len(response.songs),
        ),
        items_extractor=lambda response: response.songs,
    )

from ..core.pagination import BatchRefreshStrategy
from ..models.base import MV


@cgi_endpoint(
    key="song.get_related_mv",
    module="MvService.MvInfoProServer",
    method="GetSongRelatedMv",
    response_model=GetRelatedMvResponse,
    item_type=RelatedMv,
)
def get_related_mv(self, songid: int, last_mvid: str | None = None) -> CgiRequestData:
    """获取歌曲相关 MV."""
    return CgiRequestData(
        param={"songid": str(songid), "songtype": 1, "lastmvid": last_mvid or 0},
        pager_strategy=BatchRefreshStrategy[GetRelatedMvResponse](
            refresh_key="lastmvid",
            cursor_extractor=lambda response: response.mv[-1].id if response.mv else None,
            has_more_extractor=lambda response: bool(response.has_more),
        ),
        items_extractor=lambda response: response.mv,
    )

@cgi_endpoint(
    key="song.get_sheet",
    module="music.mir.SheetMusicSvr",
    method="GetMoreSheetMusic",
    sign=True,
)
def get_sheet(self, mid: str) -> CgiRequestData:
    """获取曲谱."""
    return CgiRequestData(
        param={"songMid": mid},
    )

# 合并到自动生成的 comm 中（默认行为）
CgiRequestData(
    ...,
    comm={"extra_key": "value"},
)

CgiRequestData(
    ...,
    comm={
        "g_tk": 5381,
        "uin": "",
        "format": "json",
        "inCharset": "utf-8",
        "outCharset": "utf-8",
        "notice": 0,
        "needNewCode": 1,
    },
    override_comm=True,
)

from ..core.exceptions import ApiDataError

try:
    ...
except KeyError as e:
    raise ApiDataError("无法解析歌曲信息") from e

"""歌曲模块测试."""

import pytest

from qqmusic_api import Client


async def test_query_song(client: Client) -> None:
    """测试根据 ID 查询歌曲."""
    result = await client.song.query_song([SongQueryInfo(mid="003w2xz20QlUZt")])
    assert result.tracks
    assert result.tracks[0].name

@pytest.mark.parametrize("page", [1, 2])
async def test_general_search(client: Client, page: int) -> None:
    """测试综合搜索翻页逻辑."""
    try:
        result = await client.search.general_search("周杰伦", page=page)
    except Exception as e:
        # 示例：优雅处理网络风控或限流 (需根据实际异常类型调整)
        if "limit" in str(e).lower() or "risk" in str(e).lower():
            pytest.skip(f"Triggered rate limit or risk control: {e}")
        raise

    assert result.song.items is not None

async def test_get_vip_info(authenticated_client: Client) -> None:
    """测试获取 VIP 信息."""
    result = await authenticated_client.user.get_vip_info()
    assert result.vip_flag is not None

async def test_search_paginate(client: Client) -> None:
    """测试搜索分页."""
    pager = client.search.search_by_type("周杰伦", num=5).pager(limit=2)

    assert pager.has_more() is True
    first_page = await pager.next()
    assert pager.has_more() is True
    second_page = await pager.next()

    assert first_page.song
    assert second_page.song