搭个舆情监控 Agent：微博 + 抖音

┌─────────────────┐     ┌──────────────┐     ┌─────────────┐     ┌──────────────┐
│ 调度             │────▶│ 拉数据       │────▶│ LLM 分类    │────▶│ Slack        │
│（每 15 分钟）    │     │ 微博 + 抖音  │     │ 相关吗？    │     │ webhook      │
└─────────────────┘     └──────────────┘     └─────────────┘     └──────────────┘
                                                    │
                                                    ▼
                                             ┌────────────┐
                                             │ 周期日志   │
                                             │（JSONL）   │
                                             └────────────┘

#!/usr/bin/env python3
"""舆情监控 agent：微博 + 抖音。

每 15 分钟检查微博热搜和抖音热门账号榜，用 LLM 判断相关性，
把告警推送到 Slack incoming webhook。
可以常驻运行，也可以用 cron 加 --once 调用。
"""

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

import requests
from openai import OpenAI

# --- 配置 ---
BRAND_KEYWORDS = ["你的品牌名", "YourBrand", "竞品A", "竞品B"]
INDUSTRY_KEYWORDS = ["AI", "大模型", "agent", "智能体"]
SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL", "")  # Slack incoming webhook
CHECK_INTERVAL = 900  # 15 分钟
LOG_DIR = Path("./listening_logs")
LOG_DIR.mkdir(exist_ok=True)

SANDBASE_API_KEY = os.environ["SANDBASE_API_KEY"]
API_BASE = "https://api.sandbase.ai/v1/api"
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,
    timeout=60,
)


# --- 拉数据：POST /v1/api/<端点>，只读 outputs[0].data ---
def call_data_api(endpoint: str, params: dict | None = None, retries: int = 3) -> dict:
    """调用一个 SandBase 数据端点；HTTP 429/5xx 按 1s、2s 退避重试。"""
    for attempt in range(retries):
        resp = requests.post(f"{API_BASE}/{endpoint}", headers=HEADERS, json=params or {}, timeout=90)
        if (resp.status_code == 429 or resp.status_code >= 500) and attempt < retries - 1:
            time.sleep(2 ** attempt)
            continue
        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')}")
        return outputs[0].get("data") or {}


def fetch_weibo_hot_search() -> list:
    """微博热搜，整理成 [{title, heat}]。"""
    data = call_data_api("weibo/web-v2/hot-search")
    # 字段名来自 2026-10-01 实测，仅供参考、非文档保证：realtime[].word / .num
    return [
        {"title": t.get("word") or t.get("note"), "heat": t.get("num")}
        for t in data.get("realtime") or []
    ]


def fetch_douyin_hot_accounts() -> list:
    """抖音热门账号榜（榜单上的账号，不是热搜话题）。"""
    data = call_data_api("douyin/billboard/hot-account-list", {"page_num": 1, "page_size": 10})
    # 字段名来自 2026-10-01 实测：data.user_list[].nick_name / .fans_cnt / .new_fans_cnt
    users = (data.get("data") or {}).get("user_list") or []
    return [
        {"title": u.get("nick_name"), "fans": u.get("fans_cnt"), "new_fans": u.get("new_fans_cnt")}
        for u in users
    ]


def fetch_weibo_keyword_posts(keyword: str) -> list:
    """可选补查，run_cycle 不调用：按命中关键词拉微博实时帖子。"""
    data = call_data_api("weibo/web-v2/realtime-search", {"query": keyword})
    # 字段名来自 2026-10-01 实测：parsed_data.results[].content / .post_url
    return (data.get("parsed_data") or {}).get("results") or []


# --- 分类 ---
def classify_relevance(weibo_topics: list, douyin_accounts: list) -> dict:
    """LLM 判断哪些话题和品牌/行业相关。"""
    prompt = f"""你是品牌舆情分析师。把下面的微博热搜话题和抖音热门账号逐条分类：
- "brand_mention"：直接提到我们品牌或竞品
- "industry_relevant"：和行业相关但非直接提及
- "irrelevant"：不相关

品牌关键词：{json.dumps(BRAND_KEYWORDS, ensure_ascii=False)}
行业关键词：{json.dumps(INDUSTRY_KEYWORDS, ensure_ascii=False)}

微博热搜（前 20）：
{json.dumps(weibo_topics[:20], ensure_ascii=False, indent=2)}

抖音热门账号（前 10）：
{json.dumps(douyin_accounts[:10], ensure_ascii=False, indent=2)}

只回复 JSON，不要 markdown：
{{
  "brand_mentions": [{{"source": "weibo|douyin", "title": "...", "heat": N, "reason": "..."}}],
  "industry_relevant": [{{"source": "weibo|douyin", "title": "...", "heat": N, "reason": "..."}}],
  "summary": "一句话总结"
}}

只列真正相关的条目。没有命中就返回空列表。"""

    resp = client.chat.completions.create(
        model="anthropic/claude-sonnet-4",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=1500,
    )
    if resp.usage:  # 打印真实 token 数，用来核对成本估算
        print(f"  LLM tokens: {resp.usage.prompt_tokens} 输入 / {resp.usage.completion_tokens} 输出")
    text = (resp.choices[0].message.content or "").strip()
    if text.startswith("```"):  # 兼容带代码块围栏的回复
        text = text.strip("`").removeprefix("json").strip()
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return {"brand_mentions": [], "industry_relevant": [], "summary": "解析失败"}


# --- 告警（Slack incoming webhook 负载：{"text": ...}） ---
def send_slack_alert(classification: dict, timestamp: str):
    """把品牌提及和行业信号推送到 Slack。"""
    if not SLACK_WEBHOOK_URL:
        print("  未配置 Slack webhook，跳过告警")
        return

    brand_mentions = classification.get("brand_mentions") or []
    industry = classification.get("industry_relevant") or []
    if not brand_mentions and not industry:
        return  # 没有要告警的

    blocks = []
    if brand_mentions:
        blocks.append("🚨 *品牌提及*\n" + "\n".join(
            f"• [{m.get('source', '?').upper()}] {m.get('title')}（热度：{m.get('heat', '?')}）— {m.get('reason', '')}"
            for m in brand_mentions
        ))
    if industry:
        blocks.append("📡 *行业信号*\n" + "\n".join(
            f"• [{i.get('source', '?').upper()}] {i.get('title')} — {i.get('reason', '')}"
            for i in industry[:5]  # 行业条目最多 5 条
        ))

    message = {
        "text": f"舆情告警 — {timestamp}\n\n" + "\n\n".join(blocks)
                + f"\n\n_{classification.get('summary', '')}_"
    }
    try:
        resp = requests.post(SLACK_WEBHOOK_URL, json=message, timeout=10)
        resp.raise_for_status()
        print(f"  告警已发送：HTTP {resp.status_code}")
    except requests.RequestException as e:
        print(f"  告警失败：{e}")


# --- 日志 ---
def log_cycle(timestamp: str, weibo: list, douyin: list, classification: dict):
    """每个周期追加一行 JSON：条数、命中条目、总结。"""
    log_entry = {
        "timestamp": timestamp,
        "weibo_count": len(weibo),
        "douyin_count": len(douyin),
        "brand_mentions": classification.get("brand_mentions") or [],
        "industry_relevant": classification.get("industry_relevant") or [],
        "summary": classification.get("summary", ""),
    }
    log_file = LOG_DIR / f"listening-{timestamp[:10]}.jsonl"
    with open(log_file, "a", encoding="utf-8") as f:
        f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")


# --- 主循环 ---
def run_cycle():
    """执行一个巡检周期：2 次数据调用 + 1 次 LLM 调用。"""
    timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    print(f"\n[{timestamp}] 巡检中...")

    weibo = fetch_weibo_hot_search()
    print(f"  微博热搜 {len(weibo)} 条")
    douyin = fetch_douyin_hot_accounts()
    print(f"  抖音热门账号 {len(douyin)} 个")

    classification = classify_relevance(weibo, douyin)
    brand_count = len(classification.get("brand_mentions") or [])
    industry_count = len(classification.get("industry_relevant") or [])
    print(f"  品牌提及 {brand_count} 条，行业信号 {industry_count} 条")

    # 品牌提及和仅有行业信号的情况都推送到 Slack
    if brand_count > 0:
        print("  ⚠️ 检测到品牌提及！发送告警...")
        send_slack_alert(classification, timestamp)
    elif industry_count > 0:
        print("  📡 检测到行业信号，发送摘要...")
        send_slack_alert(classification, timestamp)
    else:
        print("  ✓ 本周期无相关内容")

    log_cycle(timestamp, weibo, douyin, classification)
    return classification


def run_continuous():
    """常驻运行，每 15 分钟一次。"""
    print("舆情监控 Agent 启动")
    print(f"  品牌关键词：{BRAND_KEYWORDS}")
    print(f"  行业关键词：{INDUSTRY_KEYWORDS}")
    print(f"  间隔：{CHECK_INTERVAL}s")
    print(f"  Slack webhook：{'已配置' if SLACK_WEBHOOK_URL else '未配置'}")

    while True:
        try:
            run_cycle()
        except Exception as e:
            print(f"  错误：{e}")
        time.sleep(CHECK_INTERVAL)


if __name__ == "__main__":
    if "--once" in sys.argv:
        run_cycle()  # 单次运行，用于测试或 cron
    else:
        run_continuous()

POST https://api.sandbase.ai/v1/api/weibo/web-v2/hot-search
{}

{"id": "1e56f2e0-3eca-4242-b371-0a4e2f4cc060", "status": "completed", "model": "weibo/web-v2/hot-search",
 "outputs": [{"data": {"realtime": [
   {"word": "鸿蒙智行问界业务升级", "num": 1116159, "realpos": 1, "label_name": "新", ...},
   ...], "hotgov": {...}, "hotgovs": [...]}}]}

POST https://api.sandbase.ai/v1/api/douyin/billboard/hot-account-list
{"page_num": 1, "page_size": 10}

{"id": "c38e0333-4f1d-4104-b5bc-f41dbb8a06ad", "status": "completed", "model": "douyin/billboard/hot-account-list",
 "outputs": [{"data": {"code": 0, "message": "", "data": {"has_more": true, "user_list": [
   {"nick_name": "人民日报", "fans_cnt": 203210991, "new_fans_cnt": 51760, ...},
   {"nick_name": "央视新闻", "fans_cnt": 187596558, "new_fans_cnt": 38735, ...},
   ...]}}}]}

def quick_keyword_check(items: list, keywords: list) -> bool:
    """快速检查：关键词是否出现在任何条目里。不调用 API。"""
    text = " ".join(json.dumps(i, ensure_ascii=False) for i in items).lower()
    return any(kw.lower() in text for kw in keywords)


def run_cycle_optimized():
    """和 run_cycle 一样，但只在关键词出现时才调 LLM。"""
    timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    weibo = fetch_weibo_hot_search()
    douyin = fetch_douyin_hot_accounts()

    if quick_keyword_check(weibo + douyin, BRAND_KEYWORDS + INDUSTRY_KEYWORDS):
        classification = classify_relevance(weibo, douyin)
        send_slack_alert(classification, timestamp)  # 品牌或行业条目都会发
    else:
        classification = {"brand_mentions": [], "industry_relevant": [], "summary": "无关键词命中"}
    log_cycle(timestamp, weibo, douyin, classification)
    return classification

SANDBASE_API_KEY=sk-... SLACK_WEBHOOK_URL=https://hooks.slack.com/services/... python3 listener.py

*/15 * * * * SANDBASE_API_KEY=sk-... SLACK_WEBHOOK_URL=... python3 listener.py --once

on:
  schedule:
    - cron: '*/15 * * * *'  # 每 15 分钟