81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
from fastapi import APIRouter, UploadFile, File, Depends
|
|
from typing import List
|
|
import uuid
|
|
from datetime import datetime
|
|
from app.core.cos import cos_manager
|
|
from app.core.config import settings
|
|
from app.models.upload import UploadResponse, MultiUploadResponse
|
|
from app.core.response import success_response, error_response, ResponseModel
|
|
from app.api.deps import get_current_user
|
|
from app.models.user import UserDB
|
|
|
|
router = APIRouter()
|
|
|
|
async def upload_to_cos(file: UploadFile) -> str:
|
|
"""上传文件到腾讯云COS"""
|
|
# 生成唯一文件名
|
|
ext = file.filename.split('.')[-1] if '.' in file.filename else ''
|
|
filename = f"{datetime.now().strftime('%Y%m%d')}/{uuid.uuid4()}.{ext}"
|
|
|
|
# 上传文件
|
|
cos_manager.put_object(
|
|
Bucket=settings.COS_BUCKET,
|
|
Body=await file.read(),
|
|
Key=filename,
|
|
ContentType=file.content_type,
|
|
ACL="public-read"
|
|
)
|
|
|
|
# 返回文件URL
|
|
return f"https://{settings.COS_BASE_URL}/{filename}"
|
|
|
|
@router.post("/image", response_model=ResponseModel)
|
|
async def upload_image(
|
|
file: UploadFile = File(...),
|
|
current_user: UserDB = Depends(get_current_user)
|
|
):
|
|
"""上传单张图片"""
|
|
if not file.content_type.startswith('image/'):
|
|
return error_response(code=400, message="只能上传图片文件")
|
|
|
|
try:
|
|
url = await upload_to_cos(file)
|
|
return success_response(data=UploadResponse(url=url))
|
|
except Exception as e:
|
|
return error_response(code=500, message=f"上传失败: {str(e)}")
|
|
|
|
@router.post("/images", response_model=ResponseModel)
|
|
async def upload_images(
|
|
files: List[UploadFile] = File(...),
|
|
current_user: UserDB = Depends(get_current_user)
|
|
):
|
|
"""上传多张图片"""
|
|
if len(files) > 5:
|
|
return error_response(code=400, message="最多同时上传5张图片")
|
|
|
|
urls = []
|
|
try:
|
|
for file in files:
|
|
if not file.content_type.startswith('image/'):
|
|
return error_response(code=400, message="只能上传图片文件")
|
|
url = await upload_to_cos(file)
|
|
urls.append(url)
|
|
return success_response(data=MultiUploadResponse(urls=urls))
|
|
except Exception as e:
|
|
return error_response(code=500, message=f"上传失败: {str(e)}")
|
|
|
|
@router.get("/presigned-url")
|
|
async def get_presigned_url(
|
|
filename: str,
|
|
current_user: UserDB = Depends(get_current_user)
|
|
):
|
|
"""获取预签名上传URL"""
|
|
key = f"uploads/{current_user.userid}/{filename}"
|
|
try:
|
|
url = cos_manager.get_upload_url(key)
|
|
return success_response(data={
|
|
"url": url,
|
|
"key": key
|
|
})
|
|
except Exception as e:
|
|
return error_response(code=500, message=f"获取上传URL失败: {str(e)}") |