benevoles/backend/app/api/position_sheets.py

"""Feuilles de poste (PDF) : une page par poste et par jour, listant les
créneaux du jour avec les bénévoles affectés et leur téléphone — à imprimer
et afficher sur place. Complément des badges (`app/api/badges.py`) : un
badge identifie un bénévole, une feuille identifie un poste pour la
journée. Voir `app/sheets/pdf.py` pour la composition du PDF.
"""

import datetime as dt
import uuid
from collections import defaultdict

from fastapi import APIRouter, Depends, HTTPException, Response
from sqlalchemy.orm import aliased
from sqlmodel import Session, select

from app.api.deps import get_edition
from app.api.schemas import PositionSheetsRequest
from app.auth.deps import require_coordinator
from app.db import get_session
from app.models import Account, Artist, Assignment, Edition, Position, Shift, Volunteer
from app.sheets.pdf import (
    PositionDaySheet,
    SheetArtist,
    SheetShift,
    SheetVolunteer,
    generate_position_sheets_pdf,
)

router = APIRouter(
    prefix="/editions/{edition_id}/position-sheets", tags=["position-sheets"]
)


@router.post("/generate")
def generate_position_sheets(
    payload: PositionSheetsRequest,
    edition: Edition = Depends(get_edition),
    _guard: Account = Depends(require_coordinator),
    session: Session = Depends(get_session),
) -> Response:
    query = select(Position).where(Position.edition_id == edition.id)
    if payload.position_ids is not None:
        query = query.where(Position.id.in_(payload.position_ids))
    positions = list(session.exec(query.order_by(Position.name)))
    if not positions:
        raise HTTPException(
            status_code=422, detail={"code": "position_sheets.no_positions"}
        )

    position_ids = [p.id for p in positions]
    shifts = list(
        session.exec(
            select(Shift)
            .where(Shift.position_id.in_(position_ids))
            .order_by(Shift.starts_at)
        )
    )
    shift_ids = [s.id for s in shifts]

    volunteers_by_shift: dict[uuid.UUID, list[Volunteer]] = defaultdict(list)
    if shift_ids:
        rows = session.exec(
            select(Assignment.shift_id, Volunteer)
            .join(Volunteer, Volunteer.id == Assignment.volunteer_id)
            .where(Assignment.shift_id.in_(shift_ids))
        ).all()
        for shift_id, volunteer in rows:
            volunteers_by_shift[shift_id].append(volunteer)
        for vs in volunteers_by_shift.values():
            vs.sort(key=lambda v: (v.last_name, v.first_name))

    # Un jour sans aucun besoin réel (créneau 0/0 généré automatiquement,
    # jamais rempli) n'apparaît pas — évite des pages « personne affecté »
    # pour des créneaux qui ne représentent aucun besoin.
    shifts_by_position_day: dict[tuple[uuid.UUID, dt.date], list[Shift]] = defaultdict(list)
    for shift in shifts:
        undefined_need = shift.min_staff == 0 and shift.ideal_staff == 0
        if undefined_need and not volunteers_by_shift.get(shift.id):
            continue
        shifts_by_position_day[(shift.position_id, shift.starts_at.date())].append(shift)

    Referent = aliased(Volunteer)
    referent_rows = session.exec(
        select(Position.id, Referent)
        .join(Account, Account.id == Position.lead_account_id, isouter=True)
        .join(Referent, Referent.id == Account.volunteer_id, isouter=True)
        .where(Position.id.in_(position_ids))
    ).all()
    referent_by_position: dict[uuid.UUID, Volunteer] = {
        pid: referent for pid, referent in referent_rows if referent is not None
    }

    # Programmation du jour, tous postes confondus — voir `SheetArtist`.
    artists_by_date: dict[dt.date, list[Artist]] = defaultdict(list)
    for artist in session.exec(
        select(Artist)
        .where(Artist.edition_id == edition.id)
        .order_by(Artist.start_time)
    ):
        artists_by_date[artist.performance_date].append(artist)

    positions_by_id = {p.id: p for p in positions}

    def _sheet(position_id: uuid.UUID, date: dt.date, day_shifts: list[Shift]) -> PositionDaySheet:
        referent = referent_by_position.get(position_id)
        return PositionDaySheet(
            position_name=positions_by_id[position_id].name,
            position_location=positions_by_id[position_id].location,
            date=date,
            referent_name=f"{referent.first_name} {referent.last_name}" if referent else None,
            referent_phone=referent.phone if referent else None,
            artists=tuple(
                SheetArtist(name=a.name, start_time=a.start_time, end_time=a.end_time)
                for a in artists_by_date.get(date, [])
            ),
            shifts=tuple(
                SheetShift(
                    starts_at=s.starts_at,
                    ends_at=s.ends_at,
                    min_staff=s.min_staff,
                    ideal_staff=s.ideal_staff,
                    volunteers=tuple(
                        SheetVolunteer(
                            name=f"{v.first_name} {v.last_name}", phone=v.phone, photo=v.photo
                        )
                        for v in volunteers_by_shift.get(s.id, [])
                    ),
                )
                for s in sorted(day_shifts, key=lambda s: s.starts_at)
            ),
        )

    sheets = [
        _sheet(position_id, date, day_shifts)
        for (position_id, date), day_shifts in sorted(
            shifts_by_position_day.items(),
            key=lambda item: (positions_by_id[item[0][0]].name, item[0][1]),
        )
    ]
    if not sheets:
        raise HTTPException(
            status_code=422, detail={"code": "position_sheets.no_shifts"}
        )

    pdf_bytes = generate_position_sheets_pdf(sheets)
    return Response(
        content=pdf_bytes,
        media_type="application/pdf",
        headers={"Content-Disposition": 'attachment; filename="feuilles-de-poste.pdf"'},
    )