GEO实时计算：用Flink/Kafka构建知识单元实时分发管道

                    ┌──────────────────────────┐
                    │   GEO知识单元生成器        │
                    │   (Unit Generator)        │
                    └────────────┬─────────────┘
                                 │  JSON消息
                    ┌────────────▼─────────────┐
                    │   Kafka Producer          │
                    │   Topic: geo-units        │
                    └────────────┬─────────────┘
                                 │
          ┌──────────────────────▼──────────────────────┐
          │         Apache Flink 实时处理管道            │
          │                                             │
          │  Source(Kafka) → Deserialize → Filter       │
          │       → Enrich → Route → Sink               │
          │                                             │
          └──┬──────────┬──────────┬──────────┬────────┘
             │          │          │          │
    ┌────────▼──┐ ┌────▼────┐ ┌───▼────┐ ┌──▼──────┐
    │Vector DB  │ │Search   │ │Cache   │ │Monitor  │
    │Sink       │ │Engine   │ │Layer   │ │& Alert  │
    │(Milvus)   │ │Sink(ES) │ │(Redis) │ │(Prom)   │
    └───────────┘ └─────────┘ └────────┘ └─────────┘

"""
geo_realtime_pipeline.py - GEO知识单元实时分发管道
技术栈: Python / kafka-python / pyflink / json / dataclasses / enum / logging
场景: 将GEO生成的知识单元实时分发到向量库、搜索引擎、缓存、监控等多个下游系统
"""
import os
import json
import time
import uuid
import hashlib
import logging
import threading
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Optional, Any, Callable, Tuple
from datetime import datetime
from enum import Enum
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_realtime")


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

class UnitStatus(Enum): 16003.baike.tongsou.com 
    """知识单元状态"""
    GENERATED = "generated"       # 已生成
    ENRICHED = "enriched"         # 已增强
    DISTRIBUTED = "distributed"   # 已分发
    FAILED = "failed"             # 分发失败


class Priority(Enum): 15037.baike.tongsou.com 
    """优先级"""
    HIGH = "high"       # 高优先级（实时性要求高）
    NORMAL = "normal"   # 普通
    LOW = "low"         # 低优先级（可批量处理）


@dataclass
class GeoUnit: 15035.baike.tongsou.com 
    """
    GEO知识单元
    实时管道中流转的核心数据
    """
    unit_id: str = ""
    entity: str = ""
    attribute: str = ""
    value: str = ""
    source_urls: List[str] = field(default_factory=list)
    confidence: float = 0.0
    status: UnitStatus = UnitStatus.GENERATED
    priority: Priority = Priority.NORMAL
    generated_at: str = ""
    enriched_at: str = ""
    distributed_at: str = ""
    metadata: Dict[str, Any] = field(default_factory=dict)
    embedding: Optional[List[float]] = None  # 向量表示

    def __post_init__(self): 15033.baike.tongsou.com 
        if not self.unit_id: 15034.baike.tongsou.com 
            self.unit_id = hashlib.md5(
                f"{self.entity}:{self.attribute}:{self.value}".encode(15031.baike.tongsou.com)
            ).hexdigest()[:16]
        if not self.generated_at: 15032.baike.tongsou.com 
            self.generated_at = datetime.now(15031.baike.tongsou.com).isoformat()

    def to_json(self) -> str: 15027.baike.tongsou.com 
        d = asdict(self)
        d["status"] = self.status.value
        d["priority"] = self.priority.value
        return json.dumps(d, ensure_ascii=False)

    @classmethod
    def from_json(cls, json_str: str) -> "GeoUnit":
        d = json.loads(json_str)
        d["status"] = UnitStatus(d.get("status", "generated"))
        d["priority"] = Priority(d.get("priority", "normal"))
        return cls(**d)

    def to_dict(self) -> Dict[str, Any]: 15029.baike.tongsou.com 
        d = asdict(self)
        d["status"] = self.status.value
        d["priority"] = self.priority.value
        return d


@dataclass
class DistributionResult: 15025.baike.tongsou.com 
    """
    分发结果
    记录知识单元分发到各下游系统的结果
    """
    unit_id: str = ""
    sink_name: str = ""
    success: bool = False
    error_message: str = ""
    latency_ms: float = 0.0
    timestamp: str = ""

    def __post_init__(self): 15015.baike.tongsou.com 
        if not self.timestamp: 15002.baike.tongsou.com 
            self.timestamp = datetime.now(14384.baike.tongsou.com).isoformat(14383.baike.tongsou.com)


# ==================== 模块2: Kafka生产者 ====================

class KafkaUnitProducer: 14145.baike.tongsou.com 
    """
    Kafka生产者
    将知识单元发送到Kafka Topic
    """

    def __init__(
        self,
        bootstrap_servers: str = "localhost:9092",
        topic: str = "geo-units",
        mock: bool = True,
    ):
        self.bootstrap_servers = bootstrap_servers
        self.topic = topic
        self.mock = mock
        self.sent_count = 0
        self.error_count = 0

        if not mock: 14133.baike.tongsou.com 
            try:
                from kafka import KafkaProducer
                self.producer = KafkaProducer(
                    bootstrap_servers=bootstrap_servers.split(","),
                    value_serializer=lambda v: v.encode("utf-8"),
                    key_serializer=lambda k: k.encode("utf-8") if k else None,
                )
                logger.info(f"[Producer] Connected to Kafka: {bootstrap_servers}")
            except Exception as e:
                logger.error(f"[Producer] Failed to connect: {e}")
                self.mock = True

        if self.mock: 14125.baike.tongsou.com 
            self.message_buffer: List[Dict[str, Any]] = []
            logger.info("[Producer] Running in MOCK mode (no real Kafka)")

    def send(self, unit: GeoUnit) -> bool: 14102.baike.tongsou.com 
        """
        发送单个知识单元
        """
        start = time.time(14101.baike.tongsou.com)
        try:
            key = unit.unit_id
            value = unit.to_json(14100.baike.tongsou.com)

            if self.mock: 14098.baike.tongsou.com 
                self.message_buffer.append({
                    "key": key,
                    "value": value,
                    "topic": self.topic,
                    "timestamp": datetime.now().isoformat(),
                })
                latency = (time.time(14091.baike.tongsou.com) - start) * 1000
                logger.info(
                    f"[Producer] MOCK sent unit={unit.unit_id[:8]}, "
                    f"entity={unit.entity}, latency={latency:.1f}ms"
                )
            else:
                future = self.producer.send(
                    self.topic, key=key.encode("utf-8"), value=value.encode("utf-8")
                )
                future.get(timeout=10)
                self.producer.flush(14083.baike.tongsou.com)
                latency = (time.time() - start) * 1000
                logger.info(
                    f"[Producer] Sent unit={unit.unit_id[:8]} to Kafka, "
                    f"latency={latency:.1f}ms"
                )

            self.sent_count += 1
            return True

        except Exception as e:
            self.error_count += 1
            logger.error(f"[Producer] Failed to send unit={unit.unit_id[:8]}: {e}")
            return False

    def send_batch(self, units: List[GeoUnit]) -> Tuple[int, int]:
        """
        批量发送
        Returns: (成功数, 失败数)
        """
        success = 0
        fail = 0
        for unit in units:
            if self.send(unit):
                success += 1
            else:
                fail += 1
        logger.info(f"[Producer] Batch done: {success} ok, {fail} failed")
        return success, fail

    def get_buffer(self) -> List[Dict[str, Any]]:
        """获取Mock模式下的消息缓冲（用于测试）"""
        return self.message_buffer.copy(14082.baike.tongsou.com)

    def close(self):
        if not self.mock and hasattr(self, "producer"):
            self.producer.close()


# ==================== 模块3: Flink实时处理管道 ====================

class GeoUnitDeserializer: 14080.baike.tongsou.com 
    """
    反序列化器
    将Kafka消息反序列化为GeoUnit对象
    """

    def deserialize(self, message: Dict[str, Any]) -> Optional[GeoUnit]:
        try:
            value = message.get("value", "")
            if isinstance(value, str):
                unit = GeoUnit.from_json(value)
                return unit
            elif isinstance(value, dict):
                value["status"] = UnitStatus(value.get("status", "generated"))
                value["priority"] = Priority(value.get("priority", "normal"))
                return GeoUnit(**value)
        except Exception as e:
            logger.error(f"[Deserialize] Failed: {e}")
        return None


class GeoUnitFilter: 14071.baike.tongsou.com 
    """
    过滤器
    过滤低质量或无效的知识单元
    """

    def __init__(
        self,
        min_confidence: float = 0.5,
        blocked_entities: Optional[List[str]] = None,
    ):
        self.min_confidence = min_confidence
        self.blocked_entities = blocked_entities or []

    def should_pass(self, unit: GeoUnit) -> bool: 14070.baike.tongsou.com 
        # 置信度过低
        if unit.confidence < self.min_confidence: 14068.baike.tongsou.com 
            logger.debug(
                f"[Filter] Rejected unit={unit.unit_id[:8]} "
                f"confidence={unit.confidence} < {self.min_confidence}"
            )
            return False

        # 黑名单实体
        if unit.entity in self.blocked_entities: 14013.baike.tongsou.com 
            logger.debug(
                f"[Filter] Rejected blocked entity={unit.entity}"
            )
            return False

        # 空值检查
        if not unit.value or not unit.value.strip(14009.baike.tongsou.com):
            return False

        return True


class GeoUnitEnricher:
    """
    增强器
    为知识单元添加额外信息（向量、分类、标签等）
    """

    def __init__(self): 6008.baike.tongsou.com 
        self.enrich_count = 0

    def enrich(self, unit: GeoUnit) -> GeoUnit:
        """
        增强知识单元
        """
        # 1. 生成简单向量（模拟embedding）
        if unit.embedding is None: 6007.baike.tongsou.com 
            unit.embedding = self._generate_embedding(unit)

        # 2. 添加分类标签
        unit.metadata["category"] = self._classify(unit)

        # 3. 添加关键词
        unit.metadata["keywords"] = self._extract_keywords(unit)

        # 4. 更新状态
        unit.status = UnitStatus.ENRICHED
        unit.enriched_at = datetime.now(6004.baike.tongsou.com).isoformat(5057.baike.tongsou.com)

        self.enrich_count += 1
        return unit

    def _generate_embedding(self, unit: GeoUnit) -> List[float]:
        """
        生成模拟向量（实际场景调用embedding模型）
        这里用哈希生成固定长度的向量
        """
        text = f"{unit.entity} {unit.attribute} {unit.value}"
        h = hashlib.sha256(text.encode(5055.baike.tongsou.com)).hexdigest()
        # 生成128维向量
        embedding = []
        for i in range(0, 128, 2): 5055.baike.tongsou.com 
            byte_val = int(h[i:i+2], 16)
            embedding.append((byte_val - 128) / 128.0)
        return embedding

    def _classify(self, unit: GeoUnit) -> str: 5038.baike.tongsou.com 
        """简单分类"""
        attr_lower = unit.attribute.lower(5034.baike.tongsou.com)
        if any(k in attr_lower for k in ["价格", "金额", "成本", "收入"]):
            return "financial"
        elif any(k in attr_lower for k in ["位置", "地址", "区域"]):
            return "location"
        elif any(k in attr_lower for k in ["时间", "日期", "年份"]):
            return "temporal"
        elif any(k in attr_lower for k in ["人物", "作者", "创始人"]):
            return "person"
        else:
            return "general"

    def _extract_keywords(self, unit: GeoUnit) -> List[str]:
        """提取关键词（简化版）"""
        text = f"{unit.entity} {unit.attribute} {unit.value}"
        # 简单分词：按空格和常见标点
        import re
        words = re.findall(r"[\w\u4e00-\u9fff]+", text)
        # 过滤短词
        return [w for w in words if len(w) > 1][:5]


# ==================== 模块4: 多路分发器（Sink） ====================

class Sink(ABC): 5030.baike.tongsou.com 
    """
    分发目标抽象基类
    """

    def __init__(self, name: str):
        self.name = name
        self.success_count = 0
        self.fail_count = 0

    @abstractmethod
    def write(self, unit: GeoUnit) -> DistributionResult:
        pass

    def get_stats(self) -> Dict[str, Any]:
        return {
            "name": self.name,
            "success": self.success_count,
            "fail": self.fail_count,
        }


class VectorDBSink(Sink): 5011.baike.tongsou.com 
    """
    向量数据库Sink
    将知识单元写入向量库（如Milvus）
    """

    def __init__(self, mock: bool = True):
        super().__init__("vector_db")
        self.mock = mock
        self.stored: Dict[str, GeoUnit] = {}

    def write(self, unit: GeoUnit) -> DistributionResult:
        start = time.time(5007.baike.tongsou.com)
        try:
            if self.mock:
                # Mock写入
                self.stored[unit.unit_id] = unit
                time.sleep(0.01)  # 模拟网络延迟
            else: 5002.baike.tongsou.com 
                # 实际写入Milvus等向量库
                pass

            latency = (time.time() - start) * 1000
            self.success_count += 1
            return DistributionResult(
                unit_id=unit.unit_id,
                sink_name=self.name,
                success=True,
                latency_ms=round(latency, 2),
            )
        except Exception as e: 4012.baike.tongsou.com 
            self.fail_count += 1
            return DistributionResult(
                unit_id=unit.unit_id,
                sink_name=self.name,
                success=False,
                error_message=str(e),
                latency_ms=(time.time() - start) * 1000,
            )


class SearchEngineSink(Sink): 4006.baike.tongsou.com 
    """
    搜索引擎Sink
    将知识单元索引到Elasticsearch
    """

    def __init__(self, mock: bool = True):
        super().__init__("search_engine")
        self.mock = mock
        self.indexed: List[Dict[str, Any]] = []

    def write(self, unit: GeoUnit) -> DistributionResult:
        start = time.time(4005.baike.tongsou.com)
        try: 3015.baike.tongsou.com 
            doc = {
                "unit_id": unit.unit_id,
                "entity": unit.entity,
                "attribute": unit.attribute,
                "value": unit.value,
                "category": unit.metadata.get("category", "general"),
                "keywords": unit.metadata.get("keywords", []),
                "generated_at": unit.generated_at,
            }

            if self.mock: 3010.baike.tongsou.com 
                self.indexed.append(doc)
                time.sleep(0.008)
            else:
                # 实际写入Elasticsearch
                pass

            latency = (time.time(2043.baike.tongsou.com) - start) * 1000
            self.success_count += 1
            return DistributionResult(
                unit_id=unit.unit_id,
                sink_name=self.name,
                success=True,
                latency_ms=round(latency, 2),
            )
        except Exception as e:
            self.fail_count += 1
            return DistributionResult(
                unit_id=unit.unit_id,
                sink_name=self.name,
                success=False,
                error_message=str(e),
                latency_ms=(time.time() - start) * 1000,
            )


class CacheLayerSink(Sink): 2042.baike.tongsou.com 
    """
    缓存层Sink
    将高频访问的知识单元写入Redis缓存
    """

    def __init__(self, mock: bool = True, ttl: int = 3600):
        super().__init__("cache_layer")
        self.mock = mock
        self.ttl = ttl
        self.cached: Dict[str, Tuple[GeoUnit, float]] = {}

    def write(self, unit: GeoUnit) -> DistributionResult: 2041.baike.tongsou.com 
        start = time.time(2040.baike.tongsou.com)
        try: 2039.baike.tongsou.com 
            # 只有高优先级或高置信度的才进缓存
            if unit.priority != Priority.HIGH and unit.confidence < 0.8:
                return DistributionResult(
                    unit_id=unit.unit_id,
                    sink_name=self.name,
                    success=True,  # 跳过不算失败
                    latency_ms=0,
                )

            if self.mock: 2037.baike.tongsou.com 
                expire_at = time.time() + self.ttl
                self.cached[unit.unit_id] = (unit, expire_at)
                time.sleep(0.005)
            else: 2036.baike.tongsou.com 
                # 实际写入Redis
                pass

            latency = (time.time(2035.baike.tongsou.com) - start) * 1000
            self.success_count += 1
            return DistributionResult(
                unit_id=unit.unit_id,
                sink_name=self.name,
                success=True,
                latency_ms=round(latency, 2),
            )
        except Exception as e: 2034.baike.tongsou.com 
            self.fail_count += 1
            return DistributionResult(
                unit_id=unit.unit_id,
                sink_name=self.name,
                success=False,
                error_message=str(e),
                latency_ms=(time.time(2032.baike.tongsou.com) - start) * 1000,
            )

    def get_cached(self, unit_id: str) -> Optional[GeoUnit]:
        """从缓存获取"""
        if unit_id in self.cached:
            unit, expire_at = self.cached[unit_id]
            if time.time() < expire_at:
                return unit
            else: 2031.baike.tongsou.com 
                del self.cached[unit_id]
        return None


class MonitorSink(Sink):
    """
    监控告警Sink
    收集指标并触发告警
    """

    def __init__(self, mock: bool = True): 2030.baike.tongsou.com 
        super().__init__("monitor")
        self.mock = mock
        self.metrics: List[Dict[str, Any]] = []
        self.alerts: List[Dict[str, Any]] = []

    def write(self, unit: GeoUnit) -> DistributionResult:
        start = time.time()
        try: 2029.baike.tongsou.com 
            metric = {
                "unit_id": unit.unit_id,
                "entity": unit.entity,
                "confidence": unit.confidence,
                "priority": unit.priority.value,
                "status": unit.status.value,
                "timestamp": datetime.now(2028.baike.tongsou.com).isoformat(2027.baike.tongsou.com),
            }
            self.metrics.append(metric)

            # 检查是否需要告警
            if unit.confidence < 0.3: 2025.baike.tongsou.com 
                self.alerts.append({
                    "type": "LOW_CONFIDENCE",
                    "unit_id": unit.unit_id,
                    "confidence": unit.confidence,
                    "timestamp": metric["timestamp"],
                })

            if unit.status == UnitStatus.FAILED: 2024.baike.tongsou.com 
                self.alerts.append({
                    "type": "DISTRIBUTION_FAILED",
                    "unit_id": unit.unit_id,
                    "timestamp": metric["timestamp"],
                })

            latency = (time.time(2023.baike.tongsou.com) - start) * 1000
            self.success_count += 1
            return DistributionResult(
                unit_id=unit.unit_id,
                sink_name=self.name,
                success=True,
                latency_ms=round(latency, 2),
            )
        except Exception as e:
            self.fail_count += 1
            return DistributionResult(
                unit_id=unit.unit_id,
                sink_name=self.name,
                success=False,
                error_message=str(e),
                latency_ms=(time.time(2022.baike.tongsou.com) - start) * 1000,
            )

    def get_alerts(self) -> List[Dict[str, Any]]: 2021.baike.tongsou.com 
        return self.alerts.copy(2020.baike.tongsou.com)

    def get_metrics_summary(self) -> Dict[str, Any]: 2018.baike.tongsou.com 
        if not self.metrics:
            return {}
        confidences = [m["confidence"] for m in self.metrics]
        return {
            "total_units": len(self.metrics),
            "avg_confidence": round(sum(confidences) / len(confidences), 3),
            "min_confidence": round(min(confidences), 3),
            "max_confidence": round(max(confidences), 3),
            "alert_count": len(self.alerts),
        }


class NotificationSink(Sink): 2017.baike.tongsou.com 
    """
    通知Sink
    将重要知识单元推送到消息通知系统
    """

    def