feat: module media — upload, miniatures Pillow et suppression
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,3 +1,32 @@
|
||||
from fastapi import APIRouter
|
||||
from fastapi import APIRouter, UploadFile, File, Query, HTTPException
|
||||
from fastapi.responses import Response
|
||||
|
||||
from app.schemas.media import MediaUploadResponse
|
||||
from app.services.media import save_image, save_audio, delete_media, ALLOWED_IMAGE_TYPES, ALLOWED_AUDIO_TYPES
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.post("/upload", response_model=MediaUploadResponse)
|
||||
async def upload_media(
|
||||
file: UploadFile = File(...),
|
||||
context: str = Query(default="note", pattern="^(product|note|attachment)$"),
|
||||
):
|
||||
if file.content_type in ALLOWED_IMAGE_TYPES:
|
||||
result = await save_image(file, context=context)
|
||||
elif file.content_type in ALLOWED_AUDIO_TYPES:
|
||||
result = await save_audio(file)
|
||||
else:
|
||||
raise HTTPException(status_code=400, detail=f"Type de fichier non supporté : {file.content_type}")
|
||||
|
||||
return MediaUploadResponse(**result)
|
||||
|
||||
|
||||
@router.delete("/{file_id}", status_code=204)
|
||||
async def delete_media_endpoint(
|
||||
file_id: str,
|
||||
file_path: str = Query(...),
|
||||
thumbnail_path: str | None = Query(default=None),
|
||||
):
|
||||
delete_media(file_id=file_id, file_path=file_path, thumbnail_path=thumbnail_path)
|
||||
return Response(status_code=204)
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class MediaUploadResponse(BaseModel):
|
||||
file_id: str
|
||||
file_path: str
|
||||
thumbnail_path: str | None
|
||||
file_type: str
|
||||
@@ -0,0 +1,85 @@
|
||||
import io
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import HTTPException, UploadFile
|
||||
from PIL import Image
|
||||
|
||||
from app.core.config import settings
|
||||
|
||||
UPLOAD_DIR: Path = settings.upload_path
|
||||
|
||||
ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp", "image/svg+xml"}
|
||||
ALLOWED_AUDIO_TYPES = {"audio/webm", "audio/mp4"}
|
||||
|
||||
THUMBNAIL_SIZES = {
|
||||
"product": (150, 150),
|
||||
"note": (300, 300),
|
||||
"attachment": (400, 300),
|
||||
}
|
||||
|
||||
|
||||
async def save_image(file: UploadFile, context: str = "note") -> dict:
|
||||
if file.content_type not in ALLOWED_IMAGE_TYPES:
|
||||
raise HTTPException(status_code=400, detail=f"Format non supporté : {file.content_type}")
|
||||
|
||||
file_id = str(uuid.uuid4())
|
||||
content = await file.read()
|
||||
|
||||
orig_dir = UPLOAD_DIR / "images" / "originals"
|
||||
orig_dir.mkdir(parents=True, exist_ok=True)
|
||||
orig_path = orig_dir / f"{file_id}.webp"
|
||||
|
||||
if file.content_type == "image/svg+xml":
|
||||
orig_path.write_bytes(content)
|
||||
return {
|
||||
"file_id": file_id,
|
||||
"file_path": str(orig_path.relative_to(UPLOAD_DIR)),
|
||||
"thumbnail_path": None,
|
||||
"file_type": "image",
|
||||
}
|
||||
|
||||
img = Image.open(io.BytesIO(content)).convert("RGB")
|
||||
img.save(orig_path, "WEBP", quality=85)
|
||||
|
||||
thumb_dir = UPLOAD_DIR / "images" / "thumbnails"
|
||||
thumb_dir.mkdir(parents=True, exist_ok=True)
|
||||
thumb_path = thumb_dir / f"{file_id}_thumb.webp"
|
||||
|
||||
size = THUMBNAIL_SIZES.get(context, (300, 300))
|
||||
img_thumb = Image.open(io.BytesIO(content)).convert("RGB")
|
||||
img_thumb.thumbnail(size, Image.LANCZOS)
|
||||
img_thumb.save(thumb_path, "WEBP", quality=80)
|
||||
|
||||
return {
|
||||
"file_id": file_id,
|
||||
"file_path": str(orig_path.relative_to(UPLOAD_DIR)),
|
||||
"thumbnail_path": str(thumb_path.relative_to(UPLOAD_DIR)),
|
||||
"file_type": "image",
|
||||
}
|
||||
|
||||
|
||||
async def save_audio(file: UploadFile) -> dict:
|
||||
if file.content_type not in ALLOWED_AUDIO_TYPES:
|
||||
raise HTTPException(status_code=400, detail=f"Format audio non supporté : {file.content_type}")
|
||||
|
||||
file_id = str(uuid.uuid4())
|
||||
audio_dir = UPLOAD_DIR / "audio"
|
||||
audio_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
ext = ".webm" if "webm" in (file.content_type or "") else ".m4a"
|
||||
audio_path = audio_dir / f"{file_id}{ext}"
|
||||
audio_path.write_bytes(await file.read())
|
||||
|
||||
return {
|
||||
"file_id": file_id,
|
||||
"file_path": str(audio_path.relative_to(UPLOAD_DIR)),
|
||||
"thumbnail_path": None,
|
||||
"file_type": "audio",
|
||||
}
|
||||
|
||||
|
||||
def delete_media(file_id: str, file_path: str, thumbnail_path: str | None = None) -> None:
|
||||
(UPLOAD_DIR / file_path).unlink(missing_ok=True)
|
||||
if thumbnail_path:
|
||||
(UPLOAD_DIR / thumbnail_path).unlink(missing_ok=True)
|
||||
Reference in New Issue
Block a user