import asyncio
from kafka import KafkaProducer
import json
from typing import Dict, Any

class DataCollector:
    def __init__(self, kafka_servers: list):
        self.producer = KafkaProducer(
            bootstrap_servers=kafka_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            compression_type='gzip',
            batch_size=16384,
            linger_ms=10
        )
    
    async def collect_from_api(self, api_url: str, topic: str):
        """從API接口采集數(shù)據(jù)"""
        async with aiohttp.ClientSession() as session:
            while True:
                try:
                    async with session.get(api_url) as response:
                        data = await response.json()
                        self.send_to_kafka(topic, data)
                    await asyncio.sleep(1)
                except Exception as e:
                    print(f"采集錯(cuò)誤: {e}")
                    await asyncio.sleep(5)
    
    def send_to_kafka(self, topic: str, data: Dict[Any, Any]):
        """發(fā)送數(shù)據(jù)到Kafka"""
        try:
            self.producer.send(topic, value=data)
            self.producer.flush()
        except Exception as e:
            print(f"發(fā)送失敗: {e}")

from kafka import KafkaConsumer, KafkaProducer
from datetime import datetime
import json

class StreamProcessor:
    def __init__(self, input_topic: str, output_topic: str):
        self.consumer = KafkaConsumer(
            input_topic,
            bootstrap_servers=['localhost:9092'],
            value_deserializer=lambda m: json.loads(m.decode('utf-8')),
            auto_offset_reset='latest',
            enable_auto_commit=True
        )
        
        self.producer = KafkaProducer(
            bootstrap_servers=['localhost:9092'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        
        self.output_topic = output_topic
    
    def process_data(self, data: dict) -> dict:
        """數(shù)據(jù)處理邏輯"""
        # 數(shù)據(jù)清洗
        cleaned_data = self.clean_data(data)
        
        # 數(shù)據(jù)轉(zhuǎn)換
        transformed_data = self.transform_data(cleaned_data)
        
        # 數(shù)據(jù)聚合
        aggregated_data = self.aggregate_data(transformed_data)
        
        # 添加處理時(shí)間戳
        aggregated_data['processed_at'] = datetime.now().isoformat()
        
        return aggregated_data
    
    def clean_data(self, data: dict) -> dict:
        """數(shù)據(jù)清洗：去除空值、異常值"""
        return {k: v for k, v in data.items() if v is not None}
    
    def transform_data(self, data: dict) -> dict:
        """數(shù)據(jù)轉(zhuǎn)換：格式標(biāo)準(zhǔn)化"""
        # 示例：溫度單位轉(zhuǎn)換
        if 'temperature' in data:
            data['temperature_celsius'] = (data['temperature'] - 32) * 5/9
        return data
    
    def aggregate_data(self, data: dict) -> dict:
        """數(shù)據(jù)聚合：計(jì)算統(tǒng)計(jì)指標(biāo)"""
        # 這里可以添加窗口聚合邏輯
        return data
    
    def run(self):
        """啟動(dòng)流處理"""
        print("流處理引擎啟動(dòng)...")
        for message in self.consumer:
            try:
                processed_data = self.process_data(message.value)
                self.producer.send(self.output_topic, processed_data)
            except Exception as e:
                print(f"處理錯(cuò)誤: {e}")

from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
from datetime import datetime

class TimeSeriesStorage:
    def __init__(self, url: str, token: str, org: str, bucket: str):
        self.client = InfluxDBClient(url=url, token=token, org=org)
        self.write_api = self.client.write_api(write_options=SYNCHRONOUS)
        self.query_api = self.client.query_api()
        self.bucket = bucket
        self.org = org
    
    def write_data(self, measurement: str, tags: dict, fields: dict):
        """寫入時(shí)序數(shù)據(jù)"""
        point = Point(measurement)
        
        # 添加標(biāo)簽
        for tag_key, tag_value in tags.items():
            point.tag(tag_key, tag_value)
        
        # 添加字段
        for field_key, field_value in fields.items():
            point.field(field_key, field_value)
        
        point.time(datetime.utcnow())
        
        self.write_api.write(bucket=self.bucket, record=point)
    
    def query_data(self, measurement: str, time_range: str = '-1h'):
        """查詢時(shí)序數(shù)據(jù)"""
        query = f'''
            from(bucket: "{self.bucket}")
                |> range(start: {time_range})
                |> filter(fn: (r) => r._measurement == "{measurement}")
        '''
        
        tables = self.query_api.query(query, org=self.org)
        results = []
        
        for table in tables:
            for record in table.records:
                results.append({
                    'time': record.get_time(),
                    'measurement': record.get_measurement(),
                    'field': record.get_field(),
                    'value': record.get_value(),
                    'tags': record.values
                })
        
        return results
    
    def close(self):
        """關(guān)閉連接"""
        self.client.close()

from fastapi import FastAPI, WebSocket, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List, Optional
import asyncio
import json

app = FastAPI(title="實(shí)時(shí)數(shù)據(jù)處理平臺(tái)API")

# 配置CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 數(shù)據(jù)模型
class DataPoint(BaseModel):
    timestamp: str
    metric: str
    value: float
    tags: Optional[dict] = {}

class QueryRequest(BaseModel):
    measurement: str
    time_range: str = '-1h'
    filters: Optional[dict] = {}

# API端點(diǎn)
@app.get("/api/metrics/latest")
async def get_latest_metrics():
    """獲取最新指標(biāo)數(shù)據(jù)"""
    # 從Redis緩存獲取最新數(shù)據(jù)
    # 這里簡(jiǎn)化處理
    return {
        "cpu_usage": 75.5,
        "memory_usage": 68.2,
        "disk_io": 1024,
        "network_traffic": 2048
    }

@app.post("/api/query")
async def query_timeseries(request: QueryRequest):
    """查詢時(shí)序數(shù)據(jù)"""
    storage = TimeSeriesStorage(
        url="http://localhost:8086",
        token="your-token",
        org="your-org",
        bucket="your-bucket"
    )
    
    try:
        results = storage.query_data(
            measurement=request.measurement,
            time_range=request.time_range
        )
        return {"data": results}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
    finally:
        storage.close()

@app.websocket("/ws/realtime")
async def websocket_endpoint(websocket: WebSocket):
    """WebSocket實(shí)時(shí)數(shù)據(jù)推送"""
    await websocket.accept()
    
    try:
        while True:
            # 從Redis或消息隊(duì)列獲取實(shí)時(shí)數(shù)據(jù)
            data = {
                "timestamp": datetime.now().isoformat(),
                "metrics": {
                    "cpu": 75.5,
                    "memory": 68.2,
                    "requests_per_second": 1500
                }
            }
            
            await websocket.send_json(data)
            await asyncio.sleep(1)
    except Exception as e:
        print(f"WebSocket錯(cuò)誤: {e}")
    finally:
        await websocket.close()

@app.get("/api/statistics/summary")
async def get_statistics():
    """獲取統(tǒng)計(jì)摘要"""
    return {
        "total_events": 1500000,
        "events_per_second": 1500,
        "active_sources": 25,
        "processing_latency_ms": 45
    }

// RealtimeChart.vue
<template>
  <div class="realtime-dashboard">
    <div class="header">
      <h1>實(shí)時(shí)數(shù)據(jù)監(jiān)控平臺(tái)</h1>
      <div class="stats">
        <div class="stat-item">
          <span class="label">實(shí)時(shí)事件數(shù)</span>
          <span class="value">{{ stats.eventsPerSecond }}/s</span>
        </div>
        <div class="stat-item">
          <span class="label">活躍數(shù)據(jù)源</span>
          <span class="value">{{ stats.activeSources }}</span>
        </div>
        <div class="stat-item">
          <span class="label">處理延遲</span>
          <span class="value">{{ stats.latency }}ms</span>
        </div>
      </div>
    </div>
    
    <div class="charts-container">
      <div class="chart-box">
        <div ref="cpuChart" class="chart"></div>
      </div>
      <div class="chart-box">
        <div ref="memoryChart" class="chart"></div>
      </div>
      <div class="chart-box">
        <div ref="trafficChart" class="chart"></div>
      </div>
    </div>
  </div>
</template>

<script setup>
import { ref, onMounted, onUnmounted } from 'vue'
import * as echarts from 'echarts'

const cpuChart = ref(null)
const memoryChart = ref(null)
const trafficChart = ref(null)

const stats = ref({
  eventsPerSecond: 0,
  activeSources: 0,
  latency: 0
})

let ws = null
let charts = {}

// 初始化圖表
const initCharts = () => {
  // CPU使用率圖表
  charts.cpu = echarts.init(cpuChart.value)
  charts.cpu.setOption({
    title: { text: 'CPU使用率', left: 'center' },
    tooltip: { trigger: 'axis' },
    xAxis: { type: 'time', splitLine: { show: false } },
    yAxis: { type: 'value', max: 100, axisLabel: { formatter: '{value}%' } },
    series: [{
      name: 'CPU',
      type: 'line',
      smooth: true,
      data: [],
      areaStyle: { opacity: 0.3 }
    }]
  })
  
  // 內(nèi)存使用率圖表
  charts.memory = echarts.init(memoryChart.value)
  charts.memory.setOption({
    title: { text: '內(nèi)存使用率', left: 'center' },
    tooltip: { trigger: 'axis' },
    xAxis: { type: 'time', splitLine: { show: false } },
    yAxis: { type: 'value', max: 100, axisLabel: { formatter: '{value}%' } },
    series: [{
      name: 'Memory',
      type: 'line',
      smooth: true,
      data: [],
      areaStyle: { opacity: 0.3 }
    }]
  })
  
  // 網(wǎng)絡(luò)流量圖表
  charts.traffic = echarts.init(trafficChart.value)
  charts.traffic.setOption({
    title: { text: '網(wǎng)絡(luò)流量', left: 'center' },
    tooltip: { trigger: 'axis' },
    xAxis: { type: 'time', splitLine: { show: false } },
    yAxis: { type: 'value', axisLabel: { formatter: '{value} MB/s' } },
    series: [{
      name: 'Traffic',
      type: 'line',
      smooth: true,
      data: []
    }]
  })
}

// 連接WebSocket
const connectWebSocket = () => {
  ws = new WebSocket('ws://localhost:8000/ws/realtime')
  
  ws.onmessage = (event) => {
    const data = JSON.parse(event.data)
    updateCharts(data)
    updateStats(data)
  }
  
  ws.onerror = (error) => {
    console.error('WebSocket錯(cuò)誤:', error)
    setTimeout(connectWebSocket, 5000)
  }
  
  ws.onclose = () => {
    console.log('WebSocket連接關(guān)閉')
    setTimeout(connectWebSocket, 5000)
  }
}

// 更新圖表數(shù)據(jù)
const updateCharts = (data) => {
  const timestamp = new Date(data.timestamp)
  const maxDataPoints = 50
  
  // 更新CPU圖表
  const cpuOption = charts.cpu.getOption()
  cpuOption.series[0].data.push([timestamp, data.metrics.cpu])
  if (cpuOption.series[0].data.length > maxDataPoints) {
    cpuOption.series[0].data.shift()
  }
  charts.cpu.setOption(cpuOption)
  
  // 更新內(nèi)存圖表
  const memoryOption = charts.memory.getOption()
  memoryOption.series[0].data.push([timestamp, data.metrics.memory])
  if (memoryOption.series[0].data.length > maxDataPoints) {
    memoryOption.series[0].data.shift()
  }
  charts.memory.setOption(memoryOption)
  
  // 更新流量圖表
  const trafficOption = charts.traffic.getOption()
  trafficOption.series[0].data.push([timestamp, data.metrics.requests_per_second / 1000])
  if (trafficOption.series[0].data.length > maxDataPoints) {
    trafficOption.series[0].data.shift()
  }
  charts.traffic.setOption(trafficOption)
}

// 更新統(tǒng)計(jì)數(shù)據(jù)
const updateStats = (data) => {
  stats.value.eventsPerSecond = data.metrics.requests_per_second
  // 從API獲取其他統(tǒng)計(jì)數(shù)據(jù)
  fetch('/api/statistics/summary')
    .then(res => res.json())
    .then(summary => {
      stats.value.activeSources = summary.active_sources
      stats.value.latency = summary.processing_latency_ms
    })
}

onMounted(() => {
  initCharts()
  connectWebSocket()
})

onUnmounted(() => {
  if (ws) ws.close()
  Object.values(charts).forEach(chart => chart.dispose())
})
</script>

<style scoped>
.realtime-dashboard {
  padding: 20px;
  background: #0a0e27;
  color: #fff;
  min-height: 100vh;
}

.header {
  margin-bottom: 30px;
}

.header h1 {
  text-align: center;
  font-size: 32px;
  margin-bottom: 20px;
}

.stats {
  display: flex;
  justify-content: center;
  gap: 40px;
}

.stat-item {
  display: flex;
  flex-direction: column;
  align-items: center;
}

.stat-item .label {
  font-size: 14px;
  color: #8b9dc3;
  margin-bottom: 5px;
}

.stat-item .value {
  font-size: 24px;
  font-weight: bold;
  color: #00d4ff;
}

.charts-container {
  display: grid;
  grid-template-columns: repeat(auto-fit, minmax(400px, 1fr));
  gap: 20px;
}

.chart-box {
  background: #151932;
  border-radius: 8px;
  padding: 20px;
  box-shadow: 0 4px 6px rgba(0, 0, 0, 0.3);
}

.chart {
  width: 100%;
  height: 300px;
}
</style>

from dataclasses import dataclass
from enum import Enum
import smtplib
from email.mime.text import MIMEText

class AlertLevel(Enum):
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"
    CRITICAL = "critical"

@dataclass
class Alert:
    level: AlertLevel
    message: str
    metric: str
    value: float
    threshold: float

class AlertManager:
    def __init__(self):
        self.thresholds = {
            'cpu_usage': 80.0,
            'memory_usage': 85.0,
            'processing_latency': 1000.0,  # ms
            'error_rate': 0.05  # 5%
        }
    
    def check_metrics(self, metrics: dict):
        """檢查指標(biāo)并觸發(fā)告警"""
        alerts = []
        
        for metric, value in metrics.items():
            if metric in self.thresholds:
                threshold = self.thresholds[metric]
                if value > threshold:
                    level = self._determine_alert_level(value, threshold)
                    alert = Alert(
                        level=level,
                        message=f"{metric}超過(guò)閾值",
                        metric=metric,
                        value=value,
                        threshold=threshold
                    )
                    alerts.append(alert)
                    self.send_alert(alert)
        
        return alerts
    
    def _determine_alert_level(self, value: float, threshold: float) -> AlertLevel:
        """確定告警級(jí)別"""
        ratio = value / threshold
        if ratio > 1.5:
            return AlertLevel.CRITICAL
        elif ratio > 1.2:
            return AlertLevel.ERROR
        else:
            return AlertLevel.WARNING
    
    def send_alert(self, alert: Alert):
        """發(fā)送告警通知"""
        print(f"[{alert.level.value.upper()}] {alert.message}: "
              f"{alert.metric}={alert.value} (閾值: {alert.threshold})")
        
        # 這里可以集成郵件、短信、釘釘?shù)韧ㄖ?
        if alert.level in [AlertLevel.ERROR, AlertLevel.CRITICAL]:
            self.send_email_alert(alert)
    
    def send_email_alert(self, alert: Alert):
        """發(fā)送郵件告警"""
        # 郵件發(fā)送邏輯
        pass

import logging
import json
from datetime import datetime

class StructuredLogger:
    def __init__(self, name: str):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(logging.INFO)
        
        # 配置處理器
        handler = logging.StreamHandler()
        handler.setFormatter(self.JsonFormatter())
        self.logger.addHandler(handler)
    
    class JsonFormatter(logging.Formatter):
        def format(self, record):
            log_data = {
                'timestamp': datetime.utcnow().isoformat(),
                'level': record.levelname,
                'logger': record.name,
                'message': record.getMessage(),
                'module': record.module,
                'function': record.funcName,
                'line': record.lineno
            }
            
            if hasattr(record, 'extra_data'):
                log_data.update(record.extra_data)
            
            return json.dumps(log_data)
    
    def info(self, message: str, **kwargs):
        self.logger.info(message, extra={'extra_data': kwargs})
    
    def error(self, message: str, **kwargs):
        self.logger.error(message, extra={'extra_data': kwargs})

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
  
  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
  
  redis:
    image: redis:alpine
    ports:
      - "6379:6379"
  
  influxdb:
    image: influxdb:2.7
    ports:
      - "8086:8086"
    environment:
      DOCKER_INFLUXDB_INIT_MODE: setup
      DOCKER_INFLUXDB_INIT_USERNAME: admin
      DOCKER_INFLUXDB_INIT_PASSWORD: adminpassword
      DOCKER_INFLUXDB_INIT_ORG: myorg
      DOCKER_INFLUXDB_INIT_BUCKET: mybucket
  
  api:
    build: ./backend
    ports:
      - "8000:8000"
    depends_on:
      - kafka
      - redis
      - influxdb
    environment:
      KAFKA_BOOTSTRAP_SERVERS: kafka:9092
      REDIS_HOST: redis
      INFLUXDB_URL: http://influxdb:8086
  
  frontend:
    build: ./frontend
    ports:
      - "3000:80"
    depends_on:
      - api

# k8s-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: data-platform-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: data-platform-api
  template:
    metadata:
      labels:
        app: data-platform-api
    spec:
      containers:
      - name: api
        image: data-platform-api:latest
        ports:
        - containerPort: 8000
        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "1Gi"
            cpu: "1000m"
        env:
        - name: KAFKA_BOOTSTRAP_SERVERS
          value: "kafka-service:9092"
        - name: REDIS_HOST
          value: "redis-service"
---
apiVersion: v1
kind: Service
metadata:
  name: data-platform-api-service
spec:
  selector:
    app: data-platform-api
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8000
  type: LoadBalancer