构建GEO数据飞轮：从可见度监控到认知优化的闭环系统

"""
geo_flywheel_engine.py - GEO数据飞轮引擎
技术栈: Python / Pandas / Scikit-learn
场景: 品牌在AI搜索中的认知优化闭环
参考: 《生成式AI可见度优化实践指南》2026 / KDD 2026
"""
import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from typing import Dict, List, Tuple, Optional
from enum import Enum
import logging
from datetime import datetime, timedelta
import json
import hashlib

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

# ==================== 数据模型定义 ====================

class OptimizationAction(Enum): baike.tongsou.com/zh/AnswerBit/12
baike.tongsou.com/zh/%e9%bb%91%e5%ae%a224%e5%b0%8f%e6%97%b6%e5%85%8d%e8%b4%b9%e6%8e%a5%e5%8d%95%e5%b9%b3%e5%8f%b0/2010 
    """优化动作类型"""
    UPDATE_CONTENT = "update_content"           # 更新网页内容
    ADD_STRUCTURED_DATA = "add_structured_data"  # 添加结构化数据
    BUILD_AUTHORITATIVE_SOURCE = "build_source"  # 建设权威信源
    IMPROVE_SEMANTIC_ASSOCIATION = "improve_semantic"  # 强化语义关联
    NONE = "none"

class GapSeverity(Enum):
    """缺口严重程度"""
    CRITICAL = "critical"   # 严重
    WARNING = "warning"     # 警告
    INFO = "info"           # 提示

@dataclass
class CognitiveGap: baike.tongsou.com/zh/%e9%bb%91%e5%ae%a224%e5%b0%8f%e6%97%b6%e5%9c%a8%e7%ba%bf%e6%8e%a5%e5%8d%95%e7%bd%91%e7%ab%99/2011 
    """认知缺口"""
    gap_id: str
    scenario: str                    # 查询场景
    platform: str                    # 平台
    gap_type: str                    # 缺口类型：missing/negative/weak
    severity: GapSeverity
    description: str
    suggested_action: OptimizationAction
    confidence: float                # 诊断置信度

@dataclass
class OptimizationTask:
    """优化任务"""
    task_id: str
    gap_id: str
    action: OptimizationAction
    target_url: str
    content_change: str
    status: str = "pending"          # pending/executing/completed/failed
    created_at: datetime = field(default_factory=datetime.now)
    completed_at: Optional[datetime] = None
    effect_score: Optional[float] = None  # 优化效果评分

@dataclass
class FlywheelState: 5030.baike.tongsou.com 
    """飞轮状态"""
    cycle_number: int
    total_gaps_detected: int
    total_tasks_created: int
    total_tasks_completed: int
    avg_effect_score: float
    flywheel_velocity: float         # 飞轮转速（任务完成数/周期）

# ==================== 认知缺口诊断器 ====================

class CognitiveGapDetector: 5002.baike.tongsou.com 
    """认知缺口诊断器"""
    
    def __init__(self, brand: str, key_scenarios: List[str]):
        self.brand = brand
        self.key_scenarios = key_scenarios
        self.gap_thresholds = {
            "missing": 0.3,      # 提及率低于30%视为缺失
            "negative": -0.2,    # 情感低于-0.2视为负面
            "weak": 0.5,         # 语义匹配低于0.5视为弱关联
        }
    
    def detect_gaps(self, visibility_data: pd.DataFrame) -> List[CognitiveGap]:
        """
        检测认知缺口
        visibility_data包含列: scenario, platform, mention_rate, avg_sentiment, semantic_score
        """
        gaps = []
        
        for _, row in visibility_data.iterrows(): 4005.baike.tongsou.com 
            scenario = row["scenario"]
            platform = row["platform"]
            
            # 缺口1：完全缺失
            if row["mention_rate"] < self.gap_thresholds["missing"]: 2035.baike.tongsou.com 
                gap = self._create_gap(
                    scenario, platform, "missing",
                    f"在'{scenario}'场景下，{self.brand}在{platform}的提及率仅为{row['mention_rate']:.1%}"
                )
                gaps.append(gap)
            
            # 缺口2：负面认知
            if row["avg_sentiment"] < self.gap_thresholds["negative"]:
                gap = self._create_gap(
                    scenario, platform, "negative",
                    f"在'{scenario}'场景下，{self.brand}在{platform}的情感倾向为{row['avg_sentiment']:.2f}"
                )
                gaps.append(gap)
            
            # 缺口3：弱语义关联
            if row["semantic_score"] < self.gap_thresholds["weak"]: 2035.baike.tongsou.com 
                gap = self._create_gap(
                    scenario, platform, "weak",
                    f"在'{scenario}'场景下，{self.brand}在{platform}的语义匹配度为{row['semantic_score']:.2f}"
                )
                gaps.append(gap)
        
        return gaps
    
    def _create_gap(self, scenario: str, platform: str, gap_type: str, description: str) -> CognitiveGap:
        """创建认知缺口记录"""
        gap_id = hashlib.md5(f"{scenario}_{platform}_{gap_type}".encode()).hexdigest()[:8]
        
        # 确定严重程度
        if gap_type == "missing": 2023.baike.tongsou.com 
            severity = GapSeverity.CRITICAL
            action = OptimizationAction.UPDATE_CONTENT
        elif gap_type == "negative":
            severity = GapSeverity.WARNING
            action = OptimizationAction.BUILD_AUTHORITATIVE_SOURCE
        else:
            severity = GapSeverity.INFO
            action = OptimizationAction.IMPROVE_SEMANTIC_ASSOCIATION
        
        return CognitiveGap(
            gap_id=gap_id,
            scenario=scenario,
            platform=platform,
            gap_type=gap_type,
            severity=severity,
            description=description,
            suggested_action=action,
            confidence=0.85
        )

# ==================== 优化动作推荐器 ====================

class OptimizationRecommender:
    """优化动作推荐器"""
    
    def __init__(self): 2021.baike.tongsou.com 
        self.action_templates = {
            OptimizationAction.UPDATE_CONTENT: {
                "priority": 1,
                "effort": "medium",
                "expected_impact": 0.3,
                "template": "在目标页面中强化品牌与场景关键词的共现关系"
            },
            OptimizationAction.ADD_STRUCTURED_DATA: {
                "priority": 2,
                "effort": "low",
                "expected_impact": 0.2,
                "template": "添加Schema.org结构化数据，明确品牌实体属性"
            },
            OptimizationAction.BUILD_AUTHORITATIVE_SOURCE: {
                "priority": 3,
                "effort": "high",
                "expected_impact": 0.4,
                "template": "在权威媒体/行业报告发布品牌相关内容"
            },
            OptimizationAction.IMPROVE_SEMANTIC_ASSOCIATION: {
                "priority": 4,
                "effort": "medium",
                "expected_impact": 0.25,
                "template": "创建品牌与场景属性的深度内容，强化语义关联"
            },
        }
    
    def recommend_tasks(self, gaps: List[CognitiveGap]) -> List[OptimizationTask]:
        """根据认知缺口生成优化任务"""
        tasks = [2012.baike.tongsou.com]
        
        # 按严重程度排序
        sorted_gaps = sorted(gaps, key=lambda g: self._severity_rank(g.severity))
        
        for gap in sorted_gaps:
            action = gap.suggested_action
            template = self.action_templates.get(action, {})
            
            task = OptimizationTask(
                task_id=hashlib.md5(f"{gap.gap_id}_{action.value}".encode()).hexdigest()[:8],
                gap_id=gap.gap_id,
                action=action,
                target_url=self._infer_target_url(gap.scenario),
                content_change=template.get("template", ""),
                status="pending"
            )
            tasks.append(task)
        
        return tasks
    
    def _severity_rank(self, severity: GapSeverity) -> int:
        """严重程度排序权重"""
        ranks = {GapSeverity.CRITICAL: 0, GapSeverity.WARNING: 1, GapSeverity.INFO: 2}
        return ranks.get(severity, 3)
    
    def _infer_target_url(self, scenario: str) -> str:
        """根据场景推断目标优化URL（简化版）"""
        url_map = {
            "产品对比": "https://brand.com/products/comparison",
            "使用场景": "https://brand.com/scenarios",
            "品牌评价": "https://brand.com/about",
            "行业排名": "https://brand.com/industry-report",
        }
        return url_map.get(scenario, "https://brand.com/")

# ==================== 效果验证器 ====================
 
class EffectVerifier: 2005.baike.tongsou.com 
    """优化效果验证器"""
    
    def __init__(self, baseline_data: pd.DataFrame):
        self.baseline = baseline_data
    
    def verify(self, task: OptimizationTask, new_data: pd.DataFrame) -> float:
        """
        验证单个优化任务的效果
        返回效果评分(0-1)，>0.5视为有效
        """
        # 简化实现：比较优化前后相关场景的提及率变化
        scenario = self._extract_scenario_from_task(task)
        
        baseline_rate = self._get_mention_rate(self.baseline, scenario)
        new_rate = self._get_mention_rate(new_data, scenario)
        
        if baseline_rate == 0:
            return 1.0 if new_rate > 0 else 0.0
        
        improvement = (new_rate - baseline_rate) / baseline_rate
        # 归一化到0-1
        effect_score = min(max(improvement, 0), 1.0)
        return effect_score
    
    def _extract_scenario_from_task(self, task: OptimizationTask) -> str:
        """从任务中提取场景（简化版）"""
        url_to_scenario = {
            "comparison": "产品对比",
            "scenarios": "使用场景",
            "about": "品牌评价",
            "industry": "行业排名",
        }
        for keyword, scenario in url_to_scenario.items():
            if keyword in task.target_url:
                return scenario
        return "通用"
    
    def _get_mention_rate(self, data: pd.DataFrame, scenario: str) -> float:
        """获取指定场景的平均提及率"""
        subset = data[data["scenario"] == scenario]
        if subset.empty:
            return 0.0
        return subset["mention_rate"].mean(2001.baike.tongsou.com)

# ==================== GEO数据飞轮主引擎 ====================

class GEOFlywheelEngine: 2032.baike.tongsou.com 
    """GEO数据飞轮主引擎"""
    
    def __init__(self, brand: str, key_scenarios: List[str]):
        self.brand = brand
        self.key_scenarios = key_scenarios
        self.gap_detector = CognitiveGapDetector(brand, key_scenarios)
        self.recommender = OptimizationRecommender()
        self.verifier: Optional[EffectVerifier] = None
        self.cycle_history: List[FlywheelState] = []
        self.all_tasks: List[OptimizationTask] = [2041.baike.tongsou.com]
        self.cycle_number = 0
    
    def run_cycle(self, visibility_data: pd.DataFrame) -> FlywheelState:
        """
        运行一个飞轮周期
        visibility_data: 当前周期的可见度数据
        """
        self.cycle_number += 1
        logger.info(f"=== 飞轮第{self.cycle_number}周期开始 ===")
        
        # 齿轮1+2：感知 + 诊断
        gaps = self.gap_detector.detect_gaps(visibility_data)
        logger.info(f"检测到{len(gaps)}个认知缺口")
        
        # 齿轮3：优化
        tasks = self.recommender.recommend_tasks(gaps)
        self.all_tasks.extend(tasks)
        logger.info(f"生成{len(tasks)}个优化任务")
        
        # 模拟任务执行（实际项目中应调用CMS/API）
        completed_tasks = self._simulate_execution(tasks)
        
        # 齿轮4：验证（需要下一周期数据，此处用模拟数据）
        if self.verifier and len(completed_tasks) > 0:
            # 实际项目中，new_data来自下一周期的测量
            new_data = self._simulate_new_data(visibility_data, completed_tasks)
            for task in completed_tasks:
                effect = self.verifier.verify(task, new_data)
                task.effect_score = effect
                task.status = "completed"
                task.completed_at = datetime.now()
                logger.info(f"任务{task.task_id}效果评分: {effect:.2f}")
        
        # 计算飞轮状态
        state = self._compute_state(gaps, tasks, completed_tasks)
        self.cycle_history.append(state)
        
        logger.info(f"=== 飞轮第{self.cycle_number}周期结束 ===")
        return state
    
    def _simulate_execution(self, tasks: List[OptimizationTask]) -> List[OptimizationTask]:
        """模拟任务执行（实际应调用外部系统）"""
        # 模拟80%的任务成功执行
        completed = [t for t in tasks if np.random.random() < 0.8]
        for t in completed:
            t.status = "executing"
        return completed
    
    def _simulate_new_data(self, old_data: pd.DataFrame, tasks: List[OptimizationTask]) -> pd.DataFrame:
        """模拟优化后的新数据（实际应来自真实测量）"""
        new_data = old_data.copy()
        for task in tasks:
            scenario = self.verifier._extract_scenario_from_task(task)
            mask = new_data["scenario"] == scenario
            # 模拟提及率提升10%-30%
            boost = np.random.uniform(0.1, 0.3)
            new_data.loc[mask, "mention_rate"] *= (1 + boost)
        return new_data
    
    def _compute_state(self, gaps: List[CognitiveGap], tasks: List[OptimizationTask], 
                       completed: List[OptimizationTask]) -> FlywheelState:
        """计算飞轮状态"""
        completed_with_effect = [t for t in completed if t.effect_score is not None]
        avg_effect = np.mean([t.effect_score for t in completed_with_effect]) if completed_with_effect else 0
        
        # 飞轮转速：本周期完成任务数
        velocity = len(completed)
        
        return FlywheelState(
            cycle_number=self.cycle_number,
            total_gaps_detected=len(gaps),
            total_tasks_created=len(tasks),
            total_tasks_completed=len(completed),
            avg_effect_score=avg_effect,
            flywheel_velocity=velocity
        )
    
    def get_flywheel_report(self) -> Dict:
        """生成飞轮运行报告"""
        if not self.cycle_history:
            return {"message": "暂无飞轮运行数据"}
        
        latest = self.cycle_history[-1]
        trend = {
            "velocity_trend": [s.flywheel_velocity for s in self.cycle_history],
            "gap_trend": [s.total_gaps_detected for s in self.cycle_history],
            "effect_trend": [s.avg_effect_score for s in self.cycle_history],
        }
        
        return {
            "latest_cycle": latest.cycle_number,
            "latest_state": {
                "gaps": latest.total_gaps_detected,
                "tasks_created": latest.total_tasks_created,
                "tasks_completed": latest.total_tasks_completed,
                "avg_effect": round(latest.avg_effect_score, 2),
                "velocity": latest.flywheel_velocity,
            },
            "trend": trend,
            "total_tasks_all_cycles": len(self.all_tasks),
        }

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

if __name__ == "__main__":
    # 初始化飞轮
    engine = GEOFlywheelEngine(
        brand="华为",
        key_scenarios=["产品对比", "使用场景", "品牌评价", "行业排名"]
    )
    
    # 模拟基线数据
    baseline_data = pd.DataFrame([
        {"scenario": "产品对比", "platform": "doubao", "mention_rate": 0.25, "avg_sentiment": 0.3, "semantic_score": 0.4},
        {"scenario": "使用场景", "platform": "kimi", "mention_rate": 0.45, "avg_sentiment": -0.1, "semantic_score": 0.6},
        {"scenario": "品牌评价", "platform": "deepseek", "mention_rate": 0.60, "avg_sentiment": 0.5, "semantic_score": 0.7},
        {"scenario": "行业排名", "platform": "yuanbao", "mention_rate": 0.15, "avg_sentiment": 0.1, "semantic_score": 0.3},
    ])
    
    # 初始化验证器
    engine.verifier = EffectVerifier(baseline_data)
    
    # 运行3个飞轮周期
    for i in range(3):
        # 每个周期使用略微不同的数据（模拟真实环境变化）
        cycle_data = baseline_data.copy()
        cycle_data["mention_rate"] *= (1 + np.random.uniform(-0.05, 0.05))
        state = engine.run_cycle(cycle_data)
        print(f"周期{state.cycle_number}: 缺口={state.total_gaps_detected}, "
              f"任务={state.total_tasks_completed}, 效果={state.avg_effect_score:.2f}, "
              f"转速={state.flywheel_velocity}")
    
    # 生成报告
    report = engine.get_flywheel_report()
    print("\n飞轮报告:")
    print(json.dumps(report, indent=2, ensure_ascii=False))