DEV Community: Qovia

Search WeChat Official Account articles by topic in Python

Qovia — Thu, 05 Nov 2026 18:29:33 +0000

I build Qovia, the API used in this post. The search loop (query, filter, paginate, de-duplicate) works with any source that can return WeChat Official Account articles; the CSV and state-file parts are plain Python.

WeChat Official Accounts (微信公众号) are where a lot of Chinese media, brands and analysts publish long-form posts. There is no public web search page you can scrape the way you would for Reddit or X, so people usually either refresh individual accounts on their phone or give up on topic discovery entirely.

In this tutorial we'll write a small Python script that:

searches WeChat Official Account articles by topic with POST /v1/search and platforms: ["wechat"],
narrows the window with since / until (Beijing calendar days),
pages through results with page / limit and has_more,
writes hits to a CSV you can open in a spreadsheet or feed into a notebook.

A short section at the end contrasts topic search with monitoring a specific account by mp_id (the account-scoped pair GET /v1/raw/wechat/account_search + GET /v1/raw/wechat/articles).

Setup

You need Python 3.9+ and requests:

pip install requests
export QOVIA_KEY=sk-qovia-...   # key from the Qovia console

Get a key from Qovia. Requests authenticate with Authorization: Bearer <key> (an x-api-key header also works). The search reference lists every parameter with a real response example.

Step 1: a POST helper

Successful calls return HTTP 200 with {"request_id", "data", "meta"}. The payload you care about is in data; billing lives in meta (meta.charged_usd, meta.billed, meta.empty). Empty results are still 200, with meta.empty: true, and are not charged. Errors are {"error": {"code", "message", ...}} and are never charged. A 429 carries a Retry-After header — wait and retry once:

import os
import time
import requests

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

spent = 0.0


def post(path, body):
    """POST JSON; honour Retry-After once; accumulate meta.charged_usd."""
    global spent
    for attempt in range(2):
        r = requests.post(
            f"{BASE}{path}",
            headers=HEADERS,
            json=body,
            timeout=120,  # a search can take a few seconds
        )
        if r.status_code == 429 and attempt == 0:
            time.sleep(int(r.headers.get("Retry-After", "10")))
            continue
        r.raise_for_status()
        payload = r.json()
        spent += float((payload.get("meta") or {}).get("charged_usd") or 0)
        return payload.get("data") or {}

Unknown body fields are rejected with 400 invalid_params, so only send documented keys.

Step 2: WeChat-only topic search

POST /v1/search is the cross-domain search endpoint. Pass platforms: ["wechat"] to restrict hits to Official Account articles. query is required (1–200 characters). Space-separated terms are ANDed — for example 人工智能教育 means both words must appear. Put words you want out of the results in exclude_words.

def search_wechat(query, since=None, until=None, page=1, limit=20, **extra):
    body = {
        "query": query,
        "platforms": ["wechat"],
        "page": page,
        "limit": limit,  # 10 or 20 only
    }
    if since:
        body["since"] = since  # YYYY-MM-DD, Beijing time, inclusive
    if until:
        body["until"] = until  # YYYY-MM-DD, Beijing time, inclusive
    body.update({k: v for k, v in extra.items() if v is not None})
    return post("/v1/search", body)


data = search_wechat("新能源汽车")
print("total hits:", data.get("total"))
print("by_platform:", data.get("by_platform"))
for item in data.get("items", []):
    author = (item.get("author") or {}).get("name")
    print(item.get("published_at"), author, item.get("title"))
    print(" ", item.get("url"))

Each item in data.items is a search hit with fields such as id, platform (here "wechat"), site, title, text, url, published_at, type, sentiment, and author (name, sometimes followers / avatar). data.total is the hit count; data.by_platform breaks that count down per platform (useful when you later widen platforms beyond WeChat). page, limit and has_more tell you where you are in the result set.

This is the search envelope for /v1/search. Other Qovia endpoints (for example the raw WeChat account/article routes below) return their own data shapes — don't assume every endpoint looks like this one.

Step 3: date window (`since` / `until`)

Dates are whole calendar days in Beijing time (+08:00). Defaults are roughly the last 7 days (until defaults to today; since defaults to 6 days before until). The span from since to until may be at most 90 days, and until cannot be later than today.

from datetime import date, timedelta

today = date.today()  # if your machine is Asia/Singapore or UTC+8, this is fine
until = today.isoformat()
since = (today - timedelta(days=13)).isoformat()  # two weeks inclusive

data = search_wechat(
    "大模型 应用",
    since=since,
    until=until,
    exclude_words="招聘 广告",
    match=["title", "content"],  # default also includes author
)
print(f"{since} → {until}: {data.get('total')} hits, page {data.get('page')}")

Useful optional filters from the same request body:

Field	What it does
`exclude_words`	Drop results that contain these words
`author`	Only this Official Account display name
`match`	Where terms must hit: `title`, `content`, `author`, `media_text`, `audio` (default `title`+`content`+`author`)
`types`	`original` / `repost` / `comment`
`sentiment`	`negative` / `neutral` / `positive`

For brand or topic monitoring you usually want match=["title", "content"] so an account name that happens to contain your keyword doesn't pollute the list.

Step 4: pagination

limit is 10 or 20 (default 20). page starts at 1 and goes up to 50. Keep requesting the next page while has_more is true, and stop early if you only need a sample:

def collect_wechat(query, since=None, until=None, max_pages=5, limit=20, **extra):
    items = []
    for page in range(1, max_pages + 1):
        data = search_wechat(
            query, since=since, until=until, page=page, limit=limit, **extra
        )
        batch = data.get("items") or []
        items.extend(batch)
        if not data.get("has_more") or not batch:
            break
        time.sleep(0.5)  # stay under the account rate limit
    return items


hits = collect_wechat("人工智能 教育", since=since, until=until, max_pages=3)
print(len(hits), "articles; spent so far $", round(spent, 4))

Rate limit is 60 requests per minute per account (600 after the first top-up). Honour Retry-After on 429 — the helper already does that once.

Step 5: write a CSV

import csv
from pathlib import Path


def to_csv(items, path="wechat-topic-hits.csv"):
    path = Path(path)
    with path.open("w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(
            f,
            fieldnames=[
                "id",
                "published_at",
                "author",
                "title",
                "url",
                "sentiment",
                "type",
                "text",
            ],
        )
        w.writeheader()
        for it in items:
            w.writerow(
                {
                    "id": it.get("id"),
                    "published_at": it.get("published_at"),
                    "author": (it.get("author") or {}).get("name"),
                    "title": it.get("title"),
                    "url": it.get("url"),
                    "sentiment": it.get("sentiment"),
                    "type": it.get("type"),
                    "text": (it.get("text") or "")[:500],
                }
            )
    return path


path = to_csv(hits)
print("wrote", path, "rows=", len(hits), "charged_usd≈", round(spent, 4))

Run it:

python search_wechat_topic.py

You get a CSV of WeChat Official Account articles matching your topic in the date window, plus a running spent total from meta.charged_usd. Non-empty /v1/search calls are billed (documented at $0.005 per call); empty answers are free.

Topic search vs monitoring a specific account

These are two different jobs:

Goal	Endpoint(s)	You need
Discover articles about a topic across many 公众号	`POST /v1/search` with `platforms: ["wechat"]`	A `query` (and optional `since` / `until`)
Follow one account's publishing history	`GET /v1/raw/wechat/account_search` → `GET /v1/raw/wechat/articles`	The account's `mp_id` (starts with `gh_` or `wxid_`)

Topic search answers "who is writing about 新能源汽车 this week?". Account monitoring answers "what did 央视新闻 publish since yesterday?".

If you already know the account name, resolve it once and poll its article list:

def get(path, **params):
    """Same envelope rules as post(), for the raw WeChat GETs."""
    global spent
    params = {k: v for k, v in params.items() if v is not None}
    for attempt in range(2):
        r = requests.get(f"{BASE}{path}", headers=HEADERS, params=params, timeout=60)
        if r.status_code == 429 and attempt == 0:
            time.sleep(int(r.headers.get("Retry-After", "10")))
            continue
        r.raise_for_status()
        payload = r.json()
        spent += float((payload.get("meta") or {}).get("charged_usd") or 0)
        return payload.get("data") or {}


accounts = get("/v1/raw/wechat/account_search", keyword="央视新闻", kind="all")
for acc in accounts.get("items", []):
    print(acc["mp_id"], acc["name"], acc.get("type"), acc.get("verified_by"))

# pick the right mp_id, then:
page = get("/v1/raw/wechat/articles", mp_id="gh_example")
for a in page.get("items", []):
    print(a["published_at"], a["title"], a["url"])

account_search returns one page (not paginated): mp_id, name, wechat_id, avatar, description, verified, verified_by, type. kind is subscription (default), service, or all. articles returns that account's history newest-first; pass the previous page's cursor back unchanged and stop when has_more is false. Each article item has title, url, published_at, cover, digest, author, ip_region, is_original, is_paid, and in the documented example also reads_label and likes.

Those raw routes are the right tool when you care about a fixed watchlist. For "find every 公众号 article on this topic", stay on /v1/search.

Putting it together

A minimal daily job: search yesterday's Beijing day for your keywords, append new URLs to a state file, write only fresh rows to CSV.

import json
from pathlib import Path

SEEN = Path("seen-urls.json")


def load_seen():
    return set(json.loads(SEEN.read_text())) if SEEN.exists() else set()


def save_seen(seen):
    SEEN.write_text(json.dumps(sorted(seen), ensure_ascii=False))


def daily(query, keywords_extra=None):
    yesterday = (date.today() - timedelta(days=1)).isoformat()
    seen = load_seen()
    fresh = []
    for q in [query, *(keywords_extra or [])]:
        for it in collect_wechat(q, since=yesterday, until=yesterday, max_pages=3):
            url = it.get("url")
            if url and url not in seen:
                seen.add(url)
                fresh.append(it)
    save_seen(seen)
    if fresh:
        to_csv(fresh, path=f"wechat-{yesterday}.csv")
    print(f"{yesterday}: {len(fresh)} new; spent ${spent:.4f}")
    return fresh


if __name__ == "__main__":
    import sys
    daily(sys.argv[1] if len(sys.argv) > 1 else "人工智能")

Cron it once a day (Asia/Singapore or Asia/Shanghai — both sit on UTC+8, same calendar day as Beijing for these date fields).

Limits worth knowing

/v1/search spans at most 90 Beijing days; default window is about the last 7.
Max page 50 × 20 results = 1,000 hits per query window — treat it as a ranked sample, not a full crawl of WeChat.
Empty results and errors are not charged; only successful non-empty answers debit meta.charged_usd.
Account rate limit: 60 req/min (600 after first top-up). Honour Retry-After.
New accounts get trial calls before prepaid balance — see the billing notes.

Going further

Widen platforms to ["wechat", "weibo", "toutiao", "news"] when you want the same topic across Chinese social + news in one call; read data.by_platform to see where volume sits.
Combine both modes: use topic search to discover which accounts keep showing up, then add their mp_ids to an account-monitor watchlist.
Ask an assistant instead: the same endpoints are on the MCP server at https://api.qovia.ai/mcp (OAuth or the same API key as a Bearer token). See the MCP docs.

Full parameter lists live in the API reference, and there's a single-file dump at qovia.ai/llms-full.txt if you want to drop it into a model's context.

This article was written with AI assistance and reviewed against Qovia's OpenAPI. I work on Qovia.

Give Claude Code and Cursor live social media data with an MCP server

Qovia — Thu, 05 Nov 2026 05:09:58 +0000

Ask an AI coding assistant "what are people saying about oat-milk lattes on Xiaohongshu this week?" and it will either guess or tell you it can't browse. The fix is an MCP server that exposes real data endpoints as tools. This walkthrough connects the Qovia MCP server (social, e-commerce and web data across Douyin, Xiaohongshu, Weibo, Bilibili, WeChat Official Accounts, TikTok, YouTube, X, Reddit and more) to Claude Code, Cursor and VS Code, then shows how to drive the same server from a plain Python script so your research is reproducible.

Disclosure: I build Qovia, and this article was drafted with AI assistance and checked by hand against the live docs and server.

What the server looks like

URL: https://api.qovia.ai/mcp
Transport: Streamable HTTP, stateless (each JSON-RPC request stands alone, answered with JSON; no session).
Auth: the same key as the REST API, Authorization: Bearer sk-qovia-....

By default it exposes only four tools, so it doesn't flood the model's context with 150+ endpoint schemas:

Tool	What it does	Needs key
`search_endpoints`	Find endpoints by keyword (English or Chinese) and/or platform slug; returns ids, titles, price per call, live status	No
`get_endpoint`	One endpoint in full: description, JSON Schema params, example params, response example, price	No
`call_endpoint`	Run an endpoint (`endpoint_id` + `params`); returns data plus billing meta	Yes
`get_account`	Balance, plan and rate limit (not charged)	Yes

The model's loop is: search, read the schema, call. Discovery is free and keyless, so you can explore the catalog before paying for anything. New accounts get 50 trial calls, and only successful, non-empty results are billed.

Connect a client

Create a key at https://qovia.ai/console/?tab=keys and keep it in an environment variable, not in your repo.

Claude Code

export QOVIA_KEY=sk-qovia-...   # your own key
claude mcp add --transport http qovia https://api.qovia.ai/mcp \
  --header "Authorization: Bearer $QOVIA_KEY"

Cursor (~/.cursor/mcp.json)

{
  "mcpServers": {
    "qovia": {
      "url": "https://api.qovia.ai/mcp",
      "headers": { "Authorization": "Bearer sk-qovia-..." }
    }
  }
}

VS Code (.vscode/mcp.json) prompts for the key on first use and never writes it to the file, which makes it safe to commit:

{
  "servers": {
    "qovia": {
      "type": "http",
      "url": "https://api.qovia.ai/mcp",
      "headers": { "Authorization": "Bearer ${input:qovia-key}" }
    }
  },
  "inputs": [
    { "type": "promptString", "id": "qovia-key", "description": "Qovia API key", "password": true }
  ]
}

Claude Desktop uses OAuth for remote connectors, so bridge with mcp-remote:

{
  "mcpServers": {
    "qovia": {
      "command": "npx",
      "args": ["-y", "mcp-remote", "https://api.qovia.ai/mcp", "--header", "Authorization:${QOVIA_AUTH}"],
      "env": { "QOVIA_AUTH": "Bearer sk-qovia-..." }
    }
  }
}

Restart the client and qovia shows up in the tool list.

Optional: direct tools for one platform

If you work on one platform all day, you can skip the search step. Append ?tools= with platform slugs (or ?tools=all):

https://api.qovia.ai/mcp?tools=xiaohongshu,weibo

Every live endpoint on those platforms becomes its own tool, named after its id with dots turned into underscores (raw.xiaohongshu.search becomes raw_xiaohongshu_search). The four default tools stay. Keep the list short: each tool adds its schema to the context.

Prompts that work well

Research prompts go best when they name the platform, the time window and the shape of the output you want:

Using the qovia tools, search Xiaohongshu for notes about "燕麦拿铁" published this week, sorted by most liked. Give me a table of the top 10 with title, likes, saves, comments and author, then summarise the three recurring complaints.

A typical tool trace:

search_endpoints {"query": "xiaohongshu search"}
get_endpoint     {"endpoint_id": "raw.xiaohongshu.search"}
call_endpoint    {"endpoint_id": "raw.xiaohongshu.search",
                  "params": {"keyword": "燕麦拿铁", ...}}

get_endpoint is what keeps the model honest: it reads the real parameter schema (for example, Xiaohongshu search V4 takes note_type of all|video|image and note_time of all|day|week) instead of inventing flags. Unknown parameters are rejected with 400 invalid_params, which the model can see and correct.

Some other prompts worth trying:

"Find the WeChat Official Accounts that publish about cross-border payments, then list their latest articles." (wechat platform)
"Pull the Weibo hot search list and tell me which entries relate to EVs."
"Compare comments on this Bilibili video and this YouTube video on the same product."

Each call_endpoint result includes charged_usd and your balance, so you can ask the assistant to keep a running cost total.

Do the same thing from Python

Chat is great for exploration, but once a query is useful you want it in a script. Because the server is stateless Streamable HTTP, you don't even need an MCP SDK: a JSON-RPC POST per call is enough.

# qovia_mcp.py - minimal client for the Qovia MCP server
import json
import os

import requests

MCP_URL = "https://api.qovia.ai/mcp"
HEADERS = {
    "Content-Type": "application/json",
    "Accept": "application/json, text/event-stream",
}
if os.environ.get("QOVIA_KEY"):
    HEADERS["Authorization"] = f"Bearer {os.environ['QOVIA_KEY']}"

_id = 0


def rpc(method, params=None):
    global _id
    _id += 1
    body = {"jsonrpc": "2.0", "id": _id, "method": method}
    if params is not None:
        body["params"] = params
    r = requests.post(MCP_URL, headers=HEADERS, json=body, timeout=60)
    r.raise_for_status()
    msg = r.json()
    if "error" in msg:
        raise RuntimeError(msg["error"])
    return msg["result"]


def tool(name, **arguments):
    result = rpc("tools/call", {"name": name, "arguments": arguments})
    if result.get("isError"):
        raise RuntimeError(result["content"][0]["text"])
    # Prefer structured output; fall back to the text block.
    if "structuredContent" in result:
        return result["structuredContent"]
    return json.loads(result["content"][0]["text"])


if __name__ == "__main__":
    # 1. Discovery: free, no key needed
    for t in rpc("tools/list")["tools"]:
        print("tool:", t["name"])

    found = tool("search_endpoints", query="xiaohongshu search")
    for ep in found["endpoints"]:
        print(ep["id"], ep["price_usd_per_call"], "live" if ep["available"] else "soon")

    detail = tool("get_endpoint", endpoint_id="raw.xiaohongshu.search")
    print(json.dumps(detail, ensure_ascii=False)[:600])

Run it without a key first; tools/list, search_endpoints and get_endpoint all work anonymously:

tool: search_endpoints
tool: get_endpoint
tool: call_endpoint
tool: get_account
raw.xiaohongshu.search 0.005 live
raw.xiaohongshu.search_v4 0.02536 live
raw.pgy.note_search 0.01 live
raw.pgy.note_search_v2 0.02536 live

Now add a paid call. Xiaohongshu note search returns the platform's own result: items, each with model_type and a note (id, title, liked_count, collected_count, comments_count, user, ...). There is no has_more; you page until a page comes back empty, and empty pages are not charged.

# xhs_top_notes.py
import os
from qovia_mcp import tool

assert os.environ.get("QOVIA_KEY"), "set QOVIA_KEY first"

print(tool("get_account"))  # balance and rate limit, not charged

KEYWORD = "燕麦拿铁"
notes = []
for page in range(1, 4):
    out = tool("call_endpoint",
               endpoint_id="raw.xiaohongshu.search",
               params={"keyword": KEYWORD, "page": page})
    data = out.get("data") or {}
    items = [i["note"] for i in data.get("items", []) if i.get("model_type") == "note"]
    if not items:
        break
    notes.extend(items)
    print(f"page {page}: {len(items)} notes, charged {out.get('meta', {}).get('charged_usd')}")

notes.sort(key=lambda n: int(n.get("liked_count") or 0), reverse=True)
for n in notes[:10]:
    print(f"{n.get('liked_count'):>7} likes  {n.get('collected_count'):>6} saves  "
          f"{n.get('title') or '(no title)'}  @{n['user']['nickname']}")

The shape of call_endpoint's output mirrors the REST envelope (data plus meta with charged_usd). If your version prints it under a different key, print(out.keys()) once and adjust; the point is that what the assistant saw in chat and what your script sees are the same data.

Handling errors and limits

Errors come back as {"error": {"code", "message", ...}} and are never charged. Branch on code: invalid_params (fix your params), rate_limited / endpoint_busy (back off and honour Retry-After), insufficient_balance, upstream_timeout (retry later).
The account limit is 60 requests per minute (600 after the first top-up). A three-page loop is fine; a 500-keyword sweep needs a time.sleep.
Keep keys out of prompts and committed configs. The VS Code inputs pattern and environment variables are the easy wins.

When to use MCP vs. the REST API

Use the MCP server when a person is asking open questions and the model should pick the endpoint. Use the REST API (GET https://api.qovia.ai/v1/raw/xiaohongshu/search?keyword=...) in production jobs where the endpoint is fixed and you want one less moving part. Calls are billed identically either way, so prototyping in Claude or Cursor and then hardening in Python costs nothing extra.

Docs: MCP setup, API reference, and the whole catalog as one file for your model's context at llms-full.txt.

Compare how a keyword is doing on TikTok, YouTube and X in Python

Qovia — Wed, 04 Nov 2026 03:45:10 +0000

You have a product name, a campaign hashtag or a competitor, and someone asks: "Where are people actually talking about this, TikTok, YouTube or X?" Answering that usually means three different APIs, three auth schemes and three response shapes. In this tutorial we'll write one short Python script that searches all three for the same keyword, normalizes the posts into one table, and prints a side-by-side comparison: how many posts, how recent, how much reach, and the top posts per platform.

We'll use Qovia, a paid public-data API that exposes TikTok, YouTube and X search (plus about 20 other platforms) behind one key and one JSON envelope. Full disclosure: I work on Qovia. Everything below uses only the documented endpoints from the API reference.

What you need

Python 3.9+ and requests (pip install requests pandas)
A Qovia API key from the console. New accounts get 50 trial calls, which is plenty for this tutorial. Each search call below costs $0.001 after that, and empty results are not charged.

export QOVIA_KEY="sk-qovia-..."

The three endpoints

Platform	Endpoint	Useful params
TikTok	`GET /v1/raw/tiktok/search`	`keyword`, `sort` (`general` / `most_liked`), `publish_time` (`all`, `day`, `week`, `month`, `quarter`, `half_year`), `region` (e.g. `US`), `count` (max 30), integer `cursor`
YouTube	`GET /v1/raw/youtube/search`	`keyword`, `upload_date` (`hour`, `today`, `week`, `month`, `year`; default `month`), string `cursor`
X	`GET /v1/raw/x/search`	`keyword` (X search syntax works, e.g. `#hashtag`), `type` (`top` / `media`), string `cursor`

All three return the same record shape inside data.items: id, platform, text, author (with nickname, followers), metrics, and extra (with published_at in UTC and url). Pagination is the same everywhere: pass back data.cursor, stop when data.has_more is false.

The metrics differ per platform, and it's worth knowing that up front:

TikTok search records carry plays, likes, comments, shares, saves.
YouTube search records carry plays (view count) only. Likes and comments are on the video detail endpoint, raw.youtube.video.
X search records carry views, likes, comments, shares (reposts), quotes, saves.

So "reach" (plays or views) is comparable across all three, and "engagement" only across TikTok and X unless you enrich YouTube.

A tiny client

Every Qovia response is either {"request_id", "data", "meta"} on success or {"error": {"code", "message", ...}} on failure. Rate limits are 60 requests per minute on a new account, and a 429 comes with Retry-After, so the client honours that.

import os, time, requests

BASE = "https://api.qovia.ai"
SESSION = requests.Session()
SESSION.headers["Authorization"] = f"Bearer {os.environ['QOVIA_KEY']}"

def call(path, **params):
    for attempt in range(4):
        r = SESSION.get(BASE + path, params=params, timeout=60)
        if r.status_code == 429:
            time.sleep(int(r.headers.get("Retry-After", "5")))
            continue
        body = r.json()
        if "error" in body:
            code = body["error"]["code"]
            if code in ("upstream_busy", "upstream_timeout") and attempt < 3:
                time.sleep(2 * (attempt + 1))
                continue
            raise RuntimeError(f"{path}: {code}: {body['error']['message']}")
        return body["data"] or {}, body["meta"]
    raise RuntimeError(f"{path}: gave up after retries")

Two details from the docs worth respecting: unknown query parameters are rejected with 400 invalid_params (so typos fail loudly instead of being ignored), and an empty result is still HTTP 200 with meta.empty: true, which is why we fall back to {}.

Paging through each platform

def search(path, pages=3, **params):
    items, cursor = [], None
    for _ in range(pages):
        p = dict(params)
        if cursor is not None:
            p["cursor"] = cursor
        data, meta = call(path, **p)
        items += data.get("items", [])
        if not data.get("has_more"):
            break
        cursor = data.get("cursor")
    # TikTok notes adjacent pages may repeat a few posts: dedupe by id
    return list({it["id"]: it for it in items}.values())

def collect(keyword, pages=3):
    return {
        "tiktok": search("/v1/raw/tiktok/search", pages, keyword=keyword,
                         publish_time="month", region="US", count=30),
        "youtube": search("/v1/raw/youtube/search", pages, keyword=keyword,
                          upload_date="month"),
        "x": search("/v1/raw/x/search", pages, keyword=keyword, type="top"),
    }

Three pages per platform is nine calls, under one cent once the trial is used up.

One honest caveat about the time windows: TikTok and YouTube let you ask for "the last month", while X search returns X's "top" order with no date filter. We'll filter X by published_at ourselves after the fact so the comparison is fair.

Normalizing into one table

import pandas as pd

def to_rows(results):
    rows = []
    for platform, items in results.items():
        for it in items:
            m, ex, au = it.get("metrics") or {}, it.get("extra") or {}, it.get("author") or {}
            reach = m.get("plays") if platform != "x" else m.get("views")
            engagement = None
            if platform in ("tiktok", "x"):
                engagement = sum((m.get(k) or 0) for k in ("likes", "comments", "shares", "saves"))
            rows.append({
                "platform": platform,
                "id": it["id"],
                "published_at": pd.to_datetime(ex.get("published_at"), utc=True),
                "author": au.get("nickname"),
                "author_followers": au.get("followers"),
                "reach": reach,
                "engagement": engagement,
                "text": (ex.get("title") or it.get("text") or "").replace("\n", " ")[:90],
                "url": ex.get("url"),
            })
    return pd.DataFrame(rows)

YouTube records put the video title in extra.title and title-plus-snippet in text, so we prefer the title. Note that YouTube's author.followers is null in search results; the channel endpoint has subscriber counts if you need them.

The comparison

def compare(keyword, days=30):
    df = to_rows(collect(keyword))
    cutoff = pd.Timestamp.now(tz="UTC") - pd.Timedelta(days=days)
    df = df[df["published_at"] >= cutoff]

    summary = df.groupby("platform").agg(
        posts=("id", "count"),
        median_reach=("reach", "median"),
        total_reach=("reach", "sum"),
        median_engagement=("engagement", "median"),
        newest=("published_at", "max"),
    )
    print(f"\n=== '{keyword}' over the last {days} days ===")
    print(summary.to_string())

    for platform, group in df.groupby("platform"):
        print(f"\nTop 3 on {platform} by reach:")
        for _, r in group.nlargest(3, "reach").iterrows():
            print(f"  {int(r.reach or 0):>10,}  {r.author}: {r.text}")
            print(f"              {r.url}")
    return df

if __name__ == "__main__":
    import sys
    df = compare(sys.argv[1] if len(sys.argv) > 1 else "cursor ai")
    df.to_csv("comparison.csv", index=False)

Run it:

python compare.py "your product name"

You get a summary table with one row per platform plus the three biggest posts on each, and a comparison.csv you can open in a spreadsheet or feed into a notebook.

Reading the result

A few patterns show up quickly when you run this on real keywords:

Lots of TikTok posts, low median plays: the topic is being used as a hashtag by small creators. Good for UGC, weak as a signal of reach.
Few YouTube videos, high median plays: long-form explainers and reviews own the topic. That's where search traffic goes.
X dominated by a few accounts: check author_followers. If the top posts are all from accounts with huge followings, the "conversation" is really a couple of announcements and their replies.

Because the three samples are each platform's own ranking (TikTok relevance, YouTube relevance, X "top"), treat counts as a sample, not a census. The useful signal is the relative shape: where reach concentrates, how fresh the newest posts are, and who is driving it.

Going further

Engagement for YouTube: call GET /v1/raw/youtube/video?video_id=... with extra.source_id for the top videos to get likes and comment counts.
What people actually say: pull comments with raw.tiktok.comments, raw.youtube.comments and raw.x.comments for the top posts and run the same sentiment or keyword pass across all three.
Run it daily: append each run's summary with a date column and you have a cross-platform trend line for free.
Ask an assistant instead: the same endpoints are available as an MCP server at https://api.qovia.ai/mcp (OAuth or the same API key as a Bearer token), so Claude, Cursor or VS Code can run this comparison conversationally. See the MCP docs.

The full parameter list for every endpoint is in the API reference, and there's a single-file version at qovia.ai/llms-full.txt if you want to drop it into a model's context.

This article was written with AI assistance and reviewed against Qovia's documentation. I work on Qovia.

Monitor Reddit for mentions of your product in Python

Qovia — Sun, 01 Nov 2026 07:37:08 +0000

I build Qovia, the API used in this post. The monitoring logic (search, de-duplicate, filter, alert) works with any Reddit data source; the state file, CSV and Slack parts are plain Python.

If you ship a developer tool or a SaaS product, Reddit is where people ask "is there an alternative to X?", complain about your onboarding, or recommend you to someone else, often without tagging you. Finding those threads early is worth a lot: a helpful answer in the first hour of a thread does more than a week of ads. Doing it by hand means refreshing Reddit search every morning, so let's automate it.

In this tutorial we'll write a script that:

searches Reddit for your product name (and competitors) with Reddit's own search syntax,
watches the newest posts in a few subreddits for phrases like "alternative to" or "looking for",
remembers what it has already seen, so each run only reports new posts,
pulls the top comments under every new hit,
writes everything to CSV and posts a short summary to a Slack channel.

Run it from cron every hour and you have a basic social-listening tool for a few dollars a month.

Setup

You need Python 3.9+ and requests:

pip install requests
export QOVIA_KEY=sk-qovia-...   # key from the Qovia console
export SLACK_WEBHOOK_URL=https://hooks.slack.com/services/...   # optional

Get a key from Qovia. Requests authenticate with Authorization: Bearer <key> (an x-api-key header also works). The Reddit search reference shows every parameter with a real response; the other Reddit endpoints are listed next to it.

You don't need a Reddit account, OAuth app or client secret for this. The trade-off is that it's read-only and public content only (more under Limits).

Step 1: a small request helper

Every response has the same envelope: the result is in data, and meta reports the charge (meta.billed, meta.charged_usd, meta.empty). Empty results come back as HTTP 200 with meta.empty: true and are not charged. Errors are {"error": {"code", "message", ...}} and are never charged. A post or subreddit that doesn't exist answers 404, so the helper turns that into None. A 429 carries a Retry-After header, so the helper waits and retries once:

import os
import time
import requests

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

spent = 0.0


def get(path, **params):
    global spent
    params = {k: v for k, v in params.items() if v is not None}
    for attempt in range(2):
        r = requests.get(f"{BASE}{path}", headers=HEADERS, params=params, timeout=60)
        if r.status_code == 429 and attempt == 0:
            time.sleep(int(r.headers.get("Retry-After", "10")))
            continue
        if r.status_code == 404:
            return None  # missing post or subreddit, not charged
        r.raise_for_status()
        body = r.json()
        spent += float((body.get("meta") or {}).get("charged_usd") or 0)
        return body.get("data") or {}

Unknown parameters are rejected with 400 invalid_params, which is why the helper drops None values instead of sending them.

Step 2: search Reddit for your keywords

GET /v1/raw/reddit/search takes a keyword (1 to 200 characters) and Reddit's search syntax works inside it: AND, OR, NOT, quoted phrases, subreddit:, author:, title: and so on. Two more parameters matter for monitoring:

sort: relevance (the default), hot, top, new or comments. For monitoring you want new.
time_range: all (the default), year, month, week, day or hour. If you run hourly, day gives you some overlap, which the state file in Step 4 takes care of.

Results come about 7 a page. Pass cursor back for the next page and stop when has_more is false. If Reddit spell-corrected your query, corrected_query tells you what it actually searched, which is worth logging: brand names get "corrected" more often than you'd think.

Each post record has text (title and body), author (username), metrics (likes, which is the post's score, and comments) and extra with source_id (the post's fullname, t3_...), title, subreddit, published_at (UTC), flair, content_type, link_url and url.

QUERIES = {
    "brand": '"acme analytics" OR acmeanalytics',
    "competitor": '"plausible analytics" alternative',
}


def search(query, time_range="day", max_pages=2):
    posts, cursor = [], None
    for _ in range(max_pages):
        page = get("/v1/raw/reddit/search", keyword=query, sort="new",
                   time_range=time_range, cursor=cursor)
        if not page:
            break
        if page.get("corrected_query"):
            print(f"  note: Reddit searched {page['corrected_query']!r} instead of {query!r}")
        posts.extend(page.get("items", []))
        if not page.get("has_more"):
            break
        cursor = page.get("cursor")
    return posts

Quote multi-word names. Without quotes, acme analytics matches any post containing both words anywhere, which for a common word gets noisy fast.

Step 3: watch a few subreddits for intent phrases

Search finds posts that name you. The more valuable posts often don't: "what do you use for privacy-friendly analytics?" in r/webdev never mentions any product. For those, watch the newest posts in the communities your users hang out in and filter locally.

GET /v1/raw/reddit/subreddit_posts takes a subreddit (the name after r/) and a sort (hot by default; also best, new, top, controversial, rising). It's cursor-paginated: about 6 posts on the first page, then about 24. The list records carry the title and the first 300 characters of the body (extra.truncated tells you when it was cut), which is plenty for a phrase filter. Fields the list doesn't carry (language, author ID, images, upvote ratio) are null; GET /v1/raw/reddit/post has them if you need them.

import re

SUBREDDITS = ["webdev", "SaaS", "selfhosted"]
TOPIC = re.compile(r"analytics|tracking|pageviews", re.I)
INTENT = re.compile(r"alternative to|looking for|recommend|what do you use|switch(ed|ing)? from", re.I)


def intent_posts(subreddit, max_pages=2):
    hits, cursor = [], None
    for _ in range(max_pages):
        page = get("/v1/raw/reddit/subreddit_posts", subreddit=subreddit,
                   sort="new", cursor=cursor)
        if not page:
            break  # 404: no such subreddit
        for p in page.get("items", []):
            text = p.get("text") or ""
            if TOPIC.search(text) and INTENT.search(text):
                hits.append(p)
        if not page.get("has_more"):
            break
        cursor = page.get("cursor")
    return hits

The two regexes are deliberately simple: a post must be about the topic and sound like someone is choosing a tool. Tune them on a week of output; this is the part of the script you'll edit most.

Step 4: remember what you've already seen

Both sources return the same record shape with the same IDs, so one set of seen fullnames is enough. A JSON file next to the script is fine for an hourly job:

import json
from pathlib import Path

STATE = Path("reddit_seen.json")


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


def save_seen(seen):
    STATE.write_text(json.dumps(sorted(seen)))


def only_new(posts, seen):
    fresh = []
    for p in posts:
        pid = p["extra"]["source_id"]
        if pid not in seen:
            seen.add(pid)
            fresh.append(p)
    return fresh

This also de-duplicates a post that shows up in both a search and a subreddit feed during the same run.

Step 5: read the room under each new post

A title tells you a thread exists; the comments tell you whether it's worth answering. GET /v1/raw/reddit/comments takes the post's post_id (the same t3_... fullname) and an optional sort (best by default, or top, new, old, controversial). A page holds 8 comments: a few top-level ones plus the replies Reddit expands under them, flattened in Reddit's order. Each has cid, text, likes (the comment's score), reply_count, created_at (UTC), pinned, author, depth (0 for top-level) and parent_cid. The page also has total, cursor and has_more. A post without comments answers empty and isn't charged.

def top_comments(post_id, keywords=None):
    page = get("/v1/raw/reddit/comments", post_id=post_id, sort="top") or {}
    out = []
    for c in page.get("items", []):
        text = (c.get("text") or "").replace("\n", " ")
        out.append({
            "cid": c["cid"],
            "depth": c.get("depth"),
            "score": c.get("likes"),
            "replies": c.get("reply_count"),
            "mentions": bool(keywords and keywords.search(text)),
            "text": text[:300],
        })
    return out, page.get("total")

One page per post keeps the cost predictable. If you need a full thread, keep paging with cursor; replies Reddit collapses come with a replies_cursor that GET /v1/raw/reddit/comment_replies expands (it takes post_id and that cursor). For alerts, the top 8 is enough to see whether people are already recommending someone.

Step 6: CSV, Slack, and the run loop

import csv
from datetime import datetime, timezone

FIELDS = ["checked_at", "source", "post_id", "subreddit", "title", "author",
          "score", "comments", "published_at", "url", "top_comment"]
MENTION = re.compile(r"acme", re.I)  # flags comments that name you


def now():
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def row(source, p, comments):
    ex, m = p.get("extra") or {}, p.get("metrics") or {}
    best = comments[0]["text"] if comments else ""
    return {
        "checked_at": now(),
        "source": source,
        "post_id": ex.get("source_id"),
        "subreddit": ex.get("subreddit"),
        "title": ex.get("title"),
        "author": (p.get("author") or {}).get("username"),
        "score": m.get("likes"),
        "comments": m.get("comments"),
        "published_at": ex.get("published_at"),
        "url": ex.get("url"),
        "top_comment": best[:200],
    }


def append_csv(path, rows):
    path = Path(path)
    new_file = not path.exists()
    with path.open("a", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=FIELDS)
        if new_file:
            w.writeheader()
        w.writerows(rows)


def notify(rows):
    url = os.environ.get("SLACK_WEBHOOK_URL")
    if not url or not rows:
        return
    lines = [f"*{len(rows)} new Reddit posts*"]
    for r in rows[:15]:
        lines.append(f"• [{r['source']}] r/{r['subreddit']}: <{r['url']}|{r['title']}> "
                     f"({r['score']} pts, {r['comments']} comments)")
    requests.post(url, json={"text": "\n".join(lines)}, timeout=30)


def run():
    seen = load_seen()
    found = []
    for name, q in QUERIES.items():
        found += [(name, p) for p in only_new(search(q), seen)]
    for sub in SUBREDDITS:
        found += [(f"r/{sub}", p) for p in only_new(intent_posts(sub), seen)]

    rows = []
    for source, p in found:
        comments, _ = top_comments(p["extra"]["source_id"], MENTION)
        if any(c["mentions"] for c in comments):
            source += " +mentioned in comments"
        rows.append(row(source, p, comments))

    append_csv("reddit_mentions.csv", rows)
    notify(rows)
    save_seen(seen)
    print(f"{len(rows)} new posts, spent this run: ${spent:.4f}")


if __name__ == "__main__":
    run()

The state is saved only after the CSV and the Slack message go out, so a crash mid-run means the next run reports those posts again instead of losing them.

Run it once by hand, then schedule it:

python3 reddit_monitor.py
# crontab -e
# .env holds: export QOVIA_KEY=... and export SLACK_WEBHOOK_URL=...
0 * * * * cd /path/to/monitor && . ./.env && python3 reddit_monitor.py >> monitor.log 2>&1

Output looks something like this (illustrative, not real data):

  note: Reddit searched 'acme analytic' instead of 'acmeanalytics'
4 new posts, spent this run: $0.0140

And in Slack (also illustrative):

4 new Reddit posts
• [brand] r/webdev: Anyone tried Acme Analytics for a static site? (12 pts, 9 comments)
• [competitor] r/selfhosted: Plausible alternative that runs on SQLite? (31 pts, 22 comments)
• [r/SaaS] r/SaaS: Looking for privacy-friendly analytics for a B2B app (5 pts, 3 comments)
• [r/webdev +mentioned in comments] r/webdev: What do you use for pageview tracking in 2026? (48 pts, 61 comments)

The last one is the interesting kind: a post that never names you, where someone in the comments already did.

Doing the same from Claude or Cursor

For one-off research ("what did r/selfhosted say about analytics tools this month?") a script is overkill. The same endpoints are available through Qovia's MCP server at https://api.qovia.ai/mcp (Streamable HTTP, same API key in an Authorization: Bearer header, same billing as the HTTP API). In Claude Code:

claude mcp add --transport http qovia https://api.qovia.ai/mcp \
  --header "Authorization: Bearer $QOVIA_KEY"

Then ask something like "search Reddit for posts from the last week asking for a Plausible alternative, and summarize what the top comments recommend." Configs for Cursor, VS Code and Claude Desktop are in the MCP docs. The assistant is good for digging into one question; the cron job is better for not missing anything.

Cost

Every Reddit endpoint used here (search, subreddit_posts, comments, and post / comment_replies if you add them) is a fixed $0.001 per successful call ($1 per 1,000). Empty results, errors and 404s are not charged, so a quiet hour where search finds nothing costs nothing for that query.

As a rough example (arithmetic only): with 2 queries x up to 2 search pages, 3 subreddits x up to 2 pages, and one comment page per new post, a run is at most 10 calls plus one per new post. With 4 new posts that's $0.014. Run hourly, the worst case is 240 calls a day ($0.24) plus one call per new post, so roughly $7 to $8 a month; it's less in practice because a search with no new results returns empty and isn't charged. The subreddit feeds are the fixed part: one page per subreddit, or running every three hours, cuts that by half or more. The spent counter prints the real number each run. New accounts get 50 trial calls, counted only when a call returns data; after that it's a prepaid balance with no monthly fee. Details are on the pricing page, and every parameter is in the API docs.

The default rate limit is 60 requests per minute per account, far more than an hourly job needs.

Limits

Public content only, read-only: you can't post, vote or reply through this.
The Reddit endpoints are marked beta.
Search covers posts, not comments. A comment that names you under a post that doesn't is only found if that post is in one of your watched subreddits (Step 3 + Step 5).
subreddit_posts carries only the first 300 characters of the body; fetch GET /v1/raw/reddit/post when you need the full text.
Scores and comment counts are a snapshot at the time of the call; a post found an hour after it went up will have very different numbers a day later.
Reddit's own search decides what matches. Check corrected_query and test your queries on reddit.com once before trusting them.

If you build on this, tell me in the comments what you'd want it to flag besides mentions.

Drafted with AI assistance and checked against the live API docs.

Pull Kuaishou creator stats and video comments in Python

Qovia — Fri, 30 Oct 2026 23:33:17 +0000

I build Qovia, the API used in this post. The workflow itself (resolve a video, list the creator's top videos, compute engagement, collect comments, append to CSV) works with any Kuaishou data source; the CSV and pandas parts are plain Python.

Kuaishou (快手, Kwai abroad) is China's other big short-video app next to Douyin. Its audience leans more toward smaller cities and live commerce, and a lot of creators who matter for consumer brands there are barely visible on Douyin. If you're evaluating a Kuaishou creator for a campaign, or just want to know what's working in a niche, the questions are usually the same: how many plays do their best videos get, how engaged are viewers, what do they use as hashtags, and what do people say in the comments?

A note on the angle: the Kuaishou endpoints are built around a video link (its eid) or a creator's numeric uid, and from there you can walk to everything else. There's no Kuaishou-only keyword search; for finding a starting video by keyword, Qovia's cross-platform search can be limited to short-video apps (Step 2b). The hot board gives you a separate view of what's trending.

In this tutorial we'll write a script that:

takes one Kuaishou video link and fetches its detail, including the creator's numeric uid,
pulls the creator's hot videos (most liked first) and computes engagement stats,
collects the first pages of comments under their top videos,
snapshots the Kuaishou hot board,
appends everything to CSV and summarizes it with pandas.

Setup

You need Python 3.9+, requests and pandas:

pip install requests pandas
export QOVIA_KEY=sk-qovia-...   # key from the Qovia console

Get a key from Qovia. Requests authenticate with Authorization: Bearer <key> (an x-api-key header also works). The Kuaishou API page lists the five Kuaishou endpoints with a real request and response.

Step 1: a small request helper

Every response has the same envelope: the result is in data, and meta reports the charge (meta.billed, meta.charged_usd, meta.empty). Empty results come back as HTTP 200 with meta.empty: true and are not charged. Errors are {"error": {"code", "message", ...}} and are never charged. The video detail endpoint answers 404 for a post that doesn't exist or was deleted, so the helper turns that into None. A 429 carries a Retry-After header, so the helper waits and retries once:

import os
import time
import requests

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

spent = 0.0


def get(path, **params):
    global spent
    params = {k: v for k, v in params.items() if v is not None}
    for attempt in range(2):
        r = requests.get(f"{BASE}{path}", headers=HEADERS, params=params, timeout=60)
        if r.status_code == 429 and attempt == 0:
            time.sleep(int(r.headers.get("Retry-After", "10")))
            continue
        if r.status_code == 404:
            return None  # deleted or missing, not charged
        r.raise_for_status()
        body = r.json()
        spent += float((body.get("meta") or {}).get("charged_usd") or 0)
        return body.get("data") or {}

Unknown parameters are rejected with 400 invalid_params, which is why the helper drops None values instead of sending them.

Step 2: from a video link to the creator

Kuaishou video links look like https://www.kuaishou.com/short-video/3xry7mhn2ywjyqg. The last part is the video's eid, and that's what the API takes; the long numeric photo ID is not accepted.

GET /v1/raw/kuaishou/video with video_id=<eid> returns one post: text (the caption), author (nickname, numeric uid, eid, and unique_id, the Kuaishou ID), metrics (plays, likes, comments, shares, saves) and extra with eid, published_at (Beijing time), video_dur_s, topics (hashtags), cover_url, ip_location and url.

The important bit for us is author.uid: the creator endpoint takes the numeric uid, not the eid.

import re

EID_RE = re.compile(r"(3x[0-9a-z]{10,20})")


def eid_from_link(link):
    m = EID_RE.search(link)
    if not m:
        raise ValueError(f"no Kuaishou eid in {link!r}")
    return m.group(1)


def resolve(link):
    video = get("/v1/raw/kuaishou/video", video_id=eid_from_link(link))
    if not video:
        raise SystemExit("video not found (deleted or private)")
    a = video["author"]
    print(f"{a.get('nickname')} (uid {a['uid']}, Kuaishou ID {a.get('unique_id')})")
    return video

The 3x... pattern matches the format the API validates; if Kuaishou changes its link format, you can always paste the eid directly.

Step 2b (optional): find a starting video by keyword

If you don't have a link yet, POST /v1/search searches public content across Chinese platforms. With platforms: ["short_video"] (and types: ["original"] to skip reposts and comments) it is limited to "Kuaishou and other short-video apps", so the results can include other apps too; keep the ones whose url is a Kuaishou link. Dates are whole days in Beijing time (the last 7 days by default), 10 or 20 results per page. This endpoint is priced differently from the Kuaishou ones (see Cost).

def find_kuaishou_links(query, limit=20):
    global spent
    r = requests.post(f"{BASE}/v1/search", headers=HEADERS, timeout=60,
                      json={"query": query, "platforms": ["short_video"],
                            "types": ["original"], "limit": limit})
    r.raise_for_status()
    body = r.json()
    spent += float((body.get("meta") or {}).get("charged_usd") or 0)
    items = (body.get("data") or {}).get("items", [])
    return [i["url"] for i in items
            if "kuaishou.com" in (i.get("url") or "") and EID_RE.search(i["url"])]

I haven't checked what share of short_video results are Kuaishou for a given query, so treat this as a way to find candidates, then run the rest of the pipeline on a link you picked.

Step 3: the creator's hot videos and their stats

GET /v1/raw/kuaishou/creator_hot_videos takes the uid and returns the creator's hot posts, most liked first, 20 per page, as the same records as the video endpoint. Pass cursor back for the next page and stop when has_more is false.

This is the creator's hot list, not a full upload history, so treat the numbers as "what their best content does" rather than an average over everything they post. That's still the number most people want when comparing creators.

import csv
from datetime import datetime, timezone
from pathlib import Path

VIDEO_FIELDS = ["checked_at", "creator_uid", "creator", "eid", "caption", "plays",
                "likes", "comments", "shares", "saves", "engagement_rate",
                "duration_s", "topics", "published_at", "url"]


def now():
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def hot_videos(uid, max_pages=2):
    videos, seen, cursor = [], set(), None
    for _ in range(max_pages):
        page = get("/v1/raw/kuaishou/creator_hot_videos", uid=uid, cursor=cursor)
        for v in (page or {}).get("items", []):
            if v["extra"]["eid"] not in seen:
                seen.add(v["extra"]["eid"])
                videos.append(v)
        if not (page or {}).get("has_more"):
            break
        cursor = page.get("cursor")
    return videos


def video_row(v):
    m, ex, a = v.get("metrics") or {}, v.get("extra") or {}, v.get("author") or {}
    plays = m.get("plays") or 0
    actions = sum(m.get(k) or 0 for k in ("likes", "comments", "shares"))
    return {
        "checked_at": now(),
        "creator_uid": a.get("uid"),
        "creator": a.get("nickname"),
        "eid": ex.get("eid"),
        "caption": (v.get("text") or "").replace("\n", " ")[:200],
        "plays": m.get("plays"),
        "likes": m.get("likes"),
        "comments": m.get("comments"),
        "shares": m.get("shares"),
        "saves": m.get("saves"),
        "engagement_rate": round(actions / plays, 4) if plays else None,
        "duration_s": ex.get("video_dur_s"),
        "topics": "|".join(ex.get("topics") or []),
        "published_at": ex.get("published_at"),
        "url": ex.get("url"),
    }


def append_csv(path, fields, rows):
    path = Path(path)
    new_file = not path.exists()
    with path.open("a", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=fields)
        if new_file:
            w.writeheader()
        w.writerows(rows)

engagement_rate here is (likes + comments + shares) / plays. That's my definition, not Kuaishou's; pick whatever ratio fits your use, but keep it consistent across creators.

Step 4: what viewers say

GET /v1/raw/kuaishou/comments takes the same video_id (eid) and returns first-level comments with a cursor: about 30 on the first page, then about 20. Each comment has cid, text, likes, reply_count, ip_location, created_at (Beijing time), is_hot (Kuaishou's hot-comment flag), liked_by_author, images and author; the page also has total, cursor and has_more.

liked_by_author is a nice signal for creator research: it tells you which comments the creator chose to like, which often says something about how they manage their audience.

COMMENT_FIELDS = ["checked_at", "eid", "cid", "text", "likes", "reply_count",
                  "ip_location", "created_at", "is_hot", "liked_by_author"]


def comment_rows(videos, top=5, pages_per_video=1):
    rows = []
    for v in videos[:top]:
        eid, cursor = v["extra"]["eid"], None
        for _ in range(pages_per_video):
            page = get("/v1/raw/kuaishou/comments", video_id=eid, cursor=cursor)
            for c in (page or {}).get("items", []):
                rows.append({
                    "checked_at": now(),
                    "eid": eid,
                    "cid": c["cid"],
                    "text": (c.get("text") or "").replace("\n", " "),
                    "likes": c.get("likes"),
                    "reply_count": c.get("reply_count"),
                    "ip_location": c.get("ip_location"),
                    "created_at": c.get("created_at"),
                    "is_hot": c.get("is_hot"),
                    "liked_by_author": c.get("liked_by_author"),
                })
            if not (page or {}).get("has_more"):
                break
            cursor = page.get("cursor")
    return rows

Since creator_hot_videos is already sorted most liked first, videos[:top] is the creator's top five.

For a comment with a big reply_count, GET /v1/raw/kuaishou/comment_replies with video_id and comment_id=<cid> returns the thread (about 12 per page, same cursor rules). Each reply has parent_cid and reply_to (the user being answered). Kuaishou doesn't report a reply total, so just page until has_more is false. I left it out of the main loop to keep the call count predictable.

Step 5: what's trending right now

GET /v1/raw/kuaishou/hot_board takes no parameters and returns the whole main hot board in one call, about 50 entries. Each has rank, keyword, hot_value, views, label (the board's tag, such as 霸榜中 "holding the top"), pinned, word_id and a search url. Pinned entries come first and have no rank.

BOARD_FIELDS = ["checked_at", "rank", "keyword", "hot_value", "views", "label", "pinned"]


def board_rows():
    board = get("/v1/raw/kuaishou/hot_board") or {}
    ts = now()
    return [{"checked_at": ts, **{k: w.get(k) for k in BOARD_FIELDS[1:]}}
            for w in board.get("items", [])]

Snapshot it every hour or so and you have a history of what Kuaishou users were looking at, which is handy context next to a creator's numbers.

Step 6: put it together

import sys


def run(link):
    video = resolve(link)
    uid = video["author"]["uid"]
    videos = hot_videos(uid, max_pages=2)
    append_csv("ks_videos.csv", VIDEO_FIELDS, [video_row(v) for v in videos])
    comments = comment_rows(videos, top=5, pages_per_video=1)
    append_csv("ks_comments.csv", COMMENT_FIELDS, comments)
    append_csv("ks_hot_board.csv", BOARD_FIELDS, board_rows())
    print(f"{len(videos)} hot videos, {len(comments)} comments")
    print(f"spent this run: ${spent:.4f}")


if __name__ == "__main__":
    run(sys.argv[1])

python3 ks_creator.py https://www.kuaishou.com/short-video/3xry7mhn2ywjyqg

Output looks something like this (illustrative, not real data):

权少爱吃小汉堡🍔 (uid 228905802, Kuaishou ID quanshaogmod)
40 hot videos, 150 comments
spent this run: $0.0090

Step 7: summarize with pandas

import pandas as pd

videos = pd.read_csv("ks_videos.csv", dtype={"creator_uid": str, "eid": str})
comments = pd.read_csv("ks_comments.csv", dtype={"eid": str, "cid": str})

v = videos.sort_values("checked_at").drop_duplicates("eid", keep="last")
for uid, g in v.groupby("creator_uid"):
    print(f"\n{g['creator'].iloc[0]} ({len(g)} hot videos)")
    print(f"  median plays:          {g['plays'].median():,.0f}")
    print(f"  median likes:          {g['likes'].median():,.0f}")
    print(f"  median engagement:     {g['engagement_rate'].median():.2%}")
    print(f"  median duration (s):   {g['duration_s'].median():.0f}")
    print("  top hashtags:", g["topics"].dropna().str.split("|").explode().value_counts().head(5).to_dict())

print("\nTop videos by plays:")
print(v.sort_values("plays", ascending=False)[["plays", "likes", "engagement_rate", "caption", "url"]].head(10))

c = comments.drop_duplicates("cid", keep="last")
print("\nWhere commenters are (IP location):")
print(c["ip_location"].value_counts().head(10))
print(f"\nHot comments: {c['is_hot'].fillna(False).astype(bool).mean():.0%}, "
      f"liked by the creator: {c['liked_by_author'].fillna(False).astype(bool).mean():.0%}")
print("\nMost liked comments:")
print(c.sort_values("likes", ascending=False)[["likes", "text"]].head(10))

To compare several creators, run the script once per creator (one video link each); the rows accumulate in the same CSV and the groupby prints one block per creator. Because every row has a checked_at timestamp, running it weekly also lets you see whether a creator's top videos are still gaining plays.

Doing the same from Claude or Cursor

If you'd rather ask questions than maintain a script, the same endpoints are available through Qovia's MCP server at https://api.qovia.ai/mcp (Streamable HTTP, same API key in an Authorization: Bearer header, same billing as the HTTP API). In Claude Code:

claude mcp add --transport http qovia https://api.qovia.ai/mcp \
  --header "Authorization: Bearer $QOVIA_KEY"

Then ask something like "for the creator of https://www.kuaishou.com/short-video/3xry7mhn2ywjyqg, list their ten most liked videos with plays and engagement, and summarize the top comments on the first three." Configs for Cursor, VS Code and Claude Desktop are in the MCP docs, and the Kuaishou MCP page lists the Kuaishou tools. The assistant is good for one-off digging; the script is better when you want the same numbers every week.

Cost

Every Kuaishou endpoint used here (video, creator_hot_videos, comments, comment_replies, hot_board) is a fixed $0.001 per successful call ($1 per 1,000). The optional keyword search in Step 2b (POST /v1/search) is $0.005 per call ($5 per 1,000). Empty results, errors and the 404 for a missing video are not charged.

As a rough example (arithmetic only): one run above is 1 video detail, up to 2 hot-video pages, 5 comment pages and 1 hot board, so at most 9 calls, about $0.009. Checking 20 creators a week is well under $1 a month. The spent counter prints the real number each run. New accounts get 50 trial calls, counted only when a call returns data; after that it's a prepaid balance with no monthly fee. Details are on the pricing page, and every parameter is in the API docs.

The default rate limit is 60 requests per minute per account, more than enough for a job like this.

Limits

Public content only: videos, comments and the hot board.
The Kuaishou endpoints are marked beta.
No Kuaishou-only keyword search: you start from a video link (eid) or a creator's numeric uid. The cross-platform search in Step 2b covers Kuaishou together with other short-video apps.
creator_hot_videos is the creator's hot list, most liked first, not their full upload history, so stats skew toward their best videos.
author.followers can be null on video records; don't build on follower counts from these endpoints.
The video and comment endpoints take the eid (3x...), the creator endpoint takes the numeric uid. Mixing them up gets you a 400.
Counts are a snapshot at the time of the call. Compare snapshots across runs rather than treating one as final.

If you build on this, tell me in the comments what creator metrics you'd want added.

This article was written with the help of AI.

Monitor what Xiaohongshu (RedNote) users say about a brand in Python

Qovia — Fri, 30 Oct 2026 23:15:35 +0000

I build Qovia, the API used in this post. The pipeline itself (search, enrich, collect comments, append to CSV, summarize) works with any Xiaohongshu data source; the CSV and pandas parts are plain Python.

Xiaohongshu (小红书, called RedNote outside China) is where Chinese consumers post reviews, shopping hauls, "is it worth it?" comparisons and complaints. For a consumer brand that sells in China, or one thinking about it, it's one of the most direct places to hear what customers actually say, in their own words, with photos. The comment sections are often more honest than the notes themselves: that's where people ask "where did you buy it?", "mine broke after a week" or "is this an ad?".

In this tutorial we'll build a small brand monitor in Python. For one brand keyword it will:

search recent Xiaohongshu notes that mention the brand,
fetch the full detail of the most engaged notes (exact counts, topics, IP location),
collect the first pages of comments under each of those notes,
append everything to three CSV files with a timestamp,
summarize the result with pandas: daily volume, top notes, hashtags, where commenters are, and a crude keyword tally.

Setup

You need Python 3.9+, requests and pandas:

pip install requests pandas
export QOVIA_KEY=sk-qovia-...   # key from the Qovia console

Get a key from Qovia. Requests authenticate with Authorization: Bearer <key> (an x-api-key header also works). The Xiaohongshu API page lists all the Xiaohongshu endpoints with a real request and response.

Step 1: a small request helper

Every response has the same envelope: the result is in data, and meta reports the charge (meta.billed, meta.charged_usd, meta.empty). Empty results come back as HTTP 200 with meta.empty: true and are not charged. Errors are {"error": {"code", "message", ...}} and are never charged. A 429 carries a Retry-After header, so the helper waits and retries once:

import os
import time
import requests

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

spent = 0.0


def get(path, **params):
    global spent
    params = {k: v for k, v in params.items() if v is not None}
    for attempt in range(2):
        r = requests.get(f"{BASE}{path}", headers=HEADERS, params=params, timeout=60)
        if r.status_code == 429 and attempt == 0:
            time.sleep(int(r.headers.get("Retry-After", "10")))
            continue
        r.raise_for_status()
        body = r.json()
        spent += float((body.get("meta") or {}).get("charged_usd") or 0)
        return body.get("data") or {}

Unknown parameters are rejected with 400 invalid_params, which is why the helper drops None values instead of sending them.

Step 2: search notes that mention the brand

GET /v1/raw/xiaohongshu/search takes a keyword (1 to 50 characters) and returns 20 notes per page. The useful options for monitoring:

sort: general (relevance, the default), latest, most_liked, most_commented or most_saved,
note_type: all (default), video or image,
note_time: all (default), day, week, month or year,
page: 1 to 50; stop when has_more is false.

For a recurring monitor, sort=latest with note_time=week is a good default: you see what's new rather than the same viral notes every day.

Each note has text (an excerpt), author (nickname, user_id), metrics (likes, comments, saves, shares) and an extra object with source_id (the note ID), url, title, note_type, published_at, cover_url, images, and for video notes video_url and video_dur_s.

BRAND = "瑞幸"   # Luckin Coffee; use any brand or product name


def search_notes(keyword, sort="latest", note_time="week", max_pages=2):
    notes, seen = [], set()
    for page_no in range(1, max_pages + 1):
        page = get("/v1/raw/xiaohongshu/search", keyword=keyword, sort=sort,
                   note_time=note_time, page=page_no)
        for n in page.get("items", []):
            nid = n["extra"]["source_id"]
            if nid not in seen:
                seen.add(nid)
                notes.append(n)
        if not page.get("has_more"):
            break
    return notes

Each page is one billed call, so max_pages is also your cost cap. Deduplicating on the note ID is cheap insurance in case the feed shifts between pages.

Now flatten the notes into CSV rows. We'll use one writer for all three files:

import csv
from datetime import datetime, timezone
from pathlib import Path

NOTE_FIELDS = ["checked_at", "keyword", "note_id", "title", "note_type", "author",
               "author_id", "likes", "comments", "saves", "shares", "published_at",
               "url", "excerpt"]


def now():
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def note_row(keyword, n):
    ex, m, a = n.get("extra") or {}, n.get("metrics") or {}, n.get("author") or {}
    return {
        "checked_at": now(),
        "keyword": keyword,
        "note_id": ex.get("source_id"),
        "title": ex.get("title"),
        "note_type": ex.get("note_type"),
        "author": a.get("nickname"),
        "author_id": a.get("user_id"),
        "likes": m.get("likes"),
        "comments": m.get("comments"),
        "saves": m.get("saves"),
        "shares": m.get("shares"),
        "published_at": ex.get("published_at"),
        "url": ex.get("url"),
        "excerpt": (n.get("text") or "")[:200].replace("\n", " "),
    }


def append_csv(path, fields, rows):
    path = Path(path)
    new_file = not path.exists()
    with path.open("a", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=fields)
        if new_file:
            w.writeheader()
        w.writerows(rows)

Step 3: note details for the notes that matter

Search results are enough for counting mentions, but for the notes people actually engage with you want the detail record. GET /v1/raw/xiaohongshu/note takes a note_id (the 24 hex characters in xiaohongshu.com/explore/<ID>, which is extra.source_id from search) and returns the same record shape with exact like, comment, save and share counts, plus extra.topics (the note's hashtags) and extra.ip_location. A deleted or missing note answers empty and is not charged.

We rank the search hits by engagement and fetch details for the top few:

DETAIL_FIELDS = NOTE_FIELDS + ["ip_location", "topics"]


def engagement(n):
    m = n.get("metrics") or {}
    return sum(m.get(k) or 0 for k in ("likes", "comments", "saves", "shares"))


def note_details(keyword, notes, top=10):
    rows = []
    for n in sorted(notes, key=engagement, reverse=True)[:top]:
        d = get("/v1/raw/xiaohongshu/note", note_id=n["extra"]["source_id"])
        if not d:
            continue  # deleted since the search, not charged
        row = note_row(keyword, d)
        row["ip_location"] = d["extra"].get("ip_location")
        row["topics"] = "|".join(d["extra"].get("topics") or [])
        rows.append(row)
    return rows

topics is a list, so it's joined with | to fit in one CSV cell; pandas can split it again later.

Step 4: what are commenters saying?

GET /v1/raw/xiaohongshu/comments takes a note_id and returns first-level comments with a cursor. There are two orders, and they return different fields:

sort=default: Xiaohongshu's own order (pinned and popular first), about 5 per page, with like counts and the commenter's user ID.
sort=latest: newest first, 10 to 15 per page, without like counts and with only the commenter's Xiaohongshu ID (red_id).

For brand monitoring the default order is usually the better start, since the most liked comments tend to be the ones everybody else agrees with. Each comment has cid, text, likes, reply_count, ip_location, created_at, images and author; the page has cursor, has_more and total. Pass cursor back unchanged for the next page:

COMMENT_FIELDS = ["checked_at", "keyword", "note_id", "note_title", "cid", "text",
                  "likes", "reply_count", "ip_location", "created_at"]


def comment_rows(keyword, detail_rows, pages_per_note=2):
    rows = []
    for note in detail_rows:
        cursor = None
        for _ in range(pages_per_note):
            page = get("/v1/raw/xiaohongshu/comments", note_id=note["note_id"],
                       sort="default", cursor=cursor)
            for c in page.get("items", []):
                rows.append({
                    "checked_at": now(),
                    "keyword": keyword,
                    "note_id": note["note_id"],
                    "note_title": note["title"],
                    "cid": c["cid"],
                    "text": (c.get("text") or "").replace("\n", " "),
                    "likes": c.get("likes"),
                    "reply_count": c.get("reply_count"),
                    "ip_location": c.get("ip_location"),
                    "created_at": c.get("created_at"),
                })
            if not page.get("has_more"):
                break
            cursor = page.get("cursor")
    return rows

If a comment has a large reply_count, GET /v1/raw/xiaohongshu/comment_replies with the note_id and comment_id=<cid> returns the thread under it (oldest first, about 5 per page, same cursor rules). Each reply has parent_cid and reply_to_cid, so you can rebuild the conversation. I left it out of the main loop to keep the call count predictable.

Step 5: put it together

def run(keyword=BRAND):
    notes = search_notes(keyword, sort="latest", note_time="week", max_pages=2)
    append_csv("xhs_notes.csv", NOTE_FIELDS, [note_row(keyword, n) for n in notes])
    details = note_details(keyword, notes, top=10)
    append_csv("xhs_note_details.csv", DETAIL_FIELDS, details)
    comments = comment_rows(keyword, details, pages_per_note=2)
    append_csv("xhs_comments.csv", COMMENT_FIELDS, comments)
    print(f"{len(notes)} notes, {len(details)} details, {len(comments)} comments")
    print(f"spent this run: ${spent:.4f}")


if __name__ == "__main__":
    run()

Run it once a day from cron and the CSVs become a running log of the brand on Xiaohongshu:

0 9 * * * cd /path/to/monitor && QOVIA_KEY=sk-qovia-... python3 xhs_monitor.py >> monitor.log 2>&1

Output looks something like this (illustrative, not real data):

40 notes, 10 details, 96 comments
spent this run: $0.1600

Step 6: summarize with pandas

import pandas as pd

notes = pd.read_csv("xhs_notes.csv", dtype={"note_id": str})
details = pd.read_csv("xhs_note_details.csv", dtype={"note_id": str})
comments = pd.read_csv("xhs_comments.csv", dtype={"note_id": str, "cid": str})

notes = notes.drop_duplicates("note_id", keep="last")
notes["day"] = pd.to_datetime(notes["published_at"], utc=True).dt.tz_convert("Asia/Shanghai").dt.date
print("New notes per day:")
print(notes.groupby("day").size().tail(14))

latest = details.sort_values("checked_at").drop_duplicates("note_id", keep="last")
print("\nMost engaged notes:")
print(latest.sort_values("likes", ascending=False)[["likes", "comments", "saves", "title", "url"]].head(10))

print("\nTop hashtags on those notes:")
print(latest["topics"].dropna().str.split("|").explode().value_counts().head(10))

c = comments.drop_duplicates("cid", keep="last")
print("\nWhere commenters are (IP location):")
print(c["ip_location"].value_counts().head(10))

print("\nMost liked comments:")
print(c.sort_values("likes", ascending=False)[["likes", "text"]].head(10))

# A crude signal, not sentiment analysis: count comments containing a few words.
WATCH = {"好喝": "tastes good", "难喝": "tastes bad", "贵": "expensive",
         "排队": "queue", "退款": "refund", "广告": "ad?"}
print("\nComments mentioning:")
for word, label in WATCH.items():
    print(f"  {word} ({label}): {c['text'].fillna('').str.contains(word, regex=False).sum()}")

The word list is the part you'll want to tune for your brand: product names, competitor names, and the complaint words that matter in your category. Because every run appends with a checked_at timestamp, you can also compare a note's likes and comments across days to see which posts are still growing.

Doing the same from Claude or Cursor

claude mcp add --transport http qovia https://api.qovia.ai/mcp \
  --header "Authorization: Bearer $QOVIA_KEY"

Then ask something like "search Xiaohongshu for 瑞幸 notes from this week, open the five most liked, and summarize what the comments complain about." Configs for Cursor, VS Code and Claude Desktop are in the MCP docs, and the Xiaohongshu MCP page lists the Xiaohongshu tools. The assistant is good for one-off digging; the script is better when you want the same numbers every day.

Cost

Every Xiaohongshu endpoint used here (search, note, comments, comment_replies) is a fixed $0.005 per successful call ($5 per 1,000). Empty results and errors are not charged.

As a rough example (arithmetic only): one run above is up to 2 search pages, 10 note details and up to 20 comment pages, so at most 32 calls, about $0.16. Daily for one brand, that's under $5 a month. The spent counter prints the real number each run. New accounts get 50 trial calls, counted only when a call returns data; after that it's a prepaid balance with no monthly fee. Details are on the pricing page, and every parameter is in the API docs.

The default rate limit is 60 requests per minute per account, more than enough for a job like this.

Limits

Public content only: notes, comments and public profiles.
The Xiaohongshu endpoints are marked beta.
Search returns 20 notes per page and up to 50 pages; text in search results is an excerpt.
With sort=latest, comments have no like counts and the commenter has only a red_id, so the like-based summaries above need sort=default.
Default-order comments come about 5 per page, so deep comment threads cost more calls; cap pages_per_note.
Counts are a snapshot at the time of the call. Compare snapshots across runs rather than treating one as final.
The keyword tally is not sentiment analysis. It's a quick way to spot spikes worth reading by hand.

If you build on this, tell me in the comments which brand signals you'd want added.

This article was written with the help of AI.

Track what Zhihu is saying about any topic in Python (search, answers, comments to CSV)

Qovia — Fri, 30 Oct 2026 20:31:18 +0000

I build Qovia, the API used in this post. The structure (search a keyword, group the hits by question, pull the question stats and the top comments, store everything locally) works with any Zhihu data source; the CSV and pandas parts are plain Python.

Zhihu (知乎) is China's big Q&A site, closer to Quora plus a blogging platform than to a social feed. When a new phone, an AI model or a policy change lands, Zhihu is where people write long answers about whether it's any good, and where the comment threads under those answers show what ordinary readers think. If you do market research, product feedback or China-focused analysis, it's one of the most useful places to listen.

In this tutorial we'll write a small Python script that, for one keyword:

checks whether the topic is on Zhihu's hot list right now,
searches recent answers and articles about it and saves them to a CSV,
groups the answers by question and pulls each question's answer, follower and view counts,
collects the hottest comments under the most upvoted answers, with likes and IP location,
summarizes the result with pandas: which questions drive the discussion, who is answering, and where commenters are.

Setup

You need Python 3.9+ and requests (and pandas for the last step):

pip install requests pandas
export QOVIA_KEY=sk-qovia-...   # key from the Qovia console

Get a key from Qovia. Requests authenticate with Authorization: Bearer <key> (an x-api-key header also works). An overview of all Zhihu endpoints is on the Zhihu API page.

Step 1: one helper

Every response uses the same envelope: the result is in data, billing info is in meta (meta.charged_usd). Calls that come back empty are not charged. A 429 response carries a Retry-After header, so the helper waits and retries once:

import os
import time
import requests

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

spent = 0.0


def get(path, **params):
    global spent
    for attempt in range(2):
        r = requests.get(f"{BASE}{path}", headers=HEADERS, params=params, timeout=60)
        if r.status_code == 429 and attempt == 0:
            time.sleep(int(r.headers.get("Retry-After", "10")))
            continue
        r.raise_for_status()
        body = r.json()
        spent += float(body.get("meta", {}).get("charged_usd") or 0)
        return body.get("data") or {}

All the Zhihu endpoints return records in the same shape: text, author, metrics, and an extra object with the Zhihu-specific fields (source_id, content_type, title, url, published_at and so on). Once you've seen one, you've seen them all.

Step 2: is it on the hot list?

GET /v1/raw/zhihu/hot_list takes no parameters and returns about 30 questions in rank order. Each item's extra has rank, heat (the 「1848 万热度」 label already converted to a number, 18480000), heat_label, title and url; extra.source_id is the question id. metrics has answers and followers (views and comments are null on this list).

A quick check for our keyword:

KEYWORD = "DeepSeek"


def hot_matches(keyword):
    hot = get("/v1/raw/zhihu/hot_list")
    hits = []
    for q in hot.get("items", []):
        ex = q["extra"]
        if keyword.lower() in (q.get("text") or "").lower():
            hits.append(ex)
            print(f"#{ex['rank']} {ex['heat_label']}  {ex['title']}  {ex['url']}")
    return hits

I match on text because it holds the question title followed by the start of its description, so it catches a few more hits than the title alone. Most keywords won't be on the hot list most of the time; that's fine, it's a cheap "is this blowing up right now?" signal before the real work.

Step 3: search answers and articles

GET /v1/raw/zhihu/article_search is the core of the tracker. It takes a keyword and optionally:

type: all (default), answer or article,
sort: general (relevance, default), most_liked or latest,
publish_time: all (default), day, week, month, quarter, half_year or year. Zhihu applies this filter loosely, so expect a few older results.

Each result has text, author (nickname, followers, url_token), metrics (likes, comments, saves) and extra (source_id, content_type, title, question_id, published_at, url). For an answer, extra.question_id is the question it belongs to; for an article it's null.

To get more than one page, pass the previous page's cursor and search_id back and stop when has_more is false. Adjacent pages can repeat a few records, so we deduplicate by id. Each page is one billed call, so cap it:

def search(keyword, publish_time="week", sort="general", max_pages=3):
    results, seen = [], set()
    params = {"keyword": keyword, "publish_time": publish_time, "sort": sort}
    for _ in range(max_pages):
        page = get("/v1/raw/zhihu/article_search", **params)
        for item in page.get("items", []):
            if item["id"] not in seen:
                seen.add(item["id"])
                results.append(item)
        if not page.get("has_more"):
            break
        params["cursor"] = page.get("cursor")
        params["search_id"] = page.get("search_id")
    return results

Now flatten the results into rows and write them to a CSV:

import csv
from datetime import datetime, timezone
from pathlib import Path

POST_FIELDS = ["checked_at", "keyword", "id", "content_type", "answer_id", "question_id",
               "title", "author", "author_url_token", "author_followers",
               "likes", "comments", "saves", "published_at", "url", "excerpt"]


def post_row(keyword, item):
    ex, m, a = item["extra"], item.get("metrics") or {}, item.get("author") or {}
    return {
        "checked_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "keyword": keyword,
        "id": item["id"],
        "content_type": ex.get("content_type"),
        "answer_id": ex.get("source_id"),
        "question_id": ex.get("question_id"),
        "title": ex.get("title"),
        "author": a.get("nickname"),
        "author_url_token": a.get("url_token"),
        "author_followers": a.get("followers"),
        "likes": m.get("likes"),
        "comments": m.get("comments"),
        "saves": m.get("saves"),
        "published_at": ex.get("published_at"),
        "url": ex.get("url"),
        "excerpt": (item.get("text") or "")[:200].replace("\n", " "),
    }


def append_csv(path, fields, rows):
    path = Path(path)
    new_file = not path.exists()
    with path.open("a", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=fields)
        if new_file:
            w.writeheader()
        w.writerows(rows)

The answer_id column is extra.source_id, the number in the answer's URL. For articles it's the article id instead; only answers have a question_id, and only answers can be passed to the comments endpoint later.

Step 4: which questions carry the discussion?

On Zhihu the unit of conversation is the question: one question can collect hundreds of answers. Grouping the search hits by question_id shows which questions your topic actually lives in. For those, GET /v1/raw/zhihu/question takes a question_id and returns the question's metrics: answers, followers, views and comments, plus extra.title, extra.published_at and extra.url. A question that doesn't exist answers 404 (not charged), which our helper raises as an HTTPError, so we skip those:

from collections import Counter

Q_FIELDS = ["checked_at", "keyword", "question_id", "title", "answers", "followers",
            "views", "comments", "hits_in_search", "published_at", "url"]


def question_rows(keyword, posts, top=10):
    counts = Counter(p["extra"]["question_id"] for p in posts if p["extra"].get("question_id"))
    rows = []
    for qid, hits in counts.most_common(top):
        try:
            q = get("/v1/raw/zhihu/question", question_id=qid)
        except requests.HTTPError:
            continue
        if not q:
            continue
        m, ex = q.get("metrics") or {}, q.get("extra") or {}
        rows.append({
            "checked_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
            "keyword": keyword,
            "question_id": qid,
            "title": ex.get("title"),
            "answers": m.get("answers"),
            "followers": m.get("followers"),
            "views": m.get("views"),
            "comments": m.get("comments"),
            "hits_in_search": hits,
            "published_at": ex.get("published_at"),
            "url": ex.get("url"),
        })
    return rows

If you want more answers from one question than search gave you, GET /v1/raw/zhihu/answers takes a question_id, a sort (default, Zhihu's ranking, or latest) and a count up to 20. It only returns that first page (Zhihu's next page isn't reachable through the current source), but the records have the same shape as search results, so post_row() works on them unchanged:

def latest_answers(keyword, question_id, count=20):
    page = get("/v1/raw/zhihu/answers", question_id=question_id, sort="latest", count=count)
    return [post_row(keyword, a) for a in page.get("items", [])]

Step 5: what are readers saying?

Answers are written by people with an opinion to sell; comments are where readers push back. GET /v1/raw/zhihu/comments takes an answer_id and returns first-level comments, 10 per page, hottest first by default (sort=latest for newest first). Each comment has cid, text, likes, reply_count, ip_location, created_at, is_hot and author. The response also has total, cursor and has_more.

We take the first page of hot comments under the most upvoted answers:

C_FIELDS = ["checked_at", "keyword", "answer_id", "question_title", "cid", "text",
            "likes", "reply_count", "ip_location", "is_hot", "created_at"]


def comment_rows(keyword, posts, top_answers=10):
    answers = [p for p in posts if p["extra"].get("content_type") == "answer"]
    answers.sort(key=lambda p: (p.get("metrics") or {}).get("likes") or 0, reverse=True)
    rows = []
    for p in answers[:top_answers]:
        aid = p["extra"]["source_id"]
        page = get("/v1/raw/zhihu/comments", answer_id=aid, sort="hot")
        for c in page.get("items", []):
            rows.append({
                "checked_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
                "keyword": keyword,
                "answer_id": aid,
                "question_title": p["extra"].get("title"),
                "cid": c["cid"],
                "text": (c.get("text") or "").replace("\n", " "),
                "likes": c.get("likes"),
                "reply_count": c.get("reply_count"),
                "ip_location": c.get("ip_location"),
                "is_hot": c.get("is_hot"),
                "created_at": c.get("created_at"),
            })
    return rows

To go deeper on one answer, follow cursor until has_more is false, exactly like the search loop. And when a comment has a large reply_count, GET /v1/raw/zhihu/comment_replies with comment_id=<cid> returns the thread under it (up to 20 per page, sort is hot or oldest). Each reply has a parent_cid and a reply_to_cid, so you can rebuild who answered whom.

Step 6: put it together

def run(keyword=KEYWORD):
    hot_matches(keyword)
    posts = search(keyword, publish_time="week", max_pages=3)
    append_csv("zhihu_posts.csv", POST_FIELDS, [post_row(keyword, p) for p in posts])
    questions = question_rows(keyword, posts, top=10)
    append_csv("zhihu_questions.csv", Q_FIELDS, questions)
    comments = comment_rows(keyword, posts, top_answers=10)
    append_csv("zhihu_comments.csv", C_FIELDS, comments)
    print(f"{len(posts)} posts, {len(questions)} questions, {len(comments)} comments")
    print(f"spent this run: ${spent:.4f}")


if __name__ == "__main__":
    run()

Run it once a day from cron and the CSVs become a daily log of the topic:

0 9 * * * cd /path/to/tracker && QOVIA_KEY=sk-qovia-... python3 tracker.py >> tracker.log 2>&1

The output looks something like this (illustrative, not real data):

#7 412 万热度  <question title>  https://www.zhihu.com/question/...
58 posts, 10 questions, 96 comments
spent this run: $0.0240

Step 7: summarize it

With the three CSVs in place, a few lines of pandas give you the shape of the discussion:

import pandas as pd

posts = pd.read_csv("zhihu_posts.csv", dtype={"question_id": str, "answer_id": str})
questions = pd.read_csv("zhihu_questions.csv", dtype={"question_id": str})
comments = pd.read_csv("zhihu_comments.csv", dtype={"answer_id": str})

latest_q = questions.sort_values("checked_at").drop_duplicates("question_id", keep="last")
print("Questions driving the topic:")
print(latest_q.sort_values("views", ascending=False)[["title", "answers", "followers", "views"]].head(10))

print("\nMost upvoted answers and articles:")
top = posts.drop_duplicates("id").sort_values("likes", ascending=False)
print(top[["content_type", "author", "likes", "comments", "saves", "title"]].head(10))

print("\nWhere commenters are (IP location):")
print(comments.drop_duplicates("cid")["ip_location"].value_counts().head(10))

print("\nMost liked comments:")
print(comments.drop_duplicates("cid").sort_values("likes", ascending=False)[["likes", "text"]].head(10))

Because every run appends with a checked_at timestamp, you can also compare a question's views and answers from one day to the next to see whether the topic is heating up or fading.

If you want to know who the loudest voices are, GET /v1/raw/zhihu/user takes an author's url_token (already in zhihu_posts.csv) and returns followers, answers, articles, verified, verify_reason, is_org and employers. That's a quick way to separate verified professionals and organisation accounts from everyone else.

Cost

From the catalog: every Zhihu endpoint used here (hot_list, article_search, question, answers, comments, comment_replies, user) is $1 per 1,000 calls ($0.001 per call). Empty results and 404s are not charged.

As a rough example (arithmetic only): one run of the script above is 1 hot list call, up to 3 search pages, up to 10 question calls and up to 10 comment pages, so at most 24 calls, about $0.024. Running it daily for one keyword is under $1 a month. Print the spent counter on every run so you see the real number. Details are on the pricing page, and every endpoint's parameters are in the API reference.

The default rate limit is 60 requests per minute per account, which is plenty for this kind of job.

Using it from an AI assistant

The same endpoints are available through Qovia's MCP server, so you can ask Claude, Cursor or another MCP client things like "what is Zhihu saying about DeepSeek this week, and what do the top comments complain about?" without writing the loop yourself. Setup is in the MCP docs, and there's a Zhihu-specific MCP page that lists the Zhihu endpoints.

Limits

Public content only: answers, articles, questions, comments and public profiles. No private messages.
The Zhihu endpoints are marked beta in the catalog.
answers returns only the first page (up to 20) for a question; use search with sort=latest to keep finding new answers over time.
The publish_time filter on search is applied loosely by Zhihu, so filter on published_at yourself if you need an exact window.
GET /v1/raw/zhihu/answer (single answer detail) doesn't carry upvote, comment or save counts; take those from search or answers.
Comments are only available for answers (by answer_id), not for articles.
Numbers are a snapshot at the time of the call; compare snapshots across runs rather than treating one as final.

If you build on this, tell me in the comments what you'd want added.

This article was written with the help of AI.

Monitor WeChat Official Accounts in Python (new articles and read counts)

Qovia — Fri, 30 Oct 2026 18:57:29 +0000

I build Qovia, the API used in this post. The structure (find the account, poll its article list, snapshot metrics, store them locally) works with any data source for WeChat Official Accounts; the storage and diffing parts are plain Python.

WeChat Official Accounts (公众号) are where a lot of Chinese companies, media outlets and analysts actually publish. Unlike Weibo or Douyin, there is no public web search page you can point a scraper at, so people who need to follow a handful of accounts usually end up checking them by hand on their phone.

In this tutorial we'll write a small Python script that:

looks up an official account by name and gets its id,
polls the account's newest articles and spots the ones we haven't seen,
fetches reads, likes, "watching" (在看), comments and shares for each new article,
appends every snapshot to a CSV, so you can see how an article's numbers grow over time.

At the end there's a short section on searching WeChat articles by topic across all accounts, instead of following specific ones.

Setup

You need Python 3.9+ and requests:

pip install requests
export QOVIA_KEY=sk-qovia-...   # key from the Qovia console

Get a key from Qovia. Requests authenticate with Authorization: Bearer <key> (an x-api-key header also works).

Step 1: one helper

import os
import time
import requests

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

spent = 0.0


def get(path, **params):
    global spent
    for attempt in range(2):
        r = requests.get(f"{BASE}{path}", headers=HEADERS, params=params, timeout=60)
        if r.status_code == 429 and attempt == 0:
            time.sleep(int(r.headers.get("Retry-After", "10")))
            continue
        r.raise_for_status()
        body = r.json()
        spent += float(body.get("meta", {}).get("charged_usd") or 0)
        return body.get("data") or {}

Step 2: find the account id

The endpoints that list articles need the account's mp_id (it starts with gh_). GET /v1/raw/wechat/account_search takes a keyword (the account name or part of it) and returns one page of matching accounts. kind picks subscription (the default), service, or all:

accounts = get("/v1/raw/wechat/account_search", keyword="央视新闻", kind="all")

for acc in accounts.get("items", []):
    print(acc["mp_id"], acc["name"], acc["type"], acc.get("verified_by"))

Each item also has wechat_id, description, avatar and verified. Several accounts can share a similar name, so check verified_by (the organization behind a verified account) before you pick one. You only need to do this lookup once per account; save the mp_id in your config.

Step 3: poll the newest articles

GET /v1/raw/wechat/articles takes an mp_id and returns that account's articles, newest first, one page per call. Each item has title, url, published_at, digest, author, ip_region, is_original, is_paid, and in the documented example also reads_label and likes.

For monitoring, the first page is usually all you need: if you poll often enough, new articles will always be on it. We keep the URLs we've already seen in a JSON file:

import json
from pathlib import Path

SEEN_FILE = Path("seen.json")


def load_seen():
    return set(json.loads(SEEN_FILE.read_text())) if SEEN_FILE.exists() else set()


def save_seen(seen):
    SEEN_FILE.write_text(json.dumps(sorted(seen), ensure_ascii=False))


def new_articles(mp_id, seen):
    page = get("/v1/raw/wechat/articles", mp_id=mp_id)
    fresh = [a for a in page.get("items", []) if a["url"] not in seen]
    return fresh

I key on the article url because it's what the metrics endpoint takes anyway.

To backfill an account's history once (for example the first time you add it), follow the cursor: pass the previous page's cursor back unchanged and stop when has_more is false. Each page is one billed call, so cap it:

def backfill(mp_id, max_pages=5):
    items, cursor = [], None
    for _ in range(max_pages):
        params = {"mp_id": mp_id}
        if cursor:
            params["cursor"] = cursor
        page = get("/v1/raw/wechat/articles", **params)
        items.extend(page.get("items", []))
        if not page.get("has_more"):
            break
        cursor = page.get("cursor")
    return items

Step 4: snapshot the numbers

GET /v1/raw/wechat/article_metrics takes the article's url (the mp.weixin.qq.com link, full or /s/ short form) and returns reads, likes, watching, comments, shares, plus updated_at, the time the numbers were taken. You're charged only when metrics come back.

We append one row per snapshot to a CSV:

import csv
from datetime import datetime, timezone

CSV_FILE = Path("wechat_metrics.csv")
FIELDS = ["checked_at", "account", "title", "url", "published_at",
          "reads", "likes", "watching", "comments", "shares", "updated_at"]


def snapshot(account_name, article):
    m = get("/v1/raw/wechat/article_metrics", url=article["url"])
    if not m:
        return None
    row = {
        "checked_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "account": account_name,
        "title": article["title"],
        "url": article["url"],
        "published_at": article.get("published_at"),
        **{k: m.get(k) for k in ["reads", "likes", "watching", "comments", "shares", "updated_at"]},
    }
    new_file = not CSV_FILE.exists()
    with CSV_FILE.open("a", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=FIELDS)
        if new_file:
            w.writeheader()
        w.writerow(row)
    return row

Step 5: put it together

The watch list maps a label to the mp_id you found in step 2. The ids below are placeholders; replace them with real ones.

WATCH = {
    "Account A": "gh_xxxxxxxxxxxx",
    "Account B": "gh_yyyyyyyyyyyy",
}


def run():
    seen = load_seen()
    for name, mp_id in WATCH.items():
        for art in new_articles(mp_id, seen):
            row = snapshot(name, art)
            print(f"[{name}] {art['title'][:40]}  {art.get('published_at')}")
            if row:
                print(f"    reads={row['reads']} likes={row['likes']} watching={row['watching']}")
            seen.add(art["url"])
    save_seen(seen)
    print(f"spent this run: ${spent:.4f}")


if __name__ == "__main__":
    run()

(The growth section below also uses pandas: pip install pandas.)

Run it from cron, for example every hour:

0 * * * * cd /path/to/monitor && QOVIA_KEY=sk-qovia-... python3 monitor.py >> monitor.log 2>&1

The output looks something like this (illustrative, not real data):

[Account A] <article title>  2026-10-30T18:05:00+08:00
    reads=5321 likes=88 watching=12
spent this run: $0.0150

Note the first run treats every article on the first page as new. If you only want alerts going forward, run it once and discard the output, or pre-fill seen.json from backfill().

Tracking growth

A single snapshot right after publishing isn't very interesting: most reads come in over the first day or two. To get a curve, re-snapshot recent articles on later runs. A simple rule: for every URL in the CSV whose first snapshot is less than 48 hours old, call snapshot() again on each run. Then, with pandas:

import pandas as pd

df = pd.read_csv("wechat_metrics.csv", parse_dates=["checked_at"])
growth = (df.sort_values("checked_at")
            .groupby(["account", "title"])["reads"]
            .agg(["first", "last"]))
growth["gained"] = growth["last"] - growth["first"]
print(growth.sort_values("gained", ascending=False).head(10))

This tells you which articles from the accounts you follow are still picking up readers, which is often a better signal than the headline number.

Searching by topic instead of by account

Sometimes you don't know which accounts matter; you want every WeChat article about a topic. POST /v1/search is a cross-source search that covers WeChat official accounts among other sources; restrict it with platforms: ["wechat"]. Dates are whole days in Beijing time (the last 7 days by default, at most 90), and limit is 10 or 20 per page:

r = requests.post(
    f"{BASE}/v1/search",
    headers=HEADERS,
    json={
        "query": "人工智能 教育",          # space-separated terms are all required
        "platforms": ["wechat"],
        "since": "2026-10-24",
        "until": "2026-10-30",
        "limit": 20,
        "page": 1,
    },
    timeout=60,
)
r.raise_for_status()
data = r.json()["data"]
print("total hits:", data["total"])
for item in data["items"]:
    print(item["published_at"], item["author"]["name"], item["title"][:50], item["url"])

Other useful fields in the request body: exclude_words (drop results containing these words), author (only one account's content, by name) and sentiment (negative, neutral, positive). The author.name values in the results are also a quick way to discover accounts worth adding to your watch list with step 2.

Cost

From the catalog: account_search, articles and article_metrics are each $5 per 1,000 calls ($0.005 per call), and POST /v1/search is also $5 per 1,000 calls. Empty results are not charged.

As a rough example (arithmetic only): watching 5 accounts hourly is 120 articles calls a day, about $0.60, plus one article_metrics call per new article and per re-snapshot. Print the spent counter on every run so you see the real number. Details are on the pricing page, and every endpoint's parameters are in the API reference.

The default rate limit is 60 requests per minute per account, which is plenty for this kind of polling.

Using it from an AI assistant

The same endpoints are available through Qovia's MCP server, so you can ask Claude, Cursor or another MCP client things like "what did these three accounts publish this week, and which article got the most reads?" without writing the loop yourself. Setup is in the MCP docs.

Limits

Public official-account content only. No private messages, no follower lists.
The WeChat endpoints are marked beta in the catalog.
Numbers are a snapshot at updated_at, not a live counter; compare snapshots, don't treat one as final.
account_search returns a single page, so very generic keywords may not surface the account you want. Search with its exact name.

If you build on this, tell me in the comments what you'd want added.

This article was written with the help of AI.

Comparing Bilibili and YouTube comments on the same topic with Python

Qovia — Fri, 30 Oct 2026 07:49:21 +0000

I build Qovia, the API used in this post. The method here (search a topic on each platform, pull comments, compare) is generic; Qovia is just where the JSON comes from.

The same topic gets discussed very differently on Bilibili and YouTube: different audiences, different languages, different comment cultures. In this tutorial we'll pick one topic, find a popular video about it on each platform, pull the comments, and compare them side by side in a small pandas table. We won't claim anything about the platforms in general; one topic and a handful of videos is a sample, not a study.

Setup

pip install requests pandas
export QOVIA_KEY=sk-qovia-...

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

import os
import requests

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


def get(path, **params):
    global spent
    r = requests.get(f"{BASE}{path}", headers=HEADERS, params=params, timeout=60)
    r.raise_for_status()
    body = r.json()
    spent += float(body.get("meta", {}).get("charged_usd") or 0)
    return body["data"] or {}

Step 1: find one video per platform

A topic needs a keyword in each platform's main language. I'll use TSMC as the example: 台积电 on Bilibili and TSMC on YouTube.

Bilibili search (GET /v1/raw/bilibili/search) takes keyword, page, page_size and order (totalrank, click, pubdate, dm, stow or scores). Results carry the BV id in extra.source_id:

b = get("/v1/raw/bilibili/search", keyword="台积电", order="click", page_size=10)
for i, v in enumerate(b["items"]):
    print(i, v["metrics"]["plays"], v["metrics"]["comments"], v["extra"]["title"])
bvid = b["items"][0]["extra"]["source_id"]

YouTube search (GET /v1/raw/youtube/search) takes keyword, upload_date (hour, today, week, month, year; the catalog says the default is this month) and a cursor. The video id is also in extra.source_id:

y = get("/v1/raw/youtube/search", keyword="TSMC", upload_date="year")
for i, v in enumerate(y["items"]):
    print(i, v["metrics"]["plays"], v["extra"]["title"])
video_id = y["items"][0]["extra"]["source_id"]

Note that the two search sorts are not the same thing: here Bilibili is ordered by plays, while YouTube's results are in the platform's own order within the upload-date window. Choose videos by hand if you want a fair comparison, and write down how you chose them.

Step 2: video-level numbers

Video detail gives engagement counts. The catalog lists Bilibili's as plays, likes, coins, saves, comments, shares and danmaku (BV ids only), and YouTube's as likes, comments and views, plus channel subscribers.

bv = get("/v1/raw/bilibili/video", bvid=bvid)
yv = get("/v1/raw/youtube/video", video_id=video_id)
print(bv["metrics"])
print(yv["metrics"])

Bilibili's coins and danmaku (on-screen bullet comments) have no YouTube equivalent, so only compare the fields both sides have.

Step 3: pull comments

Both comment endpoints return items, a cursor and has_more:

GET /v1/raw/bilibili/comments: bvid, cursor. Top-level comments, hottest first, 20 per page. Pass the previous cursor back unchanged. IP location is not available.
GET /v1/raw/youtube/comments: video_id, sort (top or newest), cursor. About 20 per page.

def pages(path, id_param, id_value, max_pages=5, **extra):
    out, cursor = {}, None
    for _ in range(max_pages):
        params = {id_param: id_value, **extra}
        if cursor:
            params["cursor"] = cursor
        page = get(path, **params)
        for c in page.get("items", []):
            out[c["cid"]] = c
        if not page.get("has_more"):
            break
        cursor = page["cursor"]
    return list(out.values())


bili = pages("/v1/raw/bilibili/comments", "bvid", bvid)
yt = pages("/v1/raw/youtube/comments", "video_id", video_id, sort="top")

Both sides use top/hottest-first ordering here, so we are comparing the most visible comments, not a random sample. That choice shapes every number below.

Step 4: compare

Each comment has text, likes, reply_count and created_at on both platforms, so a shared table is easy:

import pandas as pd

def frame(comments, platform):
    df = pd.DataFrame(
        {"platform": platform, "text": c["text"], "likes": c["likes"], "replies": c["reply_count"]}
        for c in comments
    )
    df["chars"] = df["text"].str.len()
    return df

df = pd.concat([frame(bili, "bilibili"), frame(yt, "youtube")])
print(df.groupby("platform")[["likes", "replies", "chars"]].agg(["count", "median", "max"]))

Character counts are not directly comparable across languages (a Chinese character carries more meaning than an English one), so read chars as a rough length indicator within a platform, not across them. For content-level comparison, the next step is to run the same analysis on both: translate or use a multilingual model for sentiment, extract the most common terms, or just read the top ten by likes on each side:

for name, g in df.groupby("platform"):
    print(f"\n{name}")
    print(g.sort_values("likes", ascending=False).head(10)[["likes", "text"]].to_string(index=False))

Illustrative output shape (not real data):

          likes                 replies               chars
          count median max      count median max      count median max
platform
bilibili     <n>    <x>  <y>      <n>    <x>  <y>      <n>    <x>  <y>
youtube      <n>    <x>  <y>      <n>    <x>  <y>      <n>    <x>  <y>

Caveats worth writing in your own report

Sample size and ordering. A few pages of top comments from one video per platform says little about either platform. Use several videos per side.
Time. YouTube comment times are UTC, converted from YouTube's relative "3 days ago" and only as precise as the unit named; Bilibili's are Beijing time. Don't compare timing at fine granularity.
Language. Any text comparison needs translation or a multilingual model, with its own error.
Fields differ. Bilibili has no comment IP location here; YouTube comments carry a creator-heart flag. Compare shared fields only.

Cost

Per the catalog: Bilibili search, video and comments are $0.5 per 1,000 calls each; YouTube search, video and comments are $1 per 1,000 calls each. Calls with no results aren't charged. The spent counter above adds up meta.charged_usd so you can see what each run cost. More on the pricing page.

This article was written with the help of AI.

Douyin comments sentiment analysis in Python

Qovia — Fri, 30 Oct 2026 07:38:55 +0000

I build Qovia, the API used in this post. The approach (fetch comments as JSON, score them locally) works with any source that gives you Douyin comments, and the scoring part is plain Python.

Douyin is the Chinese version of TikTok, and its comment sections are where people say what they think of a product, a campaign or a news item. In this tutorial we'll write a script that:

searches Douyin for a keyword and picks a video,
pages through that video's comments,
scores each comment's sentiment locally,
prints a summary, with the most-liked positive and negative comments.

Setup

You need Python 3.9+, requests and snownlp (a small Chinese-text sentiment library):

pip install requests snownlp
export QOVIA_KEY=sk-qovia-...   # key from the Qovia console

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

Step 1: one helper

Responses come in an envelope: the result is in data, billing info is in meta (meta.charged_usd). One helper covers everything:

import os
import requests

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

spent = 0.0


def get(path, **params):
    global spent
    r = requests.get(f"{BASE}{path}", headers=HEADERS, params=params, timeout=60)
    r.raise_for_status()
    body = r.json()
    spent += float(body.get("meta", {}).get("charged_usd") or 0)
    return body["data"] or {}

Step 2: find a video

GET /v1/raw/douyin/search takes a keyword and returns videos and image posts. The video id you need for comments is in extra.source_id:

results = get("/v1/raw/douyin/search", keyword="新能源汽车", count=10)

for i, item in enumerate(results["items"]):
    print(i, item["metrics"]["comments"], item["text"][:40], item["extra"]["url"])

video_id = results["items"][0]["extra"]["source_id"]

Pick the video yourself rather than blindly taking the first one: the metrics.comments count tells you which ones have enough comments to be worth analyzing.

Step 3: page through the comments

GET /v1/raw/douyin/comments takes item_id (the video id), count (1 to 50) and cursor (starting at 0). Each page returns items, a cursor and has_more. Douyin can reorder comments between pages, so key them by cid to drop duplicates:

def all_comments(item_id, max_pages=10):
    comments, cursor = {}, 0
    for _ in range(max_pages):
        page = get("/v1/raw/douyin/comments", item_id=item_id, cursor=cursor, count=50)
        for c in page.get("items", []):
            comments[c["cid"]] = c
        if not page.get("has_more"):
            break
        cursor = page["cursor"]
    return list(comments.values())


comments = all_comments(video_id)
print(len(comments), "comments, spent so far $", round(spent, 6))

Each comment has cid, text, likes, reply_count, created_at, ip_location, is_hot and an author. max_pages caps the cost of a run, since every call that returns data is charged.

Step 4: score the sentiment

SnowNLP(text).sentiments returns a number between 0 and 1 (higher is more positive). It was trained on e-commerce reviews, so treat it as a rough signal for Douyin slang and emoji-heavy comments, not as ground truth. The 0.6 and 0.4 cut-offs below are my own choice; tune them on a sample you've labeled by hand.

from collections import Counter
from snownlp import SnowNLP


def label(score):
    if score >= 0.6:
        return "positive"
    if score <= 0.4:
        return "negative"
    return "neutral"


for c in comments:
    c["score"] = SnowNLP(c["text"]).sentiments
    c["label"] = label(c["score"])

counts = Counter(c["label"] for c in comments)
liked = {k: sum(c["likes"] for c in comments if c["label"] == k) for k in counts}
print("comments:", dict(counts))
print("likes by label:", liked)

Counting likes as well as comments matters: one popular negative comment can outweigh many quiet positive ones, and likes lets you see that.

Step 5: read the extremes

Numbers are a starting point. Always read the comments behind them:

for k in ("positive", "negative"):
    top = sorted((c for c in comments if c["label"] == k), key=lambda c: c["likes"], reverse=True)[:5]
    print(f"\nTop {k}")
    for c in top:
        print(f"{c['likes']:>6}  {c['text']}  ({c['ip_location']})")

Illustrative output (made-up numbers and text, to show the shape only):

comments: {'positive': 120, 'neutral': 80, 'negative': 50}
likes by label: {'positive': 3000, 'neutral': 900, 'negative': 2100}

Top negative
   932  <comment text>  (<province>)

Replies and a cross-check

Replies under a comment come from GET /v1/raw/douyin/comment_replies. Check the endpoint's reference for its parameters before using it; it's worth fetching only for comments with a high reply_count.
To check your local scores against another view, POST /v1/search is a cross-source search with a sentiment filter (negative, neutral, positive) and a platforms filter (use ["douyin"]). It searches posts and content rather than the comments of one video, so it answers a different question: how is a topic discussed on Douyin overall.

r = requests.post(
    f"{BASE}/v1/search",
    headers=HEADERS,
    json={"query": "新能源汽车", "platforms": ["douyin"], "sentiment": ["negative"], "limit": 10},
    timeout=60,
)
data = r.json()["data"]
print(data["total"], data["by_platform"])

Cost

Prices in the catalog: GET /v1/raw/douyin/search and GET /v1/raw/douyin/comments are $0.5 per 1,000 calls each; POST /v1/search is $5 per 1,000 calls. Empty results are not charged. Print the spent counter at the end of every run so the cost is never a surprise. Full details are on the pricing page.

Limits

Only public content; no private accounts or DMs.
Local sentiment models make mistakes on sarcasm, slang and mixed-language comments. Sample and read.
The comments you get are the ones the API returns for the pages you fetch, not necessarily every comment on the video.

If you build on this, tell me in the comments what you'd want added.

This article was written with the help of AI.

A Weibo hot-search tracker that alerts on your keywords

Qovia — Fri, 30 Oct 2026 06:43:41 +0000

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:

fetches the board on a schedule,
checks it against your keyword list,
sends an alert (to stdout and optionally a webhook) the first time a keyword appears, and not again until it has left the board,
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.

Analyze what Xiaohongshu (RedNote) users say about a product with Claude and MCP

Qovia — Fri, 30 Oct 2026 06:20:36 +0000

Xiaohongshu (小红书, also called RedNote) is where a lot of Chinese shoppers go to review products before they buy, and to complain after. If you sell anything in China, or compete with someone who does, the notes and comment sections there tell you a lot. The trouble is that there is no public API to sign up for, and reading hundreds of Chinese comments by hand is slow.

In this post we'll connect Claude Code to a remote MCP server, ask it what Xiaohongshu users say about a product, and let it search notes, read their comments and summarize them. At the end there's a short Python version that calls the same HTTP API without an agent.

Disclosure: I build Qovia, the API and MCP server used here. It's a paid API with a free trial.

Step 1: add the MCP server to Claude Code

Get a key from the Qovia console, then add the server with one line:

export QOVIA_KEY=sk-qovia-...
claude mcp add --transport http qovia https://api.qovia.ai/mcp --header "Authorization: Bearer $QOVIA_KEY"

The server exposes only four tools, so it doesn't fill the model's context with a hundred tool definitions:

search_endpoints: find data endpoints by keyword or platform
get_endpoint: show an endpoint's parameters, a response example and the price
call_endpoint: run an endpoint and return data plus billing info
get_account: check your key's balance and rate limit

search_endpoints and get_endpoint are free and work without an API key, so you can let the agent explore the catalog before you spend anything. Only call_endpoint needs the key. Setup details are in the MCP docs.

Step 2: ask a question

Start claude and give it something like this:

Using the qovia MCP tools, find out what Xiaohongshu users say about
the Stanley Quencher tumbler. Search in Chinese, take the 5 most
commented notes from the past month, read the first page of comments
on each, and summarize: main praise, main complaints, recurring
questions, and anything about fakes or pricing. Quote a few comments
in Chinese with an English translation. Tell me how much the run cost.

Step 3: what the agent does

The agent doesn't need to know Qovia's catalog in advance. A typical run goes like this.

1. Find the endpoints. It calls search_endpoints with platform: "xiaohongshu". That lists the Xiaohongshu endpoints, including:

Endpoint id	Path	Price per call
`raw.xiaohongshu.search`	`GET /v1/raw/xiaohongshu/search`	$0.005
`raw.xiaohongshu.comments`	`GET /v1/raw/xiaohongshu/comments`	$0.005
`raw.xiaohongshu.note`	`GET /v1/raw/xiaohongshu/note`	$0.005
`raw.xiaohongshu.comment_replies`	`GET /v1/raw/xiaohongshu/comment_replies`	$0.005

2. Read the parameters. It calls get_endpoint for the two it needs.

raw.xiaohongshu.search returns 20 notes a page:

keyword (required, up to 50 characters)
sort: general (default), latest, most_liked, most_commented or most_saved
note_time: all (default), day, week, month or year
note_type: all (default), video or image
page: from 1, up to 50; stop when has_more is false

raw.xiaohongshu.comments returns a note's first-level comments:

note_id (required, 24 hex characters)
sort: default (Xiaohongshu's own order, pinned and popular first, about 5 a page) or latest (newest first, 10 to 15 a page, without like counts)
cursor: pass back the previous page's cursor unchanged; stop when has_more is false

3. Fetch the data. It runs call_endpoint with something like:

{ "endpoint_id": "raw.xiaohongshu.search",
  "params": { "keyword": "斯坦利 保温杯", "sort": "most_commented", "note_time": "month" } }

Each note in the result carries its Xiaohongshu id in extra.source_id. Here is an abbreviated example of the structure, taken from get_endpoint's response example (not from a real run of this prompt):

{
  "page": 1,
  "items": [
    {
      "id": "soc_xia_c79d099f083d",
      "platform": "xiaohongshu",
      "text": "真假摩可纳咖啡对比（一）…",
      "author": { "user_id": "5a916db4e8ac2b54e8b53aba", "nickname": "J*y" },
      "metrics": { "likes": 1205, "saves": 966, "shares": 287, "comments": 1570 },
      "extra": {
        "title": "真假摩可纳咖啡对比（一）",
        "source_id": "69f7dc410000000035021f62",
        "url": "https://www.xiaohongshu.com/explore/69f7dc410000000035021f62",
        "note_type": "image",
        "published_at": "2026-05-04T07:37:37+08:00"
      }
    }
  ],
  "has_more": true
}

It then passes each source_id as note_id to raw.xiaohongshu.comments. Again an abbreviated example structure from get_endpoint:

{
  "items": [
    {
      "cid": "6a17eab1000000002902f300",
      "text": "喝了晚上还能睡着怎么解",
      "likes": 1918,
      "reply_count": 62,
      "ip_location": "陕西",
      "created_at": "2026-05-28T15:11:45+08:00",
      "author": { "nickname": "肉*嘟" }
    }
  ],
  "total": 1072,
  "cursor": "6a1b938c000000002b025bef",
  "has_more": true
}

4. Summarize. With five searches' worth of notes and their comments in context, Claude groups the comments into themes, picks quotes and translates them. ip_location is handy if you want a rough regional split. Every call_endpoint result includes charged_usd, so the agent can add up the cost when you ask.

A run like the prompt above is about 6 calls: one search plus five comment pages, so roughly $0.03.

Tip: search in Chinese

Xiaohongshu users write in Chinese, and they rarely use a brand's English name. Searching Stanley finds far fewer notes than 斯坦利 or 斯坦利保温杯. A few things that help:

Ask the agent to try two or three Chinese keywords: the Chinese brand name, the brand plus the product category, and any nickname people use.
Add words people use when they review: 测评 (review), 踩雷 (bad buy), 真假 (real vs fake), 平替 (cheaper alternative).
Use sort: "most_commented" to find the notes with the most discussion, and note_time to keep it recent.
Read comments with the default sort first. It puts the most-liked comments on top, which are usually the most representative.

The same thing in plain Python

If you want this as a scheduled job instead of a chat, call the HTTP API directly. Every endpoint returns the same envelope: data holds the result and meta holds billing info.

import os, requests

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

def get(path, **params):
    global spent
    r = requests.get(f"{BASE}{path}", headers=HEADERS, params=params, timeout=60)
    r.raise_for_status()
    body = r.json()
    spent += float(body.get("meta", {}).get("charged_usd", 0) or 0)
    return body["data"]

notes = get("/v1/raw/xiaohongshu/search",
            keyword="斯坦利 保温杯", sort="most_commented", note_time="month")["items"]

rows = []
for note in notes[:5]:
    comments = get("/v1/raw/xiaohongshu/comments",
                   note_id=note["extra"]["source_id"])["items"]
    for c in comments:
        rows.append((note["extra"]["title"], c.get("likes"), c["text"]))

for title, likes, text in rows:
    print(f"[{likes}] {title} | {text}")
print(f"spent ${spent:.3f}")

From there you can hand rows to any LLM with a "summarize praise and complaints" prompt, or write them to a CSV.

Pricing and trial

New accounts get 50 trial calls, enough to run the example above several times.
You're charged only when a call returns data. Empty results and errors are free, including a search with no hits or a note with no comments.
Both Xiaohongshu endpoints used here cost $0.005 per call ($5 per 1,000).
Discovery (search_endpoints, get_endpoint) is always free and needs no key.

More in the docs, the Xiaohongshu API page and the Xiaohongshu MCP page. If you try it, I'd like to hear which product you looked up and what the comments said.

DEV Community: Qovia

Search WeChat Official Account articles by topic in Python

Setup

Step 1: a POST helper

Step 2: WeChat-only topic search

Step 3: date window (since / until)

Step 4: pagination

Step 5: write a CSV

Topic search vs monitoring a specific account

Putting it together

Limits worth knowing

Going further

Give Claude Code and Cursor live social media data with an MCP server

What the server looks like

Connect a client

Optional: direct tools for one platform

Prompts that work well

Do the same thing from Python

Handling errors and limits

When to use MCP vs. the REST API

Compare how a keyword is doing on TikTok, YouTube and X in Python

What you need

The three endpoints

A tiny client

Paging through each platform

Normalizing into one table

The comparison

Reading the result

Going further

Monitor Reddit for mentions of your product in Python

Setup

Step 1: a small request helper

Step 2: search Reddit for your keywords

Step 3: watch a few subreddits for intent phrases

Step 4: remember what you've already seen

Step 5: read the room under each new post

Step 6: CSV, Slack, and the run loop

Doing the same from Claude or Cursor

Cost

Limits

Pull Kuaishou creator stats and video comments in Python

Setup

Step 1: a small request helper

Step 2: from a video link to the creator

Step 2b (optional): find a starting video by keyword

Step 3: the creator's hot videos and their stats

Step 4: what viewers say

Step 5: what's trending right now

Step 6: put it together

Step 7: summarize with pandas

Doing the same from Claude or Cursor

Cost

Limits

Monitor what Xiaohongshu (RedNote) users say about a brand in Python

Setup

Step 1: a small request helper

Step 2: search notes that mention the brand

Step 3: note details for the notes that matter

Step 4: what are commenters saying?

Step 5: put it together

Step 6: summarize with pandas

Doing the same from Claude or Cursor

Cost

Limits

Track what Zhihu is saying about any topic in Python (search, answers, comments to CSV)

Setup

Step 1: one helper

Step 2: is it on the hot list?

Step 3: search answers and articles

Step 4: which questions carry the discussion?

Step 5: what are readers saying?

Step 6: put it together

Step 7: summarize it

Cost

Using it from an AI assistant

Limits

Monitor WeChat Official Accounts in Python (new articles and read counts)

Setup

Step 1: one helper

Step 2: find the account id

Step 3: date window (`since` / `until`)