附录E.5：App与Web内容统一API设计

# 统一内容单元模型
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
from datetime import datetime

@dataclass
class ContentUnit:
    """统一内容单元"""
    id: str
    title: str
    summary: str
    body: str  # Markdown格式正文
    body_html: str  # HTML格式正文
    body_structured: Dict[str, Any]  # 结构化正文（用于AI解析）
    
    # 元数据
    author: Dict[str, str]
    publish_date: datetime
    last_modified: datetime
    category: str
    tags: List[str]
    
    # 实体
    entities: List[Dict[str, str]] = field(default_factory=list)
    
    # 媒体
    media: List[Dict[str, str]] = field(default_factory=list)
    
    # 关联内容
    related_content: List[Dict[str, str]] = field(default_factory=list)
    
    # SEO元数据
    seo_metadata: Dict[str, Any] = field(default_factory=dict)
    
    # 结构化数据（JSON-LD）
    json_ld: Dict[str, Any] = field(default_factory=dict)
    
    # 平台特定数据
    platform_data: Dict[str, Any] = field(default_factory=dict)
    # 示例：{"web": {"canonical_url": "..."}, "app": {"deep_link": "..."}}
    
    # 版本信息
    version: int = 1
    status: str = "published"  # draft, published, archived

# app.py
from fastapi import FastAPI, Query, HTTPException
from fastapi.responses import JSONResponse, HTMLResponse
from typing import Optional, List
from datetime import datetime
import json

app = FastAPI(title="统一内容API")

# 模拟数据存储
content_store = {
    "content_12345": {
        "id": "content_12345",
        "title": "如何优化Core Web Vitals",
        "summary": "本文详细介绍LCP、INP、CLS的优化方法...",
        "body": "## 引言\nCore Web Vitals是Google的重要排名因素...\n\n### LCP优化\n...",
        "body_html": "<h2>引言</h2><p>Core Web Vitals是Google的重要排名因素...</p><h3>LCP优化</h3><p>...</p>",
        "author": {"name": "张三", "url": "https://example.com/author/zhangsan"},
        "publish_date": "2024-01-15T10:00:00Z",
        "last_modified": "2024-02-20T14:30:00Z",
        "category": "SEO",
        "tags": ["Core Web Vitals", "LCP", "INP", "CLS"],
        "entities": [
            {"name": "Core Web Vitals", "type": "concept", "wikidata_id": "Q123456"},
            {"name": "LCP", "type": "concept", "wikidata_id": "Q789012"}
        ],
        "media": [
            {"type": "image", "url": "https://example.com/images/cwv-chart.png", "alt_text": "CWV指标对比图"}
        ],
        "related_content": [
            {"id": "content_12346", "title": "INP优化最佳实践", "url": "https://example.com/article/inp-optimization"}
        ],
        "seo_metadata": {
            "canonical_url": "https://example.com/article/core-web-vitals",
            "meta_description": "全面的Core Web Vitals优化指南",
            "open_graph": {
                "title": "如何优化Core Web Vitals",
                "description": "全面的Core Web Vitals优化指南",
                "image": "https://example.com/images/og-cwv.png"
            }
        },
        "json_ld": {
            "@context": "https://schema.org",
            "@type": "Article",
            "headline": "如何优化Core Web Vitals",
            "datePublished": "2024-01-15T10:00:00Z",
            "author": {"@type": "Person", "name": "张三"}
        },
        "platform_data": {
            "web": {
                "canonical_url": "https://example.com/article/core-web-vitals"
            },
            "app": {
                "deep_link": "myapp://article/core-web-vitals",
                "screen": "ArticleDetail"
            }
        },
        "version": 3,
        "status": "published"
    }
}

@app.get("/api/v1/content/{content_id}")
async def get_content(
    content_id: str,
    format: str = Query("json", enum=["json", "html", "markdown", "structured"]),
    client: str = Query("web", enum=["web", "ios", "android", "crawler"])
):
    """获取内容详情"""
    if content_id not in content_store:
        raise HTTPException(status_code=404, detail="Content not found")
    
    content = content_store[content_id]
    
    if format == "json":
        return JSONResponse(content=content)
    
    elif format == "html":
        return HTMLResponse(content=content["body_html"])
    
    elif format == "markdown":
        return JSONResponse(content={"body": content["body"]})
    
    elif format == "structured":
        # 返回针对AI爬虫的结构化数据
        structured_response = {
            "@context": "https://schema.org",
            "@type": "Article",
            "headline": content["title"],
            "author": content["author"],
            "datePublished": content["publish_date"],
            "hasPart": [
                {
                    "@type": "WebPageElement",
                    "name": "引言",
                    "text": "Core Web Vitals是Google的重要排名因素..."
                }
            ],
            "mainEntity": {
                "@type": "Question",
                "name": content["title"],
                "acceptedAnswer": {
                    "@type": "Answer",
                    "text": content["summary"]
                }
            }
        }
        return JSONResponse(content=structured_response)

@app.get("/api/v1/content/list")
async def list_content(
    category: Optional[str] = None,
    tags: Optional[List[str]] = Query(None),
    page: int = Query(1, ge=1),
    limit: int = Query(20, ge=1, le=100)
):
    """获取内容列表"""
    # 过滤逻辑
    filtered = list(content_store.values())
    if category:
        filtered = [c for c in filtered if c["category"] == category]
    if tags:
        filtered = [c for c in filtered if any(tag in c["tags"] for tag in tags)]
    
    # 分页
    start = (page - 1) * limit
    end = start + limit
    results = filtered[start:end]
    
    return JSONResponse(content={
        "total": len(filtered),
        "page": page,
        "limit": limit,
        "results": [
            {
                "id": c["id"],
                "title": c["title"],
                "summary": c["summary"],
                "category": c["category"],
                "tags": c["tags"],
                "publish_date": c["publish_date"],
                "url": c["seo_metadata"]["canonical_url"],
                "app_deep_link": c["platform_data"]["app"]["deep_link"]
            }
            for c in results
        ]
    })

@app.get("/api/v1/search")
async def search_content(
    q: str = Query(..., min_length=1),
    type: str = Query("all", enum=["all", "article", "product", "video", "question"]),
    source: str = Query("both", enum=["web", "app", "both"]),
    page: int = Query(1, ge=1)
):
    """统一搜索"""
    # 模拟搜索逻辑
    results = [
        {
            "id": "content_12345",
            "title": "如何优化Core Web Vitals",
            "snippet": "Core Web Vitals是Google的重要排名因素...",
            "url": "https://example.com/article/core-web-vitals",
            "app_deep_link": "myapp://article/core-web-vitals",
            "source": "web",
            "score": 0.95,
            "type": "article",
            "thumbnail": "https://example.com/images/cwv-thumb.png",
            "publish_date": "2024-01-15T10:00:00Z"
        }
    ]
    
    return JSONResponse(content={
        "total_results": len(results),
        "results": results,
        "facets": {
            "types": {"article": 1, "video": 0},
            "sources": {"web": 1, "app": 0}
        },
        "suggestions": ["Core Web Vitals优化", "LCP优化方法"]
    })

# 检测AI爬虫中间件
from fastapi import Request

AI_CRAWLERS = {
    "GPTBot": "OpenAI",
    "GoogleOther": "Google",
    "CCBot": "CommonCrawl",
    "ClaudeBot": "Anthropic",
    "Bytespider": "ByteDance",
    "DeepSeek-Bot": "DeepSeek",
    "Amazonbot": "Amazon"
}

@app.middleware("http")
async def detect_ai_crawler(request: Request, call_next):
    user_agent = request.headers.get("user-agent", "").lower()
    
    # 检测是否为AI爬虫
    is_ai_crawler = any(
        crawler.lower() in user_agent 
        for crawler in AI_CRAWLERS.keys()
    )
    
    if is_ai_crawler:
        # 自动返回结构化格式
        request.state.format = "structured"
        request.state.client = "crawler"
    else:
        request.state.format = request.query_params.get("format", "json")
        request.state.client = request.query_params.get("client", "web")
    
    response = await call_next(request)
    return response

# 深度链接生成器
from typing import Dict

class DeepLinkGenerator:
    """生成统一深度链接"""
    
    @staticmethod
    def generate_deep_links(content_id: str, platform_data: Dict) -> Dict:
        """生成各平台深度链接"""
        return {
            "web": platform_data.get("web", {}).get("canonical_url", f"/content/{content_id}"),
            "ios": f"myapp://content/{content_id}",
            "android": f"myapp://content/{content_id}",
            "universal": f"https://example.com/content/{content_id}"
        }
    
    @staticmethod
    def generate_app_links(content: Dict) -> Dict:
        """生成App Links和Universal Links"""
        return {
            "applinks": {
                "apps": [],
                "details": [
                    {
                        "appID": "ABCD1234.com.example.app",
                        "paths": [f"/content/{content['id']}"]
                    }
                ]
            },
            "apple-app-site-association": {
                "applinks": {
                    "apps": [],
                    "details": [
                        {
                            "appID": "ABCD1234.com.example.app",
                            "paths": [f"/content/{content['id']}"]
                        }
                    ]
                }
            }
        }

# 监控指标示例
from prometheus_client import Counter, Histogram, Gauge

# API请求计数
api_requests_total = Counter(
    'api_requests_total',
    'Total API requests',
    ['endpoint', 'method', 'status']
)

# 响应时间
api_response_time = Histogram(
    'api_response_time_seconds',
    'API response time in seconds',
    ['endpoint'],
    buckets=[0.1, 0.25, 0.5, 1, 2.5, 5, 10]
)

# 缓存命中率
cache_hit_ratio = Gauge(
    'cache_hit_ratio',
    'Cache hit ratio',
    ['cache_type']
)

# 内容版本分布
content_version_gauge = Gauge(
    'content_version',
    'Content version distribution',
    ['version']
)