用 Agent 自动筛小红书 KOL

总分 = 0.3 × 互动分
     + 0.25 × 内容匹配
     + 0.25 × 增长分
     + 0.2 × 粉丝质量信号

curl -s https://api.sandbase.ai/v1/api/xiaohongshu/pgy/blogger-detail \
  -H "Authorization: Bearer $SANDBASE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"user_id": "59ae4dfc5e87e72475fb3bf3"}'

{
  "id": "7476e66a-35d3-4e41-8c8d-a345ca4c7237",
  "status": "completed",
  "model": "xiaohongshu/pgy/blogger-detail",
  "outputs": [{"data": {"code": 0, "data": {
    "userId": "59ae4dfc5e87e72475fb3bf3",
    "name": "Perfect Diary完美日记",
    "fansNum": 1835170,
    "interMidNum": 0
  }}}]
}

{"code": 0, "data": {"fansNumInc": -9214, "list": [
  {"dateKey": "2026-08-02", "num": 1842991}, "…",
  {"dateKey": "2026-09-30", "num": 1833777}
]}}

{"id": "6abcd6f600000000020115f9", "cursor": "6abcd6f600000000020115f9",
 "title": "🦌新成员集结｜小鹿限定线上开售",
 "likes": 37, "comments_count": 33, "collected_count": 22}

#!/usr/bin/env python3
"""小红书 KOL 筛选 agent。

从蒲公英拉候选人，抽样笔记，打分，输出排名 CSV。
"""

import csv
import json
import os
import time
from dataclasses import dataclass

import requests
from openai import OpenAI

SANDBASE_API_KEY = os.environ["SANDBASE_API_KEY"]
HEADERS = {
    "Authorization": f"Bearer {SANDBASE_API_KEY}",
    "Content-Type": "application/json",
}

# OpenAI 客户端仅用于 LLM 调用
client = OpenAI(
    base_url="https://api.sandbase.ai/v1",
    api_key=SANDBASE_API_KEY,
)


@dataclass
class KOLScore:
    blogger_id: str
    nickname: str
    follower_count: int
    avg_interactions_per_note: float
    save_share: float
    growth_30d: float
    audience_score: float
    content_fit: float
    total_score: float
    reasoning: str
    red_flags: list[str]


def run_model(model: str, **params) -> dict:
    """POST /v1/api/<model>，返回 outputs[0].data 里上游的 `data` 对象。"""
    resp = requests.post(
        f"https://api.sandbase.ai/v1/api/{model}",
        headers=HEADERS,
        json=params,
        timeout=90,
    )
    resp.raise_for_status()
    body = resp.json()
    outputs = body.get("outputs") or []
    if body.get("status") != "completed" or not outputs:
        raise RuntimeError(f"SandBase call did not complete: {body.get('status')} {body.get('error')}")
    payload = outputs[0].get("data") or {}
    return payload.get("data") or {}


def fetch_candidates(brand_user_id: str, content_tag: str, min_fans: int, max_fans: int,
                     pages: int = 10, page_size: int = 20) -> list[str]:
    """翻页调用蒲公英 blogger-list，返回候选人 userId。"""
    ids = []
    for page in range(1, pages + 1):
        data = run_model(
            "xiaohongshu/pgy/blogger-list",
            brand_user_id=brand_user_id,
            blogger={"content_tag": [content_tag]},
            fans={"count": {"min": min_fans, "max": max_fans}},
            page_num=page,
            page_size=page_size,
        )
        kols = data.get("kols") or []
        ids += [k["userId"] for k in kols if k.get("userId")]
        if not kols:  # 返回条数少于 page_size 不代表到底了，空页才停
            break
    return ids


def fetch_blogger_detail(blogger_id: str) -> dict:
    """蒲公英博主概况（昵称、粉丝数、互动中位数）。"""
    return run_model("xiaohongshu/pgy/blogger-detail", user_id=blogger_id)


def fetch_fans_history(blogger_id: str) -> dict:
    """粉丝历史（参考文档里 date_type=2 约为 30 天）。"""
    return run_model("xiaohongshu/pgy/blogger-fans-history", user_id=blogger_id, date_type=2)


def fetch_recent_notes(blogger_id: str, count: int = 10) -> list:
    """抽样博主自己发布笔记的第一页（不传 cursor 即第一页）。"""
    data = run_model("xiaohongshu/app-v2/user-posted-notes", user_id=blogger_id)
    return (data.get("notes") or [])[:count]


def compute_engagement_metrics(notes: list) -> dict:
    """抽样笔记单篇平均互动，以及收藏在互动中的占比。"""
    if not notes:
        return {"avg_interactions_per_note": 0, "save_share": 0}

    total_likes = sum(n.get("likes") or 0 for n in notes)
    total_comments = sum(n.get("comments_count") or 0 for n in notes)
    total_saves = sum(n.get("collected_count") or 0 for n in notes)
    total_interactions = total_likes + total_comments + total_saves

    return {
        "avg_interactions_per_note": total_interactions / len(notes),
        "save_share": total_saves / max(total_interactions, 1),
    }


def compute_growth(fans_history: dict) -> dict:
    """最近 30 个日点的粉丝增长，外加单日暴涨检查。"""
    data_points = (fans_history.get("list") or [])[-30:]
    if len(data_points) < 2:
        return {"growth_30d": 0, "spike_detected": False}

    first = data_points[0].get("num") or 0
    last = data_points[-1].get("num") or 0
    growth_rate = (last - first) / max(first, 1)

    max_daily_change = 0
    for i in range(1, len(data_points)):
        daily_change = abs((data_points[i].get("num") or 0) - (data_points[i - 1].get("num") or 0))
        max_daily_change = max(max_daily_change, daily_change)

    spike_threshold = first * 0.1  # 单日变化超过 10% 视为可疑
    return {
        "growth_30d": growth_rate,
        "spike_detected": max_daily_change >= spike_threshold > 0,
    }


def score_content_fit(notes: list, brand_category: str) -> dict:
    """LLM 根据抽样笔记标题/正文给内容匹配度打分。"""
    if not notes:
        return {"score": 0, "reasoning": "没有笔记"}

    titles = [(n.get("title") or n.get("desc") or "")[:100] for n in notes[:10]]

    resp = client.chat.completions.create(
        model="anthropic/claude-sonnet-4",
        messages=[{
            "role": "user",
            "content": f"""给这个博主的内容和「{brand_category}」品类的匹配度打分 1-10。

近期笔记标题/正文（最多 10 条）：
{json.dumps(titles, ensure_ascii=False)}

评分标准：
- 1-3：完全无关
- 4-6：有交集但不是核心
- 7-9：品类高度对齐
- 10：完美匹配，正是目标受众

回复 JSON：{{"score": N, "reasoning": "一句话理由"}}"""
        }],
        max_tokens=200,
    )
    try:
        return json.loads(resp.choices[0].message.content)
    except json.JSONDecodeError:
        return {"score": 5, "reasoning": "无法解析 LLM 回复"}


def screen_kol(blogger_id: str, brand_category: str) -> KOLScore:
    """完整筛一个候选人（3 次数据调用 + 1 次 LLM 调用）。"""
    detail = fetch_blogger_detail(blogger_id)
    fans = fetch_fans_history(blogger_id)
    notes = fetch_recent_notes(blogger_id, count=10)

    engagement = compute_engagement_metrics(notes)
    growth = compute_growth(fans)
    content = score_content_fit(notes, brand_category)

    red_flags = []
    follower_count = detail.get("fansNum") or detail.get("fansCount") or 0
    if follower_count > 100000 and engagement["avg_interactions_per_note"] < 100:
        red_flags.append("粉丝多互动少")
    if growth.get("spike_detected"):
        red_flags.append("单日涨粉超过 10%")
    if engagement["save_share"] < 0.05:
        red_flags.append("收藏不到互动的 5%")

    # 每一项都是 0-10 分
    engagement_score = min(engagement["avg_interactions_per_note"] / 500, 10)
    growth_score = max(0, min(growth["growth_30d"] * 100, 10))  # 涨 10% = 10 分，掉粉 = 0 分
    median_interactions = detail.get("interMidNum") or 0
    audience_score = min(median_interactions / max(follower_count, 1) * 200, 10)  # 5% = 10 分
    content_score = content.get("score", 5)

    total = (
        0.3 * engagement_score
        + 0.25 * content_score
        + 0.25 * growth_score
        + 0.2 * audience_score
    )

    return KOLScore(
        blogger_id=blogger_id,
        nickname=detail.get("name") or "未知",
        follower_count=follower_count,
        avg_interactions_per_note=engagement["avg_interactions_per_note"],
        save_share=engagement["save_share"],
        growth_30d=growth["growth_30d"],
        audience_score=round(audience_score, 2),
        content_fit=content_score,
        total_score=round(total, 2),
        reasoning=content.get("reasoning", ""),
        red_flags=red_flags,
    )


def run_screening(blogger_ids: list[str], brand_category: str, output_path: str):
    """筛完所有候选人，输出排名 CSV。"""
    results: list[KOLScore] = []

    for i, blogger_id in enumerate(blogger_ids):
        print(f"[{i+1}/{len(blogger_ids)}] 筛选 {blogger_id}...")
        try:
            results.append(screen_kol(blogger_id, brand_category))
        except Exception as e:
            print(f"  出错：{e}")
        time.sleep(0.5)  # 候选人之间稍作停顿（不是文档规定的限制）

    results.sort(key=lambda x: x.total_score, reverse=True)

    with open(output_path, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.writer(f)
        writer.writerow([
            "排名", "昵称", "博主 ID", "粉丝数", "单篇平均互动",
            "收藏占比", "30 天增长", "粉丝质量信号", "内容匹配",
            "总分", "理由", "红旗"
        ])
        for rank, r in enumerate(results, 1):
            writer.writerow([
                rank, r.nickname, r.blogger_id, r.follower_count,
                f"{r.avg_interactions_per_note:.0f}", f"{r.save_share:.2%}",
                f"{r.growth_30d:.1%}", r.audience_score, r.content_fit,
                r.total_score, r.reasoning, "; ".join(r.red_flags) or "无"
            ])

    print(f"\n筛选完成，共 {len(results)} 个 KOL。")
    print(f"输出：{output_path}")
    for r in results[:5]:
        print(f"  {r.nickname}: {r.total_score}（{r.reasoning}）")


if __name__ == "__main__":
    # 完美日记的公开品牌账号 id，上面实测时用作 brand_user_id；换成你自己品牌的账号 id
    BRAND_USER_ID = "59ae4dfc5e87e72475fb3bf3"
    candidate_ids = fetch_candidates(BRAND_USER_ID, "美妆", 10_000, 500_000, pages=10)  # 最多 200 个
    run_screening(candidate_ids, brand_category="美妆护肤", output_path="kol_ranking.csv")