小红书笔记和评论怎么采集？从关键词搜索到数据保存

python3 -m pip install requests
export JUSTONEAPI_TOKEN='your_token'

import json
import os
from datetime import datetime, timezone
from pathlib import Path

import requests

BASE_URL = "https://api.justoneapi.com"
TOKEN = os.environ.get("JUSTONEAPI_TOKEN", "")


def call_api(path, method="GET", accepted_codes=(0,), **params):
    if not TOKEN or TOKEN == "your_token":
        raise RuntimeError("Set JUSTONEAPI_TOKEN before running")
    values = {"token": TOKEN, **params}
    arguments = {"data": values} if method == "POST" else {"params": values}
    try:
        response = requests.request(
            method, BASE_URL + path, timeout=120, **arguments,
        )
    except requests.RequestException:
        raise RuntimeError("Network error or timeout") from None
    if response.status_code != 200:
        raise RuntimeError(f"HTTP {response.status_code}")
    try:
        result = response.json()
    except ValueError:
        raise RuntimeError("Response is not valid JSON") from None
    if not isinstance(result, dict) or result.get("code") not in accepted_codes:
        code = result.get("code") if isinstance(result, dict) else None
        raise RuntimeError(f"API failed: code={code}; check the docs")
    return result


def output_dir(prefix):
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
    folder = Path(prefix) / stamp
    folder.mkdir(parents=True, exist_ok=False)
    return folder


def save_json(folder, filename, value):
    (folder / filename).write_text(
        json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8",
    )


KEYWORD = "防晒"
MAX_NOTES = 3
MAX_COMMENT_PAGES = 2


def main():
    folder = output_dir("xiaohongshu-output")
    search = call_api(
        "/api/xiaohongshu/search-note/v2",
        keyword=KEYWORD, page=1, sort="general", noteType="_0",
    )
    save_json(folder, "search-1.json", search)
    note_ids = []
    seen = set()
    for item in search["data"].get("items") or []:
        note_id = (item.get("note") or {}).get("id")
        if note_id and str(note_id) not in seen:
            seen.add(str(note_id))
            note_ids.append(str(note_id))
    selected = note_ids[:MAX_NOTES]
    save_json(folder, "selected-note-ids.json", selected)

    for index, note_id in enumerate(selected, start=1):
        detail = call_api("/api/xiaohongshu/get-note-detail/v1", noteId=note_id)
        save_json(folder, f"note-{index}-detail.json", detail)
        cursor = ""
        seen_cursors = set()
        for page in range(1, MAX_COMMENT_PAGES + 1):
            comments = call_api(
                "/api/xiaohongshu/get-note-comment/v2",
                noteId=note_id, sort="latest", lastCursor=cursor,
            )
            save_json(folder, f"note-{index}-comments-{page}.json", comments)
            data = comments["data"]
            if data.get("has_more") not in (True, 1, "1"):
                break
            next_cursor = data.get("cursor")
            if page < MAX_COMMENT_PAGES:
                if not next_cursor or next_cursor in seen_cursors:
                    raise RuntimeError("Missing or repeated comment cursor")
                seen_cursors.add(next_cursor)
                cursor = next_cursor
    print(f"Saved {len(selected)} note samples to {folder}")


if __name__ == "__main__":
    try:
        main()
    except RuntimeError as error:
        raise SystemExit(str(error)) from None