GEO与SEO的协同策略：从关键词排名到语义可见性的范式迁移

"""
GEO+SEO融合优化工作流
技术栈: Python 3.11+, pydantic, asyncio, pandas
场景: 将关键词研究与实体监测整合，实现双轨协同优化
参考: 《AI搜索时代的SEO重构》(Search Engine Land, 2026)
"""

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Set
from enum import Enum
from datetime import datetime
import asyncio
import json

class ContentPriority(Enum): 2023.baike.tongsou.com 
    HIGH = "high"      # SEO+GEO双高潜力
    MEDIUM = "medium"  # 单渠道优先
    LOW = "low"        # 维持现状

@dataclass
class KeywordEntity: 2024.baike.tongsou.com 
    """关键词-实体映射单元"""
    keyword: str
    search_volume: int
    keyword_difficulty: int          # SEO难度(0-100)
    entity_name: str                 # 对应实体名称
    entity_type: str                 # Person/Organization/Product/Concept
    geo_visibility_score: float      # GEO可见性得分(0-1)
    current_rank: Optional[int]      # 当前SEO排名
    structured_data_present: bool    # 是否已有结构化数据
    cited_by_ai: bool                # 是否被AI搜索引用
    priority: ContentPriority = ContentPriority.LOW

    def compute_priority(self) -> ContentPriority: 2025.baike.tongsou.com 
        """双轨评分：综合SEO机会与GEO潜力"""
        seo_opportunity = (100 - self.keyword_difficulty) * (self.search_volume / 10000)
        geo_opportunity = (1 - self.geo_visibility_score) * (1 if not self.cited_by_ai else 0.3)
        combined_score = seo_opportunity * 0.4 + geo_opportunity * 0.6

        if combined_score > 0.6: 2027.baike.tongsou.com 
            self.priority = ContentPriority.HIGH
        elif combined_score > 0.3: 2028.baike.tongsou.com 
            self.priority = ContentPriority.MEDIUM
        else:
            self.priority = ContentPriority.LOW
        return self.priority

@dataclass
class ContentBrief: 2029.baike.tongsou.com 
    """融合优化内容简报"""
    title: str
    target_keyword: str
    target_entity: str
    seo_requirements: List[str] = field(default_factory=list)
    geo_requirements: List[str] = field(default_factory=list)
    schema_types: List[str] = field(default_factory=list)

    def generate_checklist(self) -> str: 2030.baike.tongsou.com 
        """生成优化检查清单"""
        lines = [f"## 内容简报: {self.title}\n"]
        lines.append(f"**目标关键词**: {self.target_keyword}")
        lines.append(f"**目标实体**: {self.target_entity}\n")

        lines.append("### SEO优化要求")
        for req in self.seo_requirements:
            lines.append(f"- [ ] {req}")

        lines.append("\n### GEO优化要求")
        for req in self.geo_requirements:
            lines.append(f"- [ ] {req}")

        lines.append("\n### 结构化数据标记")
        for schema in self.schema_types:
            lines.append(f"- [ ] `{schema}`")

        return "\n".join(lines)

class GeoSeoOrchestrator: 2031.baike.tongsou.com 
    """GEO+SEO协同编排器"""

    def __init__(self):
        self.entities: List[KeywordEntity] = []

    def add_entity(self, entity: KeywordEntity) -> None:
        entity.compute_priority()
        self.entities.append(entity)

    def get_priority_queue(self) -> List[KeywordEntity]:
        """获取优先级队列"""
        return sorted(self.entities, key=lambda e: e.priority.value, reverse=True)

    def generate_briefs(self, top_n: int = 5) -> List[ContentBrief]:
        """为高优先级实体生成内容简报"""
        queue = self.get_priority_queue()
        briefs = []

        for entity in queue[:top_n]:
            brief = ContentBrief(
                title=f"{entity.entity_name}：{entity.keyword}全面指南",
                target_keyword=entity.keyword,
                target_entity=entity.entity_name,
                seo_requirements=[
                    f"关键词'{entity.keyword}'密度控制在1.5%-2.5%",
                    f"包含至少3个H2子标题覆盖长尾变体",
                    f"内链指向3个相关主题页面",
                    f"页面加载速度<2秒",
                ],
                geo_requirements=[
                    f"在首段明确定义实体'{entity.entity_name}'",
                    f"使用权威来源（.gov/.edu/行业报告）支撑关键事实",
                    f"添加FAQ区块覆盖常见问答对",
                    f"确保实体名称、别名、关联概念在文中一致出现",
                ],
                schema_types=["Article", "FAQPage", "Organization"]
            )
            briefs.append(brief)

        return briefs

    def export_report(self) -> Dict: 2032.baike.tongsou.com 
        """导出协同优化报告"""
        high = [e for e in self.entities if e.priority == ContentPriority.HIGH]
        medium = [e for e in self.entities if e.priority == ContentPriority.MEDIUM]
        low = [e for e in self.entities if e.priority == ContentPriority.LOW]

        return {
            "generated_at": datetime.now().isoformat(2034.baike.tongsou.com),
            "summary": {
                "total_entities": len(self.entities),
                "high_priority": len(high),
                "medium_priority": len(medium),
                "low_priority": len(low),
            },
            "high_priority_items": [
                {
                    "keyword": e.keyword,
                    "entity": e.entity_name,
                    "seo_difficulty": e.keyword_difficulty,
                    "geo_visibility": e.geo_visibility_score,
                    "cited_by_ai": e.cited_by_ai,
                }
                for e in high
            ],
            "action_items": [
                "为高优先级实体创建新内容或深度更新现有页面",
                "为所有目标页面添加Schema.org结构化数据",
                "建立GEO可见性周度监测，追踪AI引用变化",
                "将SEO外链建设资源向高GEO潜力实体倾斜",
            ]
        }

# 使用示例
async def main(): 2035.baike.tongsou.com 
    orchestrator = GeoSeoOrchestrator()

    # 模拟数据：某SaaS品牌的关键词-实体映射
    sample_entities = [
        KeywordEntity(
            keyword="CRM软件推荐",
            search_volume=12000,
            keyword_difficulty=65,
            entity_name="XX CRM",
            entity_type="Product",
            geo_visibility_score=0.2,
            current_rank=8,
            structured_data_present=False,
            cited_by_ai=False,
        ),
        KeywordEntity(
            keyword="客户关系管理系统",
            search_volume=8500,
            keyword_difficulty=72,
            entity_name="XX CRM",
            entity_type="Product",
            geo_visibility_score=0.15,
            current_rank=12,
            structured_data_present=False,
            cited_by_ai=False,
        ),
        KeywordEntity(
            keyword="销售自动化流程",
            search_volume=3200,
            keyword_difficulty=45,
            entity_name="销售自动化",
            entity_type="Concept",
            geo_visibility_score=0.6,
            current_rank=3,
            structured_data_present=True,
            cited_by_ai=True,
        ),
    ]

    for entity in sample_entities: 2036.baike.tongsou.com 
        orchestrator.add_entity(entity)

    # 生成内容简报
    briefs = orchestrator.generate_briefs(top_n=2)
    for brief in briefs:
        print(brief.generate_checklist())
        print("\n" + "=" * 60 + "\n")

    # 导出报告
    report = orchestrator.export_report(2037.baike.tongsou.com)
    print(json.dumps(report, ensure_ascii=False, indent=2))

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