from fastapi import FastAPI, UploadFile, File, Form
from fastapi.responses import StreamingResponse
import aiofiles
import hashlib
import os
from datetime import datetime
import asyncio
from typing import Optional
import io

app = FastAPI()

class ImageUploadHandler:
    def __init__(self, upload_dir: str = "./uploads"):
        self.upload_dir = upload_dir
        self.chunk_size = 1024 * 1024  # 1MB chunks
        os.makedirs(upload_dir, exist_ok=True)
    
    async def handle_chunk_upload(
        self, 
        file: UploadFile,
        chunk_index: int,
        total_chunks: int,
        file_id: str
    ):
        """处理分片上传"""
        temp_dir = os.path.join(self.upload_dir, "temp", file_id)
        os.makedirs(temp_dir, exist_ok=True)
        
        chunk_path = os.path.join(temp_dir, f"chunk_{chunk_index}")
        
        # 异步写入分片
        async with aiofiles.open(chunk_path, 'wb') as f:
            content = await file.read()
            await f.write(content)
        
        # 检查是否所有分片都已上传
        if len(os.listdir(temp_dir)) == total_chunks:
            return await self.merge_chunks(file_id, total_chunks)
        
        return {"status": "chunk_received", "chunk": chunk_index}
    
    async def merge_chunks(self, file_id: str, total_chunks: int):
        """合并分片"""
        temp_dir = os.path.join(self.upload_dir, "temp", file_id)
        final_path = os.path.join(self.upload_dir, f"{file_id}.jpg")
        
        async with aiofiles.open(final_path, 'wb') as final_file:
            for i in range(total_chunks):
                chunk_path = os.path.join(temp_dir, f"chunk_{i}")
                async with aiofiles.open(chunk_path, 'rb') as chunk_file:
                    content = await chunk_file.read()
                    await final_file.write(content)
                os.remove(chunk_path)
        
        os.rmdir(temp_dir)
        return {"status": "complete", "path": final_path}

# 实例化处理器
upload_handler = ImageUploadHandler()

@app.post("/api/upload/chunk")
async def upload_chunk(
    file: UploadFile = File(...),
    chunk_index: int = Form(...),
    total_chunks: int = Form(...),
    file_id: str = Form(...)
):
    """分片上传接口"""
    return await upload_handler.handle_chunk_upload(
        file, chunk_index, total_chunks, file_id
    )

@app.post("/api/upload/stream")
async def upload_stream(file: UploadFile = File(...)):
    """流式上传接口 - 适合小文件"""
    # 直接在内存中处理，不写入磁盘
    contents = await file.read()
    
    # 直接传递给检测模型
    result = await process_image_in_memory(contents)
    
    return {"status": "success", "result": result}

async def process_image_in_memory(image_bytes: bytes):
    """在内存中直接处理图像"""
    import cv2
    import numpy as np
    
    # 将字节转换为numpy数组
    nparr = np.frombuffer(image_bytes, np.uint8)
    img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
    
    # 执行检测（示例）
    # results = model.predict(img)
    
    return {"processed": True}

// utils/upload.js
export class ChunkedUploader {
  constructor(options = {}) {
    this.chunkSize = options.chunkSize || 1024 * 1024; // 1MB
    this.maxRetries = options.maxRetries || 3;
    this.concurrency = options.concurrency || 3;
  }
  
  async uploadFile(file, onProgress) {
    const chunks = this.createChunks(file);
    const fileId = this.generateFileId(file);
    const totalChunks = chunks.length;
    
    let uploadedChunks = 0;
    
    // 并发上传控制
    const uploadQueue = [];
    for (let i = 0; i < chunks.length; i++) {
      uploadQueue.push(
        this.uploadChunk(chunks[i], i, totalChunks, fileId)
          .then(() => {
            uploadedChunks++;
            if (onProgress) {
              onProgress((uploadedChunks / totalChunks) * 100);
            }
          })
      );
      
      // 控制并发数
      if (uploadQueue.length >= this.concurrency) {
        await Promise.race(uploadQueue);
      }
    }
    
    // 等待所有分片上传完成
    await Promise.all(uploadQueue);
    
    return { fileId, status: 'complete' };
  }
  
  createChunks(file) {
    const chunks = [];
    let start = 0;
    
    while (start < file.size) {
      const end = Math.min(start + this.chunkSize, file.size);
      chunks.push(file.slice(start, end));
      start = end;
    }
    
    return chunks;
  }
  
  generateFileId(file) {
    return `${Date.now()}_${file.name.replace(/\s/g, '_')}`;
  }
  
  async uploadChunk(chunk, index, total, fileId, retries = 0) {
    const formData = new FormData();
    formData.append('file', chunk);
    formData.append('chunk_index', index);
    formData.append('total_chunks', total);
    formData.append('file_id', fileId);
    
    try {
      const response = await fetch('/api/upload/chunk', {
        method: 'POST',
        body: formData
      });
      
      if (!response.ok) throw new Error('Upload failed');
      
      return await response.json();
    } catch (error) {
      if (retries < this.maxRetries) {
        // 指数退避重试
        await new Promise(resolve => 
          setTimeout(resolve, Math.pow(2, retries) * 1000)
        );
        return this.uploadChunk(chunk, index, total, fileId, retries + 1);
      }
      throw error;
    }
  }
}

// Vue组件中使用
<script setup>
import { ref } from 'vue'
import { ChunkedUploader } from './utils/upload'

const uploader = new ChunkedUploader({
  chunkSize: 2 * 1024 * 1024, // 2MB
  concurrency: 5
})

const uploadProgress = ref(0)

const handleUpload = async (file) => {
  const result = await uploader.uploadFile(file, (progress) => {
    uploadProgress.value = progress
  })
  
  console.log('上传完成', result)
}
</script>

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import Dict, Set
import json
import asyncio
import cv2
import base64
import numpy as np
from datetime import datetime

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        # 存储活动连接
        self.active_connections: Dict[str, WebSocket] = {}
        # 存储连接的订阅信息
        self.subscriptions: Dict[str, Set[str]] = {}
        
    async def connect(self, websocket: WebSocket, client_id: str):
        await websocket.accept()
        self.active_connections[client_id] = websocket
        self.subscriptions[client_id] = set()
        
    def disconnect(self, client_id: str):
        if client_id in self.active_connections:
            del self.active_connections[client_id]
            del self.subscriptions[client_id]
    
    async def send_personal_message(self, message: str, client_id: str):
        if client_id in self.active_connections:
            await self.active_connections[client_id].send_text(message)
    
    async def broadcast(self, message: str, channel: str = "general"):
        """向订阅了特定频道的客户端广播消息"""
        for client_id, subscribed_channels in self.subscriptions.items():
            if channel in subscribed_channels:
                await self.send_personal_message(message, client_id)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await manager.connect(websocket, client_id)
    
    try:
        while True:
            # 接收消息
            data = await websocket.receive_text()
            message = json.loads(data)
            
            # 处理不同类型的消息
            if message["type"] == "image":
                # 处理实时图像流
                await handle_image_stream(message["data"], client_id)
                
            elif message["type"] == "subscribe":
                # 订阅特定生产线
                line_id = message["line_id"]
                manager.subscriptions[client_id].add(line_id)
                
            elif message["type"] == "detect":
                # 实时检测请求
                result = await perform_detection(message["data"])
                await manager.send_personal_message(
                    json.dumps({
                        "type": "detection_result",
                        "data": result,
                        "timestamp": datetime.utcnow().isoformat()
                    }),
                    client_id
                )
                
    except WebSocketDisconnect:
        manager.disconnect(client_id)
        print(f"Client {client_id} disconnected")
    except Exception as e:
        print(f"Error: {e}")
        manager.disconnect(client_id)

async def handle_image_stream(image_data: str, client_id: str):
    """处理实时图像流"""
    # 解码base64图像
    image_bytes = base64.b64decode(image_data.split(",")[1])
    nparr = np.frombuffer(image_bytes, np.uint8)
    img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
    
    # 执行检测
    # results = model(img)
    
    # 返回结果
    result = {
        "type": "stream_result",
        "detections": [],  # 检测结果
        "fps": 30,
        "latency": 15  # ms
    }
    
    await manager.send_personal_message(
        json.dumps(result),
        client_id
    )

# 后台任务：定期推送系统状态
async def system_monitor():
    while True:
        await asyncio.sleep(5)  # 每5秒推送一次
        
        status = {
            "type": "system_status",
            "data": {
                "gpu_usage": get_gpu_usage(),
                "queue_length": get_queue_length(),
                "active_connections": len(manager.active_connections),
                "timestamp": datetime.utcnow().isoformat()
            }
        }
        
        # 广播给所有连接
        await manager.broadcast(
            json.dumps(status),
            channel="system"
        )

@app.on_event("startup")
async def startup_event():
    # 启动系统监控任务
    asyncio.create_task(system_monitor())

// services/websocket.js
export class DetectionWebSocket {
  constructor(url, clientId) {
    this.url = url
    this.clientId = clientId
    this.ws = null
    this.reconnectAttempts = 0
    this.maxReconnectAttempts = 5
    this.reconnectDelay = 1000
    this.listeners = new Map()
    this.heartbeatInterval = null
  }
  
  connect() {
    return new Promise((resolve, reject) => {
      this.ws = new WebSocket(`${this.url}/ws/${this.clientId}`)
      
      this.ws.onopen = () => {
        console.log('WebSocket connected')
        this.reconnectAttempts = 0
        this.startHeartbeat()
        resolve()
      }
      
      this.ws.onmessage = (event) => {
        const message = JSON.parse(event.data)
        this.handleMessage(message)
      }
      
      this.ws.onerror = (error) => {
        console.error('WebSocket error:', error)
        reject(error)
      }
      
      this.ws.onclose = () => {
        console.log('WebSocket disconnected')
        this.stopHeartbeat()
        this.attemptReconnect()
      }
    })
  }
  
  handleMessage(message) {
    const { type, data } = message
    
    // 触发对应类型的监听器
    if (this.listeners.has(type)) {
      this.listeners.get(type).forEach(callback => {
        callback(data)
      })
    }
  }
  
  on(type, callback) {
    if (!this.listeners.has(type)) {
      this.listeners.set(type, new Set())
    }
    this.listeners.get(type).add(callback)
    
    // 返回取消订阅函数
    return () => {
      this.listeners.get(type).delete(callback)
    }
  }
  
  send(type, data) {
    if (this.ws && this.ws.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify({ type, data }))
    } else {
      console.error('WebSocket is not connected')
    }
  }
  
  sendImage(imageBlob) {
    // 将图像转换为base64
    const reader = new FileReader()
    reader.onloadend = () => {
      this.send('image', reader.result)
    }
    reader.readAsDataURL(imageBlob)
  }
  
  subscribe(lineId) {
    this.send('subscribe', { line_id: lineId })
  }
  
  startHeartbeat() {
    this.heartbeatInterval = setInterval(() => {
      this.send('ping', { timestamp: Date.now() })
    }, 30000) // 30秒心跳
  }
  
  stopHeartbeat() {
    if (this.heartbeatInterval) {
      clearInterval(this.heartbeatInterval)
    }
  }
  
  attemptReconnect() {
    if (this.reconnectAttempts < this.maxReconnectAttempts) {
      this.reconnectAttempts++
      console.log(`Attempting to reconnect... (${this.reconnectAttempts})`)
      
      setTimeout(() => {
        this.connect()
      }, this.reconnectDelay * this.reconnectAttempts)
    }
  }
  
  disconnect() {
    this.stopHeartbeat()
    if (this.ws) {
      this.ws.close()
    }
  }
}

// Vue组件中使用
<script setup>
import { onMounted, onUnmounted, ref } from 'vue'
import { DetectionWebSocket } from './services/websocket'

const ws = ref(null)
const detectionResults = ref([])
const systemStatus = ref({})

onMounted(async () => {
  // 创建WebSocket连接
  ws.value = new DetectionWebSocket(
    'ws://localhost:8000',
    `client_${Date.now()}`
  )
  
  await ws.value.connect()
  
  // 订阅检测结果
  ws.value.on('detection_result', (data) => {
    detectionResults.value.push(data)
  })
  
  // 订阅系统状态
  ws.value.on('system_status', (data) => {
    systemStatus.value = data
  })
  
  // 订阅特定生产线
  ws.value.subscribe('line_001')
})

onUnmounted(() => {
  if (ws.value) {
    ws.value.disconnect()
  }
})

// 发送实时检测请求
const sendDetectionRequest = (imageData) => {
  ws.value.send('detect', { data: imageData })
}
</script>

import pika
import json
import uuid
from typing import Dict, Any
import asyncio
from datetime import datetime

class RabbitMQProducer:
    def __init__(self, connection_params: Dict[str, Any]):
        self.connection_params = connection_params
        self.connection = None
        self.channel = None
        self.connect()
    
    def connect(self):
        """建立连接"""
        credentials = pika.PlainCredentials(
            self.connection_params['username'],
            self.connection_params['password']
        )
        
        parameters = pika.ConnectionParameters(
            host=self.connection_params['host'],
            port=self.connection_params['port'],
            virtual_host=self.connection_params['vhost'],
            credentials=credentials,
            heartbeat=600,
            blocked_connection_timeout=300,
            connection_attempts=3,
            retry_delay=2
        )
        
        self.connection = pika.BlockingConnection(parameters)
        self.channel = self.connection.channel()
        
        # 声明交换机
        self.channel.exchange_declare(
            exchange='detection_exchange',
            exchange_type='topic',
            durable=True
        )
        
        # 声明队列
        self.channel.queue_declare(
            queue='detection_queue',
            durable=True,
            arguments={
                'x-message-ttl': 3600000,  # 消息TTL: 1小时
                'x-max-length': 10000,      # 最大消息数
                'x-overflow': 'drop-head'   # 溢出策略
            }
        )
        
        # 绑定队列到交换机
        self.channel.queue_bind(
            exchange='detection_exchange',
            queue='detection_queue',
            routing_key='detection.*'
        )
    
    def publish_detection_task(self, image_path: str, metadata: Dict[str, Any]):
        """发布检测任务"""
        task_id = str(uuid.uuid4())
        
        message = {
            'task_id': task_id,
            'image_path': image_path,
            'metadata': metadata,
            'timestamp': datetime.utcnow().isoformat(),
            'priority': metadata.get('priority', 5)
        }
        
        # 发布消息
        self.channel.basic_publish(
            exchange='detection_exchange',
            routing_key=f"detection.{metadata.get('line_id', 'default')}",
            body=json.dumps(message),
            properties=pika.BasicProperties(
                delivery_mode=2,  # 持久化消息
                priority=message['priority'],
                correlation_id=task_id,
                content_type='application/json'
            )
        )
        
        return task_id
    
    def close(self):
        if self.connection and not self.connection.is_closed:
            self.connection.close()

# FastAPI集成
from fastapi import FastAPI, UploadFile, BackgroundTasks

app = FastAPI()

producer = RabbitMQProducer({
    'host': 'localhost',
    'port': 5672,
    'vhost': 'detection_vhost',
    'username': 'admin',
    'password': 'secure_password'
})

@app.post("/api/detect/async")
async def async_detect(
    background_tasks: BackgroundTasks,
    file: UploadFile,
    line_id: str = "default",
    priority: int = 5
):
    """异步检测接口"""
    # 保存文件
    file_path = f"/tmp/{uuid.uuid4()}.jpg"
    with open(file_path, 'wb') as f:
        content = await file.read()
        f.write(content)
    
    # 发布到消息队列
    task_id = producer.publish_detection_task(
        file_path,
        {
            'line_id': line_id,
            'priority': priority,
            'original_name': file.filename
        }
    )
    
    return {
        'task_id': task_id,
        'status': 'queued',
        'message': '任务已加入队列'
    }

import pika
import json
import cv2
from ultralytics import YOLO
import traceback
import time
import signal
import sys
from threading import Thread
from concurrent.futures import ThreadPoolExecutor

class DetectionConsumer:
    def __init__(self, connection_params, model_path, num_workers=4):
        self.connection_params = connection_params
        self.model = YOLO(model_path)
        self.num_workers = num_workers
        self.executor = ThreadPoolExecutor(max_workers=num_workers)
        self.running = True
        
    def connect(self):
        """建立连接"""
        credentials = pika.PlainCredentials(
            self.connection_params['username'],
            self.connection_params['password']
        )
        
        parameters = pika.ConnectionParameters(
            host=self.connection_params['host'],
            port=self.connection_params['port'],
            virtual_host=self.connection_params['vhost'],
            credentials=credentials
        )
        
        connection = pika.BlockingConnection(parameters)
        channel = connection.channel()
        
        # 设置预取数量（负载均衡）
        channel.basic_qos(prefetch_count=self.num_workers)
        
        return connection, channel
    
    def process_detection(self, message_body):
        """处理检测任务"""
        try:
            data = json.loads(message_body)
            task_id = data['task_id']
            image_path = data['image_path']
            
            print(f"Processing task: {task_id}")
            
            # 读取图像
            img = cv2.imread(image_path)
            if img is None:
                raise ValueError(f"Cannot read image: {image_path}")
            
            # 执行检测
            results = self.model(img)
            
            # 处理结果
            detections = []
            for r in results:
                if r.boxes is not None:
                    for box in r.boxes:
                        detections.append({
                            'bbox': box.xyxy[0].tolist(),
                            'confidence': box.conf[0].item(),
                            'class': self.model.names[int(box.cls[0])]
                        })
            
            # 保存结果到数据库或发送到MES
            self.save_results(task_id, detections)
            
            # 清理临时文件
            os.remove(image_path)
            
            return True
            
        except Exception as e:
            print(f"Error processing task: {e}")
            traceback.print_exc()
            return False
    
    def callback(self, ch, method, properties, body):
        """消息回调函数"""
        try:
            # 在线程池中处理任务
            future = self.executor.submit(self.process_detection, body)
            result = future.result(timeout=30)  # 30秒超时
            
            if result:
                # 确认消息
                ch.basic_ack(delivery_tag=method.delivery_tag)
                print(f"Task completed: {properties.correlation_id}")
            else:
                # 重新入队
                ch.basic_nack(
                    delivery_tag=method.delivery_tag,
                    requeue=True
                )
                
        except Exception as e:
            print(f"Callback error: {e}")
            # 拒绝消息，不重新入队
            ch.basic_nack(
                delivery_tag=method.delivery_tag,
                requeue=False
            )
    
    def start_consuming(self):
        """开始消费消息"""
        while self.running:
            try:
                connection, channel = self.connect()
                
                # 设置消费者
                channel.basic_consume(
                    queue='detection_queue',
                    on_message_callback=self.callback,
                    auto_ack=False
                )
                
                print("Starting consumer...")
                channel.start_consuming()
                
            except KeyboardInterrupt:
                print("Stopping consumer...")
                self.running = False
                channel.stop_consuming()
                connection.close()
                self.executor.shutdown(wait=True)
                break
                
            except Exception as e:
                print(f"Consumer error: {e}")
                time.sleep(5)  # 等待后重连
    
    def save_results(self, task_id, detections):
        """保存检测结果"""
        # 这里实现结果保存逻辑
        # 可以保存到数据库、Redis或发送到MES系统
        pass

# 启动消费者
if __name__ == "__main__":
    consumer = DetectionConsumer(
        connection_params={
            'host': 'localhost',
            'port': 5672,
            'vhost': 'detection_vhost',
            'username': 'admin',
            'password': 'secure_password'
        },
        model_path='yolov8x.pt',
        num_workers=4
    )
    
    # 优雅关闭
    def signal_handler(sig, frame):
        print("Received shutdown signal")
        consumer.running = False
        sys.exit(0)
    
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)
    
    consumer.start_consuming()

# Kafka生产者
from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import KafkaError
import json
import asyncio
from typing import Dict, Any

class KafkaDetectionProducer:
    def __init__(self, bootstrap_servers=['localhost:9092']):
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            compression_type='gzip',  # 压缩
            acks='all',  # 等待所有副本确认
            retries=3,
            max_in_flight_requests_per_connection=5
        )
    
    async def send_detection_task(self, task_data: Dict[str, Any]):
        """发送检测任务到Kafka"""
        topic = f"detection-line-{task_data.get('line_id', 'default')}"
        
        future = self.producer.send(
            topic,
            value=task_data,
            partition=None,  # 自动分区
            timestamp_ms=None
        )
        
        try:
            record_metadata = future.get(timeout=10)
            return {
                'success': True,
                'topic': record_metadata.topic,
                'partition': record_metadata.partition,
                'offset': record_metadata.offset
            }
        except KafkaError as e:
            return {
                'success': False,
                'error': str(e)
            }

# Kafka消费者
class KafkaDetectionConsumer:
    def __init__(self, bootstrap_servers=['localhost:9092'], group_id='detection-group'):
        self.consumer = KafkaConsumer(
            'detection-line-*',  # 订阅所有检测主题
            bootstrap_servers=bootstrap_servers,
            group_id=group_id,
            value_deserializer=lambda m: json.loads(m.decode('utf-8')),
            auto_offset_reset='earliest',
            enable_auto_commit=False,  # 手动提交offset
            max_poll_records=10  # 每次最多拉取10条
        )
        
    async def consume_and_process(self):
        """消费并处理消息"""
        for message in self.consumer:
            try:
                # 处理消息
                result = await self.process_message(message.value)
                
                if result:
                    # 手动提交offset
                    self.consumer.commit()
                    
            except Exception as e:
                print(f"Error processing message: {e}")
                # 可以选择跳过或重试

# Dockerfile - 多阶段构建优化镜像大小
# 阶段1：构建环境
FROM python:3.10-slim as builder

WORKDIR /build

# 安装构建依赖
RUN apt-get update && apt-get install -y \
    gcc \
    g++ \
    cmake \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

# 复制requirements并安装Python包
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt

# 阶段2：运行环境
FROM nvidia/cuda:11.8.0-cudnn8-runtime-ubuntu22.04

# 设置时区
ENV TZ=Asia/Shanghai
RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone

# 安装运行时依赖
RUN apt-get update && apt-get install -y \
    python3.10 \
    python3-pip \
    libglib2.0-0 \
    libsm6 \
    libxext6 \
    libxrender-dev \
    libgomp1 \
    libgl1-mesa-glx \
    ffmpeg \
    && rm -rf /var/lib/apt/lists/*

# 创建非root用户
RUN useradd -m -u 1000 detector && \
    mkdir -p /app /data /logs && \
    chown -R detector:detector /app /data /logs

# 复制Python包从builder阶段
COPY --from=builder /root/.local /home/detector/.local

# 设置工作目录
WORKDIR /app

# 复制应用代码
COPY --chown=detector:detector . .

# 切换到非root用户
USER detector

# 设置Python路径
ENV PATH=/home/detector/.local/bin:$PATH
ENV PYTHONPATH=/app:$PYTHONPATH

# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD python3 -c "import requests; requests.get('http://localhost:8000/health')"

# 启动命令
CMD ["python3", "-m", "uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml - 完整的服务编排
version: '3.8'

services:
  # Nginx反向代理
  nginx:
    image: nginx:alpine
    container_name: detection-nginx
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
      - ./nginx/ssl:/etc/nginx/ssl:ro
      - static-content:/usr/share/nginx/html:ro
    depends_on:
      - backend
    networks:
      - detection-network
    restart: unless-stopped
    logging:
      driver: "json-file"
      options:
        max-size: "10m"
        max-file: "3"

  # 后端检测服务（多实例）
  backend:
    build:
      context: ./backend
      dockerfile: Dockerfile
    image: detection-backend:latest
    deploy:
      replicas: 3  # 运行3个实例
      resources:
        limits:
          cpus: '2.0'
          memory: 4G
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    environment:
      - DATABASE_URL=postgresql://user:pass@postgres:5432/detection_db
      - REDIS_URL=redis://redis:6379/0
      - RABBITMQ_URL=amqp://admin:pass@rabbitmq:5672/
      - MODEL_PATH=/models/yolov8x.pt
      - LOG_LEVEL=INFO
    volumes:
      - ./models:/models:ro
      - uploaded-images:/data/images
      - ./logs:/logs
    networks:
      - detection-network
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  # PostgreSQL数据库
  postgres:
    image: postgres:15-alpine
    container_name: detection-postgres
    environment:
      - POSTGRES_USER=detection_user
      - POSTGRES_PASSWORD=secure_password
      - POSTGRES_DB=detection_db
      - PGDATA=/var/lib/postgresql/data/pgdata
    volumes:
      - postgres-data:/var/lib/postgresql/data
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql:ro
    ports:
      - "5432:5432"
    networks:
      - detection-network
    restart: unless-stopped
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U detection_user"]
      interval: 10s
      timeout: 5s
      retries: 5

  # Redis缓存
  redis:
    image: redis:7-alpine
    container_name: detection-redis
    command: redis-server --appendonly yes --requirepass redis_password
    volumes:
      - redis-data:/data
    ports:
      - "6379:6379"
    networks:
      - detection-network
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 3

  # RabbitMQ消息队列
  rabbitmq:
    image: rabbitmq:3.12-management-alpine
    container_name: detection-rabbitmq
    environment:
      - RABBITMQ_DEFAULT_USER=admin
      - RABBITMQ_DEFAULT_PASS=admin_password
      - RABBITMQ_DEFAULT_VHOST=detection
    volumes:
      - rabbitmq-data:/var/lib/rabbitmq
    ports:
      - "5672:5672"
      - "15672:15672"
    networks:
      - detection-network
    restart: unless-stopped

  # 前端服务
  frontend:
    build:
      context: ./frontend
      dockerfile: Dockerfile
    image: detection-frontend:latest
    container_name: detection-frontend
    volumes:
      - ./frontend/dist:/usr/share/nginx/html:ro
    networks:
      - detection-network
    restart: unless-stopped

  # Celery Worker（异步任务处理）
  celery-worker:
    build:
      context: ./backend
      dockerfile: Dockerfile
    image: detection-backend:latest
    command: celery -A tasks worker --loglevel=info --concurrency=4
    deploy:
      replicas: 2
    environment:
      - DATABASE_URL=postgresql://user:pass@postgres:5432/detection_db
      - REDIS_URL=redis://redis:6379/0
      - CELERY_BROKER_URL=redis://redis:6379/1
    volumes:
      - ./models:/models:ro
      - uploaded-images:/data/images
    networks:
      - detection-network
    depends_on:
      - redis
      - postgres
    restart: unless-stopped

  # Celery Beat（定时任务）
  celery-beat:
    build:
      context: ./backend
      dockerfile: Dockerfile
    image: detection-backend:latest
    command: celery -A tasks beat --loglevel=info
    environment:
      - REDIS_URL=redis://redis:6379/0
      - CELERY_BROKER_URL=redis://redis:6379/1
    networks:
      - detection-network
    depends_on:
      - redis
    restart: unless-stopped

volumes:
  postgres-data:
  redis-data:
  rabbitmq-data:
  uploaded-images:
  static-content:

networks:
  detection-network:
    driver: bridge
    ipam:
      config:
        - subnet: 172.20.0.0/16

# detection-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: detection-backend
  namespace: detection-system
spec:
  replicas: 3
  selector:
    matchLabels:
      app: detection-backend
  template:
    metadata:
      labels:
        app: detection-backend
    spec:
      containers:
      - name: backend
        image: detection-backend:latest
        ports:
        - containerPort: 8000
        resources:
          requests:
            memory: "2Gi"
            cpu: "1"
            nvidia.com/gpu: 1
          limits:
            memory: "4Gi"
            cpu: "2"
            nvidia.com/gpu: 1
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: db-credentials
              key: url
        - name: REDIS_URL
          valueFrom:
            configMapKeyRef:
              name: app-config
              key: redis-url
        volumeMounts:
        - name: model-storage
          mountPath: /models
        - name: image-storage
          mountPath: /data/images
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5
      volumes:
      - name: model-storage
        persistentVolumeClaim:
          claimName: model-pvc
      - name: image-storage
        persistentVolumeClaim:
          claimName: image-pvc

---
apiVersion: v1
kind: Service
metadata:
  name: detection-backend-service
  namespace: detection-system
spec:
  selector:
    app: detection-backend
  ports:
  - port: 80
    targetPort: 8000
  type: LoadBalancer

---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: detection-backend-hpa
  namespace: detection-system
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: detection-backend
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80