133 lines
3.9 KiB
Python
133 lines
3.9 KiB
Python
import os
|
|
import uuid
|
|
from typing import List, Optional
|
|
|
|
from fastapi import APIRouter, Body, Depends, File, HTTPException, Query, UploadFile, status
|
|
from pydantic import BaseModel
|
|
from sqlmodel import Session, select
|
|
|
|
from app.config import UPLOAD_DIR
|
|
from app.database import get_session
|
|
from app.models.media import Attachment, Media
|
|
|
|
|
|
class MediaPatch(BaseModel):
|
|
entity_type: Optional[str] = None
|
|
entity_id: Optional[int] = None
|
|
titre: Optional[str] = None
|
|
|
|
router = APIRouter(tags=["media"])
|
|
|
|
|
|
def _save_webp(data: bytes, max_px: int) -> str:
|
|
try:
|
|
from PIL import Image
|
|
import io
|
|
|
|
img = Image.open(io.BytesIO(data)).convert("RGB")
|
|
img.thumbnail((max_px, max_px))
|
|
name = f"{uuid.uuid4()}.webp"
|
|
path = os.path.join(UPLOAD_DIR, name)
|
|
img.save(path, "WEBP", quality=85)
|
|
return name
|
|
except Exception:
|
|
name = f"{uuid.uuid4()}.bin"
|
|
path = os.path.join(UPLOAD_DIR, name)
|
|
with open(path, "wb") as f:
|
|
f.write(data)
|
|
return name
|
|
|
|
|
|
@router.post("/upload")
|
|
async def upload_file(file: UploadFile = File(...)):
|
|
os.makedirs(UPLOAD_DIR, exist_ok=True)
|
|
data = await file.read()
|
|
ct = file.content_type or ""
|
|
if ct.startswith("image/"):
|
|
name = _save_webp(data, 1200)
|
|
thumb = _save_webp(data, 300)
|
|
return {"url": f"/uploads/{name}", "thumbnail_url": f"/uploads/{thumb}"}
|
|
else:
|
|
name = f"{uuid.uuid4()}_{file.filename}"
|
|
path = os.path.join(UPLOAD_DIR, name)
|
|
with open(path, "wb") as f:
|
|
f.write(data)
|
|
return {"url": f"/uploads/{name}", "thumbnail_url": None}
|
|
|
|
|
|
@router.get("/media/all", response_model=List[Media])
|
|
def list_all_media(
|
|
entity_type: Optional[str] = Query(default=None),
|
|
session: Session = Depends(get_session),
|
|
):
|
|
"""Retourne tous les médias, filtrés optionnellement par entity_type."""
|
|
q = select(Media).order_by(Media.created_at.desc())
|
|
if entity_type:
|
|
q = q.where(Media.entity_type == entity_type)
|
|
return session.exec(q).all()
|
|
|
|
|
|
@router.get("/media", response_model=List[Media])
|
|
def list_media(
|
|
entity_type: str = Query(...),
|
|
entity_id: int = Query(...),
|
|
session: Session = Depends(get_session),
|
|
):
|
|
return session.exec(
|
|
select(Media).where(
|
|
Media.entity_type == entity_type, Media.entity_id == entity_id
|
|
)
|
|
).all()
|
|
|
|
|
|
@router.post("/media", response_model=Media, status_code=status.HTTP_201_CREATED)
|
|
def create_media(m: Media, session: Session = Depends(get_session)):
|
|
session.add(m)
|
|
session.commit()
|
|
session.refresh(m)
|
|
return m
|
|
|
|
|
|
@router.patch("/media/{id}", response_model=Media)
|
|
def update_media(id: int, payload: MediaPatch, session: Session = Depends(get_session)):
|
|
m = session.get(Media, id)
|
|
if not m:
|
|
raise HTTPException(404, "Media introuvable")
|
|
for k, v in payload.model_dump(exclude_none=True).items():
|
|
setattr(m, k, v)
|
|
session.add(m)
|
|
session.commit()
|
|
session.refresh(m)
|
|
return m
|
|
|
|
|
|
@router.delete("/media/{id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_media(id: int, session: Session = Depends(get_session)):
|
|
m = session.get(Media, id)
|
|
if not m:
|
|
raise HTTPException(404, "Media introuvable")
|
|
session.delete(m)
|
|
session.commit()
|
|
|
|
|
|
@router.get("/attachments", response_model=List[Attachment])
|
|
def list_attachments(
|
|
entity_type: str = Query(...),
|
|
entity_id: int = Query(...),
|
|
session: Session = Depends(get_session),
|
|
):
|
|
return session.exec(
|
|
select(Attachment).where(
|
|
Attachment.entity_type == entity_type,
|
|
Attachment.entity_id == entity_id,
|
|
)
|
|
).all()
|
|
|
|
|
|
@router.post("/attachments", response_model=Attachment, status_code=status.HTTP_201_CREATED)
|
|
def create_attachment(a: Attachment, session: Session = Depends(get_session)):
|
|
session.add(a)
|
|
session.commit()
|
|
session.refresh(a)
|
|
return a
|