benevoles/backend/app/seed.py

"""Jeu de données de démonstration.

    docker compose run --rm backend uv run python -m app.seed

Idempotent : supprime l'édition de démonstration existante puis la recrée.
Déterministe (graine fixe). Aucune donnée nominative réelle.
"""

import datetime as dt
import random
from collections import Counter
from dataclasses import dataclass

from sqlmodel import Session, select

from app.auth.passwords import hash_password
from app.db import engine
from app.importing.text import normalize
from app.models import (
    Account,
    Artist,
    Assignment,
    AssignmentSource,
    DayPart,
    Edition,
    FeePaymentStatus,
    Position,
    PositionCategory,
    Role,
    Shift,
    Volunteer,
    VolunteerArtist,
    VolunteerCompanionWish,
    VolunteerPositionWish,
)
from app.validation import Severity, validate
from app.validation.loader import build_validation_input

SEED = 20260709
DEMO_EDITION_NAME = "Festival des Lucioles"
DEMO_COORDINATION_PASSWORD = "coordination-lucioles-2026"
DEMO_YEAR = 2026
START = dt.date(2026, 10, 30)
END = dt.date(2026, 10, 31)

FIRST_NAMES = [
    "Camille",
    "Léa",
    "Hugo",
    "Nathan",
    "Emma",
    "Lucas",
    "Chloé",
    "Louis",
    "Manon",
    "Jules",
    "Sarah",
    "Tom",
    "Inès",
    "Théo",
    "Jade",
    "Raphaël",
    "Alice",
    "Gabriel",
    "Louise",
    "Adam",
    "Zoé",
    "Arthur",
    "Lina",
    "Paul",
    "Rose",
    "Noah",
    "Anna",
    "Ethan",
    "Mila",
    "Sacha",
    "Lou",
    "Naël",
    "Ambre",
    "Liam",
    "Juliette",
    "Maël",
    "Nina",
    "Aaron",
    "Éva",
    "Timéo",
]
LAST_NAMES = [
    "Martin",
    "Bernard",
    "Dubois",
    "Thomas",
    "Robert",
    "Richard",
    "Petit",
    "Durand",
    "Leroy",
    "Moreau",
    "Simon",
    "Laurent",
    "Lefebvre",
    "Michel",
    "Garcia",
    "David",
    "Bertrand",
    "Roux",
    "Vincent",
    "Fournier",
    "Morel",
    "Girard",
    "André",
    "Lefevre",
    "Mercier",
    "Dupont",
    "Lambert",
    "Bonnet",
    "François",
    "Martinez",
    "Legrand",
    "Garnier",
    "Faure",
    "Rousseau",
    "Blanc",
    "Guerin",
    "Muller",
    "Henry",
    "Roussel",
    "Nicolas",
]

_C = PositionCategory
# (nom, catégorie, lieu, _ inutilisé)
POSITIONS: list[tuple[str, PositionCategory, str, list[str]]] = [
    ("Bar central", _C.BAR, "Chapiteau", []),
    ("Bar scène", _C.BAR, "Grande scène", []),
    ("Billetterie entrée", _C.TICKETING, "Entrée principale", []),
    ("Billetterie navette", _C.TICKETING, "Parking P2", []),
    ("Accueil artistes", _C.WELCOME, "Backstage", []),
    ("Accueil public", _C.WELCOME, "Entrée principale", []),
    ("Parking P1", _C.PARKING, "Entrée nord", ["PERMIS_B"]),
    ("Parking P2", _C.PARKING, "Entrée sud", ["PERMIS_B"]),
    ("Propreté site", _C.CLEANING, "Ensemble du site", []),
    ("Propreté sanitaires", _C.CLEANING, "Zone sanitaires", []),
    ("Catering bénévoles", _C.CATERING, "Tente staff", []),
    ("Catering artistes", _C.CATERING, "Backstage", []),
    ("Montage structures", _C.SETUP, "Ensemble du site", ["CACES"]),
    ("Régie grande scène", _C.STAGE, "Grande scène", ["REGIE_SON"]),
    ("Poste de secours", _C.OTHER, "Infirmerie", ["SST"]),
]

# fenêtres horaires (heure de début, heure de fin ; > 24 ⇒ le lendemain)
_WINDOWS: dict[PositionCategory, dict[str, list[tuple[int, int]]]] = {
    _C.BAR: {"montage": [(16, 20)], "festival": [(12, 17), (17, 23)], "demontage": []},
    _C.TICKETING: {"montage": [(14, 18)], "festival": [(9, 15), (15, 21)], "demontage": [(9, 13)]},
    _C.WELCOME: {"montage": [(13, 18)], "festival": [(9, 15), (15, 21)], "demontage": [(9, 13)]},
    _C.PARKING: {"montage": [(9, 13)], "festival": [(8, 14), (14, 20)], "demontage": [(9, 15)]},
    _C.CLEANING: {
        "montage": [(8, 12)],
        "festival": [(7, 11), (23, 27)],
        "demontage": [(8, 14), (14, 18)],
    },
    _C.CATERING: {
        "montage": [(7, 10), (11, 15), (18, 22)],
        "festival": [(7, 10), (11, 15), (18, 22)],
        "demontage": [(7, 10), (11, 15)],
    },
    _C.SETUP: {
        "montage": [(8, 12), (13, 18)],
        "festival": [(9, 12)],
        "demontage": [(9, 13), (14, 18)],
    },
    _C.STAGE: {"montage": [(10, 16)], "festival": [(14, 19), (19, 26)], "demontage": [(10, 15)]},
    _C.OTHER: {"montage": [(9, 17)], "festival": [(12, 20), (20, 28)], "demontage": [(9, 15)]},
}
_STAFF: dict[PositionCategory, tuple[int, int]] = {
    _C.BAR: (3, 6),
    _C.TICKETING: (2, 3),
    _C.WELCOME: (2, 4),
    _C.PARKING: (2, 4),
    _C.CLEANING: (2, 5),
    _C.CATERING: (2, 4),
    _C.SETUP: (3, 6),
    _C.STAGE: (1, 2),
    _C.OTHER: (1, 2),
}

# Programmation : trois passages par jour de festival (pas montage/démontage).
ARTIST_NAMES = [
    "La Battante", "Orchestre Miroir", "Kntongo", "Louve",
    "Les Frères Vent", "Nuage Bleu", "Sœur Machine", "Comète",
]
ARTIST_SLOTS: list[tuple[dt.time, dt.time]] = [
    (dt.time(18, 0), dt.time(19, 0)),
    (dt.time(19, 30), dt.time(21, 0)),
    (dt.time(21, 30), dt.time(23, 0)),
]

# Adresses postales des bénévoles « mécénat » — voir `_add_demo_payments`.
MECENAT_ADDRESSES = [
    ("12 rue des Lilas", "75020", "Paris"),
    ("3 impasse du Petit Bois", "44000", "Nantes"),
    ("8 avenue de la République", "69003", "Lyon"),
    ("21 chemin des Vignes", "33000", "Bordeaux"),
]


@dataclass
class Summary:
    volunteers: int = 0
    accounts: int = 0
    positions: int = 0
    shifts: int = 0
    understaffed_shifts: int = 0
    assignments: int = 0
    artists: int = 0
    position_wishes: int = 0
    artist_wishes: int = 0
    companion_wishes: int = 0
    fee_paid_classique: int = 0
    fee_paid_mecenat: int = 0
    blocking: int = 0

    def __str__(self) -> str:
        lines = [
            "Jeu de démonstration « Festival des Lucioles » 2026",
            f"  Bénévoles              : {self.volunteers}",
            f"  Comptes (connexion)   : {self.accounts}",
            f"  Postes                : {self.positions}",
            f"  Créneaux              : {self.shifts}",
            f"  Créneaux sous l'effectif minimal (trous) : {self.understaffed_shifts}",
            f"  Affectations           : {self.assignments}",
            f"  Artistes programmés    : {self.artists}",
            f"  Vœux de poste déclarés : {self.position_wishes}",
            f"  Vœux d'artiste déclarés : {self.artist_wishes}",
            f"  Vœux de compagnon déclarés : {self.companion_wishes}",
            f"  Cotisation payée (classique) : {self.fee_paid_classique}",
            f"  Cotisation payée (mécénat)   : {self.fee_paid_mecenat}",
        ]
        lines.append(f"  Violations bloquantes : {self.blocking}")
        lines.append("")
        lines.append(
            "  Connexion coordination (mot de passe) : "
            "coordination@lucioles.test / " + DEMO_COORDINATION_PASSWORD
        )
        lines.append(
            "  Référents et bénévoles : lien magique "
            "(referent0..2@lucioles.test, <prénom>.<nom><n>@example.test)"
        )
        return "\n".join(lines)


def _slug(value: str) -> str:
    return normalize(value).replace(" ", "").replace("-", "")


def _overlap(a: Shift, b: Shift) -> bool:
    return a.starts_at < b.ends_at and b.starts_at < a.ends_at


def _wipe(session: Session) -> None:
    for edition in session.exec(select(Edition).where(Edition.name == DEMO_EDITION_NAME)):
        session.delete(edition)
    session.flush()


def build(session: Session, *, rng: random.Random | None = None) -> Summary:
    rng = rng or random.Random(SEED)
    _wipe(session)

    edition = Edition(
        year=DEMO_YEAR,
        name=DEMO_EDITION_NAME,
        start_date=START,
        end_date=END,
        retention_days=365,
    )
    session.add(edition)
    session.flush()

    day_parts = [
        DayPart(
            edition_id=edition.id,
            code=code,
            label=label,
            start_time=start,
            end_time=end,
            order=order,
        )
        for order, (code, label, start, end) in enumerate(
            [
                ("MORNING", "Matin", dt.time(6, 0), dt.time(13, 0)),
                ("AFTERNOON", "Après-midi", dt.time(13, 0), dt.time(19, 0)),
                ("EVENING", "Soirée", dt.time(19, 0), dt.time(6, 0)),
            ]
        )
    ]
    session.add_all(day_parts)
    session.flush()

    coordination = Account(
        edition_id=edition.id,
        email="coordination@lucioles.test",
        role=Role.COORDINATOR,
        password_hash=hash_password(DEMO_COORDINATION_PASSWORD),
    )
    leads = [
        Account(
            edition_id=edition.id,
            email=f"referent{i}@lucioles.test",
            role=Role.LEAD,
        )
        for i in range(3)
    ]
    session.add_all([coordination, *leads])
    session.flush()

    positions: list[Position] = []
    for index, (name, category, location, _required) in enumerate(POSITIONS):
        position = Position(
            edition_id=edition.id,
            name=name,
            category=category,
            location=location,
        )
        if index in (12, 13, 14):
            position.lead_account_id = leads[index - 12].id
        session.add(position)
        session.flush()
        positions.append(position)

    dates = [START + dt.timedelta(days=k) for k in range(4)]
    phase_of = {
        dates[0]: "montage",
        dates[1]: "festival",
        dates[2]: "festival",
        dates[3]: "demontage",
    }

    artists: list[Artist] = []
    for day in dates:
        if phase_of[day] != "festival":
            continue
        for start, end in ARTIST_SLOTS:
            artist = Artist(
                edition_id=edition.id,
                name=ARTIST_NAMES[len(artists) % len(ARTIST_NAMES)],
                performance_date=day,
                start_time=start,
                end_time=end,
            )
            session.add(artist)
            artists.append(artist)
    session.flush()

    volunteers: list[Volunteer] = []
    used_names: set[tuple[str, str]] = set()
    n_position_wishes = 0
    n_artist_wishes = 0
    for i in range(120):
        while True:
            first = rng.choice(FIRST_NAMES)
            last = rng.choice(LAST_NAMES)
            if (first, last) not in used_names:
                used_names.add((first, last))
                break
        email = f"{_slug(first)}.{_slug(last)}{i}@example.test" if rng.random() < 0.85 else None
        volunteer = Volunteer(
            edition_id=edition.id,
            first_name=first,
            last_name=last,
            email=email,
            phone=(f"06{rng.randint(10_000_000, 99_999_999)}" if rng.random() < 0.8 else None),
        )
        session.add(volunteer)
        session.flush()
        volunteers.append(volunteer)
        if email is not None:
            session.add(
                Account(
                    edition_id=edition.id,
                    email=email,
                    role=Role.VOLUNTEER,
                    volunteer_id=volunteer.id,
                )
            )
        # Vœux : poste qu'il voudrait tenir, artistes qu'il veut voir.
        if rng.random() < 0.4:
            session.add(
                VolunteerPositionWish(
                    volunteer_id=volunteer.id,
                    position_id=rng.choice(positions).id,
                )
            )
            n_position_wishes += 1
        for artist in artists:
            if rng.random() < 0.15:
                session.add(
                    VolunteerArtist(volunteer_id=volunteer.id, artist_id=artist.id)
                )
                n_artist_wishes += 1
    # Vœux de compagnon (couple, amis) : à sens unique, choisis parmi les
    # autres bénévoles déjà créés.
    n_companion_wishes = 0
    for volunteer in volunteers:
        others = [v for v in volunteers if v.id != volunteer.id]
        roll = rng.random()
        n_wishes = 2 if roll < 0.03 else (1 if roll < 0.23 else 0)
        for companion in rng.sample(others, k=min(n_wishes, len(others))):
            session.add(
                VolunteerCompanionWish(
                    volunteer_id=volunteer.id, companion_id=companion.id
                )
            )
            n_companion_wishes += 1
    shifts: list[Shift] = []
    for position, (_n, category, _l, _r) in zip(positions, POSITIONS, strict=True):
        for day in dates:
            for start_hour, end_hour in _WINDOWS[category][phase_of[day]]:
                min_staff, ideal_staff = _STAFF[category]
                min_staff = max(1, min_staff + rng.choice([-1, 0, 0, 1]))
                ideal_staff = max(min_staff + 1, ideal_staff + rng.choice([-1, 0, 1]))
                shift = Shift(
                    edition_id=edition.id,
                    position_id=position.id,
                    starts_at=dt.datetime.combine(day, dt.time(start_hour, 0)),
                    ends_at=dt.datetime.combine(day, dt.time(0, 0)) + dt.timedelta(hours=end_hour),
                    min_staff=min_staff,
                    ideal_staff=ideal_staff,
                )
                session.add(shift)
                session.flush()
                shifts.append(shift)

    assigned: dict[object, list[Shift]] = {v.id: [] for v in volunteers}

    n_assignments = 0
    order = shifts[:]
    rng.shuffle(order)
    for index, shift in enumerate(order):
        if index % 16 == 0:  # ~6 créneaux : trou d'effectif volontaire
            target = max(0, shift.min_staff - 1)
        else:
            target = rng.randint(shift.min_staff, shift.ideal_staff)
        pool = volunteers[:]
        rng.shuffle(pool)
        placed = 0
        for volunteer in pool:
            if placed >= target:
                break
            if any(_overlap(shift, s) for s in assigned[volunteer.id]):
                continue
            session.add(
                Assignment(
                    edition_id=edition.id,
                    volunteer_id=volunteer.id,
                    shift_id=shift.id,
                    source=AssignmentSource.MANUAL,
                )
            )
            assigned[volunteer.id].append(shift)
            placed += 1
            n_assignments += 1

    n_fee_classique, n_fee_mecenat = _add_demo_payments(session, volunteers, rng)
    session.flush()

    fill: Counter = Counter()
    for volunteer_shifts in assigned.values():
        for shift in volunteer_shifts:
            fill[shift.id] += 1

    session.flush()
    account_count = len(
        session.exec(
            select(Account).where(Account.edition_id == edition.id)
        ).all()
    )

    summary = Summary(
        volunteers=len(volunteers),
        accounts=account_count,
        positions=len(positions),
        shifts=len(shifts),
        assignments=n_assignments,
        understaffed_shifts=sum(1 for shift in shifts if fill[shift.id] < shift.min_staff),
        artists=len(artists),
        position_wishes=n_position_wishes,
        artist_wishes=n_artist_wishes,
        companion_wishes=n_companion_wishes,
        fee_paid_classique=n_fee_classique,
        fee_paid_mecenat=n_fee_mecenat,
    )
    _verify(session, edition, summary)
    return summary


def _add_demo_payments(
    session: Session, volunteers: list[Volunteer], rng: random.Random
) -> tuple[int, int]:
    """Quelques bénévoles ayant payé leur cotisation via SumUp — voir
    CLAUDE.md § 2, Paiement de la cotisation. `fee_payment_label`/
    `fee_amount` sont une copie figée (comme en production) : aucun
    `FormPaymentOption` n'a besoin d'exister pour autant. Pour que l'export
    Excel et les reçus fiscaux mécénat (`/editions/{id}/volunteers/
    mecenat-receipts`) aient de quoi montrer quelque chose en démonstration."""
    pool = rng.sample(volunteers, k=12)
    classique, mecenat = pool[:8], pool[8:]

    for volunteer in classique:
        volunteer.fee_payment_label = "Cotisation classique"
        volunteer.fee_amount = 15.0
        volunteer.fee_payment_status = FeePaymentStatus.PAID
        volunteer.fee_paid_at = dt.datetime.combine(
            START - dt.timedelta(days=rng.randint(5, 45)), dt.time(rng.randint(9, 21))
        )
        session.add(volunteer)

    for i, volunteer in enumerate(mecenat):
        address, postal_code, city = MECENAT_ADDRESSES[i % len(MECENAT_ADDRESSES)]
        volunteer.fee_payment_label = "Mécénat"
        volunteer.fee_amount = rng.choice([30.0, 50.0, 80.0, 120.0])
        volunteer.fee_payment_status = FeePaymentStatus.PAID
        volunteer.fee_paid_at = dt.datetime.combine(
            START - dt.timedelta(days=rng.randint(5, 45)), dt.time(rng.randint(9, 21))
        )
        volunteer.mecenat_address = address
        volunteer.mecenat_postal_code = postal_code
        volunteer.mecenat_city = city
        volunteer.mecenat_country = "France"
        session.add(volunteer)

    return len(classique), len(mecenat)


def _verify(session: Session, edition: Edition, summary: Summary) -> None:
    session.flush()
    assignments = session.exec(
        select(Assignment).where(Assignment.edition_id == edition.id)
    )
    for assignment in assignments:
        volunteer = session.get(Volunteer, assignment.volunteer_id)
        shift = session.get(Shift, assignment.shift_id)
        for violation in validate(
            build_validation_input(
                session=session,
                edition=edition,
                volunteer=volunteer,
                shift=shift,
                exclude_assignment_id=assignment.id,
            )
        ):
            if violation.severity is Severity.BLOCKING:
                summary.blocking += 1


def main() -> None:
    with Session(engine) as session:
        summary = build(session)
        session.commit()
    print(summary)


if __name__ == "__main__":
    main()