Build a Social Listening Agent: Weibo + Douyin

┌─────────────────┐     ┌──────────────┐     ┌─────────────┐     ┌──────────────┐
│ Scheduler       │────▶│ Data Fetch   │────▶│ LLM Classify│────▶│ Slack        │
│ (every 15 min)  │     │ Weibo+Douyin │     │ Relevant?   │     │ webhook      │
└─────────────────┘     └──────────────┘     └─────────────┘     └──────────────┘
                                                    │
                                                    ▼
                                             ┌────────────┐
                                             │ Cycle log  │
                                             │ (JSONL)    │
                                             └────────────┘

#!/usr/bin/env python3
"""Social listening agent: Weibo + Douyin.

Checks Weibo hot search and the Douyin hot account billboard every 15 minutes,
classifies relevance with an LLM, posts alerts to a Slack incoming webhook.
Run as a long-running process, or via cron with --once.
"""

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

import requests
from openai import OpenAI

# --- Configuration ---
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 minutes in seconds
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 client for LLM calls only
client = OpenAI(
    base_url="https://api.sandbase.ai/v1",
    api_key=SANDBASE_API_KEY,
    timeout=60,
)


# --- Data fetching: POST /v1/api/<endpoint>, read outputs[0].data ---
def call_data_api(endpoint: str, params: dict | None = None, retries: int = 3) -> dict:
    """Call one SandBase data endpoint; retry HTTP 429/5xx with 1s, 2s backoff."""
    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:
    """Get current Weibo hot search list as [{title, heat}]."""
    data = call_data_api("weibo/web-v2/hot-search")
    # Observed-only field names (tested 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:
    """Get Douyin hot account list (accounts ranked on the billboard, not topics)."""
    data = call_data_api("douyin/billboard/hot-account-list", {"page_num": 1, "page_size": 10})
    # Observed-only field names (tested 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:
    """Optional follow-up, not called by run_cycle: recent Weibo posts for a flagged keyword."""
    data = call_data_api("weibo/web-v2/realtime-search", {"query": keyword})
    # Observed-only field names (tested 2026-10-01): parsed_data.results[].content / .post_url
    return (data.get("parsed_data") or {}).get("results") or []


# --- Classification ---
def classify_relevance(weibo_topics: list, douyin_accounts: list) -> dict:
    """Use LLM to classify which topics are relevant to our brand/industry."""
    prompt = f"""You are a brand monitoring analyst. Review these Weibo trending topics and Douyin hot accounts and classify each as:
- "brand_mention": directly mentions our brand or competitors
- "industry_relevant": relevant to our industry but not a direct mention
- "irrelevant": not related

Brand keywords: {json.dumps(BRAND_KEYWORDS, ensure_ascii=False)}
Industry keywords: {json.dumps(INDUSTRY_KEYWORDS, ensure_ascii=False)}

WEIBO HOT SEARCH (top 20):
{json.dumps(weibo_topics[:20], ensure_ascii=False, indent=2)}

DOUYIN HOT ACCOUNTS (top 10):
{json.dumps(douyin_accounts[:10], ensure_ascii=False, indent=2)}

Respond in JSON format only, no markdown:
{{
  "brand_mentions": [{{"source": "weibo|douyin", "title": "...", "heat": N, "reason": "..."}}],
  "industry_relevant": [{{"source": "weibo|douyin", "title": "...", "heat": N, "reason": "..."}}],
  "summary": "One sentence overall assessment"
}}

Only include items that are genuinely relevant. Empty lists are fine if nothing matches."""

    resp = client.chat.completions.create(
        model="anthropic/claude-sonnet-4",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=1500,
    )
    if resp.usage:  # log real token counts so you can check the cost estimate
        print(f"  LLM tokens: {resp.usage.prompt_tokens} in / {resp.usage.completion_tokens} out")
    text = (resp.choices[0].message.content or "").strip()
    if text.startswith("`" * 3):  # tolerate a fenced reply
        text = text.strip("`").removeprefix("json").strip()
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return {"brand_mentions": [], "industry_relevant": [], "summary": "Parse error"}


# --- Alerting (Slack incoming webhook payload: {"text": ...}) ---
def send_slack_alert(classification: dict, timestamp: str):
    """Post brand mentions and industry signals to Slack."""
    if not SLACK_WEBHOOK_URL:
        print("  No Slack webhook configured, skipping alert")
        return

    brand_mentions = classification.get("brand_mentions") or []
    industry = classification.get("industry_relevant") or []
    if not brand_mentions and not industry:
        return  # Nothing to alert

    blocks = []
    if brand_mentions:
        blocks.append("🚨 *Brand mentions*\n" + "\n".join(
            f"• [{m.get('source', '?').upper()}] {m.get('title')} (heat: {m.get('heat', '?')}) — {m.get('reason', '')}"
            for m in brand_mentions
        ))
    if industry:
        blocks.append("📡 *Industry signals*\n" + "\n".join(
            f"• [{i.get('source', '?').upper()}] {i.get('title')} — {i.get('reason', '')}"
            for i in industry[:5]  # Cap industry items
        ))

    message = {
        "text": f"Social Listening Alert — {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"  Alert sent: HTTP {resp.status_code}")
    except requests.RequestException as e:
        print(f"  Alert failed: {e}")


# --- Logging ---
def log_cycle(timestamp: str, weibo: list, douyin: list, classification: dict):
    """Append one JSON line per cycle: counts, flagged items, summary."""
    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")


# --- Main loop ---
def run_cycle():
    """Execute one monitoring cycle: 2 data calls + 1 LLM call."""
    timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    print(f"\n[{timestamp}] Running social listening cycle...")

    weibo = fetch_weibo_hot_search()
    print(f"  Got {len(weibo)} Weibo topics")
    douyin = fetch_douyin_hot_accounts()
    print(f"  Got {len(douyin)} Douyin accounts")

    classification = classify_relevance(weibo, douyin)
    brand_count = len(classification.get("brand_mentions") or [])
    industry_count = len(classification.get("industry_relevant") or [])
    print(f"  Found: {brand_count} brand mentions, {industry_count} industry signals")

    # Brand mentions and industry-only signals both go to Slack
    if brand_count > 0:
        print("  ⚠️ Brand mention detected! Sending alert...")
        send_slack_alert(classification, timestamp)
    elif industry_count > 0:
        print("  📡 Industry signal detected, sending digest...")
        send_slack_alert(classification, timestamp)
    else:
        print("  ✓ Nothing relevant this cycle")

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


def run_continuous():
    """Run as a long-lived process with 15-minute intervals."""
    print("Social Listening Agent started")
    print(f"  Brand keywords: {BRAND_KEYWORDS}")
    print(f"  Industry keywords: {INDUSTRY_KEYWORDS}")
    print(f"  Check interval: {CHECK_INTERVAL}s")
    print(f"  Slack webhook: {'configured' if SLACK_WEBHOOK_URL else 'NOT configured'}")

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


if __name__ == "__main__":
    if "--once" in sys.argv:
        run_cycle()  # Single cycle for testing / 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:
    """Fast check: does any keyword appear in any item? No API call."""
    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():
    """Same as run_cycle, but only calls the LLM when a keyword shows up."""
    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)  # sends for brand or industry items
    else:
        classification = {"brand_mentions": [], "industry_relevant": [], "summary": "No keyword match"}
    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 * * * *'  # Every 15 minutes