# Monitor what Xiaohongshu (RedNote) users say about a brand in Python > Source: > Published: 2026-10-30 23:15:35+00:00 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: 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](https://qovia.ai/?utm_source=devto&utm_medium=article&utm_campaign=xiaohongshu-brand-monitor-python). Requests authenticate with `Authorization: Bearer ` (an `x-api-key` header also works). The [Xiaohongshu API page](https://qovia.ai/xiaohongshu-api/?utm_source=devto&utm_medium=article&utm_campaign=xiaohongshu-brand-monitor-python) lists all the Xiaohongshu endpoints with a real request and response. 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: ``` python 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. `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`: `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: ``` python 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) ``` 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/`, 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: ``` python 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. `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=` 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. ``` python 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 python 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. 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 "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](https://qovia.ai/docs/mcp/?utm_source=devto&utm_medium=article&utm_campaign=xiaohongshu-brand-monitor-python), and the [Xiaohongshu MCP page](https://qovia.ai/mcp/xiaohongshu/?utm_source=devto&utm_medium=article&utm_campaign=xiaohongshu-brand-monitor-python) lists the Xiaohongshu tools. The assistant is good for one-off digging; the script is better when you want the same numbers every day. 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](https://qovia.ai/pricing/?utm_source=devto&utm_medium=article&utm_campaign=xiaohongshu-brand-monitor-python), and every parameter is in the [API docs](https://qovia.ai/docs/?utm_source=devto&utm_medium=article&utm_campaign=xiaohongshu-brand-monitor-python). The default rate limit is 60 requests per minute per account, more than enough for a job like this. `text` in search results is an excerpt.`pages_per_note`. 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.*