皮皮虾热榜监测 API 教程

监测皮皮虾上什么在热,归根到底是三个动作:读热榜、给每个条目挂一个稳定 id、跨运行 diff 排名,这样你就能抓住正在上升的。这篇教程用 SandBase 皮皮虾 API 把这些动作串成一套工作流——一个 SandBase 密钥,不需要登录皮皮虾,也不需要 SDK。这些是同步读取,所以”监测”指的是按计划轮询。

完整的端点全景见皮皮虾公开数据 API 总览。这一篇是落地的监测工作流。

先说结论

一个端点驱动全程:pipixia/app/hot-search-board-list 返回榜单及其排名条目。

每次都是 POST /v1/api/pipixia/<path>,一个 SANDBASE_API_KEY;completed 运行带 outputs,failed/timeout 运行带 error。

皮皮虾把结果包在一个上游 { status_code, data, message } 对象里——检查 status_code == 0,再读 data.boards[].board_items。

给每个条目挂它的 item_id_str,这样你就能跨运行 diff 排名和互动。仅公开、只读数据;仍需要一个 SandBase API 密钥。

工作流全貌

步骤	做什么	怎么做
1. 读榜	`pipixia/app/hot-search-board-list`	检查 `status_code`,读 `data.boards`
2. 归一化	(在你代码里)	把 `board_items` 拍平成 `{ item_id, rank, metric }`
3. Diff	(在你代码里)	按 id → rank 和上一次运行比较

SandBase 皮皮虾端点参考,展示本工作流用到的 hot-search-board-list 端点 端点 API 参考是每个参数名和响应路径的事实来源。

第 1 步 —— 读热榜

先写一个判 status 的辅助函数,再解包上游 status_code:

import os
import requests

BASE = "https://api.sandbase.ai/v1/api/pipixia"
HEADERS = {
    "Authorization": f"Bearer {os.environ['SANDBASE_API_KEY']}",
    "Content-Type": "application/json",
}


def call(path: str, payload: dict) -> dict:
    resp = requests.post(f"{BASE}/{path}", headers=HEADERS, json=payload, timeout=90)
    resp.raise_for_status()
    body = resp.json()
    if body.get("status") != "completed":
        raise RuntimeError(body.get("error", {}).get("message", "请求未完成"))
    payload = body["outputs"][0]["data"]
    # 皮皮虾用一个上游 status_code 包裹结果;先检查它,
    # 再防御式读取,并以真实响应核对路径。
    if payload.get("status_code") != 0:
        raise RuntimeError(payload.get("message", "上游错误"))
    return payload.get("data", {})


data = call("app/hot-search-board-list", {})
boards = data.get("boards", [])
print(len(boards), "个榜")

下面是我测试于 2026-09-27(UTC)的一段裁剪后的真实响应——一个榜的 board_items 各自嵌套一个内层 item,而度量字段取决于榜(在”今日神评点赞”榜上是点赞数,在”今日播放”榜上是播放数)。数值会随时间变化,所以对每个业务字段都用 .get() 读取:

{
  "id": "139c58b8-a1e1-49ac-8e16-16a3b5254f34",
  "status": "completed",
  "model": "pipixia/app/hot-search-board-list",
  "outputs": [
    {
      "data": {
        "status_code": 0,
        "data": {
          "boards": [
            {
              "block_type": 12,
              "board_items": [
                { "item_info": "今日神评点赞", "today_digg_num": "4375", "item": { "item_id_str": "…" } }
              ]
            }
          ]
        }
      }
    }
  ]
}

SandBase 皮皮虾 hot-search-board-list API 参考,展示响应结构 hot-search-board-list 返回榜单,每个榜在上游 data 对象下带排名的 board_items。

第 2 步 —— 按 id 归一化条目

每个榜的条目嵌套一个内层 item,带一个稳定的 item_id_str。把它们拍平成一个统一行,这样排名和 diff 就不用管条目来自哪个榜:

def normalize(boards: list[dict]) -> list[dict]:
    rows = []
    for board in boards:
        board_type = board.get("block_type")
        for rank, entry in enumerate(board.get("board_items", [])):
            inner = entry.get("item", {}) if isinstance(entry, dict) else {}
            rows.append({
                "item_id": inner.get("item_id_str"),
                "board_type": board_type,
                "rank": rank,
                # 度量字段因榜而异;两个都防御式保留
                "digg": entry.get("today_digg_num"),
                "plays": entry.get("today_show_num"),
                "content": inner.get("content"),
            })
    return rows


rows = normalize(boards)
for r in rows[:5]:
    print(r["board_type"], r["rank"], r["item_id"], r["digg"] or r["plays"])

因为度量字段因榜而异——一个是点赞数、另一个是播放数——这个行两个都保留、各自防御式读取。在依赖它们之前,先以一份真实响应核对确切字段名。

SandBase 皮皮虾端点列表,展示热榜端点及其路径 读每个端点的结构;度量字段因榜类型而异。

第 3 步 —— 跨运行 diff 排名

保留上一次运行的 item_id → rank 映射。下一次轮询,比较出新条目和排名变动:

def diff(prev: dict[str, int], rows: list[dict]) -> dict:
    current = {r["item_id"]: r["rank"] for r in rows if r.get("item_id")}
    new_items = [i for i in current if i not in prev]
    movers = [
        {"item_id": i, "from": prev[i], "to": current[i]}
        for i in current
        if i in prev and current[i] != prev[i]
    ]
    return {"new": new_items, "movers": movers, "snapshot": current}


prev = {}  # 从你的存储加载
result = diff(prev, rows)
print(len(result["new"]), "新增,", len(result["movers"]), "变动")
prev = result["snapshot"]  # 存下来供下次运行用

因为一切都挂在稳定的 item_id_str 上,一个在多次轮询之间都留在榜上的条目会被识别为同一个——于是你报告的是排名变动,而不是重复。按计划跑,每一次都只浮现发生变化的部分。

把它串起来

一次最小的监测过程长这样——读、归一化、diff、持久化:

prev = {}  # 从你的存储加载

def monitor_once():
    global prev
    data = call("app/hot-search-board-list", {})
    rows = normalize(data.get("boards", []))
    result = diff(prev, rows)
    prev = result["snapshot"]  # 存下来供下次
    return result

因为每次调用共用同一个信封和同一个 call 辅助函数(带它的 status_code 检查),加重试或速率退避是一处改动的事。当你需要的不止榜单列表时——某个榜的详情或一条帖子——查线上皮皮虾列表找到合适的端点,接入前先确认它的参数。

为什么在 API 层做监测

你当然可以打开皮皮虾手动盯榜,但那不 scale,也给不了你可以做趋势的结构化数据。通过一层统一 API 来读,意味着每次轮询都返回相同的 { id, status, model, outputs } 信封、带一个内层 status_code,于是你的循环就几行、行也一致。挂在 item_id_str 上让每一次成为一个干净的 diff——新条目、排名变动、掉榜——不重复告警。你的时间花在”这个趋势意味着什么”上,而不是花在维持一个采集器上。

这种一致性让工作流可组合。换进另一个读取——比如某个榜的详情——它就用同一个辅助函数、同一个 status_code 检查接进来。对一个上升条目加一次关键词搜索,你就从监测走到了研究,只用几行。

局限与边界

仅公开、只读数据。 不发帖、不做机器人操作、不涉及私有或仅账号可见的数据。
轮询,不是流式。 这些是同步读取;按节奏轮询、按 item_id_str 做 diff。
上游 status_code。 皮皮虾把结果包在 { status_code, data, message } 里;读 data 前先检查 status_code == 0。
度量字段因榜而异。 一个榜按点赞数排,另一个按播放数排;两个都防御式读取,并以真实响应核对。
速率与量级。 把响应当作尽力而为的读取;遇到 HTTP 429 等瞬时错误时按退避策略重试,并控制轮询节奏。
以线上参考核对。 可用性和字段可能变化;在依赖某个具体端点前先确认。

常见问题

我需要皮皮虾机器人 token 或登录吗? 不需要。你用自己的 SANDBASE_API_KEY 向 SandBase 鉴权。这套工作流读的是公开榜单数据,不需要你这边有皮皮虾账号或 OAuth。

我怎么只抓变化的部分? 给每个条目挂它的 item_id_str,并保留上一次运行的 item_id → rank 映射。下一次轮询,不在映射里的是新条目,排名变了的是变动项。

为什么不同榜用不同的度量字段? 不同榜按不同信号排名——一个按点赞数(today_digg_num)、另一个按播放数(today_show_num)。两个都防御式读取,并以真实响应核对字段。