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, field

from sqlmodel import Session, select

from app.db import engine
from app.importing.text import normalize
from app.models import (
    Account,
    Assignment,
    AssignmentSource,
    AssignmentStatus,
    Availability,
    AvailabilityStatus,
    DayPart,
    Edition,
    Position,
    PositionCategory,
    Role,
    Shift,
    ShiftRequiredSkill,
    Skill,
    TshirtSize,
    Volunteer,
    VolunteerSkill,
)
from app.validation import (
    AvailabilityDeclaration,
    DayPartWindow,
    Severity,
    ShiftInfo,
    ValidationInput,
    validate,
)
from app.validation.engine import covered_slots
from app.validation.loader import build_validation_input

SEED = 20260709
DEMO_EDITION_NAME = "Festival des Lucioles"
DEMO_YEAR = 2026
START = dt.date(2026, 7, 9)
END = dt.date(2026, 7, 12)

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",
]

SKILLS: list[tuple[str, str]] = [
    ("SST", "Sauveteur secouriste du travail"),
    ("PERMIS_B", "Permis B"),
    ("REGIE_SON", "Régie son"),
    ("REGIE_LUM", "Régie lumière"),
    ("CACES", "CACES nacelle"),
    ("ELEC", "Habilitation électrique"),
    ("EN", "Anglais courant"),
    ("ES", "Espagnol courant"),
]

_C = PositionCategory
# (nom, catégorie, lieu, compétences requises sur ses créneaux)
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),
}


@dataclass
class Summary:
    volunteers: int = 0
    accounts: int = 0
    positions: int = 0
    shifts: int = 0
    understaffed_shifts: int = 0
    assignments: int = 0
    warnings: Counter = field(default_factory=Counter)
    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 actives  : {self.assignments}",
            f"  Avertissements actifs : {sum(self.warnings.values())}",
        ]
        for code, count in sorted(self.warnings.items()):
            lines.append(f"      {code} : {count}")
        lines.append(f"  Violations bloquantes : {self.blocking}")
        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()
    dp_windows = tuple(
        DayPartWindow(dp.code, dp.start_time, dp.end_time, dp.order) for dp in day_parts
    )
    dp_by_code = {dp.code: dp for dp in day_parts}

    skills = {code: Skill(edition_id=edition.id, code=code, label=label) for code, label in SKILLS}
    session.add_all(skills.values())
    session.flush()

    coordination = Account(
        edition_id=edition.id,
        email="coordination@lucioles.test",
        role=Role.COORDINATOR,
    )
    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",
    }

    volunteers: list[Volunteer] = []
    vol_skill_codes: dict[object, set[str]] = {}
    used_names: set[tuple[str, str]] = set()
    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),
            tshirt_size=(rng.choice(list(TshirtSize)) if rng.random() < 0.9 else None),
            diet=(
                rng.choice(["Végétarien", "Végétalien", "Sans gluten", "Sans porc"])
                if rng.random() < 0.25
                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,
                )
            )
        count = rng.choices([0, 1, 2, 3], weights=[0.3, 0.34, 0.24, 0.12])[0]
        codes = set(rng.sample(list(skills), k=count))
        # Le permis B est courant : on en garantit chez ~40 % des bénévoles
        # pour que le parking soit gréable.
        if i % 5 in (0, 1):
            codes.add("PERMIS_B")
        for code in codes:
            session.add(VolunteerSkill(volunteer_id=volunteer.id, skill_id=skills[code].id))
        vol_skill_codes[volunteer.id] = codes

    availabilities: dict[object, dict[tuple[dt.date, str], AvailabilityStatus]] = {}
    for volunteer in volunteers:
        declared: dict[tuple[dt.date, str], AvailabilityStatus] = {}
        for day in dates:
            for dp in day_parts:
                if rng.random() < 0.72:
                    status = rng.choices(
                        [
                            AvailabilityStatus.AVAILABLE,
                            AvailabilityStatus.IF_NEEDED,
                            AvailabilityStatus.UNAVAILABLE,
                        ],
                        weights=[0.78, 0.14, 0.08],
                    )[0]
                    session.add(
                        Availability(
                            edition_id=edition.id,
                            volunteer_id=volunteer.id,
                            date=day,
                            day_part_id=dp.id,
                            status=status,
                        )
                    )
                    declared[(day, dp.code)] = status
        availabilities[volunteer.id] = declared

    shifts: list[Shift] = []
    shift_required: dict[object, set[str]] = {}
    for position, (_n, category, _l, required) 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()
                for code in required:
                    session.add(ShiftRequiredSkill(shift_id=shift.id, skill_id=skills[code].id))
                shifts.append(shift)
                shift_required[shift.id] = set(required)

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

    # Trois bénévoles réservés aux scénarios d'avertissement : exclus de
    # l'affectation générale pour rester libres.
    reserved_unskilled = next(
        v for v in volunteers if not vol_skill_codes[v.id]
    )
    reserved = [reserved_unskilled] + [
        v for v in volunteers if v is not reserved_unskilled
    ][:2]
    reserved_ids = {v.id for v in reserved}
    assignable = [v for v in volunteers if v.id not in reserved_ids]

    def _clean(volunteer: Volunteer, shift: Shift) -> bool:
        data = ValidationInput(
            candidate_shift=ShiftInfo(
                id=str(shift.id),
                starts_at=shift.starts_at,
                ends_at=shift.ends_at,
                required_skill_codes=frozenset(shift_required[shift.id]),
            ),
            volunteer_skill_codes=frozenset(vol_skill_codes[volunteer.id]),
            existing_shifts=tuple(
                ShiftInfo(id=str(s.id), starts_at=s.starts_at, ends_at=s.ends_at)
                for s in assigned[volunteer.id]
            ),
            availabilities=tuple(
                AvailabilityDeclaration(day, code, status)
                for (day, code), status in availabilities[volunteer.id].items()
            ),
            day_parts=dp_windows,
        )
        return not validate(data)

    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 = assignable[:]
        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
            if not _clean(volunteer, shift):
                continue
            session.add(
                Assignment(
                    edition_id=edition.id,
                    volunteer_id=volunteer.id,
                    shift_id=shift.id,
                    status=(
                        AssignmentStatus.CONFIRMED
                        if rng.random() < 0.55
                        else AssignmentStatus.PROPOSED
                    ),
                    source=AssignmentSource.MANUAL,
                )
            )
            assigned[volunteer.id].append(shift)
            placed += 1
            n_assignments += 1

    n_assignments += _add_demo_warnings(
        session,
        edition,
        shifts,
        reserved,
        assigned,
        availabilities,
        dp_by_code,
        day_parts,
        shift_required,
    )
    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),
    )
    _verify(session, edition, summary)
    return summary


def _set_availability(session, edition, volunteer, day, day_part, status, availabilities) -> None:
    row = session.exec(
        select(Availability)
        .where(Availability.volunteer_id == volunteer.id)
        .where(Availability.date == day)
        .where(Availability.day_part_id == day_part.id)
    ).first()
    if row is not None:
        row.status = status
    else:
        session.add(
            Availability(
                edition_id=edition.id,
                volunteer_id=volunteer.id,
                date=day,
                day_part_id=day_part.id,
                status=status,
            )
        )
    availabilities[volunteer.id][(day, day_part.code)] = status


def _add_demo_warnings(
    session,
    edition,
    shifts,
    reserved,
    assigned,
    availabilities,
    dp_by_code,
    day_parts,
    shift_required,
) -> int:
    """Trois affectations qui déclenchent chacune un avertissement, et rien
    d'autre : compétence absente, hors disponibilité, repos insuffisant.

    `reserved` = [bénévole sans compétence, deux autres], tenus à l'écart de
    l'affectation générale.
    """
    unskilled, reluctant, tired = reserved
    added = 0

    # 1. MISSING_REQUIRED_SKILL : sur la régie, un bénévole sans compétence.
    regie = next(s for s in shifts if "REGIE_SON" in shift_required[s.id])
    for day, code in covered_slots(regie, day_parts):
        _set_availability(
            session,
            edition,
            unskilled,
            day,
            dp_by_code[code],
            AvailabilityStatus.AVAILABLE,
            availabilities,
        )
    _add_active(session, edition, unskilled, regie, assigned)
    added += 1

    # 2. OUTSIDE_AVAILABILITY : un créneau de bar sur une demi-journée déclarée
    #    indisponible.
    bar = next(
        s
        for s in shifts
        if not shift_required[s.id] and (s.ends_at - s.starts_at) <= dt.timedelta(hours=8)
    )
    for offset, (day, code) in enumerate(covered_slots(bar, day_parts)):
        _set_availability(
            session,
            edition,
            reluctant,
            day,
            dp_by_code[code],
            AvailabilityStatus.UNAVAILABLE if offset == 0 else AvailabilityStatus.AVAILABLE,
            availabilities,
        )
    _add_active(session, edition, reluctant, bar, assigned)
    added += 1

    # 3. INSUFFICIENT_REST : deux créneaux sans compétence espacés de moins de
    #    8 h, sur deux journées (pour ne pas déclencher le cumul).
    pair = _find_rest_pair(shifts, shift_required)
    if pair is not None:
        first, second = pair
        for shift in (first, second):
            for day, code in covered_slots(shift, day_parts):
                _set_availability(
                    session,
                    edition,
                    tired,
                    day,
                    dp_by_code[code],
                    AvailabilityStatus.AVAILABLE,
                    availabilities,
                )
        _add_active(session, edition, tired, first, assigned)
        _add_active(session, edition, tired, second, assigned)
        added += 2

    return added


def _find_rest_pair(shifts: list[Shift], shift_required: dict) -> tuple[Shift, Shift] | None:
    plain = sorted(
        (s for s in shifts if not shift_required[s.id]),
        key=lambda s: s.starts_at,
    )
    for i, first in enumerate(plain):
        for second in plain[i + 1 :]:
            gap = second.starts_at - first.ends_at
            if (
                first.ends_at <= second.starts_at
                and dt.timedelta(0) < gap < dt.timedelta(hours=8)
                and first.starts_at.date() != second.starts_at.date()
            ):
                return first, second
    return None


def _add_active(session, edition, volunteer, shift, assigned) -> None:
    session.add(
        Assignment(
            edition_id=edition.id,
            volunteer_id=volunteer.id,
            shift_id=shift.id,
            status=AssignmentStatus.CONFIRMED,
            source=AssignmentSource.MANUAL,
        )
    )
    assigned[volunteer.id].append(shift)


def _verify(session: Session, edition: Edition, summary: Summary) -> None:
    session.flush()
    active = session.exec(
        select(Assignment)
        .where(Assignment.edition_id == edition.id)
        .where(Assignment.status.in_((AssignmentStatus.PROPOSED, AssignmentStatus.CONFIRMED)))
    )
    for assignment in active:
        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
            else:
                summary.warnings[violation.code] += 1


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


if __name__ == "__main__":
    main()