Bruno Charest
cc99fea20a
- Implement tests for database generator to ensure proper session handling. - Create tests for EXIF extraction and conversion functions. - Add tests for image-related endpoints, ensuring proper data retrieval and isolation between clients. - Develop tests for OCR functionality, including language detection and text extraction. - Introduce tests for the image processing pipeline, covering success and failure scenarios. - Validate rate limiting functionality and ensure independent counters for different clients. - Implement scraper tests to verify HTML content fetching and error handling. - Add unit tests for various services, including storage and filename generation. - Establish worker entry point for ARQ to handle background image processing tasks.
124 lines
4.5 KiB
Python
124 lines
4.5 KiB
Python
"""
|
|
Service de stockage — sauvegarde fichiers, génération thumbnails
|
|
Multi-tenant : les fichiers sont isolés par client_id.
|
|
"""
|
|
import uuid
|
|
import logging
|
|
import aiofiles
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
from PIL import Image as PILImage
|
|
from fastapi import UploadFile, HTTPException, status
|
|
from app.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ALLOWED_MIME_TYPES = {
|
|
"image/jpeg", "image/png", "image/gif",
|
|
"image/webp", "image/bmp", "image/tiff",
|
|
}
|
|
|
|
THUMBNAIL_SIZE = (320, 320)
|
|
|
|
|
|
def _generate_filename(original: str) -> tuple[str, str]:
|
|
"""Retourne (uuid_filename, extension)."""
|
|
suffix = Path(original).suffix.lower() or ".jpg"
|
|
uid = str(uuid.uuid4())
|
|
return f"{uid}{suffix}", uid
|
|
|
|
|
|
def _get_client_upload_path(client_id: str) -> Path:
|
|
"""Retourne le répertoire d'upload pour un client donné."""
|
|
p = settings.upload_path / client_id
|
|
p.mkdir(parents=True, exist_ok=True)
|
|
return p
|
|
|
|
|
|
def _get_client_thumbnails_path(client_id: str) -> Path:
|
|
"""Retourne le répertoire de thumbnails pour un client donné."""
|
|
p = settings.thumbnails_path / client_id
|
|
p.mkdir(parents=True, exist_ok=True)
|
|
return p
|
|
|
|
|
|
async def save_upload(file: UploadFile, client_id: str) -> dict:
|
|
"""
|
|
Valide, sauvegarde le fichier uploadé et génère un thumbnail.
|
|
Les fichiers sont stockés dans uploads/{client_id}/ pour l'isolation.
|
|
Retourne un dict avec toutes les métadonnées fichier.
|
|
"""
|
|
# ── Validation MIME ───────────────────────────────────────
|
|
if file.content_type not in ALLOWED_MIME_TYPES:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
|
|
detail=f"Type non supporté : {file.content_type}. "
|
|
f"Acceptés : {', '.join(ALLOWED_MIME_TYPES)}",
|
|
)
|
|
|
|
# ── Lecture du contenu ────────────────────────────────────
|
|
content = await file.read()
|
|
|
|
if len(content) > settings.max_upload_bytes:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
|
|
detail=f"Fichier trop volumineux. Max : {settings.MAX_UPLOAD_SIZE_MB} MB",
|
|
)
|
|
|
|
# ── Nommage et chemins ────────────────────────────────────
|
|
filename, file_uuid = _generate_filename(file.filename or "image")
|
|
upload_dir = _get_client_upload_path(client_id)
|
|
thumb_dir = _get_client_thumbnails_path(client_id)
|
|
|
|
file_path = upload_dir / filename
|
|
thumb_filename = f"thumb_{filename}"
|
|
thumb_path = thumb_dir / thumb_filename
|
|
|
|
# ── Sauvegarde fichier original ───────────────────────────
|
|
async with aiofiles.open(file_path, "wb") as f:
|
|
await f.write(content)
|
|
|
|
# ── Dimensions + thumbnail ────────────────────────────────
|
|
width, height = None, None
|
|
try:
|
|
with PILImage.open(file_path) as img:
|
|
width, height = img.size
|
|
img.thumbnail(THUMBNAIL_SIZE, PILImage.LANCZOS)
|
|
# Convertit en RGB si nécessaire (ex: PNG RGBA)
|
|
if img.mode in ("RGBA", "P"):
|
|
img = img.convert("RGB")
|
|
img.save(thumb_path, "JPEG", quality=85)
|
|
except Exception as e:
|
|
# Thumbnail non bloquant
|
|
thumb_path = None
|
|
logger.warning("Erreur génération thumbnail : %s", e)
|
|
|
|
return {
|
|
"uuid": file_uuid,
|
|
"original_name": file.filename,
|
|
"filename": filename,
|
|
"file_path": str(file_path),
|
|
"thumbnail_path": str(thumb_path) if thumb_path else None,
|
|
"mime_type": file.content_type,
|
|
"file_size": len(content),
|
|
"width": width,
|
|
"height": height,
|
|
"uploaded_at": datetime.now(timezone.utc),
|
|
"client_id": client_id,
|
|
}
|
|
|
|
|
|
def delete_files(file_path: str, thumbnail_path: str | None = None) -> None:
|
|
"""Supprime le fichier original et son thumbnail du disque."""
|
|
for path_str in [file_path, thumbnail_path]:
|
|
if path_str:
|
|
p = Path(path_str)
|
|
if p.exists():
|
|
p.unlink()
|
|
|
|
|
|
def get_image_url(filename: str, client_id: str, thumb: bool = False) -> str:
|
|
"""Construit l'URL publique d'une image."""
|
|
prefix = "thumbnails" if thumb else "uploads"
|
|
return f"/static/{prefix}/{client_id}/{filename}"
|