import os from pathlib import Path from typing import List from fastapi import UploadFile from app.schemas.upload import ImageUploadResponse, FileUploadResponse from app.settings.config import settings class UploadController: """文件上传控制器""" @staticmethod async def upload_image(file: UploadFile) -> ImageUploadResponse: """ 上传图片 :param file: 上传的图片文件 :return: 图片URL和文件名 """ ext = os.path.splitext(file.filename or "")[1].lower() image_exts = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".svg"} if not (file.content_type.startswith('image/') or ext in image_exts): raise ValueError("只支持上传图片文件") # 获取项目根目录 base_dir = Path(__file__).resolve().parent.parent # 图片保存目录 upload_dir = base_dir / "static" / "images" # 确保目录存在 if not upload_dir.exists(): upload_dir.mkdir(parents=True, exist_ok=True) # 生成文件名 filename = file.filename file_path = upload_dir / filename # 如果文件已存在,重命名 counter = 1 while file_path.exists(): name, ext = os.path.splitext(filename) filename = f"{name}_{counter}{ext}" file_path = upload_dir / filename counter += 1 # 保存文件 content = await file.read() with open(file_path, "wb") as f: f.write(content) # 返回完整的可访问URL return ImageUploadResponse( url=f"{settings.BASE_URL}/static/images/{filename}", filename=filename ) @staticmethod async def upload_file(file: UploadFile) -> FileUploadResponse: allowed = { "application/pdf", "application/vnd.openxmlformats-officedocument.wordprocessingml.document", "application/msword", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "application/vnd.ms-excel", "application/zip", "application/x-zip-compressed", "application/octet-stream", "text/plain", "text/csv", "application/json", "application/vnd.ms-powerpoint", "application/vnd.openxmlformats-officedocument.presentationml.presentation", "application/x-rar-compressed", "application/x-7z-compressed", } allowed_exts = { ".pdf", ".doc", ".docx", ".xls", ".xlsx", ".zip", ".rar", ".7z", ".txt", ".csv", ".ppt", ".pptx", ".json", } ext = os.path.splitext(file.filename or "")[1].lower() if (file.content_type not in allowed) and (ext not in allowed_exts): raise ValueError("不支持的文件类型") base_dir = Path(__file__).resolve().parent.parent upload_dir = base_dir / "static" / "files" if not upload_dir.exists(): upload_dir.mkdir(parents=True, exist_ok=True) filename = file.filename file_path = upload_dir / filename counter = 1 while file_path.exists(): name, ext = os.path.splitext(filename) filename = f"{name}_{counter}{ext}" file_path = upload_dir / filename counter += 1 content = await file.read() with open(file_path, "wb") as f: f.write(content) return FileUploadResponse( url=f"{settings.BASE_URL}/static/files/{filename}", filename=filename, content_type=file.content_type, ) @staticmethod async def upload_any(file: UploadFile) -> FileUploadResponse: """ 统一上传入口,自动识别图片与非图片类型。 返回统一结构:url, filename, content_type """ image_exts = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".svg"} ext = os.path.splitext(file.filename or "")[1].lower() if (file.content_type and file.content_type.startswith("image/")) or (ext in image_exts): img = await UploadController.upload_image(file) return FileUploadResponse(url=img.url, filename=img.filename, content_type=file.content_type or "image") # 非图片类型复用原文件上传校验 return await UploadController.upload_file(file)