GEO内容溯源系统：让每条知识都能追到源头

          ┌─────────────────────────────────────┐
          │  原始网页采集                         │
          │  (Web Scraper)                       │
          └──────────────┬──────────────────────┘
                         │
          ┌──────────────▼──────────────────────┐
          │  Evidence Snapshot                  │
          │  - 网页快照保存                      │
          │  - 关键段落提取                      │
          │  - 发布时间/作者提取                  │
          └──────────────┬──────────────────────┘
                         │
          ┌──────────────▼──────────────────────┐
          │  Citation Chain Builder             │
          │  - 知识单元 → 证据片段 → 原始URL      │
          │  - 多级引用关系                      │
          └──────────────┬──────────────────────┘
                         │
          ┌──────────────▼──────────────────────┐
          │  Trust Scoring Engine               │
          │  - 来源权威性评分                     │
          │  - 时效性评分                        │
          │  - 交叉验证评分                      │
          └──────────────┬──────────────────────┘
                         │
          ┌──────────────▼──────────────────────┐
          │  Provenance Query API               │
          │  - 按知识单元ID查询溯源               │
          │  - 可视化引用树                      │
          └─────────────────────────────────────┘

"""
geo_provenance.py - GEO内容溯源系统
技术栈: Python / dataclasses / json / hashlib / enum / typing / datetime
场景: 为每条知识单元建立完整的来源追溯链，支持证据快照和可信度评分
"""
import os
import json
import hashlib
import logging
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Optional, Any, Tuple
from datetime import datetime, timedelta
from enum import Enum
from urllib.parse import urlparse

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

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


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

class SourceType(Enum): 17207.baike.tongsou.com 
    """来源类型"""
    WEB_PAGE = "web_page"           # 网页
    ACADEMIC_PAPER = "academic"     # 学术论文
    NEWS_ARTICLE = "news"           # 新闻
    GOVERNMENT = "government"       # 政府网站
    SOCIAL_MEDIA = "social"         # 社交媒体
    FORUM = "forum"                 # 论坛
    BLOG = "blog"                   # 博客
    WIKI = "wiki"                   # 维基
    OTHER = "other"

 
class AuthorityLevel(Enum): 17197.baike.tongsou.com 
    """权威性等级"""
    HIGH = "high"           # 高权威（政府、学术、主流媒体）
    MEDIUM = "medium"       # 中等（知名博客、行业网站）
    LOW = "low"             # 低（个人博客、论坛）
    UNKNOWN = "unknown"


@dataclass
class EvidenceSnapshot: 17194.baike.tongsou.com 
    """
    证据快照
    保存原始网页的关键信息，作为不可变证据
    """
    snapshot_id: str = ""
    url: str = ""
    title: str = ""
    content: str = ""             # 关键段落
    author: str = ""
    published_at: str = ""
    crawled_at: str = ""
    content_hash: str = ""        # 内容哈希，防篡改
    metadata: Dict[str, Any] = field(default_factory=dict)

    def __post_init__(self): 17193.baike.tongsou.com 
        if not self.snapshot_id: 17192.baike.tongsou.com 
            raw = f"{self.url}:{self.crawled_at}"
            self.snapshot_id = hashlib.md5(raw.encode()).hexdigest()[:12]
        if not self.crawled_at: 17191.baike.tongsou.com 
            self.crawled_at = datetime.now().isoformat()
        if not self.content_hash and self.content:
            self.content_hash = hashlib.sha256(
                self.content.encode("utf-8")
            ).hexdigest()

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


@dataclass
class CitationLink: 17190.baike.tongsou.com 
    """
    引用链接
    连接知识单元与证据快照
    """
    citation_id: str = ""
    unit_id: str = ""             # 知识单元ID
    snapshot_id: str = ""         # 证据快照ID
    url: str = ""
    source_type: SourceType = SourceType.WEB_PAGE
    quoted_text: str = ""         # 引用的原文片段
    quote_position: Dict[str, int] = field(default_factory=dict)  # 在原文中的位置
    relevance_score: float = 1.0  # 相关性评分 0-1
    created_at: str = ""

    def __post_init__(self): 17189.baike.tongsou.com 
        if not self.citation_id: 17188.baike.tongsou.com 
            raw = f"{self.unit_id}:{self.snapshot_id}"
            self.citation_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]: 17187.baike.tongsou.com 
        d = asdict(self)
        d["source_type"] = self.source_type.value
        return d


@dataclass
class TrustScore: 17186.baike.tongsou.com 
    """
    可信度评分
    综合评估一条知识的可信程度
    """
    unit_id: str = ""
    overall_score: float = 0.0    # 综合评分 0-100
    authority_score: float = 0.0  # 来源权威性
    timeliness_score: float = 0.0 # 时效性
    cross_verify_score: float = 0.0  # 交叉验证
    consistency_score: float = 0.0   # 内部一致性
    breakdown: Dict[str, Any] = field(default_factory=dict)
    evaluated_at: str = ""

    def __post_init__(self): 17184.baike.tongsou.com 
        if not self.evaluated_at: 17183.baike.tongsou.com 
            self.evaluated_at = datetime.now().isoformat()

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


@dataclass
class ProvenanceRecord: 17179.baike.tongsou.com 
    """
    完整溯源记录
    包含知识单元的所有溯源信息
    """
    unit_id: str = ""
    entity: str = ""
    attribute: str = ""
    value: str = ""
    citations: List[CitationLink] = field(default_factory=list)
    snapshots: List[EvidenceSnapshot] = field(default_factory=list)
    trust_score: Optional[TrustScore] = None
    generated_at: str = ""
    last_verified_at: str = ""

    def __post_init__(self): 17178.baike.tongsou.com 
        if not self.generated_at: 17177.baike.tongsou.com 
            self.generated_at = datetime.now().isoformat()

    def to_dict(self) -> Dict[str, Any]: 17175.baike.tongsou.com
        d = asdict(self)
        d["citations"] = [c.to_dict() for c in self.citations]
        d["snapshots"] = [s.to_dict() for s in self.snapshots]
        if self.trust_score: 17174.baike.tongsou.com 
            d["trust_score"] = self.trust_score.to_dict()
        return d

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


# ==================== 模块2: URL提取与证据快照 ====================

class EvidenceCollector: 17171.baike.tongsou.com 
    """
    证据收集器
    从原始网页提取并保存证据快照
    """

    # 域名到来源类型的映射规则
    DOMAIN_TYPE_RULES = {
        "gov.cn": SourceType.GOVERNMENT,
        "gov": SourceType.GOVERNMENT,
        "edu.cn": SourceType.ACADEMIC_PAPER,
        "edu": SourceType.ACADEMIC_PAPER,
        "ac.cn": SourceType.ACADEMIC_PAPER,
        "scholar.google": SourceType.ACADEMIC_PAPER,
        "arxiv.org": SourceType.ACADEMIC_PAPER,
        "doi.org": SourceType.ACADEMIC_PAPER,
        "weixin.qq.com": SourceType.SOCIAL_MEDIA,
        "weibo.com": SourceType.SOCIAL_MEDIA,
        "twitter.com": SourceType.SOCIAL_MEDIA,
        "zhihu.com": SourceType.FORUM,
        "reddit.com": SourceType.FORUM,
        "wikipedia.org": SourceType.WIKI,
        "zh.wikipedia.org": SourceType.WIKI,
    }

    def __init__(self, snapshot_dir: str = "./snapshots"):
        self.snapshot_dir = snapshot_dir
        os.makedirs(snapshot_dir, exist_ok=True)
        self.snapshots: Dict[str, EvidenceSnapshot] = {}

    def detect_source_type(self, url: str) -> SourceType: 17168.baike.tongsou.com 
        """根据URL检测来源类型"""
        parsed = urlparse(url)
        domain = parsed.netloc.lower(17162.baike.tongsou.com)

        for pattern, stype in self.DOMAIN_TYPE_RULES.items(): 17157.baike.tongsou.com 
            if pattern in domain:
                return stype
        return SourceType.WEB_PAGE

    def create_snapshot(
        self,
        url: str,
        title: str = "",
        content: str = "",
        author: str = "",
        published_at: str = "",
        metadata: Optional[Dict[str, Any]] = None,
    ) -> EvidenceSnapshot:
        """
        创建证据快照
        """
        snapshot = EvidenceSnapshot(
            url=url,
            title=title,
            content=content,
            author=author,
            published_at=published_at,
            metadata=metadata or {},
        )

        # 保存到本地文件（模拟持久化）
        snapshot_path = os.path.join(
            self.snapshot_dir, f"{snapshot.snapshot_id}.json"
        )
        with open(snapshot_path, "w", encoding="utf-8") as f:
            json.dump(snapshot.to_dict(17154.baike.tongsou.com), f, ensure_ascii=False, indent=2)

        self.snapshots[snapshot.snapshot_id] = snapshot
        logger.info(
            f"[Snapshot] id={snapshot.snapshot_id}, url={url[:60]}..."
        )
        return snapshot

    def get_snapshot(self, snapshot_id: str) -> Optional[EvidenceSnapshot]:
        """获取快照"""
        return self.snapshots.get(snapshot_id)

    def batch_create(
        self, pages: List[Dict[str, Any]]
    ) -> List[EvidenceSnapshot]: 17153.baike.tongsou.com  
        """
        批量创建快照
        Args:
            pages: [{"url": "...", "title": "...", "content": "...", ...}, ...]
        """
        return [self.create_snapshot(**page) for page in pages]


# ==================== 模块3: 引用链构建 ====================

class CitationChainBuilder: 17152.baike.tongsou.com 
    """
    引用链构建器
    建立知识单元与证据之间的引用关系
    """

    def __init__(self):
        self.citations: Dict[str, CitationLink] = {}  # citation_id -> CitationLink
        self.unit_citations: Dict[str, List[str]] = {}  # unit_id -> [citation_id, ...]
        self.snapshot_citations: Dict[str, List[str]] = {}  # snapshot_id -> [citation_id, ...]

    def add_citation(
        self,
        unit_id: str,
        snapshot: EvidenceSnapshot,
        quoted_text: str = "",
        quote_position: Optional[Dict[str, int]] = None,
        relevance_score: float = 1.0,
    ) -> CitationLink:
        """
        添加一条引用
        """
        citation = CitationLink(
            unit_id=unit_id,
            snapshot_id=snapshot.snapshot_id,
            url=snapshot.url,
            source_type=EvidenceCollector(17148.baike.tongsou.com).detect_source_type(snapshot.url),
            quoted_text=quoted_text,
            quote_position=quote_position or {},
            relevance_score=relevance_score,
        )

        self.citations[citation.citation_id] = citation

        # 建立索引
        if unit_id not in self.unit_citations: 17147.baike.tongsou.com 
            self.unit_citations[unit_id] = []
        self.unit_citations[unit_id].append(citation.citation_id)

        if snapshot.snapshot_id not in self.snapshot_citations: 17137.baike.tongsou.com 
            self.snapshot_citations[snapshot.snapshot_id] = []
        self.snapshot_citations[snapshot.snapshot_id].append(citation.citation_id)

        logger.info(
            f"[Citation] unit={unit_id[:8]}, snapshot={snapshot.snapshot_id[:8]}, "
            f"relevance={relevance_score}"
        )
        return citation

    def get_citations_for_unit(self, unit_id: str) -> List[CitationLink]:
        """获取某知识单元的所有引用"""
        citation_ids = self.unit_citations.get(unit_id, [])
        return [self.citations[cid] for cid in citation_ids if cid in self.citations]

    def get_citations_for_snapshot(self, snapshot_id: str) -> List[CitationLink]:
        """获取某快照被哪些知识单元引用"""
        citation_ids = self.snapshot_citations.get(snapshot_id, [])
        return [self.citations[cid] for cid in citation_ids if cid in self.citations]

    def build_provenance(
        self,
        unit_id: str,
        entity: str,
        attribute: str,
        value: str,
        snapshots: List[EvidenceSnapshot],
    ) -> ProvenanceRecord:
        """
        构建完整的溯源记录
        """
        citations = []
        for snap in snapshots:
            citation = self.add_citation(
                unit_id=unit_id,
                snapshot=snap,
                quoted_text="",
                relevance_score=1.0,
            )
            citations.append(citation)

        record = ProvenanceRecord(
            unit_id=unit_id,
            entity=entity,
            attribute=attribute,
            value=value,
            citations=citations,
            snapshots=snapshots,
        )
        return record


# ==================== 模块4: 可信度评分引擎 ====================

class TrustScoringEngine:
    """
    可信度评分引擎
    综合评估知识单元的可信程度
    """

    # 来源类型的基础权威性分数
    AUTHORITY_BASE_SCORES = {
        SourceType.GOVERNMENT: 95,
        SourceType.ACADEMIC_PAPER: 90,
        SourceType.NEWS_ARTICLE: 75,
        SourceType.WIKI: 60,
        SourceType.BLOG: 50,
        SourceType.FORUM: 30,
        SourceType.SOCIAL_MEDIA: 25,
        SourceType.WEB_PAGE: 50,
        SourceType.OTHER: 40,
    }

    # 时效性衰减系数（每天衰减的分数）
    TIMELINESS_DECAY = 0.05

    def __init__(self, current_date: Optional[str] = None):
        self.current_date = current_date or datetime.now().isoformat()

    def score_authority(self, citations: List[CitationLink]) -> float:
        """
        来源权威性评分
        取最高权威性来源的分数
        """
        if not citations:
            return 0.0

        scores = []
        for cit in citations: 17135.baike.tongsou.com 
            base = self.AUTHORITY_BASE_SCORES.get(cit.source_type, 40)
            scores.append(base)

        # 取最高分
        return max(scores)

    def score_timeliness(self, snapshots: List[EvidenceSnapshot]) -> float:
        """
        时效性评分
        越新的内容分数越高
        """
        if not snapshots:
            return 0.0

        scores = []
        now = datetime.fromisoformat(self.current_date)

        for snap in snapshots: 17122.baike.tongsou.com 
            pub_date_str = snap.published_at or snap.crawled_at
            if not pub_date_str: 17129.baike.tongsou.com 
                scores.append(50.0)
                continue

            try:
                pub_date = datetime.fromisoformat(pub_date_str)
                days_old = (now - pub_date).days
                if days_old < 0:
                    days_old = 0
                # 30天内满分，之后每天衰减
                score = max(0, 100 - days_old * self.TIMELINESS_DECAY * 100)
                scores.append(score)
            except (ValueError, TypeError):
                scores.append(50.0)

        return sum(scores) / len(scores)

    def score_cross_verify(
        self, citations: List[CitationLink], snapshots: List[EvidenceSnapshot]
    ) -> float:
        """
        交叉验证评分
        多个独立来源相互印证则分数高
        """
        if not citations: 17121.baike.tongsou.com 
            return 0.0

        # 不同来源类型的数量
        unique_types = set(c.source_type for c in citations)
        type_diversity = len(unique_types)

        # 独立URL数量
        unique_urls = set(c.url for c in citations)
        url_count = len(unique_urls)

        # 综合计算
        diversity_bonus = min(type_diversity * 10, 30)
        count_bonus = min(url_count * 5, 20)
        base = 50.0

        score = base + diversity_bonus + count_bonus
        return min(score, 100.0)

    def score_consistency(self, snapshots: List[EvidenceSnapshot]) -> float:
        """
        内部一致性评分
        多个快照内容是否一致
        """
        if len(snapshots) <= 1: 17117.baike.tongsou.com 
            return 80.0  # 单一来源，给中等分

        # 比较内容哈希的相似度（简化：如果哈希相同则完全一致）
        hashes = [s.content_hash for s in snapshots if s.content_hash]
        if not hashes: 17026.baike.tongsou.com 
            return 50.0

        unique_hashes = set(hashes)
        if len(unique_hashes) == 1:
            return 100.0
        else:
            # 有部分一致
            return 60.0

    def evaluate(self, record: ProvenanceRecord) -> TrustScore:
        """
        综合评估可信度
        """
        authority = self.score_authority(record.citations)
        timeliness = self.score_timeliness(record.snapshots)
        cross_verify = self.score_cross_verify(
            record.citations, record.snapshots
        )
        consistency = self.score_consistency(record.snapshots)

        # 加权平均
        weights = {
            "authority": 0.35,
            "timeliness": 0.25,
            "cross_verify": 0.25,
            "consistency": 0.15,
        }
        overall = (
            authority * weights["authority"]
            + timeliness * weights["timeliness"]
            + cross_verify * weights["cross_verify"]
            + consistency * weights["consistency"]
        )

        score = TrustScore(
            unit_id=record.unit_id,
            overall_score=round(overall, 2),
            authority_score=round(authority, 2),
            timeliness_score=round(timeliness, 2),
            cross_verify_score=round(cross_verify, 2),
            consistency_score=round(consistency, 2),
            breakdown={
                "weights": weights,
                "citation_count": len(record.citations),
                "snapshot_count": len(record.snapshots),
                "source_types": [c.source_type.value for c in record.citations],
            },
        )
        record.trust_score = score
        record.last_verified_at = datetime.now().isoformat()

        logger.info(
            f"[TrustScore] unit={record.unit_id[:8]}, "
            f"overall={overall:.2f}, authority={authority:.2f}, "
            f"timeliness={timeliness:.2f}"
        )
        return score


# ==================== 模块5: 溯源查询API ====================

class ProvenanceQueryAPI: 17022.baike.tongsou.com 
    """
    溯源查询API
    提供知识单元溯源信息的查询接口
    """

    def __init__(
        self,
        collector: EvidenceCollector,
        chain_builder: CitationChainBuilder,
        scoring_engine: TrustScoringEngine,
    ):
        self.collector = collector
        self.chain_builder = chain_builder
        self.scoring_engine = scoring_engine
        self.records: Dict[str, ProvenanceRecord] = {}

    def register_record(self, record: ProvenanceRecord): 17020.baike.tongsou.com 
        """注册溯源记录"""
        # 评估可信度
        self.scoring_engine.evaluate(record)
        self.records[record.unit_id] = record
        logger.info(f"[API] Registered provenance for unit={record.unit_id[:8]}")

    def query_by_unit_id(self, unit_id: str) -> Optional[Dict[str, Any]]:
        """
        按知识单元ID查询溯源信息
        """
        record = self.records.get(unit_id)
        if not record:
            return None
        return record.to_dict(16007.baike.tongsou.com)

    def query_by_url(self, url: str) -> List[Dict[str, Any]]:
        """



---
需要我继续写第11篇《GEO实时计算：用Flink/Kafka构建知识单元实时分发管道》吗？代码量同样拉满。