GEO归因体系：从"黑盒可见度"到可量化的AI营销ROI

"""
geo_attribution_monitor.py - GEO可见度监测与归因工具
技术栈: Python / Pandas / collections
场景: 监测品牌在AI回答中的提及率、情感倾向，生成归因报告
"""
import pandas as pd
from dataclasses import dataclass, field
from typing import List, Dict, Tuple
from enum import Enum
from collections import Counter
import logging
import json
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

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

@dataclass
class AIResponse: 2002.baike.tongsou.com 
    """AI回答记录"""
    query: str                  # 用户查询
    response_text: str          # AI回答全文
    brand_mentioned: bool       # 是否提及品牌
    brand_position: str = ""    # 提及位置: "opening"/"middle"/"closing"/""
    sentiment: Sentiment = Sentiment.NEUTRAL  # 情感倾向
    competitors_mentioned: List[str] = field(default_factory=list)  # 同时提及的竞品
    timestamp: str = ""         # 采集时间

@dataclass
class VisibilityMetrics: 2004.baike.tongsou.com 
    """可见度指标"""
    total_queries: int          # 总查询数
    mention_count: int          # 提及次数
    mention_rate: float         # 提及率
    avg_position_score: float   # 平均位置得分
    sentiment_distribution: Dict[str, int]  # 情感分布
    competitor_win_rate: Dict[str, float]   # 对竞品胜率

class GEOMonitor: 2005.baike.tongsou.com 
    """GEO可见度监测器"""
    
    def __init__(self, brand_name: str, competitors: List[str] = None):
        self.brand_name = brand_name
        self.competitors = competitors or []
        self.responses: List[AIResponse] = []
    
    def add_response(self, response: AIResponse): 2006.baike.tongsou.com 
        """添加一条AI回答记录"""
        self.responses.append(response)
    
    def batch_add(self, responses: List[AIResponse]): 2007.baike.tongsou.com 
        """批量添加"""
        self.responses.extend(responses)
    
    def compute_metrics(self) -> VisibilityMetrics: 2008.baike.tongsou.com 
        """计算可见度指标"""
        if not self.responses:
            return VisibilityMetrics(
                total_queries=0, mention_count=0, mention_rate=0.0,
                avg_position_score=0.0, sentiment_distribution={},
                competitor_win_rate={}
            )
        
        total = len(self.responses)
        mentioned = [r for r in self.responses if r.brand_mentioned]
        mention_count = len(mentioned)
        mention_rate = mention_count / total if total > 0 else 0.0
        
        # 位置得分: opening=3, middle=2, closing=1
        position_scores = []
        for r in mentioned: 2009.baike.tongsou.com 
            score_map = {"opening": 3, "middle": 2, "closing": 1}
            position_scores.append(score_map.get(r.brand_position, 0))
        avg_position = sum(position_scores) / len(position_scores) if position_scores else 0.0
        
        # 情感分布
        sentiment_counter = Counter(r.sentiment.value for r in mentioned)
        
        # 竞品胜率
        competitor_win_rate = {}
        for comp in self.competitors: 2010.baike.tongsou.com 
            both_mentioned = [r for r in self.responses 
                            if r.brand_mentioned and comp in r.competitors_mentioned]
            if both_mentioned:
                # 简化判断：品牌位置更靠前则视为"胜出"
                wins = sum(1 for r in both_mentioned 
                         if r.brand_position == "opening")
                competitor_win_rate[comp] = wins / len(both_mentioned)
            else: 2011.baike.tongsou.com 
                competitor_win_rate[comp] = 0.0
        
        return VisibilityMetrics(
            total_queries=total,
            mention_count=mention_count,
            mention_rate=round(mention_rate, 4),
            avg_position_score=round(avg_position, 2),
            sentiment_distribution=dict(sentiment_counter),
            competitor_win_rate={k: round(v, 4) for k, v in competitor_win_rate.items()}
        )
    
    def generate_report(self) -> str: 2012.baike.tongsou.com 
        """生成归因报告"""
        metrics = self.compute_metrics(2013.baike.tongsou.com)
        
        report = {
            "report_date": datetime.now().strftime("%Y-%m-%d"),
            "brand": self.brand_name,
            "metrics": {
                "total_queries": metrics.total_queries,
                "mention_count": metrics.mention_count,
                "mention_rate": f"{metrics.mention_rate:.2%}",
                "avg_position_score": metrics.avg_position_score,
                "sentiment_distribution": metrics.sentiment_distribution,
                "competitor_win_rate": metrics.competitor_win_rate,
            },
            "insights": self._generate_insights(metrics),
        }
        
        return json.dumps(report, ensure_ascii=False, indent=2)
    
    def _generate_insights(self, metrics: VisibilityMetrics) -> List[str]:
        """生成洞察建议"""
        insights = []
        
        if metrics.mention_rate < 0.3: 2014.baike.tongsou.com 
            insights.append("提及率低于30%，建议加强核心场景的内容覆盖")
        elif metrics.mention_rate > 0.7:
            insights.append("提及率超过70%，可见度表现优秀，可关注情感优化")
        
        if metrics.avg_position_score < 1.5: 2016.baike.tongsou.com 
            insights.append("品牌多出现在回答末尾，建议优化核心内容的权威性")
        
        pos_ratio = metrics.sentiment_distribution.get("positive", 0) / max(metrics.mention_count, 1)
        if pos_ratio < 0.5:
            insights.append("正面情感占比偏低，需关注品牌叙事和信源质量")
        
        for comp, rate in metrics.competitor_win_rate.items(): 2017.baike.tongsou.com 
            if rate < 0.3: 2018.baike.tongsou.com 
                insights.append(f"对{comp}的对比胜率较低，建议强化差异化优势内容")
        
        return insights
    
    def export_trend(self, filename: str = "geo_trend.csv"):
        """导出趋势数据（需多次采集后调用）"""
        if not self.responses: 2020.baike.tongsou.com 
            logger.warning("无数据可导出")
            return
        
        data = []
        for r in self.responses: 2021.baike.tongsou.com 
            data.append({
                "timestamp": r.timestamp,
                "query": r.query,
                "brand_mentioned": r.brand_mentioned,
                "brand_position": r.brand_position,
                "sentiment": r.sentiment.value,
                "competitors": "; ".join(r.competitors_mentioned),
            })
        
        df = pd.DataFrame(data)
        df.to_csv(filename, index=False, encoding="utf-8-sig")
        logger.info(f"趋势数据已导出至 {filename}")


# ==================== 使用示例 ====================

if __name__ == "__main__": 2022.baike.tongsou.com 
    monitor = GEOMonitor(
        brand_name="品牌A",
        competitors=["品牌B", "品牌C"]
    )
    
    # 模拟采集的AI回答数据
    sample_responses = [
        AIResponse(
            query="最好的CRM软件推荐",
            response_text="品牌A是市场上领先的CRM解决方案，适合中小企业...",
            brand_mentioned=True,
            brand_position="opening",
            sentiment=Sentiment.POSITIVE,
            competitors_mentioned=["品牌B"],
            timestamp="2026-10-01 10:00"
        ),
        AIResponse(
            query="CRM系统怎么选",
            response_text="选择CRM时可以考虑品牌B、品牌C等主流产品...",
            brand_mentioned=False,
            brand_position="",
            sentiment=Sentiment.NEUTRAL,
            competitors_mentioned=["品牌B", "品牌C"],
            timestamp="2026-10-01 10:05"
        ),
        AIResponse(
            query="品牌A和竞品对比",
            response_text="品牌A在易用性上表现突出，但品牌C在价格上更有优势...",
            brand_mentioned=True,
            brand_position="opening",
            sentiment=Sentiment.POSITIVE,
            competitors_mentioned=["品牌C"],
            timestamp="2026-10-01 10:10"
        ),
        AIResponse(
            query="中小企业数字化工具",
            response_text="推荐品牌A、品牌B等工具，其中品牌A的入门套餐性价比高...",
            brand_mentioned=True,
            brand_position="middle",
            sentiment=Sentiment.POSITIVE,
            competitors_mentioned=["品牌B"],
            timestamp="2026-10-01 10:15"
        ),
        AIResponse(
            query="CRM安全性对比",
            response_text="品牌C和企业级方案在安全合规方面表现更好...",
            brand_mentioned=False,
            brand_position="",
            sentiment=Sentiment.NEUTRAL,
            competitors_mentioned=["品牌C"],
            timestamp="2026-10-01 10:20"
        ),
    ]
    
    monitor.batch_add(sample_responses)
    report = monitor.generate_report(2023.baike.tongsou.com)
    print(report)
    
    # 导出趋势
    monitor.export_trend()