Pipixia Hot Board Monitoring API Tutorial

Monitoring what’s hot on Pipixia (皮皮虾) comes down to three moves: read the hot boards, key each item to a stable id, and diff rank across runs so you catch what’s rising. This tutorial wires those moves into one workflow with the SandBase Pipixia API — one SandBase key, no Pipixia login and no SDK. These are synchronous reads, so “monitoring” means polling on a schedule.

For the full endpoint tour, see the Pipixia public data API hub. This piece is the applied monitoring workflow.

Key takeaway

• One endpoint drives it: pipixia/app/hot-search-board-list returns the boards and their ranked items.
• Every call is POST /v1/api/pipixia/<path> with one SANDBASE_API_KEY; a completed run carries outputs, a failed/timeout run carries error.
• Pipixia wraps its result in an upstream { status_code, data, message } object — check status_code == 0, then read data.boards[].board_items.
• Key each item to its item_id_str so you can diff rank and engagement across runs. Public, read-only data only; a SandBase API key is still required.

The workflow at a glance

Step	What	How
1. Read boards	pipixia/app/hot-search-board-list	check status_code, read data.boards
2. Normalize	(in your code)	flatten board_items to { item_id, rank, metric }
3. Diff	(in your code)	compare id → rank against the previous run

The endpoint API reference is the source of truth for each parameter name and response path.

Step 1 — Read the hot boards

Start with a shared helper that branches on status, then unwraps the upstream 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", "request did not complete"))
    payload = body["outputs"][0]["data"]
    # Pipixia wraps its result with an upstream status_code; check it,
    # then read defensively and confirm paths against a live response.
    if payload.get("status_code") != 0:
        raise RuntimeError(payload.get("message", "upstream error"))
    return payload.get("data", {})


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

The block below is a trimmed real response I tested on 2026-09-27 (UTC) — a board’s board_items each nest an inner item, and the metric field depends on the board (a digg count on the “top comments” board, a play count on the “most played” board). Values move over time, so read every business field with .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": "…" } }
              ]
            }
          ]
        }
      }
    }
  ]
}

hot-search-board-list returns boards, each with ranked board_items under an upstream data object.

Step 2 — Normalize items by id

Each board’s items nest an inner item with a stable item_id_str. Flatten them into a uniform row so ranking and diffing don’t care which board an item came from:

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,
                # metric field differs by board; keep both defensively
                "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"])

Because the metric field varies by board — a digg count on one, a play count on another — the row keeps both and reads each defensively. Confirm the exact field names against a live response before you depend on them.

Read each endpoint’s schema; the metric field differs by board type.

Step 3 — Diff rank across runs

Keep the previous run’s item_id → rank map. On the next poll, compare to find new entries and rank moves:

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 = {}  # load from your store
result = diff(prev, rows)
print(len(result["new"]), "new,", len(result["movers"]), "moved")
prev = result["snapshot"]  # persist for the next run

Because everything keys off the stable item_id_str, an item that stays on the board across polls is recognized as the same one — so you report a rank move, not a duplicate. Run it on a schedule and each pass surfaces only what changed.

Putting it together

A minimal monitoring pass looks like this — read, normalize, diff, persist:

prev = {}  # load from your store

def monitor_once():
    global prev
    data = call("app/hot-search-board-list", {})
    rows = normalize(data.get("boards", []))
    result = diff(prev, rows)
    prev = result["snapshot"]  # save for next run
    return result

Because every call shares the same envelope and the same call helper (with its status_code check), adding retries or rate-limit backoff is a one-place change. When you need more than the board list — a board’s detail or a post — check the live Pipixia listing for the endpoint that fits and confirm its parameters before wiring it in.

Why run monitoring at the API layer

You could open Pipixia and watch the boards by hand, but that does not scale and gives you no structured data to trend. Reading through one uniform API means each poll returns the same { id, status, model, outputs } envelope with an inner status_code, so your loop is a few lines and your rows are consistent. Keying off item_id_str makes each pass a clean diff — new items, rank moves, and items dropping off — without duplicate alerts. Your time goes to what the trend means, not to keeping a scraper alive.

That uniformity keeps the workflow composable. Swap in another read — a board’s detail, say — and it slots in behind the same helper with the same status_code check. Add a keyword search on a rising item and you have gone from monitoring to research in a couple of lines.

Limitations and boundaries

• Public, read-only data only. No posting, bot actions, or private/account-only data.
• Polling, not streaming. These are synchronous reads; poll on a cadence and diff by item_id_str.
• Upstream status_code. Pipixia wraps its result in { status_code, data, message }; check status_code == 0 before reading data.
• The metric field differs by board. One board ranks by digg count, another by play count; read both defensively and confirm against a live response.
• Rate and volume. Treat responses as best-effort reads; retry with backoff on transient errors such as HTTP 429, and pace your polling.
• Verify against the live reference. Availability and fields can change; confirm before building on a specific endpoint.

FAQ

Do I need a Pipixia bot token or login? No. You authenticate to SandBase with your SANDBASE_API_KEY. This workflow reads public board data and needs no Pipixia account or OAuth on your side.

How do I catch only what changed? Key each item to its item_id_str and keep the previous run’s item_id → rank map. On the next poll, items not in the map are new, and items whose rank changed are movers.

Why do boards use different metric fields? Different boards rank by different signals — one by a digg count (today_digg_num), another by a play count (today_show_num). Read both defensively and confirm the fields against a live response.

Build it

Create a SandBase API key, read the hot boards, and diff by item_id_str. When you are ready:

• Get a SandBase API key
• Pipixia public data API hub
• Open the hot-search-board-list API reference