#Web视觉应用：FastAPI、图像处理与AI服务部署详解

"""
Web 视觉应用的重要性：

1. 用户体验：
   - 便捷的图形界面
   - 无需安装复杂环境
   - 跨平台访问能力

2. 服务化部署：
   - 统一的服务接口
   - 便于维护和升级
   - 支持多用户访问

3. 商业价值：
   - 降低使用门槛
   - 提高服务可访问性
   - 便于商业模式创新
"""

def web_vision_importance():
    """
    Web视觉应用的重要性
    """
    importance_factors = {
        "可访问性": "用户无需安装复杂环境即可使用AI功能",
        "可扩展性": "支持多用户并发访问和水平扩展",
        "易维护性": "集中式部署便于维护和更新",
        "商业化": "便于构建SaaS服务和API产品",
        "协作性": "支持团队协作和共享使用"
    }
    
    print("Web视觉应用的重要性:")
    for factor, desc in importance_factors.items():
        print(f"• {factor}: {desc}")

web_vision_importance()

def web_vision_architecture():
    """
    Web视觉应用架构
    """
    architecture = {
        "前端": "用户界面、图像上传、结果展示",
        "API层": "请求处理、参数验证、错误处理",
        "模型服务层": "模型推理、预处理、后处理",
        "存储层": "图像存储、结果缓存、日志记录",
        "部署层": "容器化、负载均衡、监控告警"
    }
    
    print("Web视觉应用架构:")
    for layer, desc in architecture.items():
        print(f"• {layer}: {desc}")

web_vision_architecture()

def fastapi_benefits():
    """
    FastAPI优势
    """
    benefits = {
        "高性能": "接近NodeJS和Go的性能",
        "快速开发": "减少约40%的代码编写",
        "类型安全": "基于Python类型提示的自动验证",
        "自动文档": "自动生成交互式API文档",
        "异步支持": "原生async/await支持",
        "依赖注入": "强大的依赖注入系统"
    }
    
    print("FastAPI优势:")
    for benefit, desc in benefits.items():
        print(f"• {benefit}: {desc}")

fastapi_benefits()

from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import asyncio
import uvicorn

def basic_fastapi_example():
    """
    FastAPI基础示例
    """
    print("FastAPI基础示例:")
    print("""
from fastapi import FastAPI, File, UploadFile
from fastapi.responses import StreamingResponse
import io

app = FastAPI(title="AI Vision API", version="1.0.0")

class ProcessRequest(BaseModel):
    image_url: str
    algorithm: str = "style_transfer"

@app.get("/")
async def root():
    return {"message": "Welcome to AI Vision API"}

@app.post("/process")
async def process_image(request: ProcessRequest):
    # 处理图像逻辑
    return {"status": "success", "result_url": "/result/123"}

@app.post("/upload")
async def upload_image(file: UploadFile = File(...)):
    # 处理上传的文件
    contents = await file.read()
    # ... 处理逻辑
    return {"filename": file.filename, "size": len(contents)}
""")

from typing import Optional
from fastapi import BackgroundTasks, Depends, Header, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

def advanced_fastapi_features():
    """
    FastAPI高级特性示例
    """
    print("FastAPI高级特性:")
    print("""
# 依赖注入
async def get_current_user(token: str = Depends(oauth2_scheme)):
    if token != "fake-token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"username": "admin"}

# 背景任务
def send_notification(email: str, message: str):
    # 模拟发送通知
    print(f"Sending notification to {email}: {message}")

@app.post("/send-email")
async def send_email(background_tasks: BackgroundTasks, email: str, message: str):
    background_tasks.add_task(send_notification, email, message)
    return {"message": "Notification sent in background"}

# 中间件
from starlette.middleware.base import BaseHTTPMiddleware

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        print(f"Incoming request: {request.method} {request.url}")
        response = await call_next(request)
        print(f"Response status: {response.status_code}")
        return response

app.add_middleware(LoggingMiddleware)
""")

advanced_fastapi_features()

import torch
import torchvision.transforms as transforms
from PIL import Image
import numpy as np
from typing import List, Tuple
import os

class VisionModelService:
    """
    视觉模型服务基类
    """
    def __init__(self, model_path: str, device: str = None):
        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
        self.model = self._load_model(model_path)
        self.model.eval()
        self.transform = self._get_transform()
    
    def _load_model(self, model_path: str):
        """
        加载模型
        """
        if not os.path.exists(model_path):
            raise FileNotFoundError(f"Model file not found: {model_path}")
        
        model = torch.load(model_path, map_location=self.device)
        return model.to(self.device)
    
    def _get_transform(self):
        """
        获取预处理变换
        """
        return transforms.Compose([
            transforms.Resize((224, 224)),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406], 
                               std=[0.229, 0.224, 0.225])
        ])
    
    def preprocess(self, image: Image.Image) -> torch.Tensor:
        """
        预处理图像
        """
        return self.transform(image).unsqueeze(0).to(self.device)
    
    def postprocess(self, output: torch.Tensor) -> np.ndarray:
        """
        后处理输出
        """
        return output.cpu().detach().numpy()

class StyleTransferModel(VisionModelService):
    """
    风格迁移模型服务
    """
    def __init__(self, model_path: str, device: str = None):
        super().__init__(model_path, device)
    
    def transfer_style(self, image_path: str) -> Image.Image:
        """
        执行风格迁移
        """
        try:
            # 加载图像
            image = Image.open(image_path).convert('RGB')
            
            # 预处理
            x = self.preprocess(image)
            
            # 推理
            with torch.no_grad():
                result = self.model(x)
            
            # 后处理
            result = result.squeeze(0).cpu()
            
            # 转换为图像
            denormalize = transforms.Normalize(
                mean=[-0.485/0.229, -0.456/0.224, -0.406/0.225],
                std=[1/0.229, 1/0.224, 1/0.225]
            )
            result = denormalize(result)
            result = torch.clamp(result, 0, 1)
            
            return transforms.ToPILImage()(result)
        
        except Exception as e:
            raise ValueError(f"Error during style transfer: {str(e)}")

def model_service_examples():
    """
    模型服务示例
    """
    print("模型服务示例:")
    print("""
# 初始化模型服务
model_service = StyleTransferModel("path/to/style_model.pth")

# 使用模型服务
result_image = model_service.transfer_style("input_image.jpg")

# 保存结果
result_image.save("output_image.jpg")
""")

model_service_examples()

from functools import lru_cache
import threading
from concurrent.futures import ThreadPoolExecutor
import time

class OptimizedModelService:
    """
    优化的模型服务
    """
    def __init__(self, model_path: str):
        self.model_lock = threading.Lock()
        self.executor = ThreadPoolExecutor(max_workers=4)
        self.model = self._load_model(model_path)
        self.cache = {}
    
    def _load_model(self, model_path: str):
        """
        加载模型（支持ONNX Runtime等优化）
        """
        # 这里可以集成ONNX Runtime或其他优化框架
        return torch.load(model_path)
    
    @lru_cache(maxsize=128)
    def cached_inference(self, image_hash: str, model_params: tuple):
        """
        缓存推理结果
        """
        # 实现缓存逻辑
        pass
    
    def batch_process(self, images: List[str]) -> List[Image.Image]:
        """
        批量处理图像
        """
        futures = []
        for img_path in images:
            future = self.executor.submit(self.process_single, img_path)
            futures.append(future)
        
        results = [future.result() for future in futures]
        return results
    
    def process_single(self, image_path: str) -> Image.Image:
        """
        处理单个图像
        """
        # 实现单个图像处理逻辑
        pass

def caching_and_optimization():
    """
    缓存和优化策略
    """
    strategies = [
        "LRU缓存: 缓存最近使用的推理结果",
        "批处理: 提高GPU利用率",
        "异步处理: 避免阻塞主线程",
        "模型预热: 提前加载模型到GPU",
        "内存池: 减少内存分配开销"
    ]
    
    print("模型缓存和优化策略:")
    for i, strategy in enumerate(strategies, 1):
        print(f"{i}. {strategy}")

caching_and_optimization()

from fastapi import FastAPI, File, UploadFile, Query, Form
from fastapi.responses import JSONResponse, StreamingResponse
from typing import Optional
import json
import uuid
from datetime import datetime

app = FastAPI(
    title="AI Vision API",
    description="Web视觉应用API服务",
    version="2.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)

class VisionAPIResponse(BaseModel):
    """
    API响应模型
    """
    success: bool
    message: str
    data: dict = {}
    timestamp: datetime = datetime.now()
    request_id: str = str(uuid.uuid4())

@app.post("/api/v1/style-transfer", response_model=VisionAPIResponse)
async def style_transfer_api(
    file: UploadFile = File(...),
    style: str = Form("candy"),
    strength: float = Form(1.0, ge=0.1, le=2.0),
    quality: str = Form("high")
):
    """
    风格迁移API端点
    """
    try:
        # 验证文件类型
        allowed_types = ["image/jpeg", "image/png", "image/jpg"]
        if file.content_type not in allowed_types:
            raise HTTPException(status_code=400, detail="Only image files are allowed")
        
        # 保存上传的文件
        temp_filename = f"/tmp/{uuid.uuid4()}_{file.filename}"
        with open(temp_filename, "wb") as f:
            content = await file.read()
            f.write(content)
        
        # 执行风格迁移
        style_service = StyleTransferModel("models/style_transfer.pth")
        result_image = style_service.transfer_style(temp_filename)
        
        # 生成结果文件名
        result_filename = f"/tmp/result_{uuid.uuid4()}.jpg"
        result_image.save(result_filename)
        
        # 返回结果
        return VisionAPIResponse(
            success=True,
            message="Style transfer completed successfully",
            data={
                "result_file": result_filename,
                "style": style,
                "strength": strength,
                "processing_time": "0.5s"
            }
        )
    
    except Exception as e:
        return VisionAPIResponse(
            success=False,
            message=f"Error processing request: {str(e)}",
            data={}
        )
    
    finally:
        # 清理临时文件
        if 'temp_filename' in locals():
            try:
                os.remove(temp_filename)
            except:
                pass

def restful_api_design_principles():
    """
    RESTful API设计原则
    """
    principles = {
        "资源导向": "使用名词而非动词表示资源",
        "HTTP方法": "GET/POST/PUT/DELETE对应CRUD操作",
        "状态码": "使用标准HTTP状态码",
        "版本控制": "API版本管理",
        "错误处理": "一致的错误响应格式"
    }
    
    print("RESTful API设计原则:")
    for principle, desc in principles.items():
        print(f"• {principle}: {desc}")

restful_api_design_principles()

from fastapi import Request
import time
from starlette.middleware.base import BaseHTTPMiddleware

class PerformanceMonitoringMiddleware(BaseHTTPMiddleware):
    """
    性能监控中间件
    """
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        response = await call_next(request)
        process_time = time.time() - start_time
        
        response.headers["X-Process-Time"] = str(process_time)
        print(f"Request processed in {process_time:.2f}s")
        
        return response

app.add_middleware(PerformanceMonitoringMiddleware)

def api_performance_optimization():
    """
    API性能优化策略
    """
    optimizations = [
        "异步处理: 使用async/await避免阻塞",
        "连接池: 复用数据库和模型连接",
        "缓存: 使用Redis等缓存热点数据",
        "压缩: 启用GZIP压缩响应",
        "限流: 防止API滥用",
        "CDN: 静态资源加速"
    ]
    
    print("API性能优化策略:")
    for i, opt in enumerate(optimizations, 1):
        print(f"{i}. {opt}")

api_performance_optimization()

# docker-compose.yml
"""
version: '3.8'

services:
  web-vision-app:
    build: .
    ports:
      - "8000:8000"
    volumes:
      - ./models:/app/models
      - ./uploads:/app/uploads
    environment:
      - MODEL_PATH=/app/models/style_transfer.pth
      - DEVICE=cpu
    deploy:
      resources:
        limits:
          memory: 4G
          cpus: '2'
    restart: unless-stopped

  redis:
    image: redis:alpine
    ports:
      - "6379:6379"
    restart: unless-stopped

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
      - ./ssl:/etc/nginx/ssl
    depends_on:
      - web-vision-app
    restart: unless-stopped
"""

import logging
from logging.handlers import RotatingFileHandler
import psutil
import time
from prometheus_client import Counter, Histogram, Gauge, start_http_server

# 指标定义
REQUEST_COUNT = Counter('api_requests_total', 'Total API requests', ['endpoint', 'method'])
REQUEST_LATENCY = Histogram('api_request_duration_seconds', 'Request latency')
ACTIVE_CONNECTIONS = Gauge('active_connections', 'Active connections')

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        RotatingFileHandler('app.log', maxBytes=10000000, backupCount=5),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

def monitoring_and_logging():
    """
    监控和日志策略
    """
    strategies = [
        "性能指标: 记录请求延迟、成功率等",
        "资源监控: 监控CPU、内存、GPU使用率",
        "错误追踪: 记录错误日志便于排查",
        "健康检查: 定期检查服务状态",
        "日志轮转: 防止日志文件过大",
        "告警机制: 异常时及时通知"
    ]
    
    print("监控和日志策略:")
    for i, strategy in enumerate(strategies, 1):
        print(f"{i}. {strategy}")

monitoring_and_logging()

from fastapi.security import HTTPBearer
from jose import JWTError, jwt
import hashlib

security = HTTPBearer()

def security_measures():
    """
    API安全措施
    """
    measures = [
        "认证授权: 使用JWT或API Key验证身份",
        "输入验证: 验证上传文件类型和大小",
        "速率限制: 防止API滥用",
        "CORS配置: 控制跨域访问",
        "数据加密: 敏感数据传输加密",
        "安全头: 设置安全相关的HTTP头"
    ]
    
    print("API安全措施:")
    for i, measure in enumerate(measures, 1):
        print(f"{i}. {measure}")

security_measures()

import magic
from PIL import Image
import io

def secure_file_upload():
    """
    安全文件上传检查
    """
    print("安全文件上传检查:")
    print("""
def validate_uploaded_file(file):
    # 1. 检查文件大小
    if len(file.file) > 10 * 1024 * 1024:  # 10MB限制
        raise HTTPException(status_code=400, detail="File too large")
    
    # 2. 检查文件类型（魔数检测）
    file_content = file.file.read()
    mime_type = magic.from_buffer(file_content, mime=True)
    if mime_type not in ['image/jpeg', 'image/png', 'image/jpg']:
        raise HTTPException(status_code=400, detail="Invalid file type")
    
    # 3. 重置文件指针
    file.file.seek(0)
    
    # 4. 验证图像完整性
    try:
        img = Image.open(io.BytesIO(file_content))
        img.verify()  # 验证图像完整性
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid image file")
    
    return True
""")

secure_file_upload()