Search WeChat Official Account articles by topic in Python

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:

1. searches WeChat Official Account articles by topic with POST /v1/search and platforms: ["wechat"],
2. narrows the window with since / until (Beijing calendar days),
3. pages through results with page / limit and has_more,
4. 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.