13.3 GEO专用工具（Perplexity API、Bing Chat模拟、自建答案监控）

# Perplexity API 调用示例
import requests
import json

class PerplexityGEOAnalyzer:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.perplexity.ai"
    
    def query_with_citations(self, query, max_citations=10):
        """获取生成式答案及引用来源"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": "sonar-pro",
            "messages": [
                {
                    "role": "system",
                    "content": "你是一个搜索优化分析助手。请详细列出回答中引用的所有来源URL。"
                },
                {
                    "role": "user",
                    "content": f"请回答以下问题，并明确标注每个事实的来源：{query}"
                }
            ],
            "max_tokens": 2000,
            "temperature": 0.1
        }
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=headers,
            json=payload
        )
        
        if response.status_code == 200:
            data = response.json()
            return self._parse_citations(data)
        else:
            raise Exception(f"API请求失败: {response.status_code}")
    
    def _parse_citations(self, data):
        """解析引用信息"""
        content = data['choices'][0]['message']['content']
        citations = data.get('citations', [])
        
        return {
            'answer': content,
            'citations': citations,
            'citation_count': len(citations),
            'domain_analysis': self._analyze_domains(citations)
        }
    
    def _analyze_domains(self, citations):
        """分析引用来源的域名分布"""
        from collections import Counter
        domains = []
        for url in citations:
            try:
                from urllib.parse import urlparse
                domain = urlparse(url).netloc
                domains.append(domain)
            except:
                continue
        return dict(Counter(domains))

# geo_monitor_system.py
import asyncio
import aiohttp
import json
import hashlib
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
import logging

# 数据模型
@dataclass
class QueryResult:
    query: str
    engine: str
    answer: str
    citations: List[str]
    answer_hash: str
    timestamp: datetime
    response_time_ms: int
    
@dataclass 
class CitationRecord:
    url: str
    domain: str
    first_seen: datetime
    last_seen: datetime
    appearance_count: int
    query_count: int

class AnswerMonitor:
    def __init__(self, config: Dict):
        self.config = config
        self.engines = {}
        self.db = None
        self.alert_manager = None
        self.logger = logging.getLogger(__name__)
        
    async def initialize(self):
        """初始化监控系统"""
        # 初始化数据库连接
        await self._init_database()
        
        # 初始化引擎适配器
        self.engines = {
            'perplexity': PerplexityAdapter(self.config.get('perplexity_api_key')),
            'bing_chat': BingChatAdapter(self.config.get('bing_cookie_path')),
            # 可扩展其他引擎
        }
        
        # 初始化告警管理器
        self.alert_manager = AlertManager(self.config.get('alert_channels', {}))
        
        self.logger.info("答案监控系统初始化完成")
    
    async def monitor_keywords(self, keywords: List[str], interval_minutes: int = 60):
        """监控关键词列表"""
        while True:
            tasks = []
            for keyword in keywords:
                for engine_name, engine in self.engines.items():
                    task = self._check_single_keyword(keyword, engine_name, engine)
                    tasks.append(task)
            
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            # 处理结果
            for result in results:
                if isinstance(result, Exception):
                    self.logger.error(f"监控任务失败: {result}")
                else:
                    await self._process_result(result)
            
            # 等待下一个监控周期
            await asyncio.sleep(interval_minutes * 60)
    
    async def _check_single_keyword(self, keyword: str, engine_name: str, engine) -> QueryResult:
        """检查单个关键词在指定引擎中的表现"""
        start_time = datetime.now()
        
        try:
            result = await engine.query(keyword)
            response_time = (datetime.now() - start_time).total_seconds() * 1000
            
            return QueryResult(
                query=keyword,
                engine=engine_name,
                answer=result.get('answer', ''),
                citations=result.get('citations', []),
                answer_hash=hashlib.md5(result.get('answer', '').encode()).hexdigest(),
                timestamp=datetime.now(),
                response_time_ms=int(response_time)
            )
        except Exception as e:
            self.logger.error(f"查询失败: {keyword} @ {engine_name}: {e}")
            raise
    
    async def _process_result(self, result: QueryResult):
        """处理监控结果"""
        # 1. 存储原始结果
        await self._store_result(result)
        
        # 2. 更新引用记录
        await self._update_citations(result)
        
        # 3. 检测答案变化
        await self._detect_answer_changes(result)
        
        # 4. 检查是否需要告警
        await self._check_alerts(result)
    
    async def _detect_answer_changes(self, result: QueryResult):
        """检测答案是否发生变化"""
        # 获取上一次的结果
        previous = await self._get_previous_result(result.query, result.engine)
        
        if previous and previous.answer_hash != result.answer_hash:
            # 计算相似度
            similarity = self._calculate_similarity(previous.answer, result.answer)
            
            if similarity < 0.8:  # 相似度低于80%视为重大变化
                alert = {
                    'type': 'answer_change',
                    'severity': 'high',
                    'query': result.query,
                    'engine': result.engine,
                    'old_hash': previous.answer_hash,
                    'new_hash': result.answer_hash,
                    'similarity': similarity,
                    'timestamp': result.timestamp
                }
                await self.alert_manager.send_alert(alert)
                
                self.logger.warning(f"答案发生变化: {result.query} @ {result.engine}")
    
    def _calculate_similarity(self, text1: str, text2: str) -> float:
        """计算文本相似度（简化版）"""
        # 实际项目中可以使用更复杂的算法如BERT
        from difflib import SequenceMatcher
        return SequenceMatcher(None, text1, text2).ratio()
    
    async def _check_alerts(self, result: QueryResult):
        """检查是否需要触发告警"""
        # 检查引用数量变化
        if len(result.citations) < self.config.get('min_citations', 2):
            alert = {
                'type': 'low_citations',
                'severity': 'medium',
                'query': result.query,
                'engine': result.engine,
                'citation_count': len(result.citations),
                'timestamp': result.timestamp
            }
            await self.alert_manager.send_alert(alert)
        
        # 检查响应时间
        if result.response_time_ms > self.config.get('max_response_time', 10000):
            alert = {
                'type': 'slow_response',
                'severity': 'low',
                'query': result.query,
                'engine': result.engine,
                'response_time_ms': result.response_time_ms,
                'timestamp': result.timestamp
            }
            await self.alert_manager.send_alert(alert)
    
    async def _store_result(self, result: QueryResult):
        """存储监控结果到数据库"""
        # 实际实现使用数据库操作
        self.logger.debug(f"存储结果: {result.query} @ {result.engine}")
    
    async def _update_citations(self, result: QueryResult):
        """更新引用记录"""
        for url in result.citations:
            # 检查是否已存在
            existing = await self._get_citation(url)
            if existing:
                existing.last_seen = result.timestamp
                existing.appearance_count += 1
                existing.query_count += 1
            else:
                from urllib.parse import urlparse
                domain = urlparse(url).netloc
                record = CitationRecord(
                    url=url,
                    domain=domain,
                    first_seen=result.timestamp,
                    last_seen=result.timestamp,
                    appearance_count=1,
                    query_count=1
                )
                await self._insert_citation(record)
    
    async def generate_report(self, start_date: datetime, end_date: datetime) -> Dict:
        """生成监控报告"""
        report = {
            'period': {
                'start': start_date.isoformat(),
                'end': end_date.isoformat()
            },
            'summary': {},
            'details': {}
        }
        
        # 统计各引擎表现
        for engine_name in self.engines:
            engine_data = await self._get_engine_stats(engine_name, start_date, end_date)
            report['summary'][engine_name] = engine_data
        
        # 统计引用来源
        top_citations = await self._get_top_citations(10, start_date, end_date)
        report['top_citations'] = top_citations
        
        return report

# 引擎适配器示例
class PerplexityAdapter:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.perplexity.ai"
    
    async def query(self, keyword: str) -> Dict:
        async with aiohttp.ClientSession() as session:
            headers = {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            }
            
            payload = {
                "model": "sonar-pro",
                "messages": [
                    {
                        "role": "system",
                        "content": "你是一个搜索优化分析助手。"
                    },
                    {
                        "role": "user", 
                        "content": keyword
                    }
                ]
            }
            
            async with session.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload
            ) as response:
                data = await response.json()
                return {
                    'answer': data['choices'][0]['message']['content'],
                    'citations': data.get('citations', [])
                }

class AlertManager:
    def __init__(self, channels: Dict):
        self.channels = channels
    
    async def send_alert(self, alert: Dict):
        """发送告警到配置的渠道"""
        # 邮件告警
        if 'email' in self.channels:
            await self._send_email(alert)
        
        # 钉钉/飞书/企业微信
        if 'webhook' in self.channels:
            await self._send_webhook(alert)
        
        # 短信告警
        if 'sms' in self.channels:
            await self._send_sms(alert)
    
    async def _send_webhook(self, alert: Dict):
        """发送到Webhook（钉钉/飞书）"""
        webhook_url = self.channels['webhook']
        payload = {
            "msgtype": "markdown",
            "markdown": {
                "title": f"GEO监控告警: {alert['type']}",
                "text": f"### GEO监控告警\n"
                       f"- **类型**: {alert['type']}\n"
                       f"- **严重程度**: {alert.get('severity', 'info')}\n"
                       f"- **关键词**: {alert.get('query', 'N/A')}\n"
                       f"- **引擎**: {alert.get('engine', 'N/A')}\n"
                       f"- **时间**: {alert.get('timestamp', datetime.now()).isoformat()}\n"
                       f"- **详情**: 请查看监控仪表盘获取更多信息"
            }
        }
        
        async with aiohttp.ClientSession() as session:
            await session.post(webhook_url, json=payload)

# 主程序入口
async def main():
    config = {
        'perplexity_api_key': 'your_api_key_here',
        'bing_cookie_path': './bing_cookies.json',
        'min_citations': 2,
        'max_response_time': 10000,
        'alert_channels': {
            'webhook': 'https://oapi.dingtalk.com/robot/send?access_token=xxx',
            'email': 'admin@example.com'
        }
    }
    
    monitor = AnswerMonitor(config)
    await monitor.initialize()
    
    keywords = [
        'GEO优化最佳实践',
        '生成式搜索引擎优化',
        'Perplexity引用机制'
    ]
    
    await monitor.monitor_keywords(keywords, interval_minutes=30)

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