benevoles/backend/app/solver/engine.py

"""Solveur de suggestions (OR-Tools CP-SAT) — cœur du produit, second moteur.

Ne place jamais personne : il propose, pour chaque créneau encore sous son
effectif idéal, une liste ordonnée de bénévoles candidats. C'est la
coordination qui affecte (dans l'app, via l'écran habituel — le solveur se
contente de faire remonter ces candidats en tête de liste).

Contrainte dure : aucun chevauchement de créneaux pour un même bénévole (règle
`SHIFT_OVERLAP`, cf. `app.validation`). Tout le reste (disponibilité, repos,
chaînage entre créneaux consécutifs, vœu de poste, vœu de compagnon, vœu
d'artiste, équité de charge) est un terme de l'objectif, pas une contrainte —
comme le moteur de validation, qui reste seul juge au moment où la
coordination confirme une affectation.
Pas de plafond de cumul journalier côté solveur : `DAILY_HOURS_EXCEEDED`
reste vérifiée par la validation, à la confirmation.
"""

import datetime as dt

from ortools.sat.python import cp_model

from app.solver.models import (
    ShiftSuggestion,
    SolverInput,
    SolverShift,
    SolverSolution,
    SolverVolunteer,
)
from app.validation.engine import uncovered_minutes
from app.validation.models import ShiftInfo

_EIGHT_HOURS_MIN = 8 * 60

# Poids de l'objectif (unités arbitraires, cf. docstring de `_score`).
# L'écart à un effectif minimal non atteint prime sur tout le reste : le
# solveur doit toujours proposer quelqu'un pour un créneau sous-staffé, quitte
# à ce que la disponibilité soit imparfaite (avertissement, pas un blocage —
# la validation reste seule juge à la confirmation). Au-delà du minimum, un
# candidat mal assorti peut légitimement ne pas être suggéré du tout.
_URGENT_BASE = 1000  # créneau encore sous son effectif minimal
_NORMAL_BASE = 20  # créneau entre effectif minimal et idéal
_WISH_BONUS = 30
_AVAIL_PENALTY_PER_HALF_HOUR = 5
_REST_PENALTY = 60
_FAIRNESS_PENALTY_PER_HOUR = 2
# Le créneau lui ferait manquer le passage d'un artiste qu'il veut voir.
_ARTIST_CONFLICT_PENALTY = 80
# Le bénévole a déjà tenu le créneau immédiatement précédent (même jour,
# n'importe quel poste) : limite les enchaînements de shifts plutôt que de
# faire tourner un poste en particulier.
_CHAIN_PENALTY = 40
# Par compagnon souhaité déjà affecté au créneau (0 à N vœux par bénévole).
_COMPANION_BONUS_PER_MATCH = 25


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


def _has_insufficient_rest(volunteer: SolverVolunteer, shift: SolverShift) -> bool:
    candidate = shift.shift_info
    for other in volunteer.existing_shifts:
        if _overlaps(candidate, other):
            continue
        if other.ends_at <= candidate.starts_at:
            gap = candidate.starts_at - other.ends_at
        elif candidate.ends_at <= other.starts_at:
            gap = other.starts_at - candidate.ends_at
        else:
            continue
        if gap < dt.timedelta(minutes=_EIGHT_HOURS_MIN):
            return True
    return False


def _chained_from_predecessor(volunteer: SolverVolunteer, shift: SolverShift) -> bool:
    """Le bénévole a déjà (affectation décidée) l'un des créneaux qui
    précèdent immédiatement celui-ci, le même jour — n'importe quel poste :
    il s'agit de limiter les enchaînements de shifts, pas de faire tourner un
    poste en particulier."""
    if not shift.predecessor_shift_ids:
        return False
    return any(s.id in shift.predecessor_shift_ids for s in volunteer.existing_shifts)


def _score(volunteer: SolverVolunteer, shift: SolverShift) -> int:
    """Score d'affinité (bénévole, créneau) : plus haut = meilleur candidat.

    Purement additif (aucune interaction entre deux affectations proposées),
    ce qui garde le modèle CP-SAT linéaire simple. Repos, chaînage et
    compagnons ne sont vérifiés que contre les affectations déjà décidées,
    pas entre deux suggestions du même calcul : la validation finale (à la
    confirmation) reste seule juge en cas de cas limite entre deux
    suggestions (et deux compagnons suggérés ensemble dans le même calcul ne
    se renforcent pas mutuellement, faute d'interaction entre variables).
    """
    score = _URGENT_BASE if shift.already_assigned < shift.min_staff else _NORMAL_BASE
    if shift.position_id in volunteer.wished_position_ids:
        score += _WISH_BONUS
    missed = uncovered_minutes(shift.shift_info, volunteer.availability)
    score -= (missed // 30) * _AVAIL_PENALTY_PER_HALF_HOUR
    if _has_insufficient_rest(volunteer, shift):
        score -= _REST_PENALTY
    if _chained_from_predecessor(volunteer, shift):
        score -= _CHAIN_PENALTY
    companions_present = len(volunteer.wished_companion_ids & shift.assigned_volunteer_ids)
    score += _COMPANION_BONUS_PER_MATCH * companions_present
    if any(
        _overlaps(shift.shift_info, passage) for passage in volunteer.wished_artist_windows
    ):
        score -= _ARTIST_CONFLICT_PENALTY
    score -= (_FAIRNESS_PENALTY_PER_HOUR * volunteer.existing_minutes) // 60
    return score


def solve(
    data: SolverInput,
    *,
    time_limit_seconds: float = 8.0,
    seed: int = 42,
) -> SolverSolution:
    """Calcule les suggestions. Fonction pure : ne touche jamais la base."""
    model = cp_model.CpModel()
    variables: dict[tuple[int, int], cp_model.IntVar] = {}
    scores: dict[tuple[int, int], int] = {}

    for si, shift in enumerate(data.shifts):
        gap = shift.ideal_staff - shift.already_assigned
        if gap <= 0:
            continue
        for vi, volunteer in enumerate(data.volunteers):
            if any(_overlaps(shift.shift_info, other) for other in volunteer.existing_shifts):
                continue
            key = (vi, si)
            variables[key] = model.NewBoolVar(f"x_{vi}_{si}")
            scores[key] = _score(volunteer, shift)

    if not variables:
        return SolverSolution()

    # Effectif idéal comme plafond dur — le solveur ne sur-staffe jamais.
    for si, shift in enumerate(data.shifts):
        gap = shift.ideal_staff - shift.already_assigned
        n_volunteers = len(data.volunteers)
        shift_vars = [variables[vi, si] for vi in range(n_volunteers) if (vi, si) in variables]
        if shift_vars and gap > 0:
            model.Add(sum(shift_vars) <= gap)

    # Aucun chevauchement entre deux créneaux nouvellement suggérés au même
    # bénévole (règle `SHIFT_OVERLAP`, bloquante).
    for vi in range(len(data.volunteers)):
        my_shifts = [si for si in range(len(data.shifts)) if (vi, si) in variables]
        for a in range(len(my_shifts)):
            for b in range(a + 1, len(my_shifts)):
                si_a, si_b = my_shifts[a], my_shifts[b]
                if _overlaps(data.shifts[si_a].shift_info, data.shifts[si_b].shift_info):
                    model.Add(variables[vi, si_a] + variables[vi, si_b] <= 1)

    model.Maximize(sum(score * variables[key] for key, score in scores.items()))

    solver = cp_model.CpSolver()
    solver.parameters.max_time_in_seconds = time_limit_seconds
    solver.parameters.num_search_workers = 1
    solver.parameters.random_seed = seed
    status = solver.Solve(model)

    chosen: dict[int, list[int]] = {}
    if status in (cp_model.OPTIMAL, cp_model.FEASIBLE):
        for (vi, si), var in variables.items():
            if solver.Value(var):
                chosen.setdefault(si, []).append(vi)

    suggestions: list[ShiftSuggestion] = []
    for si, shift in enumerate(data.shifts):
        picked = sorted(chosen.get(si, []), key=lambda vi: -scores[(vi, si)])
        if picked:
            suggestions.append(
                ShiftSuggestion(
                    shift_id=shift.id,
                    volunteer_ids=tuple(data.volunteers[vi].id for vi in picked),
                )
            )
    return SolverSolution(suggestions=tuple(suggestions))