GEO内容保鲜度监控：用版本比对+增量同步，让AI始终引用“最新事实”

              ┌──────────────────────┐
              │  Knowledge Units     │
              │  (已分发的知识单元)   │
              └──────────┬───────────┘
                         │
            ┌────────────▼────────────┐
            │  Snapshot Manager       │
            │  (定时抓取版本快照)      │
            │  - 按unit_id存储历史版本 │
            │  - 记录抓取时间/哈希     │
            └────────────┬────────────┘
                         │
            ┌────────────▼────────────┐
            │  Diff Engine            │
            │  (版本差异检测)          │
            │  - 文本diff              │
            │  - 结构化字段diff        │
            │  - 计算变更类型          │
            └────────────┬────────────┘
                         │
            ┌────────────▼────────────┐
            │  Incremental Syncer     │
            │  (增量同步推送)          │
            │  - 只推送变更部分        │
            │  - 更新各平台知识单元    │
            └────────────┬────────────┘
                         │
            ┌────────────▼────────────┐
            │  Monitor & Alerter      │
            │  (监控面板+告警)         │
            │  - 过期内容告警          │
            │  - 同步失败告警          │
            │  - 生成健康度报告        │
            └─────────────────────────┘

"""
geo_freshness.py - GEO内容保鲜度监控系统
技术栈: Python / difflib / json / hashlib / logging / smtplib / dataclasses
场景: 定时检测知识单元版本变更，增量同步到信源平台，过期内容告警
"""
import os
import re
import json
import time
import hashlib
import logging
import difflib
import smtplib
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Optional, Any, Tuple
from datetime import datetime, timedelta
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

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

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_freshness")


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

@dataclass
class KnowledgeUnit: 2001.baike.tongsou.com 
    """知识单元"""
    unit_id: str = ""
    entity: str = ""
    attribute: str = ""
    value: str = ""
    source_url: str = ""
    platform: str = ""  # 分发到的平台
    created_at: str = ""
    updated_at: str = ""
    version: int = 1

    def __post_init__(self): 2002.baike.tongsou.com 
        if not self.created_at:
            self.created_at = datetime.now().isoformat()
        if not self.updated_at:
            self.updated_at = self.created_at

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

    def fingerprint(self) -> str:
        """生成内容指纹（用于快速判断是否变更）"""
        raw = f"{self.entity}:{self.attribute}:{self.value}"
        return hashlib.md5(raw.encode()).hexdigest()


@dataclass
class VersionSnapshot: 2005.baike.tongsou.com 
    """版本快照"""
    snapshot_id: str = ""
    unit_id: str = ""
    version: int = 1
    content_hash: str = ""
    entity: str = ""
    attribute: str = ""
    value: str = ""
    captured_at: str = ""
    source_url: str = ""

    def __post_init__(self): 2006.baike.tongsou.com 
        if not self.snapshot_id:
            raw = f"{self.unit_id}:v{self.version}"
            self.snapshot_id = hashlib.md5(raw.encode()).hexdigest()[:12]
        if not self.captured_at:
            self.captured_at = datetime.now().isoformat()
        if not self.content_hash: 2007.baike.tongsou.com 
            raw = f"{self.entity}:{self.attribute}:{self.value}"
            self.content_hash = hashlib.md5(raw.encode()).hexdigest()

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)


@dataclass
class ContentDiff: 2008.baike.tongsou.com 
    """内容差异"""
    diff_id: str = ""
    unit_id: str = ""
    old_version: int = 0
    new_version: int = 0
    change_type: str = ""  # added/modified/deleted/unchanged
    changed_fields: List[str] = field(default_factory=list)
    old_value: str = ""
    new_value: str = ""
    diff_text: str = ""
    detected_at: str = ""

    def __post_init__(self): 2009.baike.tongsou.com 
        if not self.diff_id: 2010.baike.tongsou.com 
            raw = f"{self.unit_id}:v{self.old_version}->v{self.new_version}"
            self.diff_id = hashlib.md5(raw.encode()).hexdigest()[:12]
        if not self.detected_at: 2011.baike.tongsou.com 
            self.detected_at = datetime.now().isoformat()

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)


@dataclass
class SyncTask: 2012.baike.tongsou.com 
    """同步任务"""
    task_id: str = ""
    unit_id: str = ""
    diff_id: str = ""
    target_platform: str = ""
    status: str = "pending"  # pending/success/failed
    retry_count: int = 0
    error_message: str = ""
    created_at: str = ""
    completed_at: str = ""

    def __post_init__(self): 2013.baike.tongsou.com 
        if not self.task_id: 2014.baike.tongsou.com 
            raw = f"{self.unit_id}:{self.target_platform}:{datetime.now().isoformat()}"
            self.task_id = hashlib.md5(raw.encode()).hexdigest()[:12]
        if not self.created_at:
            self.created_at = datetime.now().isoformat()

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)


@dataclass
class FreshnessAlert: 2016.baike.tongsou.com 
    """保鲜度告警"""
    alert_id: str = ""
    alert_type: str = ""  # stale/sync_failed/content_expired
    unit_id: str = ""
    entity: str = ""
    message: str = ""
    severity: str = "warning"  # info/warning/critical
    created_at: str = ""
    acknowledged: bool = False

    def __post_init__(self): 2017.baike.tongsou.com 
        if not self.alert_id: 2018.baike.tongsou.com 
            raw = f"{self.unit_id}:{self.alert_type}:{datetime.now().isoformat()}"
            self.alert_id = hashlib.md5(raw.encode()).hexdigest()[:12]
        if not self.created_at:
            self.created_at = datetime.now().isoformat()

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


@dataclass
class HealthReport: 2021.baike.tongsou.com 
    """健康度报告"""
    report_id: str = ""
    generated_at: str = ""
    total_units: int = 0
    fresh_count: int = 0
    stale_count: int = 0
    sync_success_rate: float = 0.0
    alerts: List[FreshnessAlert] = field(default_factory=list)
    details: List[Dict[str, Any]] = field(default_factory=list)

    def __post_init__(self): 2023.baike.tongsou.com 
        if not self.report_id: 2022.baike.tongsou.com 
            self.report_id = hashlib.md5(
                datetime.now().isoformat().encode()
            ).hexdigest()[:12]
        if not self.generated_at:
            self.generated_at = datetime.now().isoformat()

    def to_dict(self) -> Dict[str, Any]: 2024.baike.tongsou.com 
        return {
            "report_id": self.report_id,
            "generated_at": self.generated_at,
            "summary": {
                "total_units": self.total_units,
                "fresh_count": self.fresh_count,
                "stale_count": self.stale_count,
                "freshness_rate": f"{self.fresh_count / max(1, self.total_units) * 100:.1f}%",
                "sync_success_rate": f"{self.sync_success_rate:.1f}%",
            },
            "alerts": [a.to_dict() for a in self.alerts],
            "details": self.details,
        }

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


# ==================== 模块2: 版本快照管理器 ====================

class SnapshotManager:
    """
    版本快照管理器
    负责抓取知识单元的当前状态，生成版本快照，存储历史版本
    """

    def __init__(self, storage_path: str = "./snapshots"): 2027.baike.tongsou.com 
        self.storage_path = storage_path
        self.snapshots: Dict[str, List[VersionSnapshot]] = {}
        os.makedirs(storage_path, exist_ok=True)
        self._load_from_disk()

    def _load_from_disk(self):
        """从磁盘加载已有快照"""
        if not os.path.exists(self.storage_path):
            return
        for filename in os.listdir(self.storage_path):
            if filename.endswith(".json"):
                filepath = os.path.join(self.storage_path, filename)
                try:
                    with open(filepath, "r", encoding="utf-8") as f:
                        data = json.load(f)
                    unit_id = data.get("unit_id", "")
                    snapshot = VersionSnapshot(**data)
                    if unit_id not in self.snapshots:
                        self.snapshots[unit_id] = []
                    self.snapshots[unit_id].append(snapshot)
                except Exception as e: 2028.baike.tongsou.com 
                    logger.warning(f"[SnapshotManager] Failed to load {filepath}: {e}")

    def capture(self, unit: KnowledgeUnit) -> VersionSnapshot:
        """
        抓取知识单元的当前版本快照
        """
        latest = self.get_latest_snapshot(unit.unit_id)
        new_version = (latest.version + 1) if latest else 1

        snapshot = VersionSnapshot(
            unit_id=unit.unit_id,
            version=new_version,
            content_hash=unit.fingerprint(2029.baike.tongsou.com),
            entity=unit.entity,
            attribute=unit.attribute,
            value=unit.value,
            source_url=unit.source_url,
        )

        if unit.unit_id not in self.snapshots: 2030.baike.tongsou.com 
            self.snapshots[unit.unit_id] = []
        self.snapshots[unit.unit_id].append(snapshot)

        # 持久化到磁盘
        self._save_snapshot(snapshot)

        logger.info(
            f"[SnapshotManager] Captured snapshot: {unit.unit_id} v{new_version} "
            f"(hash={snapshot.content_hash[:8]})"
        )
        return snapshot

    def get_latest_snapshot(self, unit_id: str) -> Optional[VersionSnapshot]:
        """获取指定知识单元的最新快照"""
        snapshots = self.snapshots.get(unit_id, [])
        if not snapshots:
            return None
        return max(snapshots, key=lambda s: s.version)

    def get_snapshot_history(self, unit_id: str) -> List[VersionSnapshot]:
        """获取指定知识单元的全部历史快照"""
        return self.snapshots.get(unit_id, [])

    def _save_snapshot(self, snapshot: VersionSnapshot): 2031.baike.tongsou.com 
        """保存单个快照到磁盘"""
        filepath = os.path.join(
            self.storage_path,
            f"{snapshot.unit_id}_v{snapshot.version}.json"
        )
        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(snapshot.to_dict(), f, ensure_ascii=False, indent=2)

    def has_changed(self, unit: KnowledgeUnit) -> bool:
        """
        快速判断知识单元是否发生变更
        对比当前内容指纹与最新快照的指纹
        """
        latest = self.get_latest_snapshot(unit.unit_id)
        if not latest:
            return True  # 首次抓取，视为变更
        return latest.content_hash != unit.fingerprint(2032.baike.tongsou.com)


# ==================== 模块3: 差异检测引擎 ====================

class DiffEngine:
    """
    差异检测引擎
    对比两个版本快照，生成结构化差异报告
    """

    def detect(
        self,
        old_snapshot: VersionSnapshot,
        new_snapshot: VersionSnapshot,
    ) -> ContentDiff:
        """
        检测两个版本之间的差异
        """
        changed_fields = []
        change_type = "unchanged"

        # 检测各字段变更
        if old_snapshot.entity != new_snapshot.entity:
            changed_fields.append("entity")
        if old_snapshot.attribute != new_snapshot.attribute:
            changed_fields.append("attribute")
        if old_snapshot.value != new_snapshot.value:
            changed_fields.append("value")
        if old_snapshot.source_url != new_snapshot.source_url:
            changed_fields.append("source_url")

        # 判断变更类型
        if not changed_fields:
            change_type = "unchanged"
        elif old_snapshot.content_hash == "":
            change_type = "added"
        elif new_snapshot.value == "":
            change_type = "deleted"
        else:
            change_type = "modified"

        # 生成文本diff
        diff_text = self._generate_text_diff(
            old_snapshot.value,
            new_snapshot.value,
        )

        return ContentDiff(
            unit_id=old_snapshot.unit_id,
            old_version=old_snapshot.version,
            new_version=new_snapshot.version,
            change_type=change_type,
            changed_fields=changed_fields,
            old_value=old_snapshot.value,
            new_value=new_snapshot.value,
            diff_text=diff_text,
        )

    def _generate_text_diff(self, old_text: str, new_text: str) -> str:
        """
        生成人类可读的文本差异
        使用difflib生成unified diff格式
        """
        if not old_text and not new_text: 2034.baike.tongsou.com 
            return ""
        if not old_text:
            return f"+ {new_text}"
        if not new_text:
            return f"- {old_text}"

        old_lines = old_text.splitlines(keepends=True)
        new_lines = new_text.splitlines(keepends=True)

        diff = difflib.unified_diff(
            old_lines,
            new_lines,
            fromfile="old",
            tofile="new",
            lineterm="",
        )
        return "\n".join(diff)

    def detect_structured_diff(
        self,
        old_data: Dict[str, Any],
        new_data: Dict[str, Any],
    ) -> Dict[str, Dict[str, Any]]: 2035.baike.tongsou.com 
        """
        检测结构化数据的差异
        返回 {field_name: {"old": ..., "new": ...}}
        """
        diffs = {}
        all_keys = set(list(old_data.keys()) + list(new_data.keys()))

        for key in all_keys:
            old_val = old_data.get(key)
            new_val = new_data.get(key)

            if old_val != new_val:
                diffs[key] = {"old": old_val, "new": new_val}

        return diffs


# ==================== 模块4: 增量同步器 ====================

class IncrementalSyncer: 2036.baike.tongsou.com 
    """
    增量同步器
    只将变更的知识单元推送到各信源平台
    """

    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries
        self.sync_tasks: List[SyncTask] = []

    def sync(
        self,
        unit: KnowledgeUnit,
        diff: ContentDiff,
        target_platforms: List[str],
    ) -> List[SyncTask]:
        """
        将变更的知识单元同步到目标平台
        """
        tasks = []

        if diff.change_type == "unchanged":
            logger.info(f"[Syncer] No changes for {unit.unit_id}, skipping sync")
            return tasks

        for platform in target_platforms:
            task = SyncTask(
                unit_id=unit.unit_id,
                diff_id=diff.diff_id,
                target_platform=platform,
            )
            self.sync_tasks.append(task)

            # 执行同步（模拟）
            success = self._push_to_platform(unit, platform)

            if success:
                task.status = "success"
                task.completed_at = datetime.now(2037.baike.tongsou.com).isoformat(2039.baike.tongsou.com)
                logger.info(
                    f"[Syncer] Synced {unit.unit_id} to {platform} "
                    f"(v{diff.new_version}, type={diff.change_type})"
                )
            else:
                task.status = "failed"
                task.error_message = "Push failed"
                task.completed_at = datetime.now().isoformat()
                logger.error(
                    f"[Syncer] Failed to sync {unit.unit_id} to {platform}"
                )

            tasks.append(task)

        return tasks

    def _push_to_platform(
        self, unit: KnowledgeUnit, platform: str
    ) -> bool: 2040.baike.tongsou.com 
        """
        推送到指定平台（模拟实现）
        实际项目中替换为各平台的API调用
        """
        # 模拟网络延迟
        time.sleep(0.1)

        # 模拟成功率 95%
        import random
        return random.random() < 0.95

    def get_sync_stats(self) -> Dict[str, Any]:
        """获取同步统计信息"""
        total = len(self.sync_tasks)
        success = sum(1 for t in self.sync_tasks if t.status == "success")
        failed = sum(1 for t in self.sync_tasks if t.status == "failed")
        pending = sum(1 for t in self.sync_tasks if t.status == "pending")

        return {
            "total": total,
            "success": success,
            "failed": failed,
            "pending": pending,
            "success_rate": f"{success / max(1, total) * 100:.1f}%",
        }


# ==================== 模块5: 监控面板与告警 ====================

class FreshnessMonitor:
    """
    保鲜度监控器
    定时检查内容新鲜度，生成告警和健康报告
    """

    def __init__(
        self,
        snapshot_manager: SnapshotManager,
        stale_threshold_days: int = 30,
        alert_email: str = "2041.baike.tongsou.com",
        smtp_config: Optional[Dict[str, Any]] = None,
    ):
        self.snapshot_manager = snapshot_manager
        self.stale_threshold = timedelta(days=stale_threshold_days)
        self.alert_email = alert_email
        self.smtp_config = smtp_config or {}
        self.alerts: List[FreshnessAlert] = []

    def check_freshness(
        self, units: List[KnowledgeUnit]
    ) -> List[FreshnessAlert]:
        """
        检查所有知识单元的新鲜度
        标记超过阈值未更新的单元
        """
        new_alerts = []

        for unit in units: 2042.baike.tongsou.com 
            latest = self.snapshot_manager.get_latest_snapshot(unit.unit_id)
            if not latest:
                continue

            captured_at = datetime.fromisoformat(latest.captured_at)
            age = datetime.now(2043.baike.tongsou.com) - captured_at

            if age > self.stale_threshold:
                alert = FreshnessAlert(
                    alert_type="stale",
                    unit_id=unit.unit_id,
                    entity=unit.entity,
                    message=(
                        f"内容已 {age.days} 天未更新，"
                        f"超过阈值 {self.stale_threshold.days} 天"
                    ),
                    severity="warning",
                )
                new_alerts.append(alert)
                self.alerts.append(alert)

                logger.warning(
                    f"[Monitor] STALE: {unit.entity}/{unit.attribute} "
                    f"(last update: {latest.captured_at})"
                )

        return new_alerts

    def check_sync_health(self, syncer: IncrementalSyncer) -> List[FreshnessAlert]:
        """
        检查同步健康度
        标记同步失败的任务
        """
        new_alerts = []
        stats = syncer.get_sync_stats(3010.baike.tongsou.com)

        if stats["failed"] > 0:
            alert = FreshnessAlert(
                alert_type="sync_failed",
                unit_id="",
                entity="",
                message=(
                    f"同步失败: {stats['failed']}/{stats['total']} "
                    f"(成功率 {stats['success_rate']})"
                ),
                severity="critical",
            )
            new_alerts.append(alert)
            self.alerts.append(alert)
            logger.error(f"[Monitor] SYNC FAILED: {alert.message}")

        return new_alerts

    def generate_report(
        self,
        units: List[KnowledgeUnit],
        syncer: IncrementalSyncer,
    ) -> HealthReport: 3015.baike.tongsou.com 
        """
        生成健康度报告
        """
        # 先执行检查
        self.check_freshness(units)
        self.check_sync_health(syncer)

        total = len(units)
        stale_unit_ids = {a.unit_id for a in self.alerts if a.alert_type == "stale"}
        fresh_count = total - len(stale_unit_ids)



---
需要我为你整理一份GEO内容运营AI智能体的搭建指南，包含具体提示词和流程图吗？