benevoles/backend/app/api/_common.py

"""Helpers partagés par les routers CRUD.

`resolve_owned_ids`/`set_artists`/`set_wished_positions` sont utilisés à la
fois par `volunteers.py` (fiche éditée par la coordination) et `forms.py`
(vœux saisis depuis le formulaire public) — un seul endroit pour la
validation « ces ids appartiennent bien à cette édition ». Idem pour
`read_photo` (photo d'un bénévole)."""

import io
import uuid

from fastapi import HTTPException, UploadFile
from PIL import Image, UnidentifiedImageError
from sqlalchemy import delete
from sqlmodel import Session, select

from app.models import Artist, Position, VolunteerArtist, VolunteerPositionWish, local_now


def owned_or_404(instance, edition_id: uuid.UUID):
    """404 si l'instance est absente ou rattachée à une autre édition."""
    if instance is None or instance.edition_id != edition_id:
        raise HTTPException(status_code=404, detail={"code": "error.not_found"})
    return instance


def replace_links(
    session: Session, link_model, *, filter_column, filter_value, pairs
) -> None:
    """Remplace toutes les lignes d'une table d'association pour une clé."""
    session.execute(
        delete(link_model).where(filter_column == filter_value)
    )
    for row in pairs:
        session.add(link_model(**row))


def resolve_owned_ids(
    session: Session, model, edition_id: uuid.UUID, ids: list[uuid.UUID], code: str
) -> list[uuid.UUID]:
    """Ordre d'entrée conservé, doublons retirés ; 422 si un id est étranger."""
    wanted = list(dict.fromkeys(ids))
    if wanted:
        found = set(
            session.exec(
                select(model.id).where(
                    model.edition_id == edition_id,
                    model.id.in_(wanted),
                )
            )
        )
        if len(found) != len(wanted):
            raise HTTPException(status_code=422, detail={"code": code})
    return wanted


def set_artists(session: Session, volunteer, artist_ids: list[uuid.UUID]) -> None:
    ids = resolve_owned_ids(
        session, Artist, volunteer.edition_id, artist_ids, "artist.unknown"
    )
    replace_links(
        session,
        VolunteerArtist,
        filter_column=VolunteerArtist.volunteer_id,
        filter_value=volunteer.id,
        pairs=[{"volunteer_id": volunteer.id, "artist_id": aid} for aid in ids],
    )


def set_wished_positions(
    session: Session, volunteer, position_ids: list[uuid.UUID]
) -> None:
    ids = resolve_owned_ids(
        session, Position, volunteer.edition_id, position_ids, "position.unknown"
    )
    replace_links(
        session,
        VolunteerPositionWish,
        filter_column=VolunteerPositionWish.volunteer_id,
        filter_value=volunteer.id,
        pairs=[{"volunteer_id": volunteer.id, "position_id": pid} for pid in ids],
    )


MAX_PHOTO_BYTES = 8 * 1024 * 1024
ALLOWED_PHOTO_TYPES = {"image/png", "image/jpeg"}


async def read_photo(photo: UploadFile | None) -> tuple[bytes, str] | None:
    """Validation partagée par le formulaire public (inscription immédiate
    ou en attente de paiement) et l'ajout d'une photo par la coordination
    depuis la fiche (`volunteers.py:set_volunteer_photo`) — `None` si aucun
    fichier fourni. `Content-Type` est
    un en-tête fourni par l'appelant, jamais vérifié par lui-même (ce
    endpoint est public, sans authentification) : un fichier qui ne se
    décode pas comme une vraie image malgré un `Content-Type` correct
    (`image/png`/`image/jpeg` revendiqué) est rejeté ici plutôt que de
    planter plus tard la génération des badges/feuilles de poste
    (`app/badges/pdf.py`/`app/sheets/pdf.py`, qui lisent ces octets sans
    filet) — même validation que l'import d'image des badges
    (`app/api/badges.py:_pixel_size`, réservé à la coordination)."""
    if photo is None or not photo.filename:
        return None
    content_type = photo.content_type or ""
    if content_type not in ALLOWED_PHOTO_TYPES:
        raise HTTPException(
            status_code=422, detail={"code": "form.photo_type_not_supported"}
        )
    content = await photo.read()
    if len(content) > MAX_PHOTO_BYTES:
        raise HTTPException(status_code=413, detail={"code": "form.photo_too_large"})
    if not content:
        return None
    try:
        with Image.open(io.BytesIO(content)) as img:
            img.verify()
    except (UnidentifiedImageError, OSError) as exc:
        raise HTTPException(
            status_code=422, detail={"code": "form.photo_type_not_supported"}
        ) from exc
    return content, content_type


def set_message(target: object, field: str, value: str | None) -> None:
    """Message affiché aux bénévoles (`Edition.volunteer_message`,
    `Position.message`) : espaces de bord retirés, vide = plus de message.
    `{field}_updated_at` n'est daté que si le texte change vraiment — il dit
    aux bénévoles de quand date le message."""
    text = (value or "").strip() or None
    if text == getattr(target, field):
        return
    setattr(target, field, text)
    setattr(target, f"{field}_updated_at", local_now() if text else None)