Blog/开发者工具/

用一个 API 搭跨平台趋势看板 | SandBase

用一个 REST API 把微博、B 站、知乎的热榜聚合成一个趋势看板——一把 SandBase key,不用登录,归一化到统一结构。

深色电影质感画面:微博、B 站、知乎热榜经由单一 API 通道汇成一个统一的趋势看板

如果你追中文互联网趋势,你盯的不止一个榜:微博热搜看突发新闻和明星八卦、B 站看视频文化、知乎看大家在问和在辩什么。读每一个通常意味着一个不同的爬虫和一份不同的数据结构。这篇教程用 SandBase API 读全部三个榜、归一化到统一结构,搭出一个跨平台趋势看板——一把 SandBase key,不用登录。每个端点参考保证的是响应信封(id、status、model,以及 outputs[0].data);下面出现的各平台业务字段是示例结构,不是保证的 schema,请以一份真实响应为准核对。

各平台的单独全景,看 微博、B 站、知乎 hub。这一篇是落地的跨平台工作流。

先说结论

  • 读三个热榜——weibo/web-v2/hot-search、bilibili/web/hot-search、zhihu/web/hot-list——归一化到一个 { platform, rank, title, score } 结构。
  • 每次都是 POST /v1/api/<vendor>/<path>,一把 SANDBASE_API_KEY;响应共用 { id, status, model, outputs } 信封。
  • 每个平台的列表嵌套方式不同,所以工作量是一个小的按平台适配器,而不是三个独立的客户端。
  • 只是公开、只读数据;你这边不用登录。

三个榜一览

平台端点列表路径标题/分数字段
微博weibo/web-v2/hot-searchdata.realtimeword / num
B 站bilibili/web/hot-searchdata.data.trending.listkeyword / heat_score
知乎zhihu/web/hot-listdata.data[].targettarget.title / detail_text

SandBase 上一个热榜端点的 API 参考,用在这套跨平台工作流里 端点的 API 参考是每个参数名和响应路径的权威来源。

一个帮助函数,三个榜

先写一个判 status 的帮助函数,再读每个榜、映射到一个共同结构。因为每个平台把列表嵌在不同路径,每个都配一个小适配器:

import os
import requests

BASE = "https://api.sandbase.ai/v1/api"
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"))
    # 只有信封是保证的;用 .get() 防御式读取 outputs[0].data。
    return (body.get("outputs") or [{}])[0].get("data", {})


# 每个适配器里嵌套的路径(realtime/word/num、trending.list/keyword/heat_score、
# target.title/detail_text)是示例结构,不是保证的 schema,所以用 .get() 防御式读取。
def weibo_board():
    data = call("weibo/web-v2/hot-search", {})
    return [{"platform": "weibo", "rank": i, "title": it.get("word"), "score": it.get("num")}
            for i, it in enumerate(data.get("realtime", []))]


def bilibili_board():
    # `limit` 对这个端点是必填的整数。
    data = call("bilibili/web/hot-search", {"limit": 20}).get("data", {})  # 上游 { code, data }
    trending = data.get("trending", {}).get("list", [])
    return [{"platform": "bilibili", "rank": i, "title": it.get("keyword"), "score": it.get("heat_score")}
            for i, it in enumerate(trending)]


def zhihu_board():
    data = call("zhihu/web/hot-list", {}).get("data", [])
    return [{"platform": "zhihu", "rank": i, "title": it.get("target", {}).get("title"), "score": it.get("detail_text")}
            for i, it in enumerate(data)]

每个适配器读一个榜,输出同样的 { platform, rank, title, score } 记录。这些源字段路径是示例结构、不是保证的 schema——以一份真实响应为准核对,因为榜单一直在变,载荷结构也可能变动。

搭出看板

把三个榜合成一个排名 feed:

dashboard = weibo_board() + bilibili_board() + zhihu_board()
for row in dashboard:
    print(f"[{row['platform']:8}] #{row['rank']:>2} {row['title']}  ({row['score']})")

现在你有了一个可以存、可以和上一轮做 diff、或喂给 agent 的单一列表。注意分数字段在平台之间不能直接比——微博的 num 和 B 站的 heat_score 是不同量纲的整数,而知乎的 detail_text 是一个像热度标签的人类可读字符串。把它们按平台分开,比较一个平台内部的排名变动,而不是跨平台比原始分数。

SandBase 上看板里第二个热榜端点的 API 参考 每个平台的热榜参考都记录了确切的列表路径和字段。

一条归一化记录的示例

下面是一条归一化行的示例结构——请把数值当作示例,以一份真实响应为准核对源字段:

{
  "platform": "weibo",
  "rank": 0,
  "title": "…",
  "score": 1172365
}

因为三个读取都走同一个 call 帮助函数和同一个 { id, status, model, outputs } 信封,唯一的平台特定代码就是每个适配器里那一小段解包。以后加第四个榜——或换掉一个——都是局部改动。

SandBase 上看板里第三个热榜端点的 API 参考 读每个端点的 schema;列表嵌套方式因平台而异。

处理粗糙的边角

  • 每个平台嵌套不同。 微博的列表在 data.realtime,B 站的在上游 data.data.trending.list,知乎把每一项的问题包在 target 下。每个平台留一个适配器。
  • 分数不可跨平台比。 比较一个平台内随时间的排名变动;别按原始分数把平台互相排名。
  • 对 status 分支。 failed 或 timeout 的请求带 error 而没有 outputs。call 帮助函数已经强制这一点。
  • 尊重速率限制。 作为客户端韧性措施,遇到 HTTP 429 这类瞬时错误时用退避重试,并把定时任务错开。
  • 只是公开数据。 在任何平台上都不登录、不发帖、不碰私密内容。

为什么在 API 层做这件事

用爬虫搭一个跨平台看板,意味着维护三个脆弱的爬虫、三种不同的失败模式。通过一个 API 读取,意味着在一个一致的信封之上写三个短适配器、用一把 key 而不是三个 cookie 罐、以及一个加重试和日志的地方。因为每个榜都归一化到 { platform, rank, title, score },每一次定时过程都能干净地落进一张表,再和上一轮做 diff——进入某个榜的新话题、排名变动,以及同时在多个平台上热起来的话题。把这个变成洞察——聚类话题、检测跨平台的峰值——是你在采集到的数据之上另跑的一个分析步骤。

把它排成定时任务

按节奏跑,并把每一轮存下来,这样你看到的是变动而不只是一张快照:

import time, json

def snapshot():
    return weibo_board() + bilibili_board() + zhihu_board()

while True:
    rows = snapshot()
    with open(f"trends-{int(time.time())}.json", "w") as f:
        json.dump(rows, f, ensure_ascii=False)
    # 和上一份快照做 diff,找出新进入的条目和排名变动
    time.sleep(1800)  # 每 30 分钟;把运行错开以尊重速率限制

把每份快照按平台和标题存下来,再对相邻两轮做 diff:在同一时间窗里出现在不止一个平台上的标题,就是你的跨平台信号;在一个平台内快速攀升的标题,就是值得用该平台的搜索端点钻进去看的。看板是采集层;你在其上搭的排名和聚类,才是产品价值所在。

常见问题

需要登录微博、B 站或知乎吗? 不需要。你用 SANDBASE_API_KEY 向 SandBase 鉴权。三个热榜都是公开、只读的读取,你这边不用平台账号,也不用 OAuth。

每个榜怎么翻页? 每个榜都是一次按端点的读取:微博和知乎的热榜一次调用就返回当前排名,B 站的 hot-search 接一个必填的 limit 请求参数。按每个端点各自记录的参数来设,而不是指望三个榜共用一个游标。

三个响应结构不一样,怎么归一化? 给每个平台写一个小适配器,解包它各自的列表路径(微博 data.realtime、B 站 data.data.trending.list、知乎 data.data[].target),映射到共享的 { platform, rank, title, score } 行。每个路径以端点参考为准;能保证统一的只有信封。

下一步

你现在有了一套可复用的跨平台趋势看板,建立在三次公开、只读的调用和一个共享结构上。你可以扩展它:在任意平台上加搜索,钻进一个热门话题;或加一次资料读取,看是谁在推动它。