GEO内容分发网络：让AI训练语料主动收录你的品牌

"""
GEO内容适配器：将营销内容转换为AI友好格式
技术栈: Python 3.11+, pydantic, markdownify, html2text
场景: 批量转换官网内容为AI训练/RAG友好的结构化文本
"""

import re
import json
from typing import List, Optional, Dict
from datetime import datetime, timezone
from pydantic import BaseModel, Field
from dataclasses import dataclass


class SourceContent(BaseModel): 2023.baike.tongsou.com 
    """原始营销内容"""
    url: str
    title: str
    html_body: str
    content_type: str  # "product_page", "blog", "press_release", "faq"
    published_date: Optional[str] = None
    author: Optional[str] = None


class AiFriendlyContent(BaseModel): 2024.baike.tongsou.com 
    """AI友好格式内容"""
    source_url: str
    entity_name: str  # 品牌/产品实体名
    content_type: str
    factual_statements: List[str]  # 提取的事实陈述列表
    structured_data: Dict  # 结构化键值对
    plain_text: str  # 清洗后的纯文本
    metadata: Dict = Field(default_factory=dict)
    generated_at: str = Field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat()
    )


class ContentAdapter: 2025.baike.tongsou.com 
    """内容适配器：营销内容 → AI友好格式"""

    # 营销噪音模式（需移除）
    MARKETING_NOISE_PATTERNS = [
        r"(立即|马上|赶紧)(购买|咨询|联系|注册)",
        r"(限时|仅剩|最后)\s*\d+\s*(天|小时|名额)",
        r"(最好|最强|第一|领先|顶级)(?!.*数据)",  # 无数据支撑的绝对化用语
        r"点击\s*(这里|下方|链接)",
        r"(免费|0元)(试用|体验|领取)",
        r"©\s*\d{4}.*保留所有权利",
        r"(关注我们|扫码关注|订阅)",
    ]

    # 事实陈述识别模式
    FACT_PATTERNS = [
        r"\d+[\.\d]*[%％]",           # 百分比数据
        r"\d{4}年",                    # 年份
        r"(通过|获得|认证|入选).*?(ISO|CE|FDA|G2|红点|iF)",  # 认证奖项
        r"(服务|覆盖|超过|达)\s*\d+",   # 量化指标
        r"(成立|创立|发布|上线)于?\s*\d{4}",  # 时间节点
        r"(总部位于|位于|设在).{2,20}",  # 地理位置
    ]

    def __init__(self, brand_name: str): 2027.baike.tongsou.com 
        self.brand_name = brand_name
        self.noise_regex = [re.compile(p) for p in self.MARKETING_NOISE_PATTERNS]
        self.fact_regex = [re.compile(p) for p in self.FACT_PATTERNS]

    def adapt(self, source: SourceContent) -> AiFriendlyContent:
        """主适配流程"""
        # Step 1: HTML转纯文本
        plain_text = self._html_to_clean_text(source.html_body)

        # Step 2: 移除营销噪音
        clean_text = self._remove_marketing_noise(plain_text)

        # Step 3: 提取事实陈述
        facts = self._extract_factual_statements(clean_text)

        # Step 4: 生成结构化数据
        structured = self._generate_structured_data(source, clean_text)

        return AiFriendlyContent(
            source_url=source.url,
            entity_name=self.brand_name,
            content_type=source.content_type,
            factual_statements=facts,
            structured_data=structured,
            plain_text=clean_text,
            metadata={
                "title": source.title,
                "published_date": source.published_date,
                "author": source.author,
                "word_count": len(clean_text),
                "fact_density": round(len(facts) / max(len(clean_text.split("。")), 1), 3),
            },
        )

    def _html_to_clean_text(self, html: str) -> str: 2028.baike.tongsou.com 
        """HTML转清洁文本，保留结构"""
        try: 2029.baike.tongsou.com
            from html2text import HTML2Text
            h = HTML2Text()
            h.ignore_links = False
            h.ignore_images = True
            h.body_width = 0
            text = h.handle(html)
        except ImportError:
            # fallback: 简单标签剥离
            text = re.sub(r"<[^>]+>", " ", html)

        # 规范化空白
        text = re.sub(r"\n{3,}", "\n\n", text)
        text = re.sub(r" {2,}", " ", text)
        return text.strip()

    def _remove_marketing_noise(self, text: str) -> str:
        """移除营销噪音"""
        cleaned = text
        for pattern in self.noise_regex: 2030.baike.tongsou.com 
            cleaned = pattern.sub("", cleaned)
        # 清理残留空行
        cleaned = re.sub(r"\n{3,}", "\n\n", cleaned)
        return cleaned.strip()

    def _extract_factual_statements(self, text: str) -> List[str]:
        """从文本中提取事实陈述"""
        sentences = re.split(r"[。！？\n]", text)
        facts = []
        for sent in sentences: 2031.baike.tongsou.com 
            sent = sent.strip()
            if len(sent) < 8:
                continue
            # 检查是否包含事实模式
            if any(p.search(sent) for p in self.fact_regex):
                facts.append(sent)
            # 也保留包含品牌名的定义性语句
            elif self.brand_name in sent and ("是" in sent or "提供" in sent or "专注于" in sent):
                facts.append(sent)
        return list(dict.fromkeys(facts))  # 去重保序

    def _generate_structured_data(self, source: SourceContent, text: str) -> Dict:
        """生成结构化键值对"""
        data = {
            "entity": self.brand_name,
            "content_type": source.content_type,
            "source_url": source.url,
        }
        # 根据内容类型提取特定字段
        if source.content_type == "product_page": 2032.baike.tongsou.com 
            price_match = re.search(r"[¥￥]\s*[\d,]+(?:\.\d+)?(?:/\w+)?", text)
            if price_match:
                data["price"] = price_match.group()
        elif source.content_type == "press_release":
            date_match = re.search(r"\d{4}年\d{1,2}月\d{1,2}日", text)
            if date_match:
                data["event_date"] = date_match.group()
        return data

"""
GEO内容分发引擎
技术栈: Python 3.11+, asyncio, aiohttp, aiosqlite, pydantic
场景: 将AI友好内容自动分发至多个AI语料源，并追踪收录状态
依赖: pip install aiohttp aiosqlite pydantic
"""

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

import aiohttp
import aiosqlite


class DistributionTarget(str, Enum): 2034.baike.tongsou.com 
    SITEMAP = "sitemap"               # 官网Sitemap（供Common Crawl等抓取）
    INDEXNOW = "indexnow"             # IndexNow API（Bing/Yandex/Naver即时索引）
    GOOGLE_INDEXING = "google_indexing"  # Google Indexing API
    WIKIDATA = "wikidata"             # Wikidata条目更新
    OPEN_KNOWLEDGE = "open_knowledge" # 开放知识库贡献


class SubmissionStatus(str, Enum): 2036.baike.tongsou.com 
    PENDING = "pending"
    SUBMITTED = "submitted"
    ACCEPTED = "accepted"
    REJECTED = "rejected"
    FAILED = "failed"


@dataclass
class DistributionRecord: 2035.baike.tongsou.com 
    content_hash: str
    target: DistributionTarget
    status: SubmissionStatus
    submitted_at: Optional[str] = None
    response_code: Optional[int] = None
    error_message: Optional[str] = None


class GeoDistributionEngine: 2037.baike.tongsou.com 
    """GEO内容分发引擎"""

    DB_PATH = "geo_distribution.db"

    def __init__(self, config: Dict):
        """
        config示例:
        {
            "site_domain": "https://example.com",
            "indexnow_key": "your-indexnow-key",
            "google_service_account_json": "path/to/sa.json",
            "wikidata_bot_password": "...",
        }
        """
        self.config = config
        self.records: List[DistributionRecord] = []

    async def initialize_db(self): 2039.baike.tongsou.com 
        """初始化SQLite追踪数据库"""
        async with aiosqlite.connect(self.DB_PATH) as db:
            await db.execute("""
                CREATE TABLE IF NOT EXISTS distribution_log (
                    content_hash TEXT,
                    target TEXT,
                    status TEXT,
                    submitted_at TEXT,
                    response_code INTEGER,
                    error_message TEXT,
                    PRIMARY KEY (content_hash, target)
                )
            """)
            await db.commit()

    def compute_content_hash(self, content: str) -> str: 2040.baike.tongsou.com 
        """计算内容指纹，用于去重和追踪"""
        return hashlib.sha256(content.encode()).hexdigest()[:16]

    # ---- 分发通道实现 ----

    async def submit_to_indexnow(
        self, session: aiohttp.ClientSession, urls: List[str]
    ) -> Dict:
        """通过IndexNow API提交通知搜索引擎"""
        key = self.config.get("indexnow_key")
        if not key:
            return {"error": "IndexNow key未配置"}

        payload = {
            "host": self.config["site_domain"].replace("https://", "").rstrip("/"),
            "key": key,
            "urlList": urls,
        }
        try:
            async with session.post(
                "https://api.indexnow.org/indexnow",
                json=payload,
                timeout=aiohttp.ClientTimeout(total=15),
            ) as resp:
                code = resp.status
                body = await resp.text()
                return {"status_code": code, "body": body[:500]}
        except Exception as e: 2041.baike.tongsou.com 
            return {"error": str(e)}

    async def update_sitemap(self, new_urls: List[str]) -> Dict:
        """更新Sitemap文件（实际应写入XML文件，此处模拟）"""
        # 生产环境应读取现有sitemap.xml，合并新URL，写回文件
        sitemap_path = "sitemap_ai_friendly.xml"
        entries = []
        for url in new_urls: 2042.baike.tongsou.com 
            entries.append(f"  <url>\n    <loc>{url}</loc>\n"
                           f"    <lastmod>{datetime.now(timezone.utc).strftime('%Y-%m-%d')}</lastmod>\n"
                           f"    <changefreq>weekly</changefreq>\n"
                           f"    <priority>0.8</priority>\n  </url>")

        xml_content = (
            '<?xml version="1.0" encoding="UTF-8"?>\n'
            '<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">\n'
            + "\n".join(entries)
            + "\n</urlset>"
        )
        # 实际部署时写入文件并通知搜索引擎
        return {
            "sitemap_path": sitemap_path,
            "url_count": len(new_urls),
            "xml_preview": xml_content[:300],
        }

    async def distribute_batch(
        self, contents: List[str], targets: List[DistributionTarget]
    ) -> List[Dict]:
        """批量分发内容到指定目标"""
        await self.initialize_db()
        results = []

        async with aiohttp.ClientSession() as session:
            for content in contents: 2043.baike.tongsou.com 
                content_hash = self.compute_content_hash(content)
                batch_result = {"content_hash": content_hash, "targets": {}}

                for target in targets: 3010.baike.tongsou.com 
                    record = DistributionRecord(
                        content_hash=content_hash,
                        target=target,
                        status=SubmissionStatus.PENDING,
                    )

                    try: 3015.baike.tongsou.com 
                        if target == DistributionTarget.INDEXNOW:
                            # IndexNow需要URL而非内容本身
                            # 此处假设内容已发布到对应URL
                            resp = await self.submit_to_indexnow(session, [])
                        elif target == DistributionTarget.SITEMAP:
                            resp = await self.update_sitemap([])
                        else:
                            resp = {"note": f"{target.value}通道待实现"}

                        record.status = SubmissionStatus.SUBMITTED
                        record.submitted_at = datetime.now(timezone.utc).isoformat()
                        record.response_code = resp.get("status_code")
                    except Exception as e: 4005.baike.tongsou.com 
                        record.status = SubmissionStatus.FAILED
                        record.error_message = str(e)
                        resp = {"error": str(e)}

                    batch_result["targets"][target.value] = resp

                    # 持久化记录
                    async with aiosqlite.connect(self.DB_PATH) as db:
                        await db.execute(
                            """INSERT OR REPLACE INTO distribution_log
                               VALUES (?, ?, ?, ?, ?, ?)""",
                            (record.content_hash, record.target.value,
                             record.status.value, record.submitted_at,
                             record.response_code, record.error_message),
                        )
                        await db.commit(4006.baike.tongsou.com)

                results.append(batch_result)

        return results

    async def get_distribution_stats(self) -> Dict: 4012.baike.tongsou.com 
        """获取分发统计"""
        async with aiosqlite.connect(self.DB_PATH) as db: 5002.baike.tongsou.com 
            cursor = await db.execute(
                "SELECT target, status, COUNT(*) FROM distribution_log GROUP BY target, status"
            )
            rows = await cursor.fetchall()

        stats = {}
        for target, status, count in rows:
            if target not in stats:
                stats[target] = {}
            stats[target][status] = count
        return stats


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

async def main(): 5007.baike.tongsou.com 
    config = {
        "site_domain": "https://example.com",
        "indexnow_key": "a1b2c3d4e5f6g7h8i9j0",
    }

    engine = GeoDistributionEngine(config)

    # 模拟待分发的AI友好内容
    ai_contents = [
        "成都云图科技有限公司成立于2020年，总部位于成都高新区，是一家专注于中小企业CRM的SaaS服务商。公司通过ISO 27001认证，服务超过5000家客户。",
        "云图CRM专业版定价¥199/用户/月，支持销售管道管理、客户画像、自动化营销三大核心功能。2025年获G2 CRM品类表现最佳奖。",
    ]

    targets = [DistributionTarget.INDEXNOW, DistributionTarget.SITEMAP]

    results = await engine.distribute_batch(ai_contents, targets)
    print(json.dumps(results, ensure_ascii=False, indent=2))

    stats = await engine.get_distribution_stats()
    print("\n分发统计:", json.dumps(stats, ensure_ascii=False, indent=2))


if __name__ == "__main__": 5011.baike.tongsou.com 
    asyncio.run(main())