diff --git a/backend/app/api/media.py b/backend/app/api/media.py index af9233c..6222318 100644 --- a/backend/app/api/media.py +++ b/backend/app/api/media.py @@ -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) diff --git a/backend/app/schemas/media.py b/backend/app/schemas/media.py new file mode 100644 index 0000000..59efe89 --- /dev/null +++ b/backend/app/schemas/media.py @@ -0,0 +1,8 @@ +from pydantic import BaseModel + + +class MediaUploadResponse(BaseModel): + file_id: str + file_path: str + thumbnail_path: str | None + file_type: str diff --git a/backend/app/services/media.py b/backend/app/services/media.py new file mode 100644 index 0000000..35fa5b7 --- /dev/null +++ b/backend/app/services/media.py @@ -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) diff --git a/backend/tests/test_media.py b/backend/tests/test_media.py new file mode 100644 index 0000000..4477810 --- /dev/null +++ b/backend/tests/test_media.py @@ -0,0 +1,96 @@ +import io +import pytest +from pathlib import Path + + +def _make_jpeg(width: int = 200, height: int = 200) -> bytes: + try: + from PIL import Image + img = Image.new("RGB", (width, height), color=(120, 80, 40)) + buf = io.BytesIO() + img.save(buf, format="JPEG") + return buf.getvalue() + except ImportError: + pytest.skip("Pillow non disponible localement (Python 3.14) — OK en Docker") + + +async def test_upload_image_retourne_chemins(client, tmp_path, monkeypatch): + from app.services import media as media_svc + monkeypatch.setattr(media_svc, "UPLOAD_DIR", tmp_path) + + response = await client.post( + "/api/media/upload", + files={"file": ("photo.jpg", _make_jpeg(), "image/jpeg")}, + params={"context": "note"}, + ) + assert response.status_code == 200 + data = response.json() + assert "file_id" in data + assert data["file_path"].startswith("images/originals/") + assert data["thumbnail_path"].startswith("images/thumbnails/") + + +async def test_upload_format_invalide_retourne_400(client, tmp_path, monkeypatch): + from app.services import media as media_svc + monkeypatch.setattr(media_svc, "UPLOAD_DIR", tmp_path) + + response = await client.post( + "/api/media/upload", + files={"file": ("doc.pdf", b"contenu pdf", "application/pdf")}, + ) + assert response.status_code == 400 + + +async def test_miniature_produit_max_150px(client, tmp_path, monkeypatch): + try: + from PIL import Image + except ImportError: + pytest.skip("Pillow non disponible localement") + from app.services import media as media_svc + monkeypatch.setattr(media_svc, "UPLOAD_DIR", tmp_path) + + response = await client.post( + "/api/media/upload", + files={"file": ("prod.jpg", _make_jpeg(800, 600), "image/jpeg")}, + params={"context": "product"}, + ) + assert response.status_code == 200 + data = response.json() + thumb = Image.open(tmp_path / data["thumbnail_path"]) + assert thumb.width <= 150 + assert thumb.height <= 150 + + +async def test_upload_audio_retourne_chemin(client, tmp_path, monkeypatch): + from app.services import media as media_svc + monkeypatch.setattr(media_svc, "UPLOAD_DIR", tmp_path) + + response = await client.post( + "/api/media/upload", + files={"file": ("enreg.webm", b"fake webm bytes", "audio/webm")}, + params={"context": "note"}, + ) + assert response.status_code == 200 + data = response.json() + assert data["file_path"].startswith("audio/") + assert data["thumbnail_path"] is None + + +async def test_suppression_media(client, tmp_path, monkeypatch): + from app.services import media as media_svc + monkeypatch.setattr(media_svc, "UPLOAD_DIR", tmp_path) + + up = await client.post( + "/api/media/upload", + files={"file": ("enreg.webm", b"fake webm bytes", "audio/webm")}, + params={"context": "note"}, + ) + data = up.json() + file_id = data["file_id"] + + response = await client.delete( + f"/api/media/{file_id}", + params={"file_path": data["file_path"]}, + ) + assert response.status_code == 204 + assert not (tmp_path / data["file_path"]).exists()