DEV Community

Qovia
Qovia

Posted on

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

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:

  1. looks up an official account by name and gets its id,
  2. polls the account's newest articles and spots the ones we haven't seen,
  3. fetches reads, likes, "watching" (在看), comments and shares for each new article,
  4. 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
Enter fullscreen mode Exit fullscreen mode

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

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 {}
Enter fullscreen mode Exit fullscreen mode

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"))
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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()
Enter fullscreen mode Exit fullscreen mode

(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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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))
Enter fullscreen mode Exit fullscreen mode

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"])
Enter fullscreen mode Exit fullscreen mode

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.

Top comments (0)