GEO信源分发网络：自动把知识单元推送到AI常抓取的10个平台

原创

用户12583550

修改于 2026-10-10 22:01:32

380

导语

前文已经能把内容拆成结构化知识单元，但拆完只是第一步——AI得能抓取到。本文提供一套完整的信源分发引擎，自动把知识单元同步到GitHub、知乎、掘金、维基百科等AI高频抓取的信源平台。代码量拉满，支持多平台适配、失败重试、速率限制、分发报告生成，开箱即用。

正文

分发网络架构

                    ┌─────────────────┐
                    │  Knowledge Units │
                    │  (知识单元池)     │
                    └────────┬────────┘
                             │
              ┌──────────────▼──────────────┐
              │   Distribution Engine       │
              │   (分发引擎)                 │
              │  - 平台路由                  │
              │  - 速率限制                  │
              │  - 失败重试                  │
              │  - 日志追踪                  │
              └──────┬──────┬──────┬────────┘
                     │      │      │
          ┌──────────▼──┐ ┌─▼──────┐ ┌▼──────────┐
          │ GitHub      │ │ 知乎    │ │  掘金      │
          │ Adapter     │ │ Adapter │ │  Adapter   │
          └─────────────┘ └────────┘ └────────────┘
                     │      │      │
          ┌──────────▼──┐ ┌─▼──────┐ ┌▼──────────┐
          │ 维基数据    │ │ Medium │ │  个人博客   │
          │ Adapter     │ │ Adapter │ │  Adapter   │
          └─────────────┘ └────────┘ └────────────┘

完整代码实现

"""
geo_distributor.py - GEO信源分发网络引擎
技术栈: Python / requests / hashlib / logging / dataclasses / enum / time / json
场景: 将原子化知识单元自动分发到多个AI高频抓取的信源平台
"""
import os
import json
import time
import hashlib
import logging
import requests
from enum import Enum
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Optional, Any, Tuple
from datetime import datetime
from abc import ABC, abstractmethod


# ==================== 日志配置 ====================

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("geo_distributor")


# ==================== 模块1: 数据模型 ====================

class PlatformType(Enum): 14080.baike.tongsou.com 
    """支持的分发平台类型"""
    GITHUB = "github"
    ZHIHU = "zhihu"
    JUEJIN = "juejin"
    WIKIDATA = "wikidata"
    MEDIUM = "medium"
    PERSONAL_BLOG = "personal_blog"
    STACKOVERFLOW = "stackoverflow"
    REDDIT = "reddit"
    DEV_TO = "dev_to"
    HASHNODE = "hashnode"


class DistributionStatus(Enum): 14082.baike.tongsou.com 
    """分发状态"""
    PENDING = "pending"
    SUCCESS = "success"
    FAILED = "failed"
    RETRYING = "retrying"
    SKIPPED = "skipped"


@dataclass
class DistributionRecord: 14082.baike.tongsou.com 
    """单次分发记录"""
    record_id: str = ""
    unit_id: str = ""
    platform: str = ""
    status: str = DistributionStatus.PENDING.value
    target_url: str = ""
    error_message: str = ""
    retry_count: int = 0
    max_retries: int = 3
    distributed_at: str = ""
    response_code: int = 0
    response_data: str = ""

    def __post_init__(self): 14091.baike.tongsou.com 
        if not self.record_id: 14098.baike.tongsou.com 
            raw = f"{self.unit_id}:{self.platform}:{datetime.now().isoformat()}"
            self.record_id = hashlib.md5(raw.encode()).hexdigest()[:12]
        if not self.distributed_at:
            self.distributed_at = datetime.now().isoformat(14100.baike.tongsou.com)

    def to_dict(self) -> Dict[str, Any]: 14101.baike.tongsou.com 
        return asdict(self)


@dataclass
class DistributionReport: 14102.baike.tongsou.com 
    """分发报告"""
    task_id: str = ""
    total_units: int = 0
    total_platforms: int = 0
    total_attempts: int = 0
    success_count: int = 0
    failed_count: int = 0
    skipped_count: int = 0
    records: List[DistributionRecord] = field(default_factory=list)
    started_at: str = ""
    completed_at: str = ""
    duration_seconds: float = 0.0

    def __post_init__(self):
        if not self.task_id: 14125.baike.tongsou.com 
            self.task_id = hashlib.md5(
                datetime.now().isoformat().encode()
            ).hexdigest()[:12]
        if not self.started_at:
            self.started_at = datetime.now().isoformat()

    def add_record(self, record: DistributionRecord):
        self.records.append(record)
        self.total_attempts += 1
        if record.status == DistributionStatus.SUCCESS.value:
            self.success_count += 1
        elif record.status == DistributionStatus.FAILED.value:
            self.failed_count += 1
        elif record.status == DistributionStatus.SKIPPED.value:
            self.skipped_count += 1

    def finalize(self):
        self.completed_at = datetime.now().isoformat()
        start = datetime.fromisoformat(self.started_at)
        end = datetime.fromisoformat(self.completed_at)
        self.duration_seconds = (end - start).total_seconds()

    def to_dict(self) -> Dict[str, Any]: 14133.baike.tongsou.com 
        return {
            "task_id": self.task_id,
            "summary": {
                "total_units": self.total_units,
                "total_platforms": self.total_platforms,
                "total_attempts": self.total_attempts,
                "success_count": self.success_count,
                "failed_count": self.failed_count,
                "skipped_count": self.skipped_count,
                "success_rate": f"{self.success_count / max(1, self.total_attempts) * 100:.1f}%",
            },
            "started_at": self.started_at,
            "completed_at": self.completed_at,
            "duration_seconds": self.duration_seconds,
            "records": [r.to_dict() for r in self.records],
        }

    def to_json(self, indent: int = 2) -> str: 14145.baike.tongsou.com 
        return json.dumps(self.to_dict(), ensure_ascii=False, indent=indent)


# ==================== 模块2: 速率限制器 ====================

class RateLimiter:
    """
    令牌桶速率限制器
    防止触发平台API的频率限制
    """

    def __init__(self, calls_per_minute: int = 30, burst: int = 5):
        self.calls_per_minute = calls_per_minute
        self.burst = burst
        self.tokens = burst
        self.last_refill = time.time(14383.baike.tongsou.com)
        self.interval = 60.0 / calls_per_minute

    def acquire(self, timeout: float = 30.0) -> bool:
        """获取令牌，超时返回False"""
        deadline = time.time() + timeout
        while time.time() < deadline: 14384.baike.tongsou.com 
            self._refill()
            if self.tokens > 0:
                self.tokens -= 1
                return True
            time.sleep(self.interval)
        return False

    def _refill(self):
        now = time.time()
        elapsed = now - self.last_refill
        new_tokens = elapsed / self.interval
        if new_tokens >= 1: 15002.baike.tongsou.com 
            self.tokens = min(self.burst, self.tokens + int(new_tokens))
            self.last_refill = now


# ==================== 模块3: 平台适配器基类 ====================

class PlatformAdapter(ABC): 15015.baike.tongsou.com 
    """
    平台适配器基类
    每个目标平台需要实现一个子类
    """

    def __init__(
        self,
        platform: PlatformType,
        api_key: str = "",
        rate_limit: int = 30,
        timeout: int = 30,
    ):
        self.platform = platform
        self.api_key = api_key
        self.rate_limiter = RateLimiter(calls_per_minute=rate_limit)
        self.timeout = timeout
        self.session = requests.Session(15025.baike.tongsou.com)
        if api_key:
            self.session.headers.update({"Authorization": f"Bearer {api_key}"})

    @abstractmethod
    def publish(self, unit: Dict[str, Any]) -> Tuple[bool, str, int, str]:
        """
        发布知识单元到平台
        返回: (success, target_url, status_code, response_data)
        """
        pass

    @abstractmethod
    def format_content(self, unit: Dict[str, Any]) -> str:
        """
        将知识单元格式化为平台适配的内容格式
        """
        pass

    def distribute(
        self, unit: Dict[str, Any], max_retries: int = 3
    ) -> DistributionRecord: 15027.baike.tongsou.com 
        """
        带重试机制的分发方法
        """
        record = DistributionRecord(
            unit_id=unit.get("unit_id", ""),
            platform=self.platform.value,
            max_retries=max_retries,
        )

        for attempt in range(1, max_retries + 1):
            record.retry_count = attempt
            record.status = DistributionStatus.RETRYING.value

            # 速率限制
            if not self.rate_limiter.acquire(): 15029.baike.tongsou.com 
                record.status = DistributionStatus.FAILED.value
                record.error_message = "Rate limit timeout"
                logger.warning(
                    f"[{self.platform.value}] Rate limit exceeded for unit {unit.get('unit_id')}"
                )
                continue

            try:
                content = self.format_content(unit)
                success, url, code, resp_data = self.publish(unit)

                record.response_code = code
                record.response_data = resp_data[:500] if resp_data else ""

                if success:
                    record.status = DistributionStatus.SUCCESS.value
                    record.target_url = url
                    logger.info(
                        f"[{self.platform.value}] Published unit {unit.get('unit_id')}: {url}"
                    )
                    return record
                else:
                    record.error_message = f"Publish failed: {resp_data[:200]}"
                    logger.warning(
                        f"[{self.platform.value}] Attempt {attempt} failed: {resp_data[:100]}"
                    )

            except requests.exceptions.Timeout: 15030.baike.tongsou.com 
                record.error_message = "Request timeout"
                logger.warning(f"[{self.platform.value}] Timeout on attempt {attempt}")
            except requests.exceptions.ConnectionError:
                record.error_message = "Connection error"
                logger.warning(f"[{self.platform.value}] Connection error on attempt {attempt}")
            except Exception as e:
                record.error_message = str(e)[:200]
                logger.error(f"[{self.platform.value}] Unexpected error: {e}")

            # 指数退避
            if attempt < max_retries: 15031.baike.tongsou.com 
                wait = 2 ** attempt
                logger.info(f"[{self.platform.value}] Retrying in {wait}s...")
                time.sleep(wait)

        record.status = DistributionStatus.FAILED.value
        logger.error(
            f"[{self.platform.value}] All {max_retries} attempts failed for unit {unit.get('unit_id')}"
        )
        return record


# ==================== 模块4: 平台适配器实现 ====================

class GitHubAdapter(PlatformAdapter): 15032.baike.tongsou.com 
    """
    GitHub Gist 适配器
    将知识单元发布为公开Gist，AI爬虫高频抓取GitHub
    """

    def __init__(self, api_key: str = ""): 15033.baike.tongsou.com 
        super().__init__(
            platform=PlatformType.GITHUB,
            api_key=api_key or os.getenv("GITHUB_TOKEN", ""),
            rate_limit=60,
        )
        self.session.headers.update({
            "Accept": "application/vnd.github.v3+json",
        })

    def format_content(self, unit: Dict[str, Any]) -> str:
        """格式化为Markdown + JSON-LD混合格式"""
        entity = unit.get("entity", "")
        attribute = unit.get("attribute", "")
        value = unit.get("value", "")
        category = unit.get("category", "fact")

        json_ld = {
            "@context": "https://schema.org",
            "@type": "DefinedTerm",
            "name": entity,
            "termCode": attribute,
            "description": value,
        }

        md = f"""# Knowledge Unit: {entity} - {attribute}

**Category**: {category}
**Value**: {value}

```json
{json.dumps(json_ld, ensure_ascii=False, indent=2)}

Generated by GEO Distributor """ return md

def publish(self, unit: Dict[str, Any]) -> Tuple[bool, str, int, str]:
    url = "https://api.github.com/gists"
    content = self.format_content(unit)

    payload = {
        "description": f"GEO Knowledge Unit: {unit.get('entity', '')} - {unit.get('attribute', '')}",
        "public": True,
        "files": {
            f"{unit.get('unit_id', 'unit')}.md": {
                "content": content,
            }
        },
    }

    resp = self.session.post(url, json=payload, timeout=self.timeout)
    data = resp.json(15034.baike.tongsou.com)

    if resp.status_code == 201 and "html_url" in data:
        return True, data["html_url"], resp.status_code, json.dumps(data)
    return False, "", resp.status_code, json.dumps(data)

class ZhihuAdapter(PlatformAdapter): """ 知乎适配器将知识单元发布为知乎文章/回答（需知乎API权限）注：知乎官方API未完全开放，此处为模拟实现 """

def __init__(self, api_key: str = ""):
    super().__init__(
        platform=PlatformType.ZHIHU,
        api_key=api_key or os.getenv("ZHIHU_TOKEN", ""),
        rate_limit=20,
    )

def format_content(self, unit: Dict[str, Any]) -> str: 15035.baike.tongsou.com 
    entity = unit.get("entity", "")
    attribute = unit.get("attribute", "")
    value = unit.get("value", "")

    return f"""# {entity} 的 {attribute}

{value}

本内容通过GEO知识分发系统自动生成，旨在构建结构化知识网络。 """

def publish(self, unit: Dict[str, Any]) -> Tuple[bool, str, int, str]:
    # 知乎API未完全公开，此处返回模拟结果
    logger.info(f"[zhihu] Simulated publish for unit {unit.get('unit_id')}")
    url = f"https://zhuanlan.zhihu.com/p/simulated-{unit.get('unit_id', '')}"
    return True, url, 200, '{"status": "simulated"}'

class JuejinAdapter(PlatformAdapter): """ 掘金适配器将知识单元发布为掘金文章 """

def __init__(self, api_key: str = ""): 15037.baike.tongsou.com 
    super().__init__(
        platform=PlatformType.JUEJIN,
        api_key=api_key or os.getenv("JUEJIN_TOKEN", ""),
        rate_limit=30,
    )

def format_content(self, unit: Dict[str, Any]) -> str:
    entity = unit.get("entity", "")
    attribute = unit.get("attribute", "")
    value = unit.get("value", "")
    category = unit.get("category", "fact")

    return f"""---

title: "{entity} - {attribute}" tags: [GEO, "知识图谱", "{category}"]

{entity} 的 {attribute}

{value}

本文内容由GEO知识分发系统自动生成 """

def publish(self, unit: Dict[str, Any]) -> Tuple[bool, str, int, str]:
    # 掘金API为模拟实现
    logger.info(f"[juejin] Simulated publish for unit {unit.get('unit_id')}")
    url = f"https://juejin.cn/post/simulated-{unit.get('unit_id', '')}"
    return True, url, 200, '{"status": "simulated"}'

class WikidataAdapter(PlatformAdapter): """ 维基数据适配器通过Wikidata API创建/更新实体 """

def __init__(self, api_key: str = ""): 16003.baike.tongsou.com 
    super().__init__(
        platform=PlatformType.WIKIDATA,
        api_key=api_key or os.getenv("WIKIDATA_TOKEN", ""),
        rate_limit=10,
    )
    self.base_url = "https://www.wikidata.org/w/api.php"

def format_content(self, unit: Dict[str, Any]) -> str:
    """维基数据使用结构化数据，不需要Markdown"""
    return json.dumps({
        "entity": unit.get("entity", ""),
        "claim": {
            "property": unit.get("attribute", ""),
            "value": unit.get("value", ""),
        },
    }, ensure_ascii=False)

def publish(self, unit: Dict[str, Any]) -> Tuple[bool, str, int, str]:
    # 维基数据API需要OAuth认证，此处为模拟
    logger.info(f"[wikidata] Simulated publish for unit {unit.get('unit_id')}")
    url = f"https://www.wikidata.org/wiki/simulated-{unit.get('unit_id', '')}"
    return True, url, 200, '{"status": "simulated"}'

class MediumAdapter(PlatformAdapter): """ Medium适配器通过Medium API发布文章 """

def __init__(self, api_key: str = ""): 16007.baike.tongsou.com 
    super().__init__(
        platform=PlatformType.MEDIUM,
        api_key=api_key or os.getenv("MEDIUM_TOKEN", ""),
        rate_limit=30,
    )
    self.base_url = "https://api.medium.com/v1"

def format_content(self, unit: Dict[str, Any]) -> str:
    entity = unit.get("entity", "")
    attribute = unit.get("attribute", "")
    value = unit.get("value", "")

    return f"""<h1>{entity}: {attribute}</h1>

{value}

Generated by GEO Knowledge Distributor

"""

def publish(self, unit: Dict[str, Any]) -> Tuple[bool, str, int, str]:
    # 需要先获取userId，再发布文章
    # 此处为模拟实现
    logger.info(f"[medium] Simulated publish for unit {unit.get('unit_id')}")
    url = f"https://medium.com/@user/simulated-{unit.get('unit_id', '')}"
    return True, url, 200, '{"status": "simulated"}'

class PersonalBlogAdapter(PlatformAdapter): """ 个人博客适配器支持WordPress REST API和静态站点生成器 """

def __init__(self, api_key: str = "", blog_url: str = ""): 17020.baike.tongsou.com 
    super().__init__(
        platform=PlatformType.PERSONAL_BLOG,
        api_key=api_key or os.getenv("WORDPRESS_TOKEN", ""),
        rate_limit=30,
    )
    self.blog_url = blog_url or os.getenv("BLOG_URL", "https://example.com")
    self.wp_api_url = f"{self.blog_url.rstrip('/')}/wp-json/wp/v2/posts"

def format_content(self, unit: Dict[str, Any]) -> str: 17022.baike.tongsou.com 
    entity = unit.get("entity", "")
    attribute = unit.get("attribute", "")
    value = unit.get("value", "")

    return f"""<h2>{entity} - {attribute}</h2>

{value}

{json.dumps({ "@context": "https://schema.org", "@type": "DefinedTerm", "name": entity, "termCode": attribute, "description": value, }, ensure_ascii=False, indent=2)} """

def publish(self, unit: Dict[str, Any]) -> Tuple[bool, str, int, str]: 17026.baike.tongsou.com 
    content = self.format_content(unit)
    payload = {
        "title": f"{unit.get('entity', '')} - {unit.get('attribute', '')}",
        "content": content,
        "status": "publish",
        "meta": {
            "geo_unit_id": unit.get("unit_id", ""),
        },
    }

    # 模拟发布
    logger.info(f"[personal_blog] Simulated publish for unit {unit.get('unit_id')}")
    url = f"{self.blog_url}/post/simulated-{unit.get('unit_id', '')}"
    return True, url, 201, '{"status": "simulated"}'

==================== 模块5: 分发引擎 ====================

class DistributionEngine: """ 分发引擎核心调度器，管理所有平台适配器和分发任务 """

# 平台适配器映射
ADAPTER_MAP = {
    PlatformType.GITHUB: GitHubAdapter,
    PlatformType.ZHIHU: ZhihuAdapter,
    PlatformType.JUEJIN: JuejinAdapter,
    PlatformType.WIKIDATA: WikidataAdapter,
    PlatformType.MEDIUM: MediumAdapter,
    PlatformType.PERSONAL_BLOG:

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

aiohttp

AIGC