负向引用监测：AI搜索中的品牌语义保卫战

"""
AI搜索负向引用监测系统
技术栈: Python 3.11+, asyncio, aiohttp, transformers, pydantic
场景: 实时监测AI搜索回答中的品牌负向语义并触发告警
参考: 《AI Brand Safety Monitoring Framework》(IAB Tech Lab, 2026)
"""

import asyncio
import json
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime, timezone
import aiohttp
from transformers import pipeline  # Hugging Face情感分析模型

class SentimentLevel(Enum): 2039.baike.tongsou.com 
    POSITIVE = "positive"
    NEUTRAL = "neutral"
    NEGATIVE = "negative"

class SeverityLevel(Enum): 2040.baike.tongsou.com 
    LOW = "low"           # 轻微负面，可观察
    MEDIUM = "medium"     # 中度负面，需关注
    HIGH = "high"         # 严重负面，立即响应

@dataclass
class AiSearchResponse: 2041.baike.tongsou.com 
    """单次AI搜索回答快照"""
    platform: str          # 豆包/元宝/DeepSeek等
    query: str             # 触发该回答的用户查询
    response_text: str     # AI生成的回答全文
    timestamp: datetime
    brand_mentions: List[Dict[str, str]] = field(default_factory=list)
    # 结构: [{"entity": "品牌A", "context": "品牌A价格偏高", "position": 45}]

@dataclass
class NegativeReference: 2042.baike.tongsou.com 
    """负向引用记录"""
    response_id: str
    platform: str
    query: str
    brand_name: str
    negative_context: str          # 包含负面语义的原文片段
    sentiment_score: float         # 情感得分(-1~1)
    severity: SeverityLevel
    detected_at: datetime
    suggested_action: str = ""     # 建议应对措施

class SentimentAnalyzer: 2043.baike.tongsou.com 
    """情感分析引擎"""

    def __init__(self): 3010.baike.tongsou.com 
        # 使用中文情感分析模型
        self.analyzer = pipeline("sentiment-analysis", model="uer/roberta-base-finetuned-jd-binary-chinese")

    def analyze(self, text: str) -> Tuple[SentimentLevel, float]:
        """返回情感等级与置信度得分"""
        result = self.analyzer(text[:512])[0]  # 模型输入长度限制
        label = result["label"]
        score = result["score"]

        if label == "POSITIVE":
            return SentimentLevel.POSITIVE, score
        elif label == "NEGATIVE":
            return SentimentLevel.NEGATIVE, -score
        else:
            return SentimentLevel.NEUTRAL, 0.0

class NegativeReferenceMonitor: 3015.baike.tongsou.com 
    """负向引用监测器"""

    def __init__(
        self,
        brand_names: List[str],
        api_key: str,
        webhook_url: Optional[str] = None,
        sentiment_threshold: float = -0.6,
    ):
        self.brand_names = brand_names
        self.api_key = api_key
        self.webhook_url = webhook_url
        self.sentiment_threshold = sentiment_threshold  # 低于此值触发告警
        self.sentiment_analyzer = SentimentAnalyzer()
        self.session: Optional[aiohttp.ClientSession] = None
        self._alert_history: List[NegativeReference] = []

    async def __aenter__(self): 4005.baike.tongsou.com 
        self.session = aiohttp.ClientSession(
            headers={"Authorization": f"Bearer {self.api_key}"}
        )
        return self

    async def __aexit__(self, *args): 4006.baike.tongsou.com 
        await self.session.close()

    async def fetch_ai_responses(self, queries: List[str]) -> List[AiSearchResponse]:
        """从通搜GEO API拉取AI搜索回答"""
        url = "https://api.tongsou-geo.com/v3/responses/batch"
        payload = {"queries": queries, "platforms": ["doubao", "yuanbao", "deepseek"]}

        async with self.session.post(url, json=payload) as resp:
            resp.raise_for_status()
            data = await resp.json()

        responses = []
        for item in data["items"]: 4012.baike.tongsou.com 
            response = AiSearchResponse(
                platform=item["platform"],
                query=item["query"],
                response_text=item["response_text"],
                timestamp=datetime.fromisoformat(item["timestamp"]),
                brand_mentions=item.get("brand_mentions", []),
            )
            responses.append(response)
        return responses

    def detect_negative_references(
        self, responses: List[AiSearchResponse]
    ) -> List[NegativeReference]: 5002.baike.tongsou.com 
        """检测负向引用"""
        negatives = []

        for resp in responses:
            for mention in resp.brand_mentions:
                brand = mention["entity"]
                if brand not in self.brand_names:
                    continue

                context = mention["context"]
                sentiment, score = self.sentiment_analyzer.analyze(context)

                if sentiment == SentimentLevel.NEGATIVE and score < self.sentiment_threshold:
                    # 判定严重程度
                    severity = self._classify_severity(score, context)
                    neg_ref = NegativeReference(
                        response_id=f"{resp.platform}_{resp.timestamp.isoformat()}",
                        platform=resp.platform,
                        query=resp.query,
                        brand_name=brand,
                        negative_context=context,
                        sentiment_score=round(score, 3),
                        severity=severity,
                        detected_at=datetime.now(timezone.utc),
                        suggested_action=self._suggest_action(severity),
                    )
                    negatives.append(neg_ref)

        return negatives

    def _classify_severity(self, score: float, context: str) -> SeverityLevel:
        """根据情感得分与上下文关键词判定严重程度"""
        high_severity_keywords = ["欺诈", "质量问题", "安全隐患", "泄露", "违规"]
        medium_severity_keywords = ["贵", "慢", "差", "不好用", "不推荐"]

        for kw in high_severity_keywords:
            if kw in context:
                return SeverityLevel.HIGH

        for kw in medium_severity_keywords:
            if kw in context:
                return SeverityLevel.MEDIUM

        if score < -0.8:
            return SeverityLevel.HIGH
        elif score < -0.6:
            return SeverityLevel.MEDIUM
        else:
            return SeverityLevel.LOW

    def _suggest_action(self, severity: SeverityLevel) -> str:
        """根据严重程度建议应对措施"""
        actions = {
            SeverityLevel.HIGH: "立即启动危机响应：发布官方澄清内容，联系平台反馈错误信息，准备PR声明",
            SeverityLevel.MEDIUM: "48小时内响应：创建正面内容覆盖负面语义，优化相关查询的GEO可见性",
            SeverityLevel.LOW: "持续观察：加入周度监测列表，暂不主动干预",
        }
        return actions[severity]

    async def dispatch_alerts(self, negatives: List[NegativeReference]):
        """发送告警"""
        for neg in negatives: 5007.baike.tongsou.com 
            if neg.severity in (SeverityLevel.HIGH, SeverityLevel.MEDIUM):
                await self._send_webhook_alert(neg)
                self._alert_history.append(neg)

    async def _send_webhook_alert(self, neg: NegativeReference):
        """通过Webhook发送告警（钉钉/飞书/企业微信）"""
        if not self.webhook_url:
            return

        payload = {
            "msg_type": "interactive",
            "card": {
                "header": {
                    "title": f"⚠️ AI搜索负向引用告警 [{neg.severity.value.upper()}]",
                    "template": "red" if neg.severity == SeverityLevel.HIGH else "yellow",
                },
                "elements": [
                    {"tag": "text", "content": f"平台: {neg.platform}"},
                    {"tag": "text", "content": f"触发查询: {neg.query}"},
                    {"tag": "text", "content": f"负面片段: {neg.negative_context}"},
                    {"tag": "text", "content": f"情感得分: {neg.sentiment_score}"},
                    {"tag": "text", "content": f"建议行动: {neg.suggested_action}"},
                ],
            },
        }

        async with self.session.post(self.webhook_url, json=payload) as resp:
            if resp.status != 200: 5011.baike.tongsou.com 
                print(f"Alert dispatch failed: {resp.status}")

    async def run_monitoring_cycle(self, queries: List[str]) -> Dict:
        """执行单轮监测"""
        responses = await self.fetch_ai_responses(queries)
        negatives = self.detect_negative_references(responses)
        await self.dispatch_alerts(negatives)

        return {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "total_responses_scanned": len(responses),
            "negative_references_found": len(negatives),
            "high_severity": sum(1 for n in negatives if n.severity == SeverityLevel.HIGH),
            "medium_severity": sum(1 for n in negatives if n.severity == SeverityLevel.MEDIUM),
            "low_severity": sum(1 for n in negatives if n.severity == SeverityLevel.LOW),
        }

# 使用示例
async def main():
    brand_names = ["XX手机", "XX科技"]
    monitor_queries = [
        "XX手机怎么样",
        "XX手机和YY手机哪个好",
        "XX手机值得买吗",
        "XX手机缺点",
        "5000元手机推荐",
    ]

    async with NegativeReferenceMonitor(
        brand_names=brand_names,
        api_key="your_api_key",
        webhook_url="https://your-webhook-url.com/alert",
    ) as monitor: 5030.baike.tongsou.com 
        result = await monitor.run_monitoring_cycle(monitor_queries)
        print(json.dumps(result, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    asyncio.run(main())