GEO 自动化工具
开发指南

# GEO AI可见度自动化诊断API - 核心实现
from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel
from typing import List, Optional, Dict
from datetime import datetime
import httpx
import asyncio
import json

app = FastAPI(title="GEO Visibility Diagnosis API", version="1.0.0")

# 请求模型定义
class DiagnosisRequest(BaseModel):
    brand_name: str
    keywords: List[str]
    platforms: List[str] = ["perplexity", "chatgpt", "kimi"]
    competitors: Optional[List[str]] = []

class BrandMention(BaseModel):
    text: str
    position: int
    sentiment: str  # positive / neutral / negative
    context: str
    is_cited: bool = False

class DiagnosisResult(BaseModel):
    brand: str
    platform: str
    query: str
    appeared: bool
    mentions: List[BrandMention]
    visibility_score: float
    timestamp: datetime

# 多平台API适配器
class PlatformAdapter:
    def __init__(self):
        self.adapters = {
            "perplexity": self._query_perplexity,
            "chatgpt": self._query_chatgpt,
            "kimi": self._query_kimi,
        }

    async def query(self, platform: str, prompt: str) -> dict:
        handler = self.adapters.get(platform)
        if not handler:
            raise ValueError(f"Unsupported platform: {platform}")
        return await handler(prompt)

    async def _query_perplexity(self, prompt: str) -> dict:
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                "https://api.perplexity.ai/chat/completions",
                headers={"Authorization": f"Bearer {PERPLEXITY_KEY}"},
                json={
                    "model": "sonar-pro",
                    "messages": [{"role": "user", "content": prompt}],
                },
                timeout=30.0,
            )
            data = resp.json()
            return {
                "answer": data["choices"][0]["message"]["content"],
                "citations": data.get("citations", []),
            }

    async def _query_chatgpt(self, prompt: str) -> dict:
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                "https://api.openai.com/v1/chat/completions",
                headers={"Authorization": f"Bearer {OPENAI_KEY}"},
                json={
                    "model": "gpt-4o",
                    "messages": [{"role": "user", "content": prompt}],
                },
                timeout=30.0,
            )
            data = resp.json()
            return {
                "answer": data["choices"][0]["message"]["content"],
                "citations": [],
            }

    async def _query_kimi(self, prompt: str) -> dict:
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                "https://api.moonshot.cn/v1/chat/completions",
                headers={"Authorization": f"Bearer {KIMI_KEY}"},
                json={
                    "model": "moonshot-v1-128k",
                    "messages": [{"role": "user", "content": prompt}],
                },
                timeout=30.0,
            )
            data = resp.json()
            return {
                "answer": data["choices"][0]["message"]["content"],
                "citations": [],
            }

# 品牌提及解析器
class BrandMentionParser:
    def __init__(self, brand_name: str, competitors: List[str]):
        self.brand_name = brand_name
        self.competitors = competitors
        self.all_entities = [brand_name] + competitors

    def parse(self, answer: str, citations: List[str]) -> List[BrandMention]:
        mentions = []
        answer_lower = answer.lower()
        for entity in self.all_entities:
            idx = 0
            while True:
                pos = answer_lower.find(entity.lower(), idx)
                if pos == -1:
                    break
                context_start = max(0, pos - 50)
                context_end = min(len(answer), pos + len(entity) + 50)
                context = answer[context_start:context_end]
                is_cited = any(entity.lower() in c.lower() for c in citations)
                mentions.append(BrandMention(
                    text=entity,
                    position=pos,
                    sentiment=self._analyze_sentiment(context),
                    context=context,
                    is_cited=is_cited,
                ))
                idx = pos + len(entity)
        return mentions

    def _analyze_sentiment(self, context: str) -> str:
        positive_words = ["优秀", "领先", "推荐", "best", "leading"]
        negative_words = ["问题", "不足", "较差", "worst", "problem"]
        ctx_lower = context.lower()
        pos_count = sum(1 for w in positive_words if w in ctx_lower)
        neg_count = sum(1 for w in negative_words if w in ctx_lower)
        if pos_count > neg_count: return "positive"
        elif neg_count > pos_count: return "negative"
        return "neutral"

# 可见度评分引擎
class VisibilityScorer:
    WEIGHTS = {
        "appearance": 0.30,
        "position": 0.20,
        "coverage": 0.15,
        "citation": 0.15,
        "sentiment": 0.10,
        "competitive": 0.10,
    }

    def calculate(self, mentions: List[BrandMention],
                    answer_length: int,
                    brand_name: str,
                    competitors: List[str]) -> float:
        if not mentions:
            return 0.0

        brand_mentions = [m for m in mentions if m.text == brand_name]
        comp_mentions = [m for m in mentions if m.text in competitors]

        appearance_score = 100.0 if brand_mentions else 0.0
        position_score = max(0, 100 - (brand_mentions[0].position / answer_length * 100)) if brand_mentions else 0.0
        coverage = sum(len(m.context) for m in brand_mentions) / answer_length if answer_length > 0 else 0.0
        coverage_score = min(100, coverage * 500)
        citation_score = 100.0 if any(m.is_cited for m in brand_mentions) else 0.0
        sentiment_map = {"positive": 100, "neutral": 60, "negative": 20}
        sentiment_score = sum(sentiment_map[m.sentiment] for m in brand_mentions) / len(brand_mentions) if brand_mentions else 0.0
        competitive_score = 80.0 if len(brand_mentions) >= len(comp_mentions) else 40.0

        total = (
            appearance_score * self.WEIGHTS["appearance"] +
            position_score * self.WEIGHTS["position"] +
            coverage_score * self.WEIGHTS["coverage"] +
            citation_score * self.WEIGHTS["citation"] +
            sentiment_score * self.WEIGHTS["sentiment"] +
            competitive_score * self.WEIGHTS["competitive"]
        )
        return round(total, 2)

# API端点
adapter = PlatformAdapter()

@app.post("/api/v1/diagnose", response_model=List[DiagnosisResult])
async def diagnose(req: DiagnosisRequest, bg: BackgroundTasks):
    results = []
    parser = BrandMentionParser(req.brand_name, req.competitors)
    scorer = VisibilityScorer()

    for platform in req.platforms:
        for keyword in req.keywords:
            query_text = f"{keyword}"
            try:
                response = await adapter.query(platform, query_text)
                mentions = parser.parse(response["answer"], response["citations"])
                score = scorer.calculate(
                    mentions, len(response["answer"]),
                    req.brand_name, req.competitors
                )
                results.append(DiagnosisResult(
                    brand=req.brand_name,
                    platform=platform,
                    query=keyword,
                    appeared=len([m for m in mentions if m.text == req.brand_name]) > 0,
                    mentions=mentions,
                    visibility_score=score,
                    timestamp=datetime.now(),
                ))
            except Exception as e:
                continue

    return results

# GEO 结构化内容自动生成器 - JSON-LD 生成模块
import json
from typing import List, Dict, Optional
from dataclasses import dataclass, field, asdict
from jinja2 import Environment, FileSystemLoader

@dataclass
class BrandInfo:
    name: str
    url: str
    logo: str
    description: str
    same_as: List[str] = field(default_factory=list)
    contact_email: Optional[str] = None
    contact_phone: Optional[str] = None
    founding_date: Optional[str] = None
    address: Optional[Dict] = None

@dataclass
class ProductInfo:
    name: str
    description: str
    brand: str
    url: str
    image: str
    price: Optional[str] = None
    price_currency: str = "CNY"
    availability: str = "InStock"
    rating_value: Optional[float] = None
    review_count: Optional[int] = None

@dataclass
class FAQItem:
    question: str
    answer: str

class JsonLdGenerator:
    """GEO JSON-LD 结构化标记自动生成器"""

    def generate_organization(self, brand: BrandInfo) -> dict:
        """生成 Organization 类型 JSON-LD"""
        org = {
            "@context": "https://schema.org",
            "@type": "Organization",
            "name": brand.name,
            "url": brand.url,
            "logo": {
                "@type": "ImageObject",
                "url": brand.logo,
            },
            "description": brand.description,
        }
        if brand.same_as:
            org["sameAs"] = brand.same_as
        if brand.contact_email:
            org["contactPoint"] = {
                "@type": "ContactPoint",
                "email": brand.contact_email,
                "contactType": "customer service",
            }
        if brand.founding_date:
            org["foundingDate"] = brand.founding_date
        if brand.address:
            org["address"] = {
                "@type": "PostalAddress",
                **brand.address,
            }
        return org

    def generate_product(self, product: ProductInfo) -> dict:
        """生成 Product 类型 JSON-LD"""
        prod = {
            "@context": "https://schema.org",
            "@type": "Product",
            "name": product.name,
            "description": product.description,
            "brand": {
                "@type": "Brand",
                "name": product.brand,
            },
            "url": product.url,
            "image": product.image,
            "offers": {
                "@type": "Offer",
                "priceCurrency": product.price_currency,
                "availability": f"https://schema.org/{product.availability}",
            },
        }
        if product.price:
            prod["offers"]["price"] = product.price
        if product.rating_value:
            prod["aggregateRating"] = {
                "@type": "AggregateRating",
                "ratingValue": str(product.rating_value),
                "reviewCount": str(product.review_count or 1),
            }
        return prod

    def generate_faq(self, faq_items: List[FAQItem]) -> dict:
        """生成 FAQPage 类型 JSON-LD"""
        return {
            "@context": "https://schema.org",
            "@type": "FAQPage",
            "mainEntity": [
                {
                    "@type": "Question",
                    "name": item.question,
                    "acceptedAnswer": {
                        "@type": "Answer",
                        "text": item.answer,
                    },
                }
                for item in faq_items
            ],
        }

    def generate_llms_txt(self, brand: BrandInfo,
                         pages: List[Dict],
                         summary: str) -> str:
        """生成 llms.txt 文件内容"""
        lines = [
            f"# {brand.name}",
            f"> {brand.description}",
            "",
            "## 重要链接",
            "",
        ]
        for page in pages:
            lines.append(f"- [{page['title']}]({page['url']}): {page.get('desc', '')}")
        lines.extend(["", "## 摘要", "", summary])
        return "\n".join(lines)

    def to_script_tag(self, data: dict) -> str:
        """将 JSON-LD 数据转换为 HTML script 标签"""
        return f'<script type="application/ld+json">\n{json.dumps(data, ensure_ascii=False, indent=2)}\n</script>'


# 使用示例
if __name__ == "__main__":
    gen = JsonLdGenerator()

    brand = BrandInfo(
        name="示例科技",
        url="https://example.com",
        logo="https://example.com/logo.png",
        description="领先的AI解决方案提供商，专注于企业智能化转型",
        same_as=["https://zhihu.com/org/example", "https://github.com/example"],
        contact_email="info@example.com",
        founding_date="2020",
    )

    faqs = [
        FAQItem("示例科技提供哪些服务？", "我们提供AI咨询、模型定制、数据平台建设等全链路服务。"),
        FAQItem("如何联系示例科技？", "您可以通过 info@example.com 或官网在线表单联系我们。"),
    ]

    # 生成并输出
    print(gen.to_script_tag(gen.generate_organization(brand)))
    print(gen.to_script_tag(gen.generate_faq(faqs)))

# GEO AI爬虫日志解析与识别系统
import re
from dataclasses import dataclass
from datetime import datetime
from typing import List, Dict, Optional
from collections import defaultdict

@dataclass
class LogEntry:
    ip: str
    timestamp: datetime
    method: str
    path: str
    status: int
    size: int
    user_agent: str
    referer: Optional[str] = None

@dataclass
class CrawlerHit:
    crawler_name: str
    crawler_type: str  # ai / search / unknown
    platform: str
    entry: LogEntry

# AI爬虫识别规则库
AI_CRAWLER_PATTERNS = {
    "PerplexityBot": {
        "pattern": re.compile(r"PerplexityBot", re.I),
        "platform": "Perplexity",
        "type": "ai",
    },
    "GPTBot": {
        "pattern": re.compile(r"GPTBot", re.I),
        "platform": "OpenAI",
        "type": "ai",
    },
    "Google-Extended": {
        "pattern": re.compile(r"Google-Extended|GoogleOther", re.I),
        "platform": "Google AI",
        "type": "ai",
    },
    "ClaudeBot": {
        "pattern": re.compile(r"ClaudeBot|anthropic-ai", re.I),
        "platform": "Anthropic",
        "type": "ai",
    },
    "CCBot": {
        "pattern": re.compile(r"CCBot", re.I),
        "platform": "Common Crawl",
        "type": "ai",
    },
    "Bytespider": {
        "pattern": re.compile(r"Bytespider", re.I),
        "platform": "ByteDance",
        "type": "ai",
    },
}

class NginxLogParser:
    """Nginx combined日志格式解析器"""
    LOG_PATTERN = re.compile(
        r'(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] '
        r'"(?P<method>\S+) (?P<path>\S+) \S+" '
        r'(?P<status>\d+) (?P<size>\d+) '
        r'"(?P<referer>[^"]*)" '
        r'"(?P<ua>[^"]*)"'
    )

    def parse_line(self, line: str) -> Optional[LogEntry]:
        match = self.LOG_PATTERN.match(line.strip())
        if not match:
            return None
        d = match.groupdict()
        return LogEntry(
            ip=d["ip"],
            timestamp=datetime.strptime(d["time"], "%d/%b/%Y:%H:%M:%S %z"),
            method=d["method"],
            path=d["path"],
            status=int(d["status"]),
            size=int(d["size"]),
            user_agent=d["ua"],
            referer=d["referer"] if d["referer"] != "-" else None,
        )

class CrawlerIdentifier:
    """AI爬虫识别器"""

    def identify(self, entry: LogEntry) -> Optional[CrawlerHit]:
        ua = entry.user_agent
        for name, rule in AI_CRAWLER_PATTERNS.items():
            if rule["pattern"].search(ua):
                return CrawlerHit(
                    crawler_name=name,
                    crawler_type=rule["type"],
                    platform=rule["platform"],
                    entry=entry,
                )
        return None

class CrawlerAnalytics:
    """AI爬虫访问统计分析"""

    def __init__(self):
        self.hits: List[CrawlerHit] = []
        self._by_crawler = defaultdict(list)
        self._by_path = defaultdict(list)
        self._by_status = defaultdict(int)

    def add_hit(self, hit: CrawlerHit):
        self.hits.append(hit)
        self._by_crawler[hit.crawler_name].append(hit)
        self._by_path[hit.entry.path].append(hit)
        self._by_status[hit.entry.status] += 1

    def summary(self) -> Dict:
        total = len(self.hits)
        return {
            "total_ai_crawler_hits": total,
            "by_crawler": {
                name: len(hits) for name, hits in self._by_crawler.items()
            },
            "top_paths": sorted(
                [(path, len(hits)) for path, hits in self._by_path.items()],
                key=lambda x: x[1], reverse=True
            )[:20],
            "status_distribution": dict(self._by_status),
            "unique_ips_per_crawler": {
                name: len(set(h.entry.ip for h in hits))
                for name, hits in self._by_crawler.items()
            },
        }

    def coverage_rate(self, all_paths: set) -> Dict:
        crawled_paths = set(hit.entry.path for hit in self.hits)
        covered = crawled_paths & all_paths
        return {
            "total_pages": len(all_paths),
            "crawled_pages": len(covered),
            "coverage_rate": round(len(covered) / len(all_paths) * 100, 2) if all_paths else 0,
            "uncrawled_pages": list(all_paths - crawled_paths),
        }

# GEO 语义密度计算器
import re
from typing import List, Dict, Tuple
from dataclasses import dataclass

@dataclass
class Entity:
    text: str
    entity_type: str
    start: int
    end: int
    attributes: List[str] = None

@dataclass
class SemanticDensityResult:
    entity_density: float
    relation_density: float
    attribute_density: float
    completeness_score: float
    overall_score: float
    missing_dimensions: List[str]
    entities: List[Entity]

class SemanticDensityCalculator:
    """GEO语义密度分析器"""

    # 5W1H信息完整性维度
    COMPLETENESS_DIMS = {
        "what": ["是什么", "什么是", "定义", "what is", "defined as"],
        "who": ["谁", "创始人", "团队", "who", "founded by"],
        "how": ["如何", "怎么", "方法", "how to", "approach"],
        "why": ["为什么", "原因", "优势", "why", "benefit"],
        "when": ["何时", "时间", "历史", "when", "since"],
        "where": ["哪里", "地点", "区域", "where", "located"],
    }

    def calculate(self, text: str, entities: List[Entity]) -> SemanticDensityResult:
        char_count = len(text.replace(" ", ""))
        if char_count == 0:
            return SemanticDensityResult(0, 0, 0, 0, 0, [], entities)

        # 实体密度：每100字的命名实体数
        entity_density = len(entities) / (char_count / 100)

        # 关系密度：基于句式模式匹配
        relation_patterns = [
            r".*提供.*服务", r".*属于.*类别", r".*支持.*功能",
            r".*是.*的", r".*包含.*", r".*适用于.*",
        ]
        relations = sum(1 for p in relation_patterns if re.search(p, text))
        relation_density = relations / (char_count / 100) * 10

        # 属性密度：每个实体的平均属性数
        total_attrs = sum(len(e.attributes or []) for e in entities)
        attribute_density = total_attrs / len(entities) if entities else 0

        # 信息完整性检查
        text_lower = text.lower()
        covered = []
        missing = []
        for dim, keywords in self.COMPLETENESS_DIMS.items():
            if any(kw in text_lower for kw in keywords):
                covered.append(dim)
            else:
                missing.append(dim)
        completeness_score = len(covered) / len(self.COMPLETENESS_DIMS) * 100

        # 综合评分（加权平均）
        overall = (
            min(entity_density * 5, 100) * 0.30 +
            min(relation_density * 10, 100) * 0.25 +
            min(attribute_density * 20, 100) * 0.20 +
            completeness_score * 0.25
        )

        return SemanticDensityResult(
            entity_density=round(entity_density, 2),
            relation_density=round(relation_density, 2),
            attribute_density=round(attribute_density, 2),
            completeness_score=round(completeness_score, 1),
            overall_score=round(overall, 1),
            missing_dimensions=missing,
            entities=entities,
        )

// GEO效果监控Dashboard - 可见度趋势图表组件
import React, { useState, useEffect } from 'react';
import ReactECharts from 'echarts-for-react';

const VisibilityChart = ({ brandId, dateRange }) => {
  const [chartData, setChartData] = useState({});
  const [loading, setLoading] = useState(true);

  useEffect(() => {
    fetchVisibilityData(brandId, dateRange);
  }, [brandId, dateRange]);

  const fetchVisibilityData = async (brandId, range) => {
    setLoading(true);
    const resp = await fetch(
      `/api/v1/visibility/trend?brand=${brandId}&range=${range}`
    );
    const data = await resp.json();
    setChartData(data);
    setLoading(false);
  };

  const getOption = () => ({
    tooltip: {
      trigger: 'axis',
      backgroundColor: 'rgba(10, 17, 40, 0.95)',
      borderColor: 'rgba(0, 212, 255, 0.3)',
      textStyle: { color: '#e8ecf4', fontSize: 12 },
    },
    legend: {
      data: ['Perplexity', 'ChatGPT', 'Kimi', '综合评分'],
      textStyle: { color: '#8b9dc3' },
    },
    grid: { left: '3%', right: '4%', bottom: '3%', containLabel: true },
    xAxis: {
      type: 'category',
      data: chartData.dates || [],
      axisLine: { lineStyle: { color: 'rgba(0, 212, 255, 0.2)' } },
      axisLabel: { color: '#8b9dc3' },
    },
    yAxis: {
      type: 'value',
      min: 0, max: 100,
      axisLine: { lineStyle: { color: 'rgba(0, 212, 255, 0.2)' } },
      axisLabel: { color: '#8b9dc3' },
      splitLine: { lineStyle: { color: 'rgba(0, 212, 255, 0.06)' } },
    },
    series: [
      {
        name: 'Perplexity',
        type: 'line',
        data: chartData.perplexity || [],
        smooth: true,
        lineStyle: { color: '#00d4ff', width: 2 },
        itemStyle: { color: '#00d4ff' },
        areaStyle: { color: 'rgba(0, 212, 255, 0.05)' },
      },
      {
        name: 'ChatGPT',
        type: 'line',
        data: chartData.chatgpt || [],
        smooth: true,
        lineStyle: { color: '#7c3aed', width: 2 },
        itemStyle: { color: '#7c3aed' },
      },
      {
        name: 'Kimi',
        type: 'line',
        data: chartData.kimi || [],
        smooth: true,
        lineStyle: { color: '#06d6a0', width: 2 },
        itemStyle: { color: '#06d6a0' },
      },
      {
        name: '综合评分',
        type: 'line',
        data: chartData.overall || [],
        smooth: true,
        lineStyle: { color: '#f59e0b', width: 3 },
        itemStyle: { color: '#f59e0b' },
      },
    ],
  });

  if (loading) return <div>Loading...</div>;

  return (
    <div className="chart-container">
      <h3>AI可见度趋势</h3>
      <ReactECharts option={getOption()} style={{ height: '400px' }} />
    </div>
  );
};

export default VisibilityChart;

# GEO 多平台内容分发调度器
from typing import List, Dict, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import asyncio

class Platform(Enum):
    WEBSITE = "website"
    ZHIHU = "zhihu"
    CSDN = "csdn"
    WECHAT = "wechat"
    BAIKE = "baike"
    JUEJIN = "juejin"
    JIANSHU = "jianshu"

@dataclass
class ContentPiece:
    title: str
    body: str
    tags: List[str] = field(default_factory=list)
    category: Optional[str] = None
    original_url: Optional[str] = None

@dataclass
class DistributionTask:
    content: ContentPiece
    platform: Platform
    scheduled_time: datetime
    adapted_body: Optional[str] = None
    status: str = "pending"  # pending / published / failed
    result_url: Optional[str] = None
    error_message: Optional[str] = None

class ContentAdapter:
    """内容平台适配器"""

    ADAPT_PROMPTS = {
        Platform.ZHIHU: "请将以下内容改写为知乎深度回答风格，增加专业论述深度和数据引用：",
        Platform.CSDN: "请将以下内容改写为CSDN技术博客风格，增加代码示例和架构说明：",
        Platform.WECHAT: "请将以下内容改写为微信公众号文章风格，段落简短，适合移动端阅读：",
        Platform.BAIKE: "请将以下内容改写为百科词条风格，中立客观，去除主观评价：",
    }

    async def adapt(self, content: ContentPiece,
                      platform: Platform) -> str:
        """使用LLM适配内容到目标平台"""
        if platform in [Platform.WEBSITE, Platform.JUEJIN, Platform.JIANSHU]:
            return content.body  # Markdown平台无需改写

        prompt = self.ADAPT_PROMPTS.get(platform, "")
        # 调用LLM API进行改写（此处简化）
        adapted = await self._call_llm(prompt + content.body)
        return adapted

    async def _call_llm(self, prompt: str) -> str:
        # LLM API调用实现
        pass

class DistributionScheduler:
    """多平台分发调度器"""

    # 发布节奏配置（天数偏移）
    SCHEDULE_OFFSETS = {
        Platform.WEBSITE: 0,   # Day 1 首发
        Platform.ZHIHU: 1,    # Day 2
        Platform.CSDN: 2,     # Day 3
        Platform.JUEJIN: 2,   # Day 3
        Platform.WECHAT: 3,   # Day 4
        Platform.JIANSHU: 4,  # Day 5
        Platform.BAIKE: 6,    # Day 7
    }

    def __init__(self, adapter: ContentAdapter):
        self.adapter = adapter
        self.tasks: List[DistributionTask] = []

    def create_schedule(self, content: ContentPiece,
                         platforms: List[Platform],
                         start_date: datetime = None) -> List[DistributionTask]:
        """创建分发排程"""
        start = start_date or datetime.now()
        tasks = []
        for platform in platforms:
            offset = self.SCHEDULE_OFFSETS.get(platform, 0)
            scheduled = start + timedelta(days=offset)
            tasks.append(DistributionTask(
                content=content,
                platform=platform,
                scheduled_time=scheduled,
            ))
        self.tasks.extend(tasks)
        return tasks

    async def execute_schedule(self):
        """执行排程中的待发布任务"""
        now = datetime.now()
        pending = [
            t for t in self.tasks
            if t.status == "pending" and t.scheduled_time <= now
        ]
        for task in pending:
            try:
                # 1. 适配内容
                adapted = await self.adapter.adapt(
                    task.content, task.platform
                )
                task.adapted_body = adapted

                # 2. 调用平台API发布
                result = await self._publish_to_platform(task)
                task.status = "published"
                task.result_url = result.get("url")
            except Exception as e:
                task.status = "failed"
                task.error_message = str(e)

    async def _publish_to_platform(self, task: DistributionTask) -> dict:
        """调用平台API发布内容"""
        # 实际实现根据各平台API对接
        pass

openapi: "3.0.3"
info:
  title: GEO Automation API
  description: GEO自动化工具API - AI可见度诊断、结构化内容生成、爬虫监控、语义分析、内容分发
  version: "1.0.0"
  contact:
    name: GEO API Support
    email: api@geo-tool.dev

servers:
  - url: https://api.geo-tool.dev
    description: Production
  - url: https://staging-api.geo-tool.dev
    description: Staging

security:
  - BearerAuth: []

paths:
  /api/v1/diagnose:
    post:
      summary: 触发AI可见度诊断
      operationId: createDiagnosis
      tags: [Diagnosis]
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/DiagnosisRequest'
      responses:
        "200":
          description: 诊断任务创建成功
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/DiagnosisResponse'
        "401":
          description: 未授权
        "429":
          description: 请求频率超限

  /api/v1/generate/jsonld:
    post:
      summary: 生成JSON-LD结构化标记
      operationId: generateJsonLd
      tags: [Generation]
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/JsonLdRequest'
      responses:
        "200":
          description: JSON-LD生成成功
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/JsonLdResponse'

components:
  securitySchemes:
    BearerAuth:
      type: http
      scheme: bearer
      bearerFormat: JWT

  schemas:
    DiagnosisRequest:
      type: object
      required: [brand_name, keywords]
      properties:
        brand_name:
          type: string
          example: "示例科技"
        keywords:
          type: array
          items:
            type: string
          example: ["AI解决方案", "企业智能化"]
        platforms:
          type: array
          items:
            type: string
            enum: [perplexity, chatgpt, kimi, metaso]
          default: [perplexity, chatgpt, kimi]
        competitors:
          type: array
          items:
            type: string

    DiagnosisResponse:
      type: object
      properties:
        code:
          type: integer
          example: 200
        message:
          type: string
          example: "诊断任务已创建"
        data:
          type: object
          properties:
            task_id:
              type: string
              format: uuid
            status:
              type: string
              enum: [pending, running, completed, failed]
            estimated_time:
              type: integer
              description: 预计完成时间（秒）

version: "3.8"

services:
  geo-api:
    build:
      context: .
      dockerfile: Dockerfile.api
    container_name: geo-api
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://geo:geo_pass@postgres:5432/geo_db
      - REDIS_URL=redis://redis:6379/0
      - PERPLEXITY_API_KEY=${PERPLEXITY_API_KEY}
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - KIMI_API_KEY=${KIMI_API_KEY}
      - JWT_SECRET=${JWT_SECRET}
    depends_on:
      - postgres
      - redis
    deploy:
      resources:
        limits:
          cpus: "2.0"
          memory: 4G
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  geo-dashboard:
    build:
      context: ./dashboard
      dockerfile: Dockerfile
    container_name: geo-dashboard
    ports:
      - "3000:80"
    depends_on:
      - geo-api
    restart: unless-stopped

  geo-celery-worker:
    build:
      context: .
      dockerfile: Dockerfile.worker
    container_name: geo-worker
    command: celery -A geo_diagnosis.celery worker -l info -c 4
    environment:
      - DATABASE_URL=postgresql://geo:geo_pass@postgres:5432/geo_db
      - REDIS_URL=redis://redis:6379/0
      - PERPLEXITY_API_KEY=${PERPLEXITY_API_KEY}
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - KIMI_API_KEY=${KIMI_API_KEY}
    depends_on:
      - postgres
      - redis
    deploy:
      resources:
        limits:
          cpus: "2.0"
          memory: 4G
    restart: unless-stopped

  geo-celery-beat:
    build:
      context: .
      dockerfile: Dockerfile.worker
    container_name: geo-beat
    command: celery -A geo_diagnosis.celery beat -l info
    environment:
      - DATABASE_URL=postgresql://geo:geo_pass@postgres:5432/geo_db
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - redis
    restart: unless-stopped

  geo-log-parser:
    build:
      context: .
      dockerfile: Dockerfile.log-parser
    container_name: geo-log-parser
    volumes:
      - /var/log/nginx:/var/log/nginx:ro
    environment:
      - REDIS_URL=redis://redis:6379/0
      - DATABASE_URL=postgresql://geo:geo_pass@postgres:5432/geo_db
      - LOG_PATH=/var/log/nginx/access.log
    depends_on:
      - redis
      - postgres
    restart: unless-stopped

  postgres:
    image: postgres:16-alpine
    container_name: geo-postgres
    environment:
      - POSTGRES_DB=geo_db
      - POSTGRES_USER=geo
      - POSTGRES_PASSWORD=geo_pass
    volumes:
      - pgdata:/var/lib/postgresql/data
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
    ports:
      - "5432:5432"
    deploy:
      resources:
        limits:
          cpus: "2.0"
          memory: 8G
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    container_name: geo-redis
    command: redis-server --maxmemory 1gb --maxmemory-policy allkeys-lru
    ports:
      - "6379:6379"
    volumes:
      - redisdata:/data
    restart: unless-stopped

volumes:
  pgdata:
  redisdata: