小红书蒲公英 API 教程：达人筛选、合作报价与粉丝画像

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, **params):
    try:
        response = requests.get(
            BASE_URL + path,
            params={"token": TOKEN, **params},
            timeout=120, allow_redirects=False,
        )
    except requests.RequestException:
        raise RuntimeError("API network error or timeout") from None
    if response.status_code != 200:
        raise RuntimeError(f"API HTTP {response.status_code}")
    try:
        result = response.json()
    except ValueError:
        raise RuntimeError("API response is not JSON") from None
    if not isinstance(result, dict) or result.get("code") != 0:
        code = result.get("code") if isinstance(result, dict) else None
        raise RuntimeError(f"API failed: code={code}; check the docs")
    data = result.get("data")
    if not isinstance(data, dict):
        raise RuntimeError("Unexpected data shape; inspect the docs")
    inner = data.get("base_resp")
    if isinstance(inner, dict) and inner.get("status_code") not in (0, "0"):
        raise RuntimeError("Platform response was unsuccessful")
    return result


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


def require_list(data, key):
    value = data.get(key)
    if not isinstance(value, list):
        raise RuntimeError(f"Expected a list at {key}; inspect saved JSON")
    return value


def output_folder(topic):
    if not TOKEN or TOKEN == "your_token":
        raise RuntimeError("Set JUSTONEAPI_TOKEN before running")
    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
    folder = Path(topic + "-output") / run_id
    folder.mkdir(parents=True, exist_ok=False)
    return folder


MAX_PAGES = 2
MAX_CREATORS = 3
SEARCH = "/api/xiaohongshu-pgy/api/solar/cooperator/blogger/v2/v1"
PROFILE = "/api/xiaohongshu-pgy/api/solar/cooperator/user/blogger/userId/v1"
AUDIENCE = "/api/xiaohongshu-pgy/api/solar/kol/data/userId/fans_profile/v1"


def main():
    folder = output_folder("pugongying")
    ids, seen = [], set()
    for page in range(1, MAX_PAGES + 1):
        result = call_api(
            SEARCH, page=page, sort="FANS",
            fansNumberLower=10000, fansNumberUpper=100000,
        )
        save_json(folder, f"search-{page}.json", result)
        rows = require_list(result["data"], "kols")
        if not rows:
            break
        previous_count = len(ids)
        for row in rows:
            uid = row.get("userId") if isinstance(row, dict) else None
            if uid and str(uid) not in seen:
                seen.add(str(uid))
                ids.append(str(uid))
        if len(ids) == previous_count:
            break

    selected = ids[:MAX_CREATORS]
    save_json(folder, "selected-creators.json", selected)
    summary = []
    for index, uid in enumerate(selected, start=1):
        profile = call_api(PROFILE, userId=uid)
        save_json(folder, f"creator-{index}-profile.json", profile)
        audience = call_api(AUDIENCE, userId=uid)
        save_json(folder, f"creator-{index}-audience.json", audience)
        data = profile["data"]
        summary.append({
            "userId": uid, "name": data.get("name"),
            "fansCount": data.get("fansCount"),
            "picturePrice": data.get("picturePrice"),
            "videoPrice": data.get("videoPrice"),
            "showPrice": data.get("showPrice"),
            "audienceDate": audience["data"].get("dateKey"),
        })
    save_json(folder, "shortlist.json", summary)
    print(f"Saved {len(selected)} creator samples to {folder}")


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