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

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

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 {} 

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 

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) 

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 

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 

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() 

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

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

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)) Source：https://dev.to/qovia/monitor-what-xiaohongshu-rednote-users-say-about-a-brand-in-python-348
My favorites