73 lines
2.7 KiB
Python
73 lines
2.7 KiB
Python
import mimetypes
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from fastapi import APIRouter, HTTPException, Request, status
|
|
from starlette.datastructures import UploadFile
|
|
|
|
from app.config import oss_settings
|
|
from app.schemas import UploadResponse
|
|
|
|
|
|
router = APIRouter(prefix="/uploads", tags=["uploads"])
|
|
|
|
MAX_IMAGE_SIZE = 5 * 1024 * 1024
|
|
ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp", "image/gif"}
|
|
|
|
|
|
def get_public_url(object_key: str) -> str:
|
|
if oss_settings.public_base_url:
|
|
return f"{oss_settings.public_base_url.rstrip('/')}/{object_key}"
|
|
|
|
endpoint = oss_settings.endpoint.replace("https://", "").replace("http://", "").rstrip("/")
|
|
return f"https://{oss_settings.bucket_name}.{endpoint}/{object_key}"
|
|
|
|
|
|
def build_object_key(filename: str | None, folder: str) -> str:
|
|
suffix = Path(filename or "").suffix.lower()
|
|
if not suffix:
|
|
suffix = mimetypes.guess_extension("image/jpeg") or ".jpg"
|
|
|
|
safe_folder = folder.strip("/ ") or oss_settings.upload_prefix
|
|
return f"{oss_settings.upload_prefix.strip('/')}/{safe_folder}/{uuid4().hex}{suffix}"
|
|
|
|
|
|
@router.post("/images", response_model=UploadResponse)
|
|
async def upload_image(request: Request) -> UploadResponse:
|
|
if not oss_settings.is_configured:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="OSS 配置未完成，请先填写 app/config.py 中的 OSS 配置。",
|
|
)
|
|
|
|
form = await request.form()
|
|
file = form.get("file")
|
|
folder_value = form.get("folder")
|
|
folder = str(folder_value) if folder_value else "products"
|
|
|
|
if not isinstance(file, UploadFile):
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请选择要上传的图片")
|
|
|
|
content_type = file.content_type or "application/octet-stream"
|
|
if content_type not in ALLOWED_IMAGE_TYPES:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="只支持上传图片文件")
|
|
|
|
data = await file.read()
|
|
if len(data) > MAX_IMAGE_SIZE:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="图片大小不能超过 5MB")
|
|
|
|
try:
|
|
import oss2
|
|
except ImportError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="缺少 oss2 依赖，请先安装 requirements.txt 中的依赖。",
|
|
) from exc
|
|
|
|
object_key = build_object_key(file.filename, folder)
|
|
auth = oss2.Auth(oss_settings.access_key_id, oss_settings.access_key_secret)
|
|
bucket = oss2.Bucket(auth, oss_settings.upload_endpoint, oss_settings.bucket_name)
|
|
bucket.put_object(object_key, data, headers={"Content-Type": content_type})
|
|
|
|
return UploadResponse(url=get_public_url(object_key), object_key=object_key)
|