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.models import AvailabilityStatus
from app.validation.models import (
    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))
    violations.extend(_check_missing_skills(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 covered_slots(
    shift: ShiftInfo, day_parts: tuple
) -> list[tuple[dt.date, str]]:
    """Demi-journées (date, code) dont la fenêtre horaire recoupe le créneau."""
    if not day_parts:
        return []
    slots: list[tuple[dt.date, str]] = []
    day = shift.starts_at.date() - dt.timedelta(days=1)
    last = shift.ends_at.date()
    while day <= last:
        for part in day_parts:
            window_start = dt.datetime.combine(day, part.start_time)
            if part.end_time > part.start_time:
                window_end = dt.datetime.combine(day, part.end_time)
            else:
                window_end = dt.datetime.combine(
                    day + dt.timedelta(days=1), part.end_time
                )
            if shift.starts_at < window_end and window_start < shift.ends_at:
                slots.append((day, part.code))
        day += dt.timedelta(days=1)
    return slots


def _check_outside_availability(data: ValidationInput) -> list[Violation]:
    slots = covered_slots(data.candidate_shift, data.day_parts)
    if not slots:
        return []
    declared = {
        (a.date, a.day_part_code): a.status for a in data.availabilities
    }
    out: list[Violation] = []
    for day, code in slots:
        status = declared.get((day, code))
        if status is None:
            out.append(
                Violation.of(
                    ViolationCode.OUTSIDE_AVAILABILITY,
                    date=day.isoformat(),
                    day_part_code=code,
                    status="UNDECLARED",
                )
            )
        elif status is AvailabilityStatus.UNAVAILABLE:
            out.append(
                Violation.of(
                    ViolationCode.OUTSIDE_AVAILABILITY,
                    date=day.isoformat(),
                    day_part_code=code,
                    status="UNAVAILABLE",
                )
            )
    return out


# --------------------------------------------------------------------------- #
# DAILY_HOURS_EXCEEDED (avertissement)                                        #
# --------------------------------------------------------------------------- #
def _day_boundary(day_parts: tuple) -> dt.time:
    if not day_parts:
        return dt.time(0, 0)
    return min(day_parts, key=lambda part: part.order).start_time


def _day_key(moment: dt.datetime, boundary: dt.time) -> dt.date:
    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 = _day_key(candidate.starts_at, boundary)

    total = candidate.ends_at - candidate.starts_at
    for other in data.existing_shifts:
        if _day_key(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


# --------------------------------------------------------------------------- #
# MISSING_REQUIRED_SKILL (avertissement)                                     #
# --------------------------------------------------------------------------- #
def _check_missing_skills(data: ValidationInput) -> list[Violation]:
    missing = sorted(
        data.candidate_shift.required_skill_codes
        - data.volunteer_skill_codes
    )
    return [
        Violation.of(ViolationCode.MISSING_REQUIRED_SKILL, skill_code=code)
        for code in missing
    ]