.
This commit is contained in:
Ross
@@ -5,12 +5,270 @@ from .models import Worker, Leave, RotaSchedule
|
||||
import importlib
|
||||
import json
|
||||
|
||||
# Try to import typed helper models from rota_generator to validate complex
|
||||
# worker option structures. If unavailable, fall back to permissive behavior.
|
||||
try:
|
||||
from rota_generator.workers import (
|
||||
NonWorkingDays,
|
||||
OutOfProgramme,
|
||||
NotAvailableToWork,
|
||||
PreferenceNotToWork,
|
||||
WorkRequests,
|
||||
HardDayDependency,
|
||||
HardDayExclusion,
|
||||
)
|
||||
except Exception:
|
||||
NonWorkingDays = OutOfProgramme = NotAvailableToWork = PreferenceNotToWork = None
|
||||
WorkRequests = HardDayDependency = HardDayExclusion = None
|
||||
|
||||
|
||||
class WorkerForm(forms.ModelForm):
|
||||
# Expose core model fields plus a set of richer worker options. Complex
|
||||
# structures (lists/dicts) are edited as JSON in textareas; common simple
|
||||
# options are provided as dedicated form controls for convenience.
|
||||
class Meta:
|
||||
model = Worker
|
||||
fields = ["name", "email", "site", "grade", "fte", "active"]
|
||||
|
||||
# Simple option fields
|
||||
locum = forms.BooleanField(required=False, label="Locum")
|
||||
locum_max_shifts = forms.IntegerField(required=False, label="Locum max shifts")
|
||||
locum_max_shifts_per_week = forms.IntegerField(required=False, label="Locum max shifts per week")
|
||||
locum_on_nwds = forms.BooleanField(required=False, label="Locum on non-working days")
|
||||
remote_site = forms.CharField(required=False, label="Remote site")
|
||||
pair = forms.CharField(required=False, label="Pair")
|
||||
bank_holiday_extra = forms.IntegerField(required=False, label="Bank holiday extra")
|
||||
weekend_shift_target_number = forms.IntegerField(required=False, label="Weekend shift target number")
|
||||
prefer_multi_shift_together = forms.IntegerField(required=False, label="Prefer multi-shift together weight")
|
||||
max_days_worked_per_week = forms.IntegerField(required=False, label="Max days worked per week")
|
||||
|
||||
# Complex structured options edited as JSON
|
||||
assign_as_block_preferences = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON mapping shift_name->weight", label="Block preferences (JSON)")
|
||||
shift_fte_overrides = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON mapping shift_name->fte", label="Shift FTE overrides (JSON)")
|
||||
previous_shifts = forms.CharField(required=False, widget=forms.Textarea, help_text="Free JSON structure for previous shifts", label="Previous shifts (JSON)")
|
||||
shift_balance_extra = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON structure for extra shift-balance penalties", label="Shift balance extra (JSON)")
|
||||
allowed_multi_shift_sets = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON list of sets (e.g. [[\"a\",\"b\"], ...])", label="Allowed multi-shift sets (JSON)")
|
||||
hard_day_dependencies = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON list of HardDayDependency objects", label="Hard day dependencies (JSON)")
|
||||
hard_day_exclusions = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON list of HardDayExclusion objects", label="Hard day exclusions (JSON)")
|
||||
force_assign_with = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON mapping of shift->list", label="Force assign with (JSON)")
|
||||
max_shifts_per_week_by_shift_name = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON mapping shift_name->max", label="Max shifts per week by shift (JSON)")
|
||||
forced_assignments = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON list of tuples (week, day, shift)", label="Forced assignments (JSON)")
|
||||
forced_assignments_by_date = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON list of tuples (date, shift)", label="Forced assignments by date (JSON)")
|
||||
|
||||
# Availability / request lists
|
||||
nwds = forms.CharField(required=False, widget=forms.Textarea, help_text="Non-working days JSON list", label="Non-working days (JSON)")
|
||||
oop = forms.CharField(required=False, widget=forms.Textarea, help_text="Out of programme JSON list", label="Out of programme (JSON)")
|
||||
not_available_to_work = forms.CharField(required=False, widget=forms.Textarea, help_text="Not-available JSON list", label="Not available to work (JSON)")
|
||||
pref_not_to_work = forms.CharField(required=False, widget=forms.Textarea, help_text="Preference not to work JSON list", label="Preferred not to work (JSON)")
|
||||
work_requests = forms.CharField(required=False, widget=forms.Textarea, help_text="Work requests JSON list", label="Work requests (JSON)")
|
||||
locum_availability = forms.CharField(required=False, widget=forms.Textarea, help_text="Locum availability JSON list", label="Locum availability (JSON)")
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
instance = kwargs.get("instance")
|
||||
super().__init__(*args, **kwargs)
|
||||
# Populate initial values for the extra fields from instance.options
|
||||
initial_opts = {}
|
||||
if instance is not None:
|
||||
try:
|
||||
initial_opts = instance.options or {}
|
||||
except Exception:
|
||||
initial_opts = {}
|
||||
|
||||
# Simple field initialisation
|
||||
for fld in [
|
||||
"locum",
|
||||
"locum_max_shifts",
|
||||
"locum_max_shifts_per_week",
|
||||
"locum_on_nwds",
|
||||
"remote_site",
|
||||
"pair",
|
||||
"bank_holiday_extra",
|
||||
"weekend_shift_target_number",
|
||||
"prefer_multi_shift_together",
|
||||
"max_days_worked_per_week",
|
||||
]:
|
||||
if fld in self.fields:
|
||||
self.fields[fld].initial = initial_opts.get(fld, self.fields[fld].initial)
|
||||
|
||||
# JSON fields: pretty-print if present
|
||||
for json_fld in [
|
||||
"assign_as_block_preferences",
|
||||
"shift_fte_overrides",
|
||||
"previous_shifts",
|
||||
"shift_balance_extra",
|
||||
"allowed_multi_shift_sets",
|
||||
"hard_day_dependencies",
|
||||
"hard_day_exclusions",
|
||||
"force_assign_with",
|
||||
"max_shifts_per_week_by_shift_name",
|
||||
"forced_assignments",
|
||||
"forced_assignments_by_date",
|
||||
"nwds",
|
||||
"oop",
|
||||
"not_available_to_work",
|
||||
"pref_not_to_work",
|
||||
"work_requests",
|
||||
"locum_availability",
|
||||
]:
|
||||
if json_fld in self.fields:
|
||||
val = initial_opts.get(json_fld, None)
|
||||
if val is None:
|
||||
self.fields[json_fld].initial = ""
|
||||
else:
|
||||
try:
|
||||
self.fields[json_fld].initial = json.dumps(val, indent=2, default=str)
|
||||
except Exception:
|
||||
self.fields[json_fld].initial = str(val)
|
||||
|
||||
def save(self, commit=True):
|
||||
"""Save Worker model and persist extra options into `instance.options`."""
|
||||
instance = super().save(commit=False)
|
||||
opts = instance.options or {}
|
||||
|
||||
# Simple options
|
||||
simple_opts = [
|
||||
"locum",
|
||||
"locum_max_shifts",
|
||||
"locum_max_shifts_per_week",
|
||||
"locum_on_nwds",
|
||||
"remote_site",
|
||||
"pair",
|
||||
"bank_holiday_extra",
|
||||
"weekend_shift_target_number",
|
||||
"prefer_multi_shift_together",
|
||||
"max_days_worked_per_week",
|
||||
]
|
||||
for k in simple_opts:
|
||||
if k in self.cleaned_data:
|
||||
opts[k] = self.cleaned_data.get(k)
|
||||
|
||||
# JSON fields: parse where possible
|
||||
json_fields = [
|
||||
"assign_as_block_preferences",
|
||||
"shift_fte_overrides",
|
||||
"previous_shifts",
|
||||
"shift_balance_extra",
|
||||
"allowed_multi_shift_sets",
|
||||
"hard_day_dependencies",
|
||||
"hard_day_exclusions",
|
||||
"force_assign_with",
|
||||
"max_shifts_per_week_by_shift_name",
|
||||
"forced_assignments",
|
||||
"forced_assignments_by_date",
|
||||
"nwds",
|
||||
"oop",
|
||||
"not_available_to_work",
|
||||
"pref_not_to_work",
|
||||
"work_requests",
|
||||
"locum_availability",
|
||||
]
|
||||
for k in json_fields:
|
||||
if k in self.cleaned_data:
|
||||
val = self.cleaned_data.get(k)
|
||||
if val is None or val == "":
|
||||
opts[k] = []
|
||||
else:
|
||||
try:
|
||||
parsed = json.loads(val)
|
||||
except Exception:
|
||||
parsed = val
|
||||
opts[k] = parsed
|
||||
|
||||
instance.options = opts
|
||||
if commit:
|
||||
instance.save()
|
||||
try:
|
||||
self.save_m2m()
|
||||
except Exception:
|
||||
pass
|
||||
return instance
|
||||
|
||||
def clean(self):
|
||||
"""Validate and parse typed JSON worker option fields when possible.
|
||||
|
||||
This will parse JSON strings into Python structures and, where we can
|
||||
import the typed helper models from `rota_generator.workers`, will
|
||||
validate each list item against the corresponding Pydantic model.
|
||||
"""
|
||||
cleaned = super().clean()
|
||||
|
||||
def _parse_and_validate(field_name, model_cls=None):
|
||||
val = cleaned.get(field_name)
|
||||
if val is None or val == "":
|
||||
return []
|
||||
# If already parsed (not a str), accept it
|
||||
if not isinstance(val, str):
|
||||
parsed = val
|
||||
else:
|
||||
try:
|
||||
parsed = json.loads(val)
|
||||
except Exception:
|
||||
# leave as raw string to preserve old behaviour
|
||||
parsed = val
|
||||
|
||||
# If a model class is provided, validate each element
|
||||
if model_cls and parsed and isinstance(parsed, (list, tuple)):
|
||||
validated = []
|
||||
for i, item in enumerate(parsed):
|
||||
try:
|
||||
# If item is already an instance, accept it
|
||||
if hasattr(model_cls, "model_dump") or hasattr(model_cls, "__fields__"):
|
||||
# pydantic v2 or v1 compatible
|
||||
validated.append(model_cls(**item) if isinstance(item, dict) else model_cls.parse_obj(item) if hasattr(model_cls, 'parse_obj') else model_cls(**item))
|
||||
else:
|
||||
validated.append(item)
|
||||
except Exception as e:
|
||||
self.add_error(field_name, forms.ValidationError(f"Invalid item in {field_name}[{i}]: {e}"))
|
||||
validated.append(item)
|
||||
return validated
|
||||
|
||||
return parsed
|
||||
|
||||
# Typed list fields
|
||||
cleaned["nwds"] = _parse_and_validate("nwds", NonWorkingDays)
|
||||
cleaned["oop"] = _parse_and_validate("oop", OutOfProgramme)
|
||||
cleaned["not_available_to_work"] = _parse_and_validate("not_available_to_work", NotAvailableToWork)
|
||||
cleaned["pref_not_to_work"] = _parse_and_validate("pref_not_to_work", PreferenceNotToWork)
|
||||
cleaned["work_requests"] = _parse_and_validate("work_requests", WorkRequests)
|
||||
cleaned["locum_availability"] = _parse_and_validate("locum_availability", WorkRequests)
|
||||
|
||||
# Hard day dependency/exclusion structures
|
||||
cleaned["hard_day_dependencies"] = _parse_and_validate("hard_day_dependencies", HardDayDependency)
|
||||
cleaned["hard_day_exclusions"] = _parse_and_validate("hard_day_exclusions", HardDayExclusion)
|
||||
|
||||
# forced assignments: expect list of tuples [week, day, shift]
|
||||
fa = cleaned.get("forced_assignments")
|
||||
if isinstance(fa, str):
|
||||
try:
|
||||
fa_parsed = json.loads(fa)
|
||||
except Exception:
|
||||
fa_parsed = fa
|
||||
else:
|
||||
fa_parsed = fa
|
||||
if isinstance(fa_parsed, list):
|
||||
# basic validation
|
||||
for i, t in enumerate(fa_parsed):
|
||||
if not (isinstance(t, (list, tuple)) and len(t) == 3):
|
||||
self.add_error("forced_assignments", forms.ValidationError(f"forced_assignments[{i}] must be a tuple/list of (week, day, shift)"))
|
||||
cleaned["forced_assignments"] = fa_parsed
|
||||
|
||||
# forced_assignments_by_date: expect list of tuples [date, shift]
|
||||
fabd = cleaned.get("forced_assignments_by_date")
|
||||
if isinstance(fabd, str):
|
||||
try:
|
||||
fabd_parsed = json.loads(fabd)
|
||||
except Exception:
|
||||
fabd_parsed = fabd
|
||||
else:
|
||||
fabd_parsed = fabd
|
||||
if isinstance(fabd_parsed, list):
|
||||
for i, t in enumerate(fabd_parsed):
|
||||
if not (isinstance(t, (list, tuple)) and len(t) == 2):
|
||||
self.add_error("forced_assignments_by_date", forms.ValidationError(f"forced_assignments_by_date[{i}] must be a tuple/list of (date, shift)"))
|
||||
cleaned["forced_assignments_by_date"] = fabd_parsed
|
||||
|
||||
return cleaned
|
||||
|
||||
|
||||
class LeaveForm(forms.ModelForm):
|
||||
class Meta:
|
||||
|
||||
Reference in New Issue
Block a user