Étape 6 : jeu de données de démonstration
- app/seed.py, exécutable via python -m app.seed, idempotent et déterministe (graine fixe), sans donnée nominative réelle - Édition « Festival des Lucioles » 2026 : 120 bénévoles, 15 postes, 4 jours (montage/festival/démontage), 3 demi-journées, 8 compétences, ~100 créneaux, ~280 affectations actives, disponibilités tri-état - ~12 créneaux volontairement sous l'effectif minimal (trous d'effectif) - 3 affectations à avertissement (compétence absente, hors disponibilité, repos insuffisant), 0 violation bloquante - Passe de vérification finale via le moteur de validation : le résumé imprime le décompte des avertissements par règle et le nombre de bloquantes - covered_slots rendue publique dans validation.engine (réutilisée par le seed) - 2 tests (dataset réaliste, déterminisme/idempotence) ; 106 tests au total Fin du premier incrément backend. 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
74e35f0195
commit
49ac6a4726
@@ -0,0 +1,674 @@
|
||||
"""Jeu de données de démonstration.
|
||||
|
||||
docker compose run --rm backend uv run python -m app.seed
|
||||
|
||||
Idempotent : supprime l'édition de démonstration existante puis la recrée.
|
||||
Déterministe (graine fixe). Aucune donnée nominative réelle.
|
||||
"""
|
||||
|
||||
import datetime as dt
|
||||
import random
|
||||
from collections import Counter
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from app.db import engine
|
||||
from app.importing.text import normalize
|
||||
from app.models import (
|
||||
Account,
|
||||
Assignment,
|
||||
AssignmentSource,
|
||||
AssignmentStatus,
|
||||
Availability,
|
||||
AvailabilityStatus,
|
||||
DayPart,
|
||||
Edition,
|
||||
Position,
|
||||
PositionCategory,
|
||||
Role,
|
||||
Shift,
|
||||
ShiftRequiredSkill,
|
||||
Skill,
|
||||
TshirtSize,
|
||||
Volunteer,
|
||||
VolunteerSkill,
|
||||
)
|
||||
from app.validation import (
|
||||
AvailabilityDeclaration,
|
||||
DayPartWindow,
|
||||
Severity,
|
||||
ShiftInfo,
|
||||
ValidationInput,
|
||||
validate,
|
||||
)
|
||||
from app.validation.engine import covered_slots
|
||||
from app.validation.loader import build_validation_input
|
||||
|
||||
SEED = 20260709
|
||||
DEMO_EDITION_NAME = "Festival des Lucioles"
|
||||
DEMO_YEAR = 2026
|
||||
START = dt.date(2026, 7, 9)
|
||||
END = dt.date(2026, 7, 12)
|
||||
|
||||
FIRST_NAMES = [
|
||||
"Camille",
|
||||
"Léa",
|
||||
"Hugo",
|
||||
"Nathan",
|
||||
"Emma",
|
||||
"Lucas",
|
||||
"Chloé",
|
||||
"Louis",
|
||||
"Manon",
|
||||
"Jules",
|
||||
"Sarah",
|
||||
"Tom",
|
||||
"Inès",
|
||||
"Théo",
|
||||
"Jade",
|
||||
"Raphaël",
|
||||
"Alice",
|
||||
"Gabriel",
|
||||
"Louise",
|
||||
"Adam",
|
||||
"Zoé",
|
||||
"Arthur",
|
||||
"Lina",
|
||||
"Paul",
|
||||
"Rose",
|
||||
"Noah",
|
||||
"Anna",
|
||||
"Ethan",
|
||||
"Mila",
|
||||
"Sacha",
|
||||
"Lou",
|
||||
"Naël",
|
||||
"Ambre",
|
||||
"Liam",
|
||||
"Juliette",
|
||||
"Maël",
|
||||
"Nina",
|
||||
"Aaron",
|
||||
"Éva",
|
||||
"Timéo",
|
||||
]
|
||||
LAST_NAMES = [
|
||||
"Martin",
|
||||
"Bernard",
|
||||
"Dubois",
|
||||
"Thomas",
|
||||
"Robert",
|
||||
"Richard",
|
||||
"Petit",
|
||||
"Durand",
|
||||
"Leroy",
|
||||
"Moreau",
|
||||
"Simon",
|
||||
"Laurent",
|
||||
"Lefebvre",
|
||||
"Michel",
|
||||
"Garcia",
|
||||
"David",
|
||||
"Bertrand",
|
||||
"Roux",
|
||||
"Vincent",
|
||||
"Fournier",
|
||||
"Morel",
|
||||
"Girard",
|
||||
"André",
|
||||
"Lefevre",
|
||||
"Mercier",
|
||||
"Dupont",
|
||||
"Lambert",
|
||||
"Bonnet",
|
||||
"François",
|
||||
"Martinez",
|
||||
"Legrand",
|
||||
"Garnier",
|
||||
"Faure",
|
||||
"Rousseau",
|
||||
"Blanc",
|
||||
"Guerin",
|
||||
"Muller",
|
||||
"Henry",
|
||||
"Roussel",
|
||||
"Nicolas",
|
||||
]
|
||||
|
||||
SKILLS: list[tuple[str, str]] = [
|
||||
("SST", "Sauveteur secouriste du travail"),
|
||||
("PERMIS_B", "Permis B"),
|
||||
("REGIE_SON", "Régie son"),
|
||||
("REGIE_LUM", "Régie lumière"),
|
||||
("CACES", "CACES nacelle"),
|
||||
("ELEC", "Habilitation électrique"),
|
||||
("EN", "Anglais courant"),
|
||||
("ES", "Espagnol courant"),
|
||||
]
|
||||
|
||||
_C = PositionCategory
|
||||
# (nom, catégorie, lieu, compétences requises sur ses créneaux)
|
||||
POSITIONS: list[tuple[str, PositionCategory, str, list[str]]] = [
|
||||
("Bar central", _C.BAR, "Chapiteau", []),
|
||||
("Bar scène", _C.BAR, "Grande scène", []),
|
||||
("Billetterie entrée", _C.TICKETING, "Entrée principale", []),
|
||||
("Billetterie navette", _C.TICKETING, "Parking P2", []),
|
||||
("Accueil artistes", _C.WELCOME, "Backstage", []),
|
||||
("Accueil public", _C.WELCOME, "Entrée principale", []),
|
||||
("Parking P1", _C.PARKING, "Entrée nord", ["PERMIS_B"]),
|
||||
("Parking P2", _C.PARKING, "Entrée sud", ["PERMIS_B"]),
|
||||
("Propreté site", _C.CLEANING, "Ensemble du site", []),
|
||||
("Propreté sanitaires", _C.CLEANING, "Zone sanitaires", []),
|
||||
("Catering bénévoles", _C.CATERING, "Tente staff", []),
|
||||
("Catering artistes", _C.CATERING, "Backstage", []),
|
||||
("Montage structures", _C.SETUP, "Ensemble du site", ["CACES"]),
|
||||
("Régie grande scène", _C.STAGE, "Grande scène", ["REGIE_SON"]),
|
||||
("Poste de secours", _C.OTHER, "Infirmerie", ["SST"]),
|
||||
]
|
||||
|
||||
# fenêtres horaires (heure de début, heure de fin ; > 24 ⇒ le lendemain)
|
||||
_WINDOWS: dict[PositionCategory, dict[str, list[tuple[int, int]]]] = {
|
||||
_C.BAR: {"montage": [(16, 20)], "festival": [(12, 17), (17, 23)], "demontage": []},
|
||||
_C.TICKETING: {"montage": [(14, 18)], "festival": [(9, 15), (15, 21)], "demontage": [(9, 13)]},
|
||||
_C.WELCOME: {"montage": [(13, 18)], "festival": [(9, 15), (15, 21)], "demontage": [(9, 13)]},
|
||||
_C.PARKING: {"montage": [(9, 13)], "festival": [(8, 14), (14, 20)], "demontage": [(9, 15)]},
|
||||
_C.CLEANING: {
|
||||
"montage": [(8, 12)],
|
||||
"festival": [(7, 11), (23, 27)],
|
||||
"demontage": [(8, 14), (14, 18)],
|
||||
},
|
||||
_C.CATERING: {
|
||||
"montage": [(7, 10), (11, 15), (18, 22)],
|
||||
"festival": [(7, 10), (11, 15), (18, 22)],
|
||||
"demontage": [(7, 10), (11, 15)],
|
||||
},
|
||||
_C.SETUP: {
|
||||
"montage": [(8, 12), (13, 18)],
|
||||
"festival": [(9, 12)],
|
||||
"demontage": [(9, 13), (14, 18)],
|
||||
},
|
||||
_C.STAGE: {"montage": [(10, 16)], "festival": [(14, 19), (19, 26)], "demontage": [(10, 15)]},
|
||||
_C.OTHER: {"montage": [(9, 17)], "festival": [(12, 20), (20, 28)], "demontage": [(9, 15)]},
|
||||
}
|
||||
_STAFF: dict[PositionCategory, tuple[int, int]] = {
|
||||
_C.BAR: (3, 6),
|
||||
_C.TICKETING: (2, 3),
|
||||
_C.WELCOME: (2, 4),
|
||||
_C.PARKING: (2, 4),
|
||||
_C.CLEANING: (2, 5),
|
||||
_C.CATERING: (2, 4),
|
||||
_C.SETUP: (3, 6),
|
||||
_C.STAGE: (1, 2),
|
||||
_C.OTHER: (1, 2),
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Summary:
|
||||
volunteers: int = 0
|
||||
positions: int = 0
|
||||
shifts: int = 0
|
||||
understaffed_shifts: int = 0
|
||||
assignments: int = 0
|
||||
warnings: Counter = field(default_factory=Counter)
|
||||
blocking: int = 0
|
||||
|
||||
def __str__(self) -> str:
|
||||
lines = [
|
||||
"Jeu de démonstration « Festival des Lucioles » 2026",
|
||||
f" Bénévoles : {self.volunteers}",
|
||||
f" Postes : {self.positions}",
|
||||
f" Créneaux : {self.shifts}",
|
||||
f" Créneaux sous l'effectif minimal (trous) : {self.understaffed_shifts}",
|
||||
f" Affectations actives : {self.assignments}",
|
||||
f" Avertissements actifs : {sum(self.warnings.values())}",
|
||||
]
|
||||
for code, count in sorted(self.warnings.items()):
|
||||
lines.append(f" {code} : {count}")
|
||||
lines.append(f" Violations bloquantes : {self.blocking}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _slug(value: str) -> str:
|
||||
return normalize(value).replace(" ", "").replace("-", "")
|
||||
|
||||
|
||||
def _overlap(a: Shift, b: Shift) -> bool:
|
||||
return a.starts_at < b.ends_at and b.starts_at < a.ends_at
|
||||
|
||||
|
||||
def _wipe(session: Session) -> None:
|
||||
for edition in session.exec(select(Edition).where(Edition.name == DEMO_EDITION_NAME)):
|
||||
session.delete(edition)
|
||||
session.flush()
|
||||
|
||||
|
||||
def build(session: Session, *, rng: random.Random | None = None) -> Summary:
|
||||
rng = rng or random.Random(SEED)
|
||||
_wipe(session)
|
||||
|
||||
edition = Edition(
|
||||
year=DEMO_YEAR,
|
||||
name=DEMO_EDITION_NAME,
|
||||
start_date=START,
|
||||
end_date=END,
|
||||
retention_days=365,
|
||||
)
|
||||
session.add(edition)
|
||||
session.flush()
|
||||
|
||||
day_parts = [
|
||||
DayPart(
|
||||
edition_id=edition.id,
|
||||
code=code,
|
||||
label=label,
|
||||
start_time=start,
|
||||
end_time=end,
|
||||
order=order,
|
||||
)
|
||||
for order, (code, label, start, end) in enumerate(
|
||||
[
|
||||
("MORNING", "Matin", dt.time(6, 0), dt.time(13, 0)),
|
||||
("AFTERNOON", "Après-midi", dt.time(13, 0), dt.time(19, 0)),
|
||||
("EVENING", "Soirée", dt.time(19, 0), dt.time(6, 0)),
|
||||
]
|
||||
)
|
||||
]
|
||||
session.add_all(day_parts)
|
||||
session.flush()
|
||||
dp_windows = tuple(
|
||||
DayPartWindow(dp.code, dp.start_time, dp.end_time, dp.order) for dp in day_parts
|
||||
)
|
||||
dp_by_code = {dp.code: dp for dp in day_parts}
|
||||
|
||||
skills = {code: Skill(edition_id=edition.id, code=code, label=label) for code, label in SKILLS}
|
||||
session.add_all(skills.values())
|
||||
session.flush()
|
||||
|
||||
coordination = Account(
|
||||
edition_id=edition.id,
|
||||
email="coordination@lucioles.test",
|
||||
role=Role.COORDINATOR,
|
||||
)
|
||||
leads = [
|
||||
Account(
|
||||
edition_id=edition.id,
|
||||
email=f"referent{i}@lucioles.test",
|
||||
role=Role.LEAD,
|
||||
)
|
||||
for i in range(3)
|
||||
]
|
||||
session.add_all([coordination, *leads])
|
||||
session.flush()
|
||||
|
||||
positions: list[Position] = []
|
||||
for index, (name, category, location, _required) in enumerate(POSITIONS):
|
||||
position = Position(
|
||||
edition_id=edition.id,
|
||||
name=name,
|
||||
category=category,
|
||||
location=location,
|
||||
)
|
||||
if index in (12, 13, 14):
|
||||
position.lead_account_id = leads[index - 12].id
|
||||
session.add(position)
|
||||
session.flush()
|
||||
positions.append(position)
|
||||
|
||||
dates = [START + dt.timedelta(days=k) for k in range(4)]
|
||||
phase_of = {
|
||||
dates[0]: "montage",
|
||||
dates[1]: "festival",
|
||||
dates[2]: "festival",
|
||||
dates[3]: "demontage",
|
||||
}
|
||||
|
||||
volunteers: list[Volunteer] = []
|
||||
vol_skill_codes: dict[object, set[str]] = {}
|
||||
used_names: set[tuple[str, str]] = set()
|
||||
for i in range(120):
|
||||
while True:
|
||||
first = rng.choice(FIRST_NAMES)
|
||||
last = rng.choice(LAST_NAMES)
|
||||
if (first, last) not in used_names:
|
||||
used_names.add((first, last))
|
||||
break
|
||||
email = f"{_slug(first)}.{_slug(last)}{i}@example.test" if rng.random() < 0.85 else None
|
||||
volunteer = Volunteer(
|
||||
edition_id=edition.id,
|
||||
first_name=first,
|
||||
last_name=last,
|
||||
email=email,
|
||||
phone=(f"06{rng.randint(10_000_000, 99_999_999)}" if rng.random() < 0.8 else None),
|
||||
tshirt_size=(rng.choice(list(TshirtSize)) if rng.random() < 0.9 else None),
|
||||
diet=(
|
||||
rng.choice(["Végétarien", "Végétalien", "Sans gluten", "Sans porc"])
|
||||
if rng.random() < 0.25
|
||||
else None
|
||||
),
|
||||
)
|
||||
session.add(volunteer)
|
||||
session.flush()
|
||||
volunteers.append(volunteer)
|
||||
count = rng.choices([0, 1, 2, 3], weights=[0.3, 0.34, 0.24, 0.12])[0]
|
||||
codes = set(rng.sample(list(skills), k=count))
|
||||
# Le permis B est courant : on en garantit chez ~40 % des bénévoles
|
||||
# pour que le parking soit gréable.
|
||||
if i % 5 in (0, 1):
|
||||
codes.add("PERMIS_B")
|
||||
for code in codes:
|
||||
session.add(VolunteerSkill(volunteer_id=volunteer.id, skill_id=skills[code].id))
|
||||
vol_skill_codes[volunteer.id] = codes
|
||||
|
||||
availabilities: dict[object, dict[tuple[dt.date, str], AvailabilityStatus]] = {}
|
||||
for volunteer in volunteers:
|
||||
declared: dict[tuple[dt.date, str], AvailabilityStatus] = {}
|
||||
for day in dates:
|
||||
for dp in day_parts:
|
||||
if rng.random() < 0.72:
|
||||
status = rng.choices(
|
||||
[
|
||||
AvailabilityStatus.AVAILABLE,
|
||||
AvailabilityStatus.IF_NEEDED,
|
||||
AvailabilityStatus.UNAVAILABLE,
|
||||
],
|
||||
weights=[0.78, 0.14, 0.08],
|
||||
)[0]
|
||||
session.add(
|
||||
Availability(
|
||||
edition_id=edition.id,
|
||||
volunteer_id=volunteer.id,
|
||||
date=day,
|
||||
day_part_id=dp.id,
|
||||
status=status,
|
||||
)
|
||||
)
|
||||
declared[(day, dp.code)] = status
|
||||
availabilities[volunteer.id] = declared
|
||||
|
||||
shifts: list[Shift] = []
|
||||
shift_required: dict[object, set[str]] = {}
|
||||
for position, (_n, category, _l, required) in zip(positions, POSITIONS, strict=True):
|
||||
for day in dates:
|
||||
for start_hour, end_hour in _WINDOWS[category][phase_of[day]]:
|
||||
min_staff, ideal_staff = _STAFF[category]
|
||||
min_staff = max(1, min_staff + rng.choice([-1, 0, 0, 1]))
|
||||
ideal_staff = max(min_staff + 1, ideal_staff + rng.choice([-1, 0, 1]))
|
||||
shift = Shift(
|
||||
edition_id=edition.id,
|
||||
position_id=position.id,
|
||||
starts_at=dt.datetime.combine(day, dt.time(start_hour, 0)),
|
||||
ends_at=dt.datetime.combine(day, dt.time(0, 0)) + dt.timedelta(hours=end_hour),
|
||||
min_staff=min_staff,
|
||||
ideal_staff=ideal_staff,
|
||||
)
|
||||
session.add(shift)
|
||||
session.flush()
|
||||
for code in required:
|
||||
session.add(ShiftRequiredSkill(shift_id=shift.id, skill_id=skills[code].id))
|
||||
shifts.append(shift)
|
||||
shift_required[shift.id] = set(required)
|
||||
|
||||
assigned: dict[object, list[Shift]] = {v.id: [] for v in volunteers}
|
||||
|
||||
# Trois bénévoles réservés aux scénarios d'avertissement : exclus de
|
||||
# l'affectation générale pour rester libres.
|
||||
reserved_unskilled = next(
|
||||
v for v in volunteers if not vol_skill_codes[v.id]
|
||||
)
|
||||
reserved = [reserved_unskilled] + [
|
||||
v for v in volunteers if v is not reserved_unskilled
|
||||
][:2]
|
||||
reserved_ids = {v.id for v in reserved}
|
||||
assignable = [v for v in volunteers if v.id not in reserved_ids]
|
||||
|
||||
def _clean(volunteer: Volunteer, shift: Shift) -> bool:
|
||||
data = ValidationInput(
|
||||
candidate_shift=ShiftInfo(
|
||||
id=str(shift.id),
|
||||
starts_at=shift.starts_at,
|
||||
ends_at=shift.ends_at,
|
||||
required_skill_codes=frozenset(shift_required[shift.id]),
|
||||
),
|
||||
volunteer_skill_codes=frozenset(vol_skill_codes[volunteer.id]),
|
||||
existing_shifts=tuple(
|
||||
ShiftInfo(id=str(s.id), starts_at=s.starts_at, ends_at=s.ends_at)
|
||||
for s in assigned[volunteer.id]
|
||||
),
|
||||
availabilities=tuple(
|
||||
AvailabilityDeclaration(day, code, status)
|
||||
for (day, code), status in availabilities[volunteer.id].items()
|
||||
),
|
||||
day_parts=dp_windows,
|
||||
)
|
||||
return not validate(data)
|
||||
|
||||
n_assignments = 0
|
||||
order = shifts[:]
|
||||
rng.shuffle(order)
|
||||
for index, shift in enumerate(order):
|
||||
if index % 16 == 0: # ~6 créneaux : trou d'effectif volontaire
|
||||
target = max(0, shift.min_staff - 1)
|
||||
else:
|
||||
target = rng.randint(shift.min_staff, shift.ideal_staff)
|
||||
pool = assignable[:]
|
||||
rng.shuffle(pool)
|
||||
placed = 0
|
||||
for volunteer in pool:
|
||||
if placed >= target:
|
||||
break
|
||||
if any(_overlap(shift, s) for s in assigned[volunteer.id]):
|
||||
continue
|
||||
if not _clean(volunteer, shift):
|
||||
continue
|
||||
session.add(
|
||||
Assignment(
|
||||
edition_id=edition.id,
|
||||
volunteer_id=volunteer.id,
|
||||
shift_id=shift.id,
|
||||
status=(
|
||||
AssignmentStatus.CONFIRMED
|
||||
if rng.random() < 0.55
|
||||
else AssignmentStatus.PROPOSED
|
||||
),
|
||||
source=AssignmentSource.MANUAL,
|
||||
)
|
||||
)
|
||||
assigned[volunteer.id].append(shift)
|
||||
placed += 1
|
||||
n_assignments += 1
|
||||
|
||||
n_assignments += _add_demo_warnings(
|
||||
session,
|
||||
edition,
|
||||
shifts,
|
||||
reserved,
|
||||
assigned,
|
||||
availabilities,
|
||||
dp_by_code,
|
||||
day_parts,
|
||||
shift_required,
|
||||
)
|
||||
session.flush()
|
||||
|
||||
fill: Counter = Counter()
|
||||
for volunteer_shifts in assigned.values():
|
||||
for shift in volunteer_shifts:
|
||||
fill[shift.id] += 1
|
||||
|
||||
summary = Summary(
|
||||
volunteers=len(volunteers),
|
||||
positions=len(positions),
|
||||
shifts=len(shifts),
|
||||
assignments=n_assignments,
|
||||
understaffed_shifts=sum(1 for shift in shifts if fill[shift.id] < shift.min_staff),
|
||||
)
|
||||
_verify(session, edition, summary)
|
||||
return summary
|
||||
|
||||
|
||||
def _set_availability(session, edition, volunteer, day, day_part, status, availabilities) -> None:
|
||||
row = session.exec(
|
||||
select(Availability)
|
||||
.where(Availability.volunteer_id == volunteer.id)
|
||||
.where(Availability.date == day)
|
||||
.where(Availability.day_part_id == day_part.id)
|
||||
).first()
|
||||
if row is not None:
|
||||
row.status = status
|
||||
else:
|
||||
session.add(
|
||||
Availability(
|
||||
edition_id=edition.id,
|
||||
volunteer_id=volunteer.id,
|
||||
date=day,
|
||||
day_part_id=day_part.id,
|
||||
status=status,
|
||||
)
|
||||
)
|
||||
availabilities[volunteer.id][(day, day_part.code)] = status
|
||||
|
||||
|
||||
def _add_demo_warnings(
|
||||
session,
|
||||
edition,
|
||||
shifts,
|
||||
reserved,
|
||||
assigned,
|
||||
availabilities,
|
||||
dp_by_code,
|
||||
day_parts,
|
||||
shift_required,
|
||||
) -> int:
|
||||
"""Trois affectations qui déclenchent chacune un avertissement, et rien
|
||||
d'autre : compétence absente, hors disponibilité, repos insuffisant.
|
||||
|
||||
`reserved` = [bénévole sans compétence, deux autres], tenus à l'écart de
|
||||
l'affectation générale.
|
||||
"""
|
||||
unskilled, reluctant, tired = reserved
|
||||
added = 0
|
||||
|
||||
# 1. MISSING_REQUIRED_SKILL : sur la régie, un bénévole sans compétence.
|
||||
regie = next(s for s in shifts if "REGIE_SON" in shift_required[s.id])
|
||||
for day, code in covered_slots(regie, day_parts):
|
||||
_set_availability(
|
||||
session,
|
||||
edition,
|
||||
unskilled,
|
||||
day,
|
||||
dp_by_code[code],
|
||||
AvailabilityStatus.AVAILABLE,
|
||||
availabilities,
|
||||
)
|
||||
_add_active(session, edition, unskilled, regie, assigned)
|
||||
added += 1
|
||||
|
||||
# 2. OUTSIDE_AVAILABILITY : un créneau de bar sur une demi-journée déclarée
|
||||
# indisponible.
|
||||
bar = next(
|
||||
s
|
||||
for s in shifts
|
||||
if not shift_required[s.id] and (s.ends_at - s.starts_at) <= dt.timedelta(hours=8)
|
||||
)
|
||||
for offset, (day, code) in enumerate(covered_slots(bar, day_parts)):
|
||||
_set_availability(
|
||||
session,
|
||||
edition,
|
||||
reluctant,
|
||||
day,
|
||||
dp_by_code[code],
|
||||
AvailabilityStatus.UNAVAILABLE if offset == 0 else AvailabilityStatus.AVAILABLE,
|
||||
availabilities,
|
||||
)
|
||||
_add_active(session, edition, reluctant, bar, assigned)
|
||||
added += 1
|
||||
|
||||
# 3. INSUFFICIENT_REST : deux créneaux sans compétence espacés de moins de
|
||||
# 8 h, sur deux journées (pour ne pas déclencher le cumul).
|
||||
pair = _find_rest_pair(shifts, shift_required)
|
||||
if pair is not None:
|
||||
first, second = pair
|
||||
for shift in (first, second):
|
||||
for day, code in covered_slots(shift, day_parts):
|
||||
_set_availability(
|
||||
session,
|
||||
edition,
|
||||
tired,
|
||||
day,
|
||||
dp_by_code[code],
|
||||
AvailabilityStatus.AVAILABLE,
|
||||
availabilities,
|
||||
)
|
||||
_add_active(session, edition, tired, first, assigned)
|
||||
_add_active(session, edition, tired, second, assigned)
|
||||
added += 2
|
||||
|
||||
return added
|
||||
|
||||
|
||||
def _find_rest_pair(shifts: list[Shift], shift_required: dict) -> tuple[Shift, Shift] | None:
|
||||
plain = sorted(
|
||||
(s for s in shifts if not shift_required[s.id]),
|
||||
key=lambda s: s.starts_at,
|
||||
)
|
||||
for i, first in enumerate(plain):
|
||||
for second in plain[i + 1 :]:
|
||||
gap = second.starts_at - first.ends_at
|
||||
if (
|
||||
first.ends_at <= second.starts_at
|
||||
and dt.timedelta(0) < gap < dt.timedelta(hours=8)
|
||||
and first.starts_at.date() != second.starts_at.date()
|
||||
):
|
||||
return first, second
|
||||
return None
|
||||
|
||||
|
||||
def _add_active(session, edition, volunteer, shift, assigned) -> None:
|
||||
session.add(
|
||||
Assignment(
|
||||
edition_id=edition.id,
|
||||
volunteer_id=volunteer.id,
|
||||
shift_id=shift.id,
|
||||
status=AssignmentStatus.CONFIRMED,
|
||||
source=AssignmentSource.MANUAL,
|
||||
)
|
||||
)
|
||||
assigned[volunteer.id].append(shift)
|
||||
|
||||
|
||||
def _verify(session: Session, edition: Edition, summary: Summary) -> None:
|
||||
session.flush()
|
||||
active = session.exec(
|
||||
select(Assignment)
|
||||
.where(Assignment.edition_id == edition.id)
|
||||
.where(Assignment.status.in_((AssignmentStatus.PROPOSED, AssignmentStatus.CONFIRMED)))
|
||||
)
|
||||
for assignment in active:
|
||||
volunteer = session.get(Volunteer, assignment.volunteer_id)
|
||||
shift = session.get(Shift, assignment.shift_id)
|
||||
for violation in validate(
|
||||
build_validation_input(
|
||||
session=session,
|
||||
edition=edition,
|
||||
volunteer=volunteer,
|
||||
shift=shift,
|
||||
exclude_assignment_id=assignment.id,
|
||||
)
|
||||
):
|
||||
if violation.severity is Severity.BLOCKING:
|
||||
summary.blocking += 1
|
||||
else:
|
||||
summary.warnings[violation.code] += 1
|
||||
|
||||
|
||||
def main() -> None:
|
||||
with Session(engine) as session:
|
||||
summary = build(session)
|
||||
session.commit()
|
||||
print(summary)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user