A Weibo hot-search tracker that alerts on your keywords

I build Qovia, the API used in this post. The script is plain Python, so you can swap in any source that returns the hot-search board as JSON.

Weibo's hot-search board is a live list of what China is talking about. If you track a brand, a person or a topic, you probably want to know the moment it shows up, not when you next remember to look. In this tutorial we'll build a small poller that:

1. fetches the board on a schedule,
2. checks it against your keyword list,
3. sends an alert (to stdout and optionally a webhook) the first time a keyword appears, and not again until it has left the board,
4. pulls a few recent posts for the matched term so the alert has context.

Setup

pip install requests
export QOVIA_KEY=sk-qovia-...
export ALERT_WEBHOOK=https://hooks.example.com/...   # optional, any URL that accepts a JSON POST

Get a key from Qovia. Requests authenticate with Authorization: Bearer <key>.

Step 1: fetch the board

GET /v1/raw/weibo/hot_search takes no parameters. The catalog describes it as Weibo's realtime board in rank order, with the term, its heat, Weibo's label, category and whether it is a topic, plus the topics pinned above rank 1 in pinned. Ad slots are left out, and the board changes about every minute.

import os
import requests

BASE = "https://api.qovia.ai"
HEADERS = {"Authorization": f"Bearer {os.environ['QOVIA_KEY']}"}


def get(path, **params):
    r = requests.get(f"{BASE}{path}", headers=HEADERS, params=params, timeout=60)
    r.raise_for_status()
    return r.json()["data"] or {}


board = get("/v1/raw/weibo/hot_search")
for item in board["items"][:5]:
    print(item["rank"], item["word"], item["heat"], item["label"], item["url"])

Each item has rank, word, heat, label, category, is_topic and url. Pinned items have rank: null, so treat rank as optional.

Step 2: match keywords and avoid repeat alerts

The part that makes an alert tracker usable is not firing every minute for a term that stays on the board for hours. We keep a small JSON state file of terms we have already alerted on, and drop a term from it once it leaves the board, so it can alert again if it comes back.

import json
import pathlib

KEYWORDS = ["新能源", "小米", "苹果"]          # substrings to watch for
STATE = pathlib.Path("alerted.json")


def load_state():
    return set(json.loads(STATE.read_text())) if STATE.exists() else set()


def check(board, alerted):
    entries = board.get("items", []) + board.get("pinned", [])
    on_board = {e["word"]: e for e in entries}
    hits = []
    for word, entry in on_board.items():
        if word in alerted:
            continue
        matched = [k for k in KEYWORDS if k in word]
        if matched:
            hits.append((matched[0], entry))
    new_state = alerted & set(on_board) | {e["word"] for _, e in hits}
    return hits, new_state

Matching is a plain substring test. If you need fuzzier matching (variant spellings, English names), this is the function to extend.

Step 3: add context to the alert

A hot term on its own doesn't tell you why. GET /v1/raw/weibo/search takes a keyword and, optionally, start_date and end_date (YYYY-MM-DD, Beijing time) and a page. Search for the matched board term and include a few post snippets:

from datetime import date


def context(word, n=3):
    today = date.today().isoformat()
    res = get("/v1/raw/weibo/search", keyword=word, start_date=today, end_date=today, page=1)
    return [p["text"][:120] for p in res.get("items", [])[:n]]

Per the catalog, search text may be cut and results carry no interaction counts, so use them for context, not for ranking. Date handling is deliberately simple here: date.today() uses your machine's timezone, and the API's dates are Beijing time, so near midnight the two can disagree.

Step 4: notify

def notify(keyword, entry, snippets):
    lines = [f"Weibo hot search: '{keyword}' matched '{entry['word']}'"]
    if entry.get("rank"):
        lines[0] += f" (rank {entry['rank']})"
    lines.append(entry["url"])
    lines += [f"- {s}" for s in snippets]
    text = "\n".join(lines)
    print(text, "\n")
    hook = os.environ.get("ALERT_WEBHOOK")
    if hook:
        requests.post(hook, json={"text": text}, timeout=30)

Many chat tools accept a JSON body with a text field; adjust the payload to whatever your destination expects.

Step 5: the loop

import time

INTERVAL = 300  # seconds

alerted = load_state()
while True:
    try:
        board = get("/v1/raw/weibo/hot_search")
        hits, alerted = check(board, alerted)
        for keyword, entry in hits:
            notify(keyword, entry, context(entry["word"]))
        STATE.write_text(json.dumps(sorted(alerted), ensure_ascii=False))
    except requests.RequestException as e:
        print("request failed, will retry:", e)
    time.sleep(INTERVAL)

For production, run the loop body from cron or a scheduled job instead of while True, and load the state file each time; the logic is the same.

What it costs

In the catalog, GET /v1/raw/weibo/hot_search and GET /v1/raw/weibo/search are each $1 per 1,000 calls, and a call that returns no data is not charged. So the polling cost is simply your poll rate times that price: as arithmetic, one poll every 5 minutes is 288 board calls a day. Context lookups only happen when a keyword hits. See the pricing page for the current terms.

The board changes about every minute, so a 5-minute interval can miss a term that appears and drops off between polls. Shorten INTERVAL if that matters to you; it is a cost versus latency trade-off.

Ideas to extend it

• Alert only when a term enters the top N ranks, using rank.
• Track heat between polls to alert on fast risers.
• Store every poll to build a history of when your keywords trended.

---

This article was written with the help of AI.