benevoles/backend/app/api/stats.py

"""Statistiques d'une édition : inscriptions (nombre, évolution), réponses au
formulaire, paiements de la cotisation, besoins face aux disponibilités.

Lecture seule, recalculée à chaque appel (quelques centaines de lignes au
plus pour un festival de 50 à 200 bénévoles) — rien n'est stocké.
"""

import datetime as dt
from collections import Counter, defaultdict

from fastapi import APIRouter, Depends
from sqlmodel import Session, select

from app.api.deps import get_edition
from app.api.schemas import (
    EditionStats,
    StatsAnswers,
    StatsArtistWish,
    StatsAttendance,
    StatsAttendanceCounts,
    StatsAttendancePosition,
    StatsNeedDay,
    StatsNeedPosition,
    StatsNeeds,
    StatsPaymentOption,
    StatsPayments,
    StatsRegistrationDay,
    StatsRegistrations,
)
from app.auth.deps import require_coordinator
from app.db import get_session
from app.models import (
    Account,
    Artist,
    Assignment,
    AttendanceStatus,
    DayPart,
    Edition,
    FeePaymentStatus,
    PendingFormSubmission,
    Position,
    Shift,
    Volunteer,
    VolunteerArtist,
    VolunteerCompanionWish,
    VolunteerPositionWish,
    local_now,
)
from app.validation.engine import day_boundary, festival_day

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


def _hours(starts_at: dt.datetime, ends_at: dt.datetime) -> float:
    return (ends_at - starts_at).total_seconds() / 3600


def _round(value: float) -> float:
    return round(value, 2)


def _registrations(volunteers: list[Volunteer]) -> StatsRegistrations:
    per_day = Counter(v.created_at.date() for v in volunteers)
    by_day: list[StatsRegistrationDay] = []
    if per_day:
        day, last, cumulative = min(per_day), max(per_day), 0
        while day <= last:
            cumulative += per_day[day]
            by_day.append(
                StatsRegistrationDay(date=day, count=per_day[day], cumulative=cumulative)
            )
            day += dt.timedelta(days=1)
    since = local_now() - dt.timedelta(days=7)
    return StatsRegistrations(
        total=len(volunteers),
        last_7_days=sum(1 for v in volunteers if v.created_at >= since),
        by_day=by_day,
    )


def _payments(
    volunteers: list[Volunteer], pending_unfinished: int
) -> StatsPayments:
    # Même règle que `lib/fee.ts:feeMode` / `exporting/xlsx.py:_payment_mode`.
    online = [v for v in volunteers if v.fee_payment_status is not None]
    paid = [v for v in online if v.fee_payment_status == FeePaymentStatus.PAID]
    by_label: dict[str, list[Volunteer]] = defaultdict(list)
    for v in paid:
        by_label[v.fee_payment_label or ""].append(v)
    mecenat = [v for v in paid if v.mecenat_address]
    return StatsPayments(
        online=len(online),
        on_site=sum(
            1 for v in volunteers if v.fee_payment_status is None and v.willing_to_pay_fee
        ),
        none=sum(
            1 for v in volunteers if v.fee_payment_status is None and not v.willing_to_pay_fee
        ),
        pending_unfinished=pending_unfinished,
        total_collected=_round(sum(v.fee_amount or 0 for v in paid)),
        mecenat_count=len(mecenat),
        mecenat_total=_round(sum(v.fee_amount or 0 for v in mecenat)),
        by_option=[
            StatsPaymentOption(
                label=label,
                mecenat=any(v.mecenat_address for v in group),
                count=len(group),
                amount_total=_round(sum(v.fee_amount or 0 for v in group)),
            )
            for label, group in sorted(by_label.items())
        ],
    )


def _count(counts: StatsAttendanceCounts, assignment: Assignment) -> None:
    counts.assigned += 1
    if assignment.attendance == AttendanceStatus.PRESENT:
        counts.present += 1
    elif assignment.attendance == AttendanceStatus.ABSENT:
        counts.absent += 1
    else:
        counts.unmarked += 1


def _attendance(
    positions: list[Position], shifts: list[Shift], assignments: list[Assignment]
) -> StatsAttendance:
    """Pointage par poste, sur les créneaux terminés et ceux en cours."""
    now = local_now()
    past = StatsAttendanceCounts()
    live = StatsAttendanceCounts()
    by_position = {
        p.id: StatsAttendancePosition(
            position_id=p.id,
            name=p.name,
            past=StatsAttendanceCounts(),
            live=StatsAttendanceCounts(),
            live_shifts=0,
        )
        for p in positions
    }
    # Créneau terminé ⇒ « past », en cours ⇒ « live », à venir : ignoré.
    phase: dict = {}
    for s in shifts:
        if s.ends_at <= now:
            phase[s.id] = ("past", s.position_id)
        elif s.starts_at <= now:
            phase[s.id] = ("live", s.position_id)
            by_position[s.position_id].live_shifts += 1
    for a in assignments:
        if a.shift_id not in phase:
            continue
        when, position_id = phase[a.shift_id]
        row = by_position[position_id]
        _count(past if when == "past" else live, a)
        _count(row.past if when == "past" else row.live, a)
    return StatsAttendance(
        as_of=now,
        past=past,
        live=live,
        live_shifts=sum(r.live_shifts for r in by_position.values()),
        by_position=[
            r for r in by_position.values() if r.past.assigned or r.live.assigned or r.live_shifts
        ],
    )


@router.get("", response_model=EditionStats)
def get_stats(
    edition: Edition = Depends(get_edition),
    _guard: Account = Depends(require_coordinator),
    session: Session = Depends(get_session),
) -> EditionStats:
    volunteers = list(session.exec(select(Volunteer).where(Volunteer.edition_id == edition.id)))
    volunteer_ids = [v.id for v in volunteers]
    positions = list(
        session.exec(
            select(Position).where(Position.edition_id == edition.id).order_by(Position.name)
        )
    )
    shifts = list(session.exec(select(Shift).where(Shift.edition_id == edition.id)))
    assignments = list(
        session.exec(select(Assignment).where(Assignment.edition_id == edition.id))
    )
    artists = list(
        session.exec(
            select(Artist)
            .where(Artist.edition_id == edition.id)
            .order_by(Artist.performance_date, Artist.start_time)
        )
    )
    position_wishes = (
        list(
            session.exec(
                select(VolunteerPositionWish).where(
                    VolunteerPositionWish.volunteer_id.in_(volunteer_ids)
                )
            )
        )
        if volunteer_ids
        else []
    )
    artist_wishes = (
        list(
            session.exec(
                select(VolunteerArtist).where(VolunteerArtist.volunteer_id.in_(volunteer_ids))
            )
        )
        if volunteer_ids
        else []
    )
    companion_wishers = (
        set(
            session.exec(
                select(VolunteerCompanionWish.volunteer_id).where(
                    VolunteerCompanionWish.volunteer_id.in_(volunteer_ids)
                )
            )
        )
        if volunteer_ids
        else set()
    )
    pending_unfinished = len(
        session.exec(
            select(PendingFormSubmission.id).where(
                PendingFormSubmission.edition_id == edition.id,
                PendingFormSubmission.promoted_volunteer_id.is_(None),
            )
        ).all()
    )
    boundary = day_boundary(
        session.exec(select(DayPart).where(DayPart.edition_id == edition.id))
    )

    answers = StatsAnswers(
        total=len(volunteers),
        email=sum(1 for v in volunteers if v.email),
        phone=sum(1 for v in volunteers if v.phone),
        photo=sum(1 for v in volunteers if v.photo is not None),
        comment=sum(1 for v in volunteers if v.comment),
        companion_note=sum(1 for v in volunteers if v.companion_wish_note),
        available=sum(1 for v in volunteers if v.self_declared_available),
        position_wishes=len({w.volunteer_id for w in position_wishes}),
        artist_wishes=len({w.volunteer_id for w in artist_wishes}),
        companion_wishes=len(companion_wishers),
    )

    fans = Counter(w.artist_id for w in artist_wishes)
    artist_stats = [
        StatsArtistWish(id=a.id, name=a.name, date=a.performance_date, count=fans[a.id])
        for a in artists
    ]

    # Besoins par journée de festival : une soirée qui passe minuit reste sur
    # le jour où elle a commencé (CLAUDE.md § 3).
    shifts_by_id = {s.id: s for s in shifts}
    day_rows: dict[dt.date, dict] = defaultdict(
        lambda: {
            "shifts": 0, "min_hours": 0.0, "ideal_hours": 0.0, "assigned_hours": 0.0,
        }
    )
    for s in shifts:
        row = day_rows[festival_day(s.starts_at, boundary)]
        hours = _hours(s.starts_at, s.ends_at)
        row["shifts"] += 1
        row["min_hours"] += hours * s.min_staff
        row["ideal_hours"] += hours * s.ideal_staff
    for a in assignments:
        s = shifts_by_id.get(a.shift_id)
        if s is not None:
            day_rows[festival_day(s.starts_at, boundary)]["assigned_hours"] += _hours(
                s.starts_at, s.ends_at
            )

    wishers = Counter(w.position_id for w in position_wishes)
    assigned_by_shift = Counter(a.shift_id for a in assignments)
    shifts_by_position: dict = defaultdict(list)
    for s in shifts:
        shifts_by_position[s.position_id].append(s)

    needs = StatsNeeds(
        by_day=[
            StatsNeedDay(
                date=day,
                shifts=row["shifts"],
                min_hours=_round(row["min_hours"]),
                ideal_hours=_round(row["ideal_hours"]),
                assigned_hours=_round(row["assigned_hours"]),
            )
            for day, row in sorted(day_rows.items())
        ],
        by_position=[
            StatsNeedPosition(
                position_id=p.id,
                name=p.name,
                shifts=len(shifts_by_position[p.id]),
                min_slots=sum(s.min_staff for s in shifts_by_position[p.id]),
                ideal_slots=sum(s.ideal_staff for s in shifts_by_position[p.id]),
                assigned=sum(assigned_by_shift[s.id] for s in shifts_by_position[p.id]),
                filled=sum(
                    min(assigned_by_shift[s.id], s.ideal_staff) for s in shifts_by_position[p.id]
                ),
                wishes=wishers[p.id],
            )
            for p in positions
        ],
        volunteers_with_assignment=len({a.volunteer_id for a in assignments}),
    )

    return EditionStats(
        registrations=_registrations(volunteers),
        answers=answers,
        payments=_payments(volunteers, pending_unfinished),
        artist_wishes=artist_stats,
        needs=needs,
        attendance=_attendance(positions, shifts, assignments),
    )