benevoles/backend/app/validation/engine.py

"""Moteur de validation des affectations — cœur du produit.

Fonction pure : une `ValidationInput` en entrée, une liste de `Violation` en
sortie. Règles et sémantique figées : voir CLAUDE.md § 3.
"""

import datetime as dt

from app.validation.models import (
    AvailabilityWindow,
    Severity,
    ShiftInfo,
    ValidationInput,
    Violation,
    ViolationCode,
)

_EIGHT_HOURS = dt.timedelta(hours=8)


def validate(data: ValidationInput) -> list[Violation]:
    violations: list[Violation] = []
    violations.extend(_check_overlap(data))
    violations.extend(_check_outside_availability(data))
    violations.extend(_check_daily_hours(data))
    violations.extend(_check_insufficient_rest(data))
    return violations


def has_blocking(violations: list[Violation]) -> bool:
    return any(v.severity is Severity.BLOCKING for v in violations)


# --------------------------------------------------------------------------- #
# SHIFT_OVERLAP (bloquante)                                                    #
# --------------------------------------------------------------------------- #
def _overlaps(a: ShiftInfo, b: ShiftInfo) -> bool:
    # Bornes strictes : deux créneaux jointifs (fin == début) ne se chevauchent
    # pas.
    return a.starts_at < b.ends_at and b.starts_at < a.ends_at


def _check_overlap(data: ValidationInput) -> list[Violation]:
    candidate = data.candidate_shift
    return [
        Violation.of(
            ViolationCode.SHIFT_OVERLAP,
            other_shift_id=other.id,
            other_starts_at=other.starts_at.isoformat(),
            other_ends_at=other.ends_at.isoformat(),
        )
        for other in data.existing_shifts
        if _overlaps(candidate, other)
    ]


# --------------------------------------------------------------------------- #
# OUTSIDE_AVAILABILITY (avertissement)                                        #
# --------------------------------------------------------------------------- #
def _window_interval(w: AvailabilityWindow) -> tuple[dt.datetime, dt.datetime]:
    start = dt.datetime.combine(w.date, w.start_time)
    end_day = w.date if w.end_time > w.start_time else w.date + dt.timedelta(days=1)
    return start, dt.datetime.combine(end_day, w.end_time)


def uncovered_minutes(
    shift: ShiftInfo, windows: tuple[AvailabilityWindow, ...]
) -> int:
    """Minutes du créneau qui ne tombent dans aucune fenêtre de disponibilité.

    Le bénévole peut déclarer plusieurs fenêtres ; on regarde leur union.
    """
    cursor = shift.starts_at
    intervals = sorted(
        (iv for iv in map(_window_interval, windows) if iv[1] > iv[0]),
        key=lambda iv: iv[0],
    )
    gap = dt.timedelta()
    for start, end in intervals:
        if end <= cursor:
            continue
        if start > shift.ends_at:
            break
        if start > cursor:
            gap += min(start, shift.ends_at) - cursor
        cursor = max(cursor, end)
        if cursor >= shift.ends_at:
            return int(gap.total_seconds() // 60)
    if cursor < shift.ends_at:
        gap += shift.ends_at - cursor
    return int(gap.total_seconds() // 60)


def _check_outside_availability(data: ValidationInput) -> list[Violation]:
    missing = uncovered_minutes(data.candidate_shift, data.availabilities)
    if missing == 0:
        return []
    return [
        Violation.of(
            ViolationCode.OUTSIDE_AVAILABILITY,
            date=data.candidate_shift.starts_at.date().isoformat(),
            uncovered_minutes=missing,
        )
    ]


# --------------------------------------------------------------------------- #
# DAILY_HOURS_EXCEEDED (avertissement)                                        #
# --------------------------------------------------------------------------- #
def day_boundary(day_parts) -> dt.time:
    """Début d'une « journée de festival » : début du premier créneau type
    (`DayPart`, par `order`), minuit à défaut — voir CLAUDE.md § 3. Accepte
    aussi bien les `DayPartInfo` de ce module que les lignes `DayPart`."""
    day_parts = list(day_parts)
    if not day_parts:
        return dt.time(0, 0)
    return min(day_parts, key=lambda part: part.order).start_time


def festival_day(moment: dt.datetime, boundary: dt.time) -> dt.date:
    """Journée de festival d'un instant : avant `boundary`, c'est encore la
    nuit de la veille (ex. 01:00 le dimanche ⇒ samedi). Partagé avec le
    solveur et les feuilles de poste, pour qu'une soirée qui passe minuit
    reste d'un seul tenant partout."""
    if moment.time() >= boundary:
        return moment.date()
    return moment.date() - dt.timedelta(days=1)


def _check_daily_hours(data: ValidationInput) -> list[Violation]:
    boundary = day_boundary(data.day_parts)
    candidate = data.candidate_shift
    key = festival_day(candidate.starts_at, boundary)

    total = candidate.ends_at - candidate.starts_at
    for other in data.existing_shifts:
        if festival_day(other.starts_at, boundary) == key:
            total += other.ends_at - other.starts_at

    if total > _EIGHT_HOURS:
        return [
            Violation.of(
                ViolationCode.DAILY_HOURS_EXCEEDED,
                date=key.isoformat(),
                total_minutes=int(total.total_seconds() // 60),
            )
        ]
    return []


# --------------------------------------------------------------------------- #
# INSUFFICIENT_REST (avertissement)                                          #
# --------------------------------------------------------------------------- #
def _check_insufficient_rest(data: ValidationInput) -> list[Violation]:
    candidate = data.candidate_shift
    out: list[Violation] = []
    for other in data.existing_shifts:
        if _overlaps(candidate, other):
            continue
        if other.ends_at <= candidate.starts_at:
            gap = candidate.starts_at - other.ends_at
            direction = "BEFORE"
        elif candidate.ends_at <= other.starts_at:
            gap = other.starts_at - candidate.ends_at
            direction = "AFTER"
        else:  # pragma: no cover - inatteignable : non-chevauchement ⇒ avant ou après
            continue
        if gap < _EIGHT_HOURS:
            out.append(
                Violation.of(
                    ViolationCode.INSUFFICIENT_REST,
                    other_shift_id=other.id,
                    direction=direction,
                    gap_minutes=int(gap.total_seconds() // 60),
                )
            )
    return out