邹方成
f536178428
feat(交易记录): 新增交易记录管理页面与API接口 feat(上传): 添加统一上传接口支持自动识别文件类型 feat(用户管理): 为用户模型添加备注字段并更新相关接口 feat(邮件): 实现SMTP邮件发送功能并添加测试脚本 feat(短信): 增强短信服务配置灵活性与日志记录 fix(发票): 修复发票列表时间筛选功能 fix(nginx): 调整上传大小限制与超时配置 docs: 添加多个功能模块的说明文档 docs(估值): 补充估值计算流程与API提交数据说明 chore: 更新依赖与Docker镜像版本
102 lines
3.5 KiB
Python
102 lines
3.5 KiB
Python
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和文件名
|
||
"""
|
||
# 检查文件类型
|
||
if not file.content_type.startswith('image/'):
|
||
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",
|
||
}
|
||
if file.content_type not in allowed:
|
||
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
|
||
"""
|
||
if file.content_type and file.content_type.startswith("image/"):
|
||
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) |