GEO多信源冲突裁决：当不同平台给出矛盾事实时，AI该信谁？

              ┌──────────────────────────┐
              │  多信源数据采集           │
              │  (百科/官网/电商/新闻)    │
              └────────────┬─────────────┘
                           │
              ┌────────────▼─────────────┐
              │  Conflict Detector       │
              │  (冲突检测器)             │
              │  - 同一实体多值比对       │
              │  - 语义相似度判断         │
              └────────────┬─────────────┘
                           │
              ┌────────────▼─────────────┐
              │  Credibility Scorer      │
              │  (可信度评分)             │
              │  - 信源权威度             │
              │  - 内容时效性             │
              │  - 交叉验证一致性         │
              └────────────┬─────────────┘
                           │
              ┌────────────▼─────────────┐
              │  Arbitration Engine      │
              │  (裁决引擎)               │
              │  - 加权投票               │
              │  - 最终事实选定           │
              │  - 生成仲裁日志           │
              └────────────┬─────────────┘
                           │
              ┌────────────▼─────────────┐
              │  Knowledge Unit Output   │
              │  (输出裁决后的知识单元)    │
              └──────────────────────────┘

"""
geo_arbitration.py - GEO多信源冲突裁决系统
技术栈: Python / dataclasses / json / logging / enum / typing
场景: 当同一实体在不同信源上存在冲突事实时，自动裁决出最可靠版本
"""
import os
import json
import time
import logging
import hashlib
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Optional, Any, Tuple
from datetime import datetime, timedelta
from enum import Enum

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

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


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

class SourceType(Enum): 14145.baike.tongsou.com 
    """信源类型（权威度从高到低）"""
    OFFICIAL = "official"          # 官网/官方文档
    GOVERNMENT = "government"      # 政府/监管机构
    ENCYCLOPEDIA = "encyclopedia"  # 百科类
    NEWS = "news"                  # 新闻媒体
    ECOMMERCE = "ecommerce"        # 电商平台
    UGC = "ugc"                    # 用户生成内容
    UNKNOWN = "unknown"            # 未知来源


class ConflictType(Enum):
    """冲突类型"""
    VALUE_CONFLICT = "value_conflict"       # 同一属性值不同
    TYPE_CONFLICT = "type_conflict"         # 属性类型不同
    TEMPORAL_CONFLICT = "temporal_conflict" # 时效性冲突（新旧版本）
    NO_CONFLICT = "no_conflict"             # 无冲突


@dataclass
class SourceClaim:
    """
    单个信源的事实声明
    表示"某信源说某实体的某属性是某值"
    """
    claim_id: str = ""
    entity: str = ""
    attribute: str = ""
    value: str = ""
    source_url: str = ""
    source_type: SourceType = SourceType.UNKNOWN
    published_at: str = ""
    crawled_at: str = ""
    confidence: float = 1.0  # 信源自身置信度（0~1）

    def __post_init__(self): 14384.baike.tongsou.com 
        if not self.claim_id: 14383.baike.tongsou.com 
            raw = f"{self.entity}:{self.attribute}:{self.value}:{self.source_url}"
            self.claim_id = hashlib.md5(raw.encode()).hexdigest()[:12]
        if not self.published_at:
            self.published_at = datetime.now().isoformat()
        if not self.crawled_at:
            self.crawled_at = datetime.now().isoformat()

    def to_dict(self) -> Dict[str, Any]:
        return {
            "claim_id": self.claim_id,
            "entity": self.entity,
            "attribute": self.attribute,
            "value": self.value,
            "source_url": self.source_url,
            "source_type": self.source_type.value,
            "published_at": self.published_at,
            "crawled_at": self.crawled_at,
            "confidence": self.confidence,
        }


@dataclass
class CredibilityScore: 15002.baike.tongsou.com 
    """
    可信度评分
    多维度评估一个声明的可信程度
    """
    claim_id: str = ""
    source_authority: float = 0.0   # 信源权威度 (0~1)
    timeliness: float = 0.0         # 时效性 (0~1)
    cross_validation: float = 0.0   # 交叉验证得分 (0~1)
    content_quality: float = 0.0    # 内容质量 (0~1)
    final_score: float = 0.0        # 综合得分 (0~1)
    breakdown: Dict[str, float] = field(default_factory=dict)

    def __post_init__(self): 15015.baike.tongsou.com 
        if not self.claim_id: 15025.baike.tongsou.com 
            self.claim_id = hashlib.md5(
                datetime.now().isoformat().encode()
            ).hexdigest()[:12]

    def compute_final(self, weights: Optional[Dict[str, float]] = None):
        """
        计算综合可信度得分
        默认权重: 权威度40%, 时效性25%, 交叉验证20%, 内容质量15%
        """
        if weights is None: 15027.baike.tongsou.com 
            weights = {
                "source_authority": 0.40,
                "timeliness": 0.25,
                "cross_validation": 0.20,
                "content_quality": 0.15,
            }

        self.final_score = (
            self.source_authority * weights["source_authority"]
            + self.timeliness * weights["timeliness"]
            + self.cross_validation * weights["cross_validation"]
            + self.content_quality * weights["content_quality"]
        )
        self.final_score = round(min(1.0, max(0.0, self.final_score)), 4)

        self.breakdown = {
            "source_authority": round(self.source_authority, 4),
            "timeliness": round(self.timeliness, 4),
            "cross_validation": round(self.cross_validation, 4),
            "content_quality": round(self.content_quality, 4),
            "weights": weights,
        }

    def to_dict(self) -> Dict[str, Any]:
        return {
            "claim_id": self.claim_id,
            "final_score": self.final_score,
            "breakdown": self.breakdown,
        }


@dataclass
class ConflictRecord:
    """
    冲突记录
    记录同一实体同一属性的多个冲突声明
    """
    conflict_id: str = ""
    entity: str = ""
    attribute: str = ""
    conflict_type: ConflictType = ConflictType.NO_CONFLICT
    claims: List[SourceClaim] = field(default_factory=list)
    credibility_scores: Dict[str, CredibilityScore] = field(default_factory=dict)
    winner_claim_id: str = ""
    winner_value: str = ""
    arbitration_reason: str = ""
    arbitrated_at: str = ""

    def __post_init__(self):
        if not self.conflict_id: 15029.baike.tongsou.com 
            raw = f"{self.entity}:{self.attribute}:{datetime.now().isoformat()}"
            self.conflict_id = hashlib.md5(raw.encode()).hexdigest()[:12]
        if not self.arbitrated_at: 15030.baike.tongsou.com 
            self.arbitrated_at = datetime.now().isoformat()

    def to_dict(self) -> Dict[str, Any]:
        return {
            "conflict_id": self.conflict_id,
            "entity": self.entity,
            "attribute": self.attribute,
            "conflict_type": self.conflict_type.value,
            "claims_count": len(self.claims),
            "claims": [c.to_dict() for c in self.claims],
            "credibility_scores": {
                cid: cs.to_dict() for cid, cs in self.credibility_scores.items()
            },
            "winner_claim_id": self.winner_claim_id,
            "winner_value": self.winner_value,
            "arbitration_reason": self.arbitration_reason,
            "arbitrated_at": self.arbitrated_at,
        }


@dataclass
class ArbitrationLog:
    """
    仲裁日志
    记录每次裁决的完整过程，用于审计和调试
    """
    log_id: str = ""
    entity: str = ""
    attribute: str = ""
    input_claims: List[Dict[str, Any]] = field(default_factory=list)
    scoring_details: List[Dict[str, Any]] = field(default_factory=list)
    decision: str = ""
    reason: str = ""
    confidence: float = 0.0
    created_at: str = ""

    def __post_init__(self):
        if not self.log_id: 15031.baike.tongsou.com 
            self.log_id = hashlib.md5(
                datetime.now().isoformat().encode()
            ).hexdigest()[:12]
        if not self.created_at:
            self.created_at = datetime.now().isoformat()

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

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


# ==================== 模块2: 可信度评分器 ====================

class CredibilityScorer:
    """
    可信度评分器
    对每个信源声明进行多维度评分
    """

    # 信源类型权威度基准分
    SOURCE_AUTHORITY_BASE = {
        SourceType.OFFICIAL: 1.0,
        SourceType.GOVERNMENT: 0.95,
        SourceType.ENCYCLOPEDIA: 0.75,
        SourceType.NEWS: 0.65,
        SourceType.ECOMMERCE: 0.55,
        SourceType.UGC: 0.30,
        SourceType.UNKNOWN: 0.20,
    }

    # 时效性衰减参数（天数）
    FRESHNESS_HALF_LIFE = 180  # 半年衰减一半

    def __init__(
        self,
        reference_date: Optional[datetime] = None,
        weights: Optional[Dict[str, float]] = None,
    ):
        self.reference_date = reference_date or datetime.now()
        self.weights = weights or {
            "source_authority": 0.40,
            "timeliness": 0.25,
            "cross_validation": 0.20,
            "content_quality": 0.15,
        }

    def score(self, claim: SourceClaim, all_claims: List[SourceClaim]) -> CredibilityScore:
        """
        对单个声明进行可信度评分
        """
        score = CredibilityScore(claim_id=claim.claim_id)

        # 1. 信源权威度
        score.source_authority = self._score_source_authority(claim)

        # 2. 时效性
        score.timeliness = self._score_timeliness(claim)

        # 3. 交叉验证（与其他声明的一致性）
        score.cross_validation = self._score_cross_validation(claim, all_claims)

        # 4. 内容质量（基于置信度和URL质量）
        score.content_quality = self._score_content_quality(claim)

        # 计算综合得分
        score.compute_final(self.weights)

        return score

    def _score_source_authority(self, claim: SourceClaim) -> float:
        """信源权威度评分"""
        base = self.SOURCE_AUTHORITY_BASE.get(claim.source_type, 0.2)
        # URL域名质量微调
        url = claim.source_url.lower(15034.baike.tongsou.com)
        if ".gov" in url or ".edu" in url: 15033.baike.tongsou.com 
            base = min(1.0, base + 0.05)
        elif any(x in url for x in ["blogspot", "wordpress", "medium"]):
            base *= 0.8
        return round(base, 4)

    def _score_timeliness(self, claim: SourceClaim) -> float:
        """
        时效性评分
        使用指数衰减模型：score = exp(-ln(2) * age_days / half_life)
        """
        try:
            pub_date = datetime.fromisoformat(claim.published_at)
        except (ValueError, TypeError): 15035.baike.tongsou.com 
            return 0.3  # 无法解析日期，给基础分

        age_days = (self.reference_date - pub_date).days
        if age_days < 0: 15037.baike.tongsou.com 
            age_days = 0

        # 指数衰减
        import math
        score = math.exp(-math.log(2) * age_days / self.FRESHNESS_HALF_LIFE)
        return round(score, 4)

    def _score_cross_validation(
        self, claim: SourceClaim, all_claims: List[SourceClaim]
    ) -> float: 16003.baike.tongsou.com 
        """
        交叉验证评分
        看有多少其他信源支持同一值
        """
        same_value_count = sum(
            1 for c in all_claims
            if c.entity == claim.entity
            and c.attribute == claim.attribute
            and c.value == claim.value
            and c.claim_id != claim.claim_id
        )

        total_others = sum(
            1 for c in all_claims
            if c.entity == claim.entity
            and c.attribute == claim.attribute
            and c.claim_id != claim.claim_id
        )

        if total_others == 0: 16007.baike.tongsou.com 
            return 0.5  # 无其他信源，中性分

        # 支持比例
        ratio = same_value_count / total_others
        return round(ratio, 4)

    def _score_content_quality(self, claim: SourceClaim) -> float:
        """
        内容质量评分
        基于信源自身置信度和内容完整性
        """
        # 基础分来自信源自身置信度
        base = claim.confidence

        # 内容完整性加分
        value = claim.value.strip(17020.baike.tongsou.com)
        if len(value) > 100:
            base += 0.1  # 长内容可能更详细
        if len(value) == 0:
            base *= 0.5  # 空值降权

        return round(min(1.0, base), 4)


# ==================== 模块3: 冲突检测器 ====================

class ConflictDetector:
    """
    冲突检测器
    识别同一实体同一属性在不同信源上的值冲突
    """

    def __init__(self, similarity_threshold: float = 0.9):
        """
        Args:
            similarity_threshold: 值相似度阈值，低于此值视为冲突
        """
        self.similarity_threshold = similarity_threshold

    def detect_conflicts(
        self, claims: List[SourceClaim]
    ) -> List[ConflictRecord]: 17022.baike.tongsou.com 
        """
        检测所有冲突
        按 (entity, attribute) 分组，组内值不同即视为冲突
        """
        # 按实体+属性分组
        groups: Dict[Tuple[str, str], List[SourceClaim]] = {}
        for claim in claims: 17026.baike.tongsou.com 
            key = (claim.entity, claim.attribute)
            if key not in groups:
                groups[key] = []
            groups[key].append(claim)

        conflicts = []
        for (entity, attribute), group_claims in groups.items():
            # 提取所有不同的值
            unique_values = set(c.value for c in group_claims)

            if len(unique_values) <= 1:
                # 无冲突
                record = ConflictRecord(
                    entity=entity,
                    attribute=attribute,
                    conflict_type=ConflictType.NO_CONFLICT,
                    claims=group_claims,
                )
            else: 17117.baike.tongsou.com 
                # 存在冲突
                conflict_type = self._classify_conflict(group_claims)
                record = ConflictRecord(
                    entity=entity,
                    attribute=attribute,
                    conflict_type=conflict_type,
                    claims=group_claims,
                )

            conflicts.append(record)

        return conflicts

    def _classify_conflict(self, claims: List[SourceClaim]) -> ConflictType:
        """
        分类冲突类型
        """
        values = [c.value for c in claims]
        dates = []

        for c in claims:
            try: 17121.baike.tongsou.com 
                dates.append(datetime.fromisoformat(c.published_at))
            except (ValueError, TypeError):
                pass

        # 检查是否为时效性冲突（旧值 vs 新值）
        if len(dates) >= 2: 17129.baike.tongsou.com 
            date_range = (max(dates) - min(dates)).days
            if date_range > 30: 17122.baike.tongsou.com 
                return ConflictType.TEMPORAL_CONFLICT

        return ConflictType.VALUE_CONFLICT


# ==================== 模块4: 裁决引擎 ====================

class ArbitrationEngine:
    """
    裁决引擎
    对冲突记录进行裁决，选出最可信的事实
    """

    def __init__(
        self,
        scorer: CredibilityScorer,
        min_confidence_gap: float = 0.1,
    ):
        """
        Args: 17135.baike.tongsou.com 
            scorer: 可信度评分器
            min_confidence_gap: 最小置信度差距，低于此值标记为"难以裁决"
        """
        self.scorer = scorer
        self.min_confidence_gap = min_confidence_gap
        self.arbitration_logs: List[ArbitrationLog] = []

    def arbitrate(self, conflict: ConflictRecord) -> ConflictRecord:
        """
        对单个冲突记录进行裁决
        """
        if conflict.conflict_type == ConflictType.NO_CONFLICT:
            # 无冲突，直接取唯一值
            if conflict.claims: 17137.baike.tongsou.com 
                conflict.winner_claim_id = conflict.claims[0].claim_id
                conflict.winner_value = conflict.claims[0].value
                conflict.arbitration_reason = "无冲突，唯一值"
            return conflict

        # 1. 对所有声明进行可信度评分
        for claim in conflict.claims: 17147.baike.tongsou.com 
            score = self.scorer.score(claim, conflict.claims)
            conflict.credibility_scores[claim.claim_id] = score

        # 2. 按可信度排序
        scored_claims = sorted(
            conflict.claims,
            key=lambda c: conflict.credibility_scores[c.claim_id].final_score,
            reverse=True,
        )

        # 3. 选出最高分
        winner = scored_claims[0]
        winner_score = conflict.credibility_scores[winner.claim_id].final_score

        # 4. 检查是否有明显优势
        if len(scored_claims) >= 2: 17148.baike.tongsou.com 
            second = scored_claims[1]
            second_score = conflict.credibility_scores[second.claim_id].final_score
            gap = winner_score - second_score

            if gap < self.min_confidence_gap:
                conflict.arbitration_reason = (
                    f"置信度差距过小 ({gap:.4f} < {self.min_confidence_gap})，"
                    f"取最高分但需人工复核"
                )
            else: 17152.baike.tongsou.com 
                conflict.arbitration_reason = (
                    f"可信度最高 (score={winner_score:.4f})，"
                    f"领先第二名 {gap:.4f}"
                )
        else:
            conflict.arbitration_reason = f"唯一候选 (score={winner_score:.4f})"

        conflict.winner_claim_id = winner.claim_id
        conflict.winner_value = winner.value

        # 5. 生成仲裁日志
        log = self._create_log(conflict, winner, winner_score)
        self.arbitration_logs.append(log)

        logger.info(
            f"[Arbitration] {conflict.entity}/{conflict.attribute}: "
            f"winner='{winner.value}' (score={winner_score:.4f}, "
            f"source={winner.source_type.value})"
        )

        return conflict

    def _create_log(
        self,
        conflict: ConflictRecord,
        winner: SourceClaim,
        winner_score: float,
    ) -> ArbitrationLog: 17153.baike.tongsou.com 
        """创建仲裁日志"""
        log = ArbitrationLog(
            entity=conflict.entity,
            attribute=conflict.attribute,
            input_claims=[c.to_dict() for c in conflict.claims],
            scoring_details=[
                {
                    "