如何用 Python 監測小紅書用戶對品牌的評價

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:

1. search recent Xiaohongshu notes that mention the brand,
2. fetch the full detail of the most engaged notes (exact counts, topics, IP location),
3. collect the first pages of comments under each of those notes,
4. append everything to three CSV files with a timestamp,
5. 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

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, 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.