Étape 4 : moteur de validation des affectations
- Module pur app/validation/ : aucune dépendance à la base, entrée ValidationInput (dataclasses), sortie list[Violation] (code + params, pas de phrase) - 5 règles conformes à CLAUDE.md § 3 : SHIFT_OVERLAP (bloquante), OUTSIDE_AVAILABILITY, DAILY_HOURS_EXCEEDED, INSUFFICIENT_REST, MISSING_REQUIRED_SKILL (avertissements) - Sémantique figée : bornes strictes (créneaux jointifs non chevauchants), frontière de journée au 1er DayPart, cumul > 8 h strict, repos < 8 h strict, statuts PROPOSED/CONFIRMED côté appelant - has_blocking() pour l'appelant - TDD : 23 tests, un cas passant + un cas déclenchant par règle + cas limites (minuit, frontière de journée, multi-créneaux, combinaisons) ; couverture 99 % - Clés de traduction validation.<CODE> dans locales/fr.json Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017uUsRg7XzopkCPWimwg9ZM
This commit is contained in:
AdamThomas
co-authored by
Claude Sonnet 5
Claude Sonnet 5
parent
bb919554ff
commit
c96e85af37
@@ -0,0 +1,192 @@
|
||||
"""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
|
||||
]
|
||||
Reference in New Issue
Block a user