"""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) # # --------------------------------------------------------------------------- # _MINUTES_PER_DAY = 24 * 60 def day_boundary(day_parts) -> dt.time: """Début d'une « journée de festival » : début du plus long creux que laissent les créneaux types (`DayPart`) sur le cadran de 24 h — l'heure à laquelle le dernier de la journée se termine —, minuit à défaut — voir CLAUDE.md § 3. Avec Matin 08 h–13 h, Après-midi 13 h–19 h, Soirée 19 h–02 h, le creux 02 h–08 h est le plus long : la journée commence à 02 h, la soirée qui passe minuit reste sur le jour où elle a commencé. Indépendant de `order`, qui n'est qu'un ordre d'affichage et ne suit pas les heures quand on les modifie (un créneau type d'ordre 0 déplacé à 21 h faisait commencer chaque journée à 21 h). Sans creux (les créneaux types couvrent 24 h), début du premier par `order`. Accepte aussi bien les `DayPartInfo` de ce module que les lignes `DayPart`. Miroir front : `lib/planning.ts:dayBoundaryFromDayParts`.""" day_parts = list(day_parts) if not day_parts: return dt.time(0, 0) covered = [False] * _MINUTES_PER_DAY for part in day_parts: start = part.start_time.hour * 60 + part.start_time.minute end = part.end_time.hour * 60 + part.end_time.minute length = (end - start) % _MINUTES_PER_DAY or _MINUTES_PER_DAY for offset in range(length): covered[(start + offset) % _MINUTES_PER_DAY] = True if all(covered): return min(day_parts, key=lambda part: part.order).start_time # Parcours circulaire à partir d'une minute couverte : chaque creux est # vu d'un seul tenant, même s'il enjambe minuit. origin = covered.index(True) best_start, best_length = 0, 0 run_start, run_length = 0, 0 for offset in range(1, _MINUTES_PER_DAY + 1): minute = (origin + offset) % _MINUTES_PER_DAY if covered[minute]: run_length = 0 continue if run_length == 0: run_start = minute run_length += 1 if run_length > best_length: best_start, best_length = run_start, run_length return dt.time(best_start // 60, best_start % 60) 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