""" 文件上传路由 """ import os import uuid from datetime import datetime from fastapi import APIRouter, UploadFile, File, Depends, HTTPException from ..config import get_settings from ..utils.jwt_utils import get_current_user_id router = APIRouter(prefix="/upload", tags=["上传"]) # 允许的图片格式 ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'webp'} # 最大文件大小 (5MB) MAX_FILE_SIZE = 5 * 1024 * 1024 def allowed_file(filename: str) -> bool: """检查文件扩展名是否允许""" return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS @router.post("/avatar") async def upload_avatar( file: UploadFile = File(...), user_id: int = Depends(get_current_user_id) ): """上传用户头像""" # 检查文件类型 if not file.filename or not allowed_file(file.filename): raise HTTPException(status_code=400, detail="不支持的文件格式") # 读取文件内容 content = await file.read() # 检查文件大小 if len(content) > MAX_FILE_SIZE: raise HTTPException(status_code=400, detail="文件大小超过限制(5MB)") # 生成唯一文件名 ext = file.filename.rsplit('.', 1)[1].lower() filename = f"{user_id}_{uuid.uuid4().hex[:8]}.{ext}" # 创建上传目录 settings = get_settings() upload_dir = os.path.join(settings.upload_path, "avatars") os.makedirs(upload_dir, exist_ok=True) # 保存文件 file_path = os.path.join(upload_dir, filename) with open(file_path, "wb") as f: f.write(content) # 返回访问URL # 使用相对路径,前端拼接 baseUrl avatar_url = f"/uploads/avatars/{filename}" return { "code": 0, "data": { "url": avatar_url, "filename": filename }, "message": "上传成功" } @router.post("/image") async def upload_image( file: UploadFile = File(...), user_id: int = Depends(get_current_user_id) ): """上传通用图片""" # 检查文件类型 if not file.filename or not allowed_file(file.filename): raise HTTPException(status_code=400, detail="不支持的文件格式") # 读取文件内容 content = await file.read() # 检查文件大小 if len(content) > MAX_FILE_SIZE: raise HTTPException(status_code=400, detail="文件大小超过限制(5MB)") # 生成唯一文件名 ext = file.filename.rsplit('.', 1)[1].lower() date_str = datetime.now().strftime("%Y%m%d") filename = f"{date_str}_{uuid.uuid4().hex[:8]}.{ext}" # 创建上传目录 settings = get_settings() upload_dir = os.path.join(settings.upload_path, "images") os.makedirs(upload_dir, exist_ok=True) # 保存文件 file_path = os.path.join(upload_dir, filename) with open(file_path, "wb") as f: f.write(content) # 返回访问URL image_url = f"/uploads/images/{filename}" return { "code": 0, "data": { "url": image_url, "filename": filename }, "message": "上传成功" }