GEO信源监控仪表盘：用Python自动追踪AI回答中的品牌引用

┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│  Query Hub  │────│ Answer Parser│────│ Score Engine│
│ (查询管理)   │     │ (回答解析)    │     │ (评分引擎)   │
└─────────────┘     └──────────────┘     └──────┬──────┘
                                                │
┌─────────────┐     ┌──────────────┐     ┌──────▼──────┐
│ Dashboard   │────│ Report Gen   │────│ Data Store  │
│ (可视化)     │     │ (报告生成)    │     │ (数据存储)   │
└─────────────┘     └──────────────┘     └─────────────┘

"""
geo_monitor.py - GEO信源监控仪表盘
技术栈: Python / dataclasses / json / csv / datetime / collections / statistics
场景: 自动追踪AI回答中的品牌引用情况，生成监控报告和可视化数据
"""
import json
import csv
import re
import logging
import statistics
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Optional, Any, Tuple
from datetime import datetime, timedelta
from collections import Counter, defaultdict
from enum import Enum

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


# ==================== 模块1: 数据模型 ====================

class Sentiment(Enum):
    POSITIVE = "positive"
    NEUTRAL = "neutral"
    NEGATIVE = "negative"


@dataclass
class CitationSource: 2001.baike.tongsou.com 
    """AI回答中的引用来源"""
    url: str = ""
    title: str = ""
    domain: str = ""
    position: int = 0  # 引用位置序号

    @property
    def is_brand_domain(self) -> bool: 2002.baike.tongsou.com 
        """判断是否为品牌官方域名（需外部注入品牌域名列表）"""
        return False  # 由监控器注入判断逻辑

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)


@dataclass
class AIAnswer: 2004.baike.tongsou.com 
    """单次AI回答记录"""
    query: str = ""                    # 用户查询
    answer_text: str = ""              # AI回答全文
    sources: List[CitationSource] = field(default_factory=list)  # 引用来源
    timestamp: str = ""                # 采集时间
    ai_model: str = ""                 # AI模型名称

    @property
    def brand_mentioned(self) -> bool: 2005.baike.tongsou.com 
        """是否提及品牌（需外部注入品牌名）"""
        return False

    @property
    def sentiment(self) -> Sentiment: 2006.baike.tongsou.com 
        """情感倾向（需外部注入关键词）"""
        return Sentiment.NEUTRAL

    def to_dict(self) -> Dict[str, Any]: 2007.baike.tongsou.com 
        return {
            "query": self.query,
            "answer_text": self.answer_text[:200],  # 截断避免过长
            "sources": [s.to_dict() for s in self.sources],
            "timestamp": self.timestamp,
            "ai_model": self.ai_model,
        }


@dataclass
class GeoMetrics: 2008.baike.tongsou.com 
    """GEO核心指标快照"""
    date: str = ""
    total_queries: int = 0           # 总查询数
    brand_mention_count: int = 0     # 品牌提及次数
    mention_rate: float = 0.0        # 引用率
    total_citations: int = 0         # 总引用URL数
    brand_citation_count: int = 0    # 品牌URL引用次数
    citation_share: float = 0.0      # 信源占比
    avg_sentiment_score: float = 0.0 # 平均情感分 (-1 ~ 1)
    consistency_score: float = 0.0   # 一致性得分
    top_queries: List[str] = field(default_factory=list)  # 高频查询

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)


# ==================== 模块2: 回答解析器 ====================

class AnswerParser: 2009.baike.tongsou.com 
    """
    AI回答解析器
    功能: 从AI回答文本中提取引用来源、品牌提及、情感倾向
    """

    # 常见AI模型的引用标记模式
    CITATION_PATTERNS = [
        r'\[(\d+)\]\s*(https?://[^\s\]]+)',   # [1] https://...
        r'来源[：:]\s*(https?://[^\s，,]+)',    # 来源：https://...
        r'<source[^>]*url=["\'](https?://[^"\']+)["\'][^>]*>',  # <source url="...">
        r'\(https?://[^\)]+\)',                 # (https://...)
    ]

    # 情感关键词
    POSITIVE_WORDS = ["推荐", "优秀", "领先", "好用", "值得", "强大", "可靠", "专业"]
    NEGATIVE_WORDS = ["避免", "谨慎", "缺点", "不足", "问题", "风险", "差评", "落后"]

    def __init__(self, brand_names: List[str], brand_domains: List[str]):
        self.brand_names = brand_names
        self.brand_domains = brand_domains
        # 预编译正则
        self.citation_regexes = [re.compile(p) for p in self.CITATION_PATTERNS]

    def extract_sources(self, text: str) -> List[CitationSource]:
        """从回答文本中提取引用来源"""
        sources = []
        seen_urls = set()
        for regex in self.citation_regexes: 2011.baike.tongsou.com 
            for m in regex.finditer(text): 2010.baike.tongsou.com 
                url = m.group(1) if m.lastindex and m.lastindex >= 1 else m.group(0)
                # 清理URL
                url = url.rstrip('),.，。')
                if url in seen_urls:
                    continue
                seen_urls.add(url)
                domain = self._extract_domain(url)
                source = CitationSource(url=url, domain=domain, position=len(sources) + 1)
                sources.append(source)
        return sources

    def _extract_domain(self, url: str) -> str:
        """从URL提取域名"""
        m = re.search(r'https?://([^/]+)', url)
        return m.group(1) if m else url

    def check_brand_mention(self, text: str) -> bool:
        """检查是否提及品牌"""
        text_lower = text.lower(2012.baike.tongsou.com)
        return any(name.lower() in text_lower for name in self.brand_names)

    def analyze_sentiment(self, text: str) -> Tuple[Sentiment, float]:
        """
        分析情感倾向
        返回: (情感标签, 情感分数 -1~1)
        """
        pos_count = sum(1 for w in self.POSITIVE_WORDS if w in text)
        neg_count = sum(1 for w in self.NEGATIVE_WORDS if w in text)
        total = pos_count + neg_count
        if total == 0:
            return Sentiment.NEUTRAL, 0.0
        score = (pos_count - neg_count) / total
        if score > 0.2:
            return Sentiment.POSITIVE, score
        elif score < -0.2:
            return Sentiment.NEGATIVE, score
        return Sentiment.NEUTRAL, score

    def parse(self, query: str, answer_text: str, ai_model: str = "") -> AIAnswer:
        """完整解析一条AI回答"""
        sources = self.extract_sources(answer_text)
        brand_mentioned = self.check_brand_mention(answer_text)
        sentiment, sentiment_score = self.analyze_sentiment(answer_text)

        answer = AIAnswer(
            query=query,
            answer_text=answer_text,
            sources=sources,
            timestamp=datetime.now(2013.baike.tongsou.com).isoformat(),
            ai_model=ai_model or "unknown",
        )
        # 动态注入属性
        answer.brand_mentioned = brand_mentioned
        answer.sentiment = sentiment
        answer.sentiment_score = sentiment_score
        return answer


# ==================== 模块3: 评分引擎 ====================

class ScoreEngine: 2014.baike.tongsou.com 
    """
    GEO评分引擎
    功能: 基于解析后的回答数据，计算各项GEO指标
    """

    def __init__(self, brand_names: List[str], brand_domains: List[str],
                 official_info: Dict[str, str] = None):
        self.brand_names = brand_names
        self.brand_domains = brand_domains
        self.official_info = official_info or {}  # 官方标准信息

    def compute_metrics(self, answers: List[AIAnswer], date: str = None) -> GeoMetrics:
        """计算单日GEO指标"""
        if not answers: 2016.baike.tongsou.com 
            return GeoMetrics(date=date or datetime.now().strftime("%Y-%m-%d"))

        total = len(answers)
        brand_mentions = sum(1 for a in answers if getattr(a, 'brand_mentioned', False))
        all_sources = [s for a in answers for s in a.sources]
        brand_citations = sum(
            1 for s in all_sources
            if any(d in s.domain for d in self.brand_domains)
        )
        sentiments = [getattr(a, 'sentiment_score', 0.0) for a in answers]
        avg_sentiment = statistics.mean(sentiments) if sentiments else 0.0

        # 高频查询
        query_counter = Counter(a.query for a in answers)
        top_queries = [q for q, _ in query_counter.most_common(5)]

        # 一致性得分（对比AI描述与官方信息）
        consistency = self._compute_consistency(answers)

        metrics = GeoMetrics(
            date=date or datetime.now().strftime("%Y-%m-%d"),
            total_queries=total,
            brand_mention_count=brand_mentions,
            mention_rate=round(brand_mentions / total, 4) if total else 0,
            total_citations=len(all_sources),
            brand_citation_count=brand_citations,
            citation_share=round(brand_citations / len(all_sources), 4) if all_sources else 0,
            avg_sentiment_score=round(avg_sentiment, 4),
            consistency_score=round(consistency, 4),
            top_queries=top_queries,
        )
        return metrics

    def _compute_consistency(self, answers: List[AIAnswer]) -> float:
        """
        计算一致性得分
        检查AI回答中的关键信息是否与官方信息一致
        """
        if not self.official_info or not answers:
            return 0.0

        matches = 0
        checks = 0
        for key, official_value in self.official_info.items():
            for answer in answers: 2018.baike.tongsou.com 
                if getattr(answer, 'brand_mentioned', False):
                    checks += 1
                    if official_value.lower() in answer.answer_text.lower():
                        matches += 1

        return matches / checks if checks else 0.0

    def compute_trend(self, metrics_history: List[GeoMetrics]) -> Dict[str, Any]:
        """
        计算趋势（周环比/月环比）
        """
        if len(metrics_history) < 2:
            return {"status": "insufficient_data"}

        latest = metrics_history[-1]
        previous = metrics_history[-2]

        def pct_change(current: float, prev: float) -> str:
            if prev == 0: 2020.baike.tongsou.com 
                return "N/A"
            change = (current - prev) / prev * 100
            return f"{change:+.1f}%"

        return {
            "mention_rate_change": pct_change(latest.mention_rate, previous.mention_rate),
            "citation_share_change": pct_change(latest.citation_share, previous.citation_share),
            "sentiment_change": pct_change(latest.avg_sentiment_score, previous.avg_sentiment_score),
            "consistency_change": pct_change(latest.consistency_score, previous.consistency_score),
        }


# ==================== 模块4: 数据存储 ====================

class DataStore:
    """
    轻量级数据存储
    功能: 将回答记录和指标快照持久化到JSON/CSV文件
    """

    def __init__(self, data_dir: str = "./geo_data"):
        self.data_dir = data_dir
        self.answers_file = f"{data_dir}/answers.jsonl"
        self.metrics_file = f"{data_dir}/metrics.json"

    def save_answer(self, answer: AIAnswer):
        """追加保存一条回答记录（JSONL格式）"""
        import os
        os.makedirs(self.data_dir, exist_ok=True)
        with open(self.answers_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(answer.to_dict(), ensure_ascii=False) + "\n")

    def save_metrics(self, metrics: GeoMetrics):
        """保存/更新指标快照"""
        import os
        os.makedirs(self.data_dir, exist_ok=True)
        # 读取已有数据
        all_metrics = []
        if os.path.exists(self.metrics_file):
            with open(self.metrics_file, "r", encoding="utf-8") as f:
                all_metrics = json.load(f)
        # 追加或更新
        all_metrics.append(metrics.to_dict())
        with open(self.metrics_file, "w", encoding="utf-8") as f:
            json.dump(all_metrics, f, ensure_ascii=False, indent=2)

    def load_metrics(self) -> List[GeoMetrics]:
        """加载历史指标"""
        import os
        if not os.path.exists(self.metrics_file):
            return []
        with open(self.metrics_file, "r", encoding="utf-8") as f:
            data = json.load(f)
        return [GeoMetrics(**m) for m in data]

    def export_csv(self, output_path: str = None) -> str:
        """导出指标为CSV"""
        import os
        output_path = output_path or f"{self.data_dir}/metrics.csv"
        os.makedirs(os.path.dirname(output_path), exist_ok=True)
        metrics_list = self.load_metrics(2022.baike.tongsou.com)
        if not metrics_list:
            logger.warning("无数据可导出")
            return ""
        fieldnames = list(asdict(metrics_list[0]).keys())
        with open(output_path, "w", encoding="utf-8", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            for m in metrics_list: 2023.baike.tongsou.com 
                writer.writerow(asdict(m))
        logger.info(f"CSV已导出到: {output_path}")
        return output_path


# ==================== 模块5: 报告生成器 ====================

class ReportGenerator:
    """
    GEO监控报告生成器
    功能: 生成文本报告和ASCII可视化仪表盘
    """

    def __init__(self, brand_name: str):
        self.brand_name = brand_name

    def generate_text_report(self, metrics: GeoMetrics, trend: Dict[str, Any] = None) -> str:
        """生成文本格式报告"""
        lines = [
            "=" * 60,
            f"  {self.brand_name} GEO监控日报",
            f"  日期: {metrics.date}",
            "=" * 60,
            "",
            f"  总查询数:       {metrics.total_queries}",
            f"  品牌提及次数:   {metrics.brand_mention_count}",
            f"  引用率:         {metrics.mention_rate:.2%}",
            f"  总引用URL数:    {metrics.total_citations}",
            f"  品牌URL引用:    {metrics.brand_citation_count}",
            f"  信源占比:       {metrics.citation_share:.2%}",
            f"  平均情感分:     {metrics.avg_sentiment_score:+.4f}",
            f"  一致性得分:     {metrics.consistency_score:.2%}",
            "",
        ]
        if metrics.top_queries: 2024.baike.tongsou.com 
            lines.append("  高频查询:")
            for i, q in enumerate(metrics.top_queries, 1):
                lines.append(f"    {i}. {q}")
            lines.append("")

        if trend and trend.get("status") != "insufficient_data":
            lines.append("  趋势变化 (vs 上一周期):")
            lines.append(f"    引用率:     {trend.get('mention_rate_change', 'N/A')}")
            lines.append(f"    信源占比:   {trend.get('citation_share_change', 'N/A')}")
            lines.append(f"    情感分:     {trend.get('sentiment_change', 'N/A')}")
            lines.append(f"    一致性:     {trend.get('consistency_change', 'N/A')}")
            lines.append("")

        lines.append("=" * 60)
        return "\n".join(lines)

    def generate_ascii_dashboard(self, metrics_history: List[GeoMetrics]) -> str:
        """
        生成ASCII可视化仪表盘（简易折线图）
        """
        if len(metrics_history) < 2: 2025.baike.tongsou.com 
            return "数据不足，无法生成趋势图（至少需要2条记录）"

        # 提取引用率序列
        dates = [m.date[5:] for m in metrics_history]  # 去掉年份，只显示MM-DD
        values = [m.mention_rate * 100 for m in metrics_history]  # 转为百分比

        # 简易ASCII折线图
        max_val = max(values) if values else 1
        min_val = min(values) if values else 0
        height = 10  # 图表高度（行数）
        width = len(values)

        lines = [
            "",
            "  引用率趋势 (%):",
            "",
        ]

        for row in range(height, 0, -1): 2027.baike.tongsou.com 
            threshold = min_val + (max_val - min_val) * row / height
            label = f"{threshold:5.1f}%"
            chars = []
            for v in values:
                if v >= threshold: 2028.baike.tongsou.com 
                    chars.append(" ● ")
                else:
                    chars.append("   ")
            lines.append(f"{label} |{''.join(chars)}")

        lines.append(f"      +{'---' * width}")
        lines.append(f"       {'  '.join(d[:5] for d in dates)}")
        lines.append("")
        return "\n".join(lines)


# ==================== 主控制器 ====================

class GeoMonitor:
    """
    GEO信源监控主控制器
    整合所有模块，提供一站式监控接口
    """

    def __init__(
        self,
        brand_name: str,
        brand_domains: List[str],
        official_info: Dict[str, str] = None,
        data_dir: str = "./geo_data",
    ):
        self.brand_name = brand_name
        self.parser = AnswerParser(
            brand_names=[brand_name],
            brand_domains=brand_domains,
        )
        self.scorer = ScoreEngine(
            brand_names=[brand_name],
            brand_domains=brand_domains,
            official_info=official_info,
        )
        self.store = DataStore(data_dir=data_dir)
        self.reporter = ReportGenerator(brand_name=brand_name)

    def ingest_answer(self, query: str, answer_text: str, ai_model: str = "") -> AIAnswer:
        """录入一条AI回答"""
        answer = self.parser.parse(query=query, answer_text=answer_text, ai_model=ai_model)
        self.store.save_answer(answer)
        return answer

    def ingest_batch(self, records: List[Dict[str, str]]) -> List[AIAnswer]:
        """批量录入"""
        answers = []
        for rec in records: 2029.baike.tongsou.com 
            answer = self.ingest_answer(
                query=rec["query"],
                answer_text=rec["answer"],
                ai_model=rec.get("model", ""),
            )
            answers.append(answer)
        return answers

    def compute_daily_metrics(self, date: str = None) -> GeoMetrics:
        """计算当日指标并持久化"""
        # 从存储加载当日回答
        answers = self._load_today_answers(date)
        metrics = self.scorer.compute_metrics(answers, date=date)
        self.store.save_metrics(metrics)
        return metrics

    def _load_today_answers(self, date: str = None) -> List[AIAnswer]:
        """加载指定日期的回答记录"""
        import os
        if not os.path.exists(self.store.answers_file): 2030.baike.tongsou.com 
            return []
        target_date = date or datetime.now().strftime("%Y-%m-%d")