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

先说结论 — 一个 Python agent，每 15 分钟查微博热搜 + 抖音热榜，用 LLM 判断是否和你的品牌/行业相关，命中就发 webhook 告警。费用：数据 $0.19/天 + LLM $0.96/天 ≈ $1.15/天（$36/月）。加上关键词预过滤可以压到 $12/月。完整代码在下面。

品牌危机不等你早上 9 点上班。凌晨 2 点的一条负面爆帖，到 9 点可能已经 5000 万播放。7 小时的空窗是「反应迅速」和「被打了个措手不及」的区别。

这个 agent 把空窗压到 15 分钟，月费 $36。

Agent 做什么

每 15 分钟，7×24：

1. 拉微博当前热搜（50 个话题 + 热度分）
2. 拉抖音热榜/热门账号
3. 两份列表送给 LLM，带你的品牌关键词和分类规则
4. 命中品牌相关 → 发 webhook（Slack、飞书、Discord、邮件）
5. 全部记日志做趋势分析

费用

组件	每周期	每天（96 周期）	日费用
微博热搜	$0.001	96	$0.096
抖音热榜	$0.001	96	$0.096
LLM 分类	~$0.01	96	$0.960
详情补查（可选）	$0.002	~20	$0.040
合计	~$0.013		~$1.19/天

月费：~$36/月 7×24 双平台监控。

LLM 占 80% 费用。优化方案：先做关键词预过滤，只在命中时才调 LLM。假设 30% 周期有关键词命中，优化后 ~$12/月。

核心代码

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

import json, os, time
from datetime import datetime, timezone
from pathlib import Path
from urllib.request import Request, urlopen
from openai import OpenAI

BRAND_KEYWORDS = ["你的品牌", "竞品A", "竞品B"]
INDUSTRY_KEYWORDS = ["AI", "大模型", "agent", "智能体"]
WEBHOOK_URL = os.environ.get("ALERT_WEBHOOK_URL", "")
CHECK_INTERVAL = 900  # 15 分钟

client = OpenAI(base_url="https://api.sandbase.ai/v1", api_key=os.environ["SANDBASE_API_KEY"])
LOG_DIR = Path("./logs"); LOG_DIR.mkdir(exist_ok=True)


def fetch_weibo():
    r = client.chat.completions.create(model="weibo/web-v2/hot-search", messages=[])
    data = json.loads(r.choices[0].message.content)
    return data.get("data", data) if isinstance(data, dict) else data

def fetch_douyin():
    r = client.chat.completions.create(model="douyin/billboard/hot-account-list", messages=[])
    data = json.loads(r.choices[0].message.content)
    return data.get("data", data) if isinstance(data, dict) else data

def quick_keyword_check(topics, keywords):
    """免费预过滤：关键词是否出现在任何话题里。"""
    text = " ".join(str(t) for t in topics).lower()
    return any(kw.lower() in text for kw in keywords)

def classify(weibo, douyin):
    """LLM 判断哪些话题和品牌/行业相关。"""
    prompt = f"""你是品牌舆情分析师。分类这些热门话题：
- "brand_mention"：直接提到我们品牌或竞品
- "industry_relevant"：和行业相关但非直接提及
- "irrelevant"：不相关

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

微博热搜前 20：
{json.dumps(weibo[:20], ensure_ascii=False)}

抖音热榜前 10：
{json.dumps(douyin[:10], ensure_ascii=False)}

回复 JSON：
{{"brand_mentions": [{{"source":"weibo|douyin","title":"...","reason":"..."}}],
 "industry_relevant": [{{"source":"weibo|douyin","title":"...","reason":"..."}}],
 "summary": "一句话总结"}}"""

    r = client.chat.completions.create(model="anthropic/claude-sonnet-4", messages=[{"role":"user","content":prompt}], max_tokens=1500)
    try: return json.loads(r.choices[0].message.content)
    except: return {"brand_mentions":[], "industry_relevant":[], "summary":"解析失败"}

def send_alert(result, ts):
    if not WEBHOOK_URL: return
    mentions = result.get("brand_mentions", [])
    if not mentions and not result.get("industry_relevant"): return
    lines = [f"🚨 舆情告警 — {ts}"]
    for m in mentions:
        lines.append(f"• [{m['source']}] {m['title']} — {m['reason']}")
    msg = {"text": "\n".join(lines)}
    req = Request(WEBHOOK_URL, data=json.dumps(msg).encode(), headers={"Content-Type":"application/json"}, method="POST")
    try: urlopen(req, timeout=10)
    except Exception as e: print(f"  告警失败: {e}")

def run_cycle():
    ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    print(f"\n[{ts}] 巡检中...")

    weibo = fetch_weibo()
    douyin = fetch_douyin()
    print(f"  微博 {len(weibo)} 条, 抖音 {len(douyin)} 条")

    # 预过滤：没命中关键词就跳过 LLM（省 $0.01）
    all_kw = BRAND_KEYWORDS + INDUSTRY_KEYWORDS
    if not quick_keyword_check(weibo + douyin, all_kw):
        print("  ✓ 无关键词命中，跳过 LLM")
        return

    result = classify(weibo, douyin)
    brand_n = len(result.get("brand_mentions", []))
    print(f"  品牌提及: {brand_n}, 行业信号: {len(result.get('industry_relevant', []))}")

    if brand_n > 0:
        print("  ⚠️ 检测到品牌提及！发送告警...")
        send_alert(result, ts)

def main():
    print(f"舆情监控启动 | 品牌: {BRAND_KEYWORDS} | 间隔: {CHECK_INTERVAL}s")
    while True:
        try: run_cycle()
        except Exception as e: print(f"  错误: {e}")
        time.sleep(CHECK_INTERVAL)

if __name__ == "__main__":
    import sys
    if "--once" in sys.argv: run_cycle()
    else: main()

部署方式

方式 1：常驻进程（VPS/容器）

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

内存约 10MB，CPU 可忽略。用 systemd 或 Docker 保活。

方式 2：Cron 每 15 分钟

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

方式 3：GitHub Actions

on:
  schedule:
    - cron: '*/15 * * * *'

注意 GitHub cron 有最多 15 分钟延迟，时效敏感场景不太合适。

告警触发后怎么办

Agent 检测，人来决策。建议响应流程：

告警类型	响应	时限
品牌提及 + 负面 + 高热	启动危机预案	30 分钟内
品牌提及 + 中性	PR 团队评估	2 小时内
行业信号	纳入周策略简报	本周内
竞品提及	记录竞品情报	下次复盘

扩展

加情感深度：品牌命中后调 weibo/web-v2/topic-posts 拉帖子做情感分析，每次告警加 ~$0.01。

加小红书：把 xiaohongshu/web-v3/hot-list 作为第三个源，同样模式，每周期多 $0.001。

历史趋势：积累 30 天日志后让 LLM 分析”哪些行业话题周环比热度在涨”——从巡检数据里提炼策略洞察。

局限

• 15 分钟粒度：不是实时。危机可以在 15 分钟内快速发酵。更高要求改成 5 分钟间隔（费用 3 倍）
• LLM 分类不完美：会有误报。设计上宁可多报不漏报——最终由人判断
• 热搜轮转快：一个话题可能第 1 分钟在榜第 14 分钟就掉了。考虑加品牌专搜做补充
• 不分析视频内容：只看热榜元数据，不看视频本身。深度分析可以对命中项追加 douyin/app-v3/video-detail

FAQ

能只监控特定关键词而不是热搜吗？

能。用 weibo/web-v2/search 配特定关键词，替代或补充热搜。能抓到不上榜但相关的内容。

多快能发现品牌危机？

最慢：话题上热搜后 15 分钟。最快：当前周期内就抓到。平均约 8 分钟从上榜到告警送达。

支持什么 webhook？

任何接受 POST JSON 的 URL。已测试：Slack（incoming webhook）、Discord（webhook）、飞书（机器人 webhook）、钉钉、通用 HTTP 端点。

能同时监控多个品牌吗？

能。方式一：扩大关键词列表，让 LLM 按品牌分类。方式二：跑多个实例各用不同关键词。前者更省钱。

微博接口详情看 微博+小红书上线。抖音接口看 310 个抖音接口。