Ross
1678 lines
55 KiB
Python
1678 lines
55 KiB
Python
from django.shortcuts import render, get_object_or_404, redirect
|
|
from django.urls import reverse
|
|
|
|
from .models import RotaSchedule, Worker
|
|
from .forms import WorkerForm, LeaveForm, RotaScheduleForm
|
|
from .forms import WorkerSelfServiceForm
|
|
from .services import run_rota_async
|
|
from .forms import ShiftForm
|
|
import json
|
|
from django.views.decorators.http import require_http_methods
|
|
from django.template.loader import render_to_string
|
|
from django.http import HttpResponse, HttpResponseBadRequest
|
|
from django.http import JsonResponse
|
|
from django.views.decorators.http import require_POST
|
|
from django.shortcuts import HttpResponseRedirect
|
|
import traceback
|
|
|
|
|
|
|
|
def _get_constraint_description(key: str):
|
|
"""Return the description for a constraint key from the typed model metadata.
|
|
|
|
This is defensive because different pydantic versions expose field metadata
|
|
to the Python API slightly differently.
|
|
"""
|
|
if not key:
|
|
return None
|
|
try:
|
|
import importlib
|
|
mod = importlib.import_module("rota_generator.shifts")
|
|
RotaConstraintOptions = getattr(mod, "RotaConstraintOptions", None)
|
|
meta = {}
|
|
if RotaConstraintOptions is not None:
|
|
meta = getattr(RotaConstraintOptions, "model_fields", {}) or {}
|
|
else:
|
|
RotaBuilder = getattr(mod, "RotaBuilder", None)
|
|
if RotaBuilder is not None:
|
|
try:
|
|
rb = RotaBuilder()
|
|
opts_model = getattr(rb, "constraint_options_model", None)
|
|
meta = getattr(opts_model.__class__, "model_fields", {}) or {}
|
|
except Exception:
|
|
meta = {}
|
|
# meta should be a dict-like mapping of field names -> field info
|
|
if not isinstance(meta, dict):
|
|
try:
|
|
meta = dict(meta)
|
|
except Exception:
|
|
meta = {}
|
|
if key not in meta:
|
|
return None
|
|
info = meta[key]
|
|
# Try several access patterns to extract a description
|
|
desc = None
|
|
try:
|
|
if hasattr(info, "description") and info.description:
|
|
desc = info.description
|
|
except Exception:
|
|
desc = None
|
|
if not desc:
|
|
try:
|
|
# dict-like
|
|
desc = info.get("description")
|
|
except Exception:
|
|
desc = None
|
|
if not desc:
|
|
try:
|
|
fi = getattr(info, "field_info", None)
|
|
if fi is not None and getattr(fi, "description", None):
|
|
desc = fi.description
|
|
except Exception:
|
|
desc = None
|
|
if not desc:
|
|
try:
|
|
extra = getattr(info, "extra", None)
|
|
if isinstance(extra, dict) and extra.get("description"):
|
|
desc = extra.get("description")
|
|
except Exception:
|
|
pass
|
|
return desc
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def index(request):
|
|
rotas = RotaSchedule.objects.all().order_by("-created_at")
|
|
return render(request, "rota/index.html", {"rotas": rotas})
|
|
|
|
|
|
def rota_detail(request, rota_id):
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
# Prepare available builder constraint defaults
|
|
available_constraints = {}
|
|
try:
|
|
# Try to import RotaBuilder to surface its default constraint options
|
|
import importlib
|
|
|
|
mod = importlib.import_module("rota_generator.shifts")
|
|
RotaBuilder = getattr(mod, "RotaBuilder")
|
|
try:
|
|
rb = RotaBuilder()
|
|
available_constraints = getattr(rb, "constraint_options", {})
|
|
except Exception:
|
|
# fallback to class attribute if instantiation fails
|
|
available_constraints = {}
|
|
except Exception:
|
|
available_constraints = {}
|
|
if request.method == "POST":
|
|
# trigger a background run
|
|
generate_builder = request.POST.get("generate_builder") == "1"
|
|
builder_solve = request.POST.get("builder_solve") == "1"
|
|
run = run_rota_async(rota.id, generate_builder=generate_builder, builder_solve=builder_solve)
|
|
return redirect(reverse("rota:rota_run_detail", args=(run.id,)))
|
|
|
|
# Build a richer representation of available constraints (default, type, description)
|
|
available_constraints_rich = {}
|
|
try:
|
|
mod = importlib.import_module("rota_generator.shifts")
|
|
# Prefer reading metadata from the typed RotaConstraintOptions class
|
|
RotaConstraintOptions = getattr(mod, "RotaConstraintOptions", None)
|
|
meta = {}
|
|
defaults = {}
|
|
if RotaConstraintOptions is not None:
|
|
# class-level model_fields avoids instantiating RotaBuilder
|
|
try:
|
|
meta = getattr(RotaConstraintOptions, "model_fields", {}) or {}
|
|
except Exception:
|
|
meta = {}
|
|
try:
|
|
# instantiate a model to obtain defaults
|
|
defaults = RotaConstraintOptions().model_dump()
|
|
except Exception:
|
|
defaults = {}
|
|
else:
|
|
# Fallback: try to use RotaBuilder's constraint_options_model if present
|
|
RotaBuilder = getattr(mod, "RotaBuilder", None)
|
|
if RotaBuilder is not None:
|
|
try:
|
|
rb = RotaBuilder()
|
|
opts_model = getattr(rb, "constraint_options_model", None)
|
|
if opts_model is not None:
|
|
defaults = opts_model.model_dump()
|
|
meta = getattr(opts_model.__class__, "model_fields", {}) or {}
|
|
except Exception:
|
|
defaults = {}
|
|
# iterate defaults and build rich info dict
|
|
for k, v in defaults.items():
|
|
# Use unified helper to extract a stable string description for the key
|
|
desc = _get_constraint_description(k)
|
|
# include the current value saved on the rota (if any)
|
|
try:
|
|
current = (rota.options or {}).get(k, v)
|
|
except Exception:
|
|
current = v
|
|
available_constraints_rich[k] = {"default": v, "description": desc, "current": current}
|
|
except Exception:
|
|
available_constraints_rich = {k: {"default": v} for k, v in available_constraints.items()}
|
|
|
|
# Build a mapping of configured constraint keys (from rota.options).
|
|
# Include all keys present in rota.options so they can be displayed and
|
|
# edited individually on the main page, even if the typed defaults are
|
|
# unavailable. Mark whether each key appears in the typed defaults.
|
|
configured_constraints = {}
|
|
try:
|
|
opts = rota.options or {}
|
|
for k, v in opts.items():
|
|
# Only surface keys that correspond to the typed RotaConstraintOptions
|
|
# (exclude builder/constructor args and other non-constraint keys).
|
|
if k not in available_constraints_rich:
|
|
continue
|
|
# pretty-print complex values where appropriate
|
|
try:
|
|
pretty = json.dumps(v, indent=2, default=str)
|
|
except Exception:
|
|
pretty = str(v)
|
|
# Include description from the typed defaults when available so the
|
|
# UI can show tooltips and the modal can display helpful text.
|
|
desc = None
|
|
try:
|
|
desc = available_constraints_rich.get(k, {}).get("description")
|
|
except Exception:
|
|
desc = None
|
|
default_val = available_constraints_rich.get(k, {}).get("default")
|
|
configured_constraints[k] = {
|
|
"value": v,
|
|
"pretty": pretty,
|
|
"typed": (k in available_constraints_rich),
|
|
"description": desc,
|
|
"default": default_val,
|
|
}
|
|
except Exception:
|
|
configured_constraints = {}
|
|
|
|
# Compute a small rota-builder summary for display in the template
|
|
try:
|
|
# Prefer explicit value stored in rota.options['weeks'] when present.
|
|
opts = rota.options or {}
|
|
if "weeks" in opts:
|
|
try:
|
|
weeks_to_rota = max(1, int(opts.get("weeks")))
|
|
except Exception:
|
|
weeks_to_rota = None
|
|
else:
|
|
delta = rota.end_date - rota.start_date
|
|
weeks_to_rota = max(1, int(delta.days // 7))
|
|
except Exception:
|
|
weeks_to_rota = None
|
|
|
|
try:
|
|
shift_count = len(rota.shifts or [])
|
|
except Exception:
|
|
shift_count = None
|
|
|
|
try:
|
|
worker_count = rota.workers.count()
|
|
except Exception:
|
|
worker_count = None
|
|
|
|
# Extract solver settings saved on the rota (if any) for display.
|
|
try:
|
|
opts = rota.options or {}
|
|
solver = opts.get("solver")
|
|
so = opts.get("solver_options")
|
|
# Normalize solver_options to a dict when possible and extract ratio
|
|
solver_options = None
|
|
solver_ratio = None
|
|
if isinstance(so, dict):
|
|
solver_options = so
|
|
solver_ratio = so.get("ratio")
|
|
elif isinstance(so, str):
|
|
try:
|
|
solver_options = json.loads(so)
|
|
except Exception:
|
|
solver_options = so
|
|
if isinstance(solver_options, dict):
|
|
solver_ratio = solver_options.get("ratio")
|
|
# fallback to legacy top-level 'ratio' key
|
|
if solver_ratio is None:
|
|
solver_ratio = opts.get("ratio")
|
|
# Pretty JSON for display when dict
|
|
try:
|
|
if isinstance(solver_options, dict):
|
|
solver_options_pretty = json.dumps(solver_options, indent=2)
|
|
else:
|
|
solver_options_pretty = str(solver_options) if solver_options is not None else None
|
|
except Exception:
|
|
solver_options_pretty = None
|
|
except Exception:
|
|
solver = None
|
|
solver_options_pretty = None
|
|
solver_ratio = None
|
|
|
|
# legacy RotaOptionsForm not used here; we provide a typed RotaScheduleForm below
|
|
|
|
# Provide a typed RotaScheduleForm instance so the template can render
|
|
# the constraint fields (prefixed with `opt__`) inline and submit them
|
|
# to the existing `rota_options` view which understands typed forms.
|
|
from .forms import RotaScheduleForm
|
|
try:
|
|
typed_form = RotaScheduleForm(instance=rota)
|
|
except Exception:
|
|
typed_form = None
|
|
|
|
return render(
|
|
request,
|
|
"rota/rota_detail.html",
|
|
{
|
|
"rota": rota,
|
|
"options_form": typed_form,
|
|
"available_constraints": available_constraints_rich,
|
|
"configured_constraints": configured_constraints,
|
|
"weeks_to_rota": weeks_to_rota,
|
|
"shift_count": shift_count,
|
|
"worker_count": worker_count,
|
|
"solver": solver,
|
|
"solver_options": solver_options_pretty,
|
|
"solver_ratio": solver_ratio,
|
|
},
|
|
)
|
|
|
|
|
|
def rota_create(request):
|
|
if request.method == "POST":
|
|
form = RotaScheduleForm(request.POST)
|
|
if form.is_valid():
|
|
rota = form.save()
|
|
return redirect("rota:rota_detail", rota_id=rota.id)
|
|
else:
|
|
form = RotaScheduleForm()
|
|
|
|
return render(request, "rota/rota_form.html", {"form": form})
|
|
|
|
|
|
def rota_edit(request, rota_id):
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
if request.method == "POST":
|
|
form = RotaScheduleForm(request.POST, instance=rota)
|
|
if form.is_valid():
|
|
form.save()
|
|
return redirect("rota:rota_detail", rota_id=rota.id)
|
|
else:
|
|
form = RotaScheduleForm(instance=rota)
|
|
|
|
return render(request, "rota/rota_form.html", {"form": form, "editing": True})
|
|
|
|
|
|
def worker_create(request):
|
|
# Support both full-page and HTMX modal flows. If the request comes from
|
|
# HTMX (header `HX-Request: true`) we return the modal partial so it can
|
|
# be injected into the `#modal` container. On successful HTMX POST we
|
|
# return an out-of-band swap to refresh the worker list and clear the
|
|
# modal.
|
|
is_hx = request.headers.get("HX-Request", "false").lower() == "true"
|
|
|
|
if request.method == "POST":
|
|
form = WorkerForm(request.POST)
|
|
if form.is_valid():
|
|
worker = form.save()
|
|
if is_hx:
|
|
# Return updated worker list and close modal
|
|
# If a rota context was supplied via querystring, we do not need it
|
|
# here because the partial will re-render from DB.
|
|
# Render worker list OOB and clear modal container
|
|
# We need a rota to render the list — try to use `rota_id` query param
|
|
# rota_id may be provided as a querystring or as a hidden form field.
|
|
rota_id = request.GET.get("rota_id") or request.POST.get("rota_id")
|
|
rota = None
|
|
if rota_id:
|
|
try:
|
|
rota = get_object_or_404(RotaSchedule, pk=int(rota_id))
|
|
except Exception:
|
|
rota = None
|
|
|
|
# If we created a worker and a rota_id was provided, create Assignment
|
|
if rota is not None:
|
|
from .models import Assignment
|
|
try:
|
|
Assignment.objects.create(worker=worker, rota=rota)
|
|
except Exception:
|
|
# best-effort: try using the M2M add as fallback
|
|
try:
|
|
rota.workers.add(worker)
|
|
except Exception:
|
|
pass
|
|
|
|
if rota is not None:
|
|
worker_list_html = render_to_string(
|
|
"rota/partials/worker_list.html",
|
|
{"rota": rota},
|
|
request=request,
|
|
)
|
|
resp = f'<div id="worker-list" hx-swap-oob="true">{worker_list_html}</div>'
|
|
response = HttpResponse(resp)
|
|
# close modal via HX-Trigger to avoid racing OOB swaps that target #modal
|
|
response["HX-Trigger"] = "closeModal"
|
|
return response
|
|
# If we couldn't render the worker list (no rota), just close modal via trigger
|
|
response = HttpResponse('')
|
|
response["HX-Trigger"] = "closeModal"
|
|
return response
|
|
# Non-HTMX POST -> redirect
|
|
return redirect("/")
|
|
else:
|
|
form = WorkerForm()
|
|
|
|
# HTMX GET: return modal partial
|
|
if is_hx:
|
|
modal_html = render_to_string(
|
|
"rota/partials/worker_form_modal.html",
|
|
{"form": form, "form_action": request.get_full_path()},
|
|
request=request,
|
|
)
|
|
return HttpResponse(modal_html)
|
|
|
|
# Non-HTMX GET: full page
|
|
return render(request, "rota/worker_form.html", {"form": form})
|
|
|
|
|
|
@require_http_methods(["GET", "POST"])
|
|
def worker_import(request, rota_id=None):
|
|
"""HTMX endpoint: import workers from CSV or pasted text.
|
|
|
|
Accepts a file upload named 'file' (CSV) or a textarea 'workers_text'.
|
|
CSV columns supported: name,email,site,grade,fte (headers optional).
|
|
When `rota_id` is provided, the created workers will be assigned to
|
|
that rota.
|
|
"""
|
|
is_hx = request.headers.get("HX-Request", "false").lower() == "true"
|
|
|
|
rota = None
|
|
if rota_id is not None:
|
|
try:
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
except Exception:
|
|
rota = None
|
|
|
|
if request.method == "GET":
|
|
# render modal partial
|
|
return render(request, "rota/partials/worker_import_modal.html", {"rota": rota})
|
|
|
|
# POST: parse uploaded file or textarea
|
|
created = []
|
|
errors = []
|
|
assign_to_rota = request.POST.get("assign_to_rota") == "1"
|
|
|
|
import csv
|
|
try:
|
|
upload = request.FILES.get("file")
|
|
if upload:
|
|
# decode bytes to text
|
|
try:
|
|
text = upload.read().decode("utf-8")
|
|
except Exception:
|
|
text = upload.read().decode("latin-1")
|
|
else:
|
|
text = request.POST.get("workers_text", "")
|
|
|
|
# Normalize lines: if simple newline-separated names/emails, make a CSV
|
|
reader = csv.DictReader(text.splitlines())
|
|
rows = list(reader)
|
|
if not rows:
|
|
# Try parsing as simple CSV without headers
|
|
reader2 = csv.reader([line for line in text.splitlines() if line.strip()])
|
|
for r in reader2:
|
|
# Map positional columns: name,email,site,grade,fte
|
|
while len(r) < 5:
|
|
r.append("")
|
|
rows.append({"name": r[0].strip(), "email": r[1].strip(), "site": r[2].strip(), "grade": r[3].strip(), "fte": r[4].strip()})
|
|
|
|
from .models import Assignment
|
|
|
|
for i, row in enumerate(rows):
|
|
try:
|
|
# normalize keys lower-case
|
|
if not isinstance(row, dict):
|
|
continue
|
|
r = {k.strip().lower(): (v if v is not None else "") for k, v in row.items()}
|
|
name = r.get("name") or r.get("full_name") or r.get("username")
|
|
if not name or str(name).strip() == "":
|
|
errors.append(f"Row {i+1}: missing name")
|
|
continue
|
|
email = r.get("email") or ""
|
|
site = r.get("site") or ""
|
|
try:
|
|
grade = int(r.get("grade")) if r.get("grade") not in (None, "") else None
|
|
except Exception:
|
|
grade = None
|
|
try:
|
|
fte_raw = r.get("fte")
|
|
if fte_raw in (None, ""):
|
|
fte = 1.0
|
|
else:
|
|
fte = float(fte_raw)
|
|
except Exception:
|
|
fte = 1.0
|
|
|
|
w = Worker.objects.create(name=name.strip(), email=email.strip(), site=site.strip(), grade=grade, fte=fte)
|
|
created.append(w)
|
|
if assign_to_rota and rota is not None:
|
|
try:
|
|
Assignment.objects.create(worker=w, rota=rota)
|
|
except Exception:
|
|
try:
|
|
rota.workers.add(w)
|
|
except Exception:
|
|
pass
|
|
except Exception as exc:
|
|
errors.append(f"Row {i+1}: {exc}")
|
|
except Exception as exc:
|
|
errors.append(str(exc))
|
|
|
|
# If HTMX, return updated worker list partial and close modal
|
|
if is_hx and rota is not None:
|
|
worker_list_html = render_to_string(
|
|
"rota/partials/worker_list.html",
|
|
{"rota": rota},
|
|
request=request,
|
|
)
|
|
# Return both new worker list and a trigger to close modal
|
|
resp = f'<div id="worker-list" hx-swap-oob="true">{worker_list_html}</div>'
|
|
response = HttpResponse(resp)
|
|
response["HX-Trigger"] = "closeModal"
|
|
return response
|
|
|
|
# Non-HTMX: redirect back to rota detail with message (flash messages not added here)
|
|
return redirect("rota:rota_detail", rota_id=rota.id if rota is not None else None)
|
|
|
|
|
|
def worker_detail(request, worker_id):
|
|
worker = get_object_or_404(Worker, pk=worker_id)
|
|
if request.method == "POST":
|
|
form = LeaveForm(request.POST)
|
|
if form.is_valid():
|
|
leave = form.save(commit=False)
|
|
leave.worker = worker
|
|
leave.save()
|
|
# For non-HTMX (full-page) submissions just redirect back to the
|
|
# worker detail page. Token-aware redirects are handled by the
|
|
# tokenized `request_leave` view (the public token page).
|
|
return redirect("rota:worker_detail", worker_id=worker.id)
|
|
else:
|
|
# Pre-fill dates when `date` query param is provided (ISO YYYY-MM-DD)
|
|
date_q = request.GET.get("date") or request.POST.get("date")
|
|
if date_q:
|
|
try:
|
|
import datetime
|
|
|
|
d = datetime.date.fromisoformat(date_q)
|
|
form = LeaveForm(initial={"start_date": d, "end_date": d})
|
|
except Exception:
|
|
form = LeaveForm()
|
|
else:
|
|
form = LeaveForm()
|
|
|
|
# Build a JSON-serializable list of leaves for the calendar widget to consume
|
|
leaves = []
|
|
try:
|
|
for leave in worker.leaves.all():
|
|
try:
|
|
s = leave.start_date.isoformat()
|
|
except Exception:
|
|
s = str(leave.start_date)
|
|
try:
|
|
e = leave.end_date.isoformat()
|
|
except Exception:
|
|
e = str(leave.end_date)
|
|
leaves.append({"start": s, "end": e})
|
|
except Exception:
|
|
leaves = []
|
|
|
|
# Provide a JSON-encoded string for safe embedding in the template
|
|
try:
|
|
leaves_json = json.dumps(leaves)
|
|
except Exception:
|
|
leaves_json = '[]'
|
|
|
|
return render(request, "rota/worker_detail.html", {"worker": worker, "form": form, "leaves": leaves, "leaves_json": leaves_json})
|
|
|
|
|
|
@require_http_methods(["GET", "POST"])
|
|
def worker_leave_add(request, worker_id):
|
|
"""HTMX endpoint: show a leave form modal for a worker, or create the leave on POST."""
|
|
worker = get_object_or_404(Worker, pk=worker_id)
|
|
is_hx = request.headers.get("HX-Request", "false").lower() == "true"
|
|
|
|
if request.method == "POST":
|
|
form = LeaveForm(request.POST)
|
|
if form.is_valid():
|
|
leave = form.save(commit=False)
|
|
leave.worker = worker
|
|
leave.save()
|
|
if is_hx:
|
|
# Close modal and optionally trigger an update; leave list can be refreshed separately
|
|
response = HttpResponse("")
|
|
response["HX-Trigger"] = "closeModal"
|
|
return response
|
|
return redirect("rota:worker_detail", worker_id=worker.id)
|
|
|
|
|
|
|
|
# invalid: return modal with errors
|
|
if is_hx:
|
|
modal_html = render_to_string("rota/partials/leave_form_modal.html", {"form": form, "worker": worker, "form_action": request.path}, request=request)
|
|
return HttpResponse(modal_html)
|
|
# non-HTMX fallthrough
|
|
# GET: render modal partial
|
|
form = LeaveForm()
|
|
if is_hx:
|
|
modal_html = render_to_string("rota/partials/leave_form_modal.html", {"form": form, "worker": worker, "form_action": request.path}, request=request)
|
|
return HttpResponse(modal_html)
|
|
return redirect("rota:worker_detail", worker_id=worker.id)
|
|
|
|
|
|
@require_http_methods(["GET"])
|
|
def worker_overview(request, rota_id, worker_id):
|
|
"""Return a read-only modal overview of the worker (HTMX GET)."""
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
worker = get_object_or_404(Worker, pk=worker_id)
|
|
try:
|
|
pretty_options = json.dumps(worker.options or {}, indent=2, default=str)
|
|
except Exception:
|
|
pretty_options = str(worker.options or {})
|
|
|
|
modal_html = render_to_string(
|
|
"rota/partials/worker_overview_modal.html",
|
|
{"worker": worker, "rota": rota, "pretty_options": pretty_options},
|
|
request=request,
|
|
)
|
|
return HttpResponse(modal_html)
|
|
|
|
|
|
@require_http_methods(["GET", "POST"])
|
|
def worker_edit(request, rota_id, worker_id):
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
worker = get_object_or_404(Worker, pk=worker_id)
|
|
|
|
if request.method == "POST":
|
|
form = WorkerForm(request.POST, instance=worker)
|
|
if form.is_valid():
|
|
form.save()
|
|
# Return updated worker list OOB and close modal
|
|
worker_list_html = render_to_string("rota/partials/worker_list.html", {"rota": rota}, request=request)
|
|
resp = f'<div id="worker-list" hx-swap-oob="true">{worker_list_html}</div>'
|
|
response = HttpResponse(resp)
|
|
response["HX-Trigger"] = "closeModal"
|
|
return response
|
|
|
|
# invalid: return form modal with errors
|
|
modal_html = render_to_string(
|
|
"rota/partials/worker_form_modal.html",
|
|
{"form": form, "form_action": request.get_full_path(), "rota": rota, "focus": request.GET.get('focus') or request.POST.get('focus')},
|
|
request=request,
|
|
)
|
|
return HttpResponse(modal_html)
|
|
|
|
# GET: populate form and return modal
|
|
form = WorkerForm(instance=worker)
|
|
modal_html = render_to_string(
|
|
"rota/partials/worker_form_modal.html",
|
|
{"form": form, "form_action": request.path, "rota": rota, "worker": worker, "token_link": request.build_absolute_uri(reverse('rota:request_leave_token', args=[worker.request_token])) if worker and getattr(worker, 'request_token', None) else '', "focus": request.GET.get('focus')},
|
|
request=request,
|
|
)
|
|
return HttpResponse(modal_html)
|
|
|
|
|
|
@require_POST
|
|
def regenerate_worker_token(request, worker_id):
|
|
# Only allow staff users to regenerate tokens from the admin modal
|
|
if not (request.user and request.user.is_authenticated and request.user.is_staff):
|
|
return HttpResponseBadRequest("Permission denied")
|
|
|
|
worker = get_object_or_404(Worker, pk=worker_id)
|
|
# generate unique token
|
|
import uuid
|
|
|
|
new_token = uuid.uuid4()
|
|
# guard against collision
|
|
while Worker.objects.filter(request_token=new_token).exclude(pk=worker.pk).exists():
|
|
new_token = uuid.uuid4()
|
|
worker.request_token = new_token
|
|
worker.save(update_fields=["request_token"])
|
|
|
|
# render only the token area so HTMX can swap it
|
|
token_html = render_to_string(
|
|
"rota/partials/worker_token_area.html",
|
|
{"worker": worker, "token_link": request.build_absolute_uri(reverse('rota:request_leave_token', args=[worker.request_token])) if getattr(worker, 'request_token', None) else ''},
|
|
request=request,
|
|
)
|
|
response = HttpResponse(token_html)
|
|
response["HX-Trigger"] = "workerTokenRegenerated"
|
|
return response
|
|
|
|
|
|
@require_http_methods(["GET", "POST"])
|
|
def worker_delete(request, rota_id, worker_id):
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
worker = get_object_or_404(Worker, pk=worker_id)
|
|
|
|
if request.method == "POST":
|
|
# remove assignment between worker and rota
|
|
try:
|
|
from .models import Assignment
|
|
|
|
Assignment.objects.filter(worker=worker, rota=rota).delete()
|
|
except Exception:
|
|
try:
|
|
rota.workers.remove(worker)
|
|
except Exception:
|
|
return HttpResponseBadRequest("Unable to remove worker from rota")
|
|
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
worker_list_html = render_to_string("rota/partials/worker_list.html", {"rota": rota}, request=request)
|
|
resp = f'<div id="worker-list" hx-swap-oob="true">{worker_list_html}</div>'
|
|
response = HttpResponse(resp)
|
|
response["HX-Trigger"] = "closeModal"
|
|
return response
|
|
|
|
# GET: confirmation modal
|
|
modal_html = render_to_string(
|
|
"rota/partials/worker_confirm_delete.html",
|
|
{"rota": rota, "worker": worker},
|
|
request=request,
|
|
)
|
|
return HttpResponse(modal_html)
|
|
|
|
|
|
def rota_run_detail(request, run_id):
|
|
from .models import RotaRun
|
|
|
|
run = get_object_or_404(RotaRun, pk=run_id)
|
|
# Prepare a pretty-printed JSON representation of the result for easier
|
|
# debugging in templates. Also expose any builder HTML present in the
|
|
# persisted `export_html` or embedded in the `result` under
|
|
# `builder_html` so it can be viewed inline.
|
|
result_pretty = None
|
|
builder_html = None
|
|
try:
|
|
if run.result is not None:
|
|
import json as _json
|
|
result_pretty = _json.dumps(run.result, indent=2, default=str)
|
|
# prefer persisted export_html if present
|
|
builder_html = getattr(run, "export_html", None) or (run.result.get("builder_html") if isinstance(run.result, dict) else None)
|
|
else:
|
|
builder_html = getattr(run, "export_html", None)
|
|
except Exception:
|
|
result_pretty = str(run.result)
|
|
|
|
return render(request, "rota/rota_run_detail.html", {"run": run, "result_pretty": result_pretty, "builder_html": builder_html})
|
|
|
|
|
|
def rota_run_export_html(request, run_id):
|
|
from .models import RotaRun
|
|
|
|
run = get_object_or_404(RotaRun, pk=run_id)
|
|
# If the export HTML was persisted with the run, return it directly.
|
|
if getattr(run, "export_html", None):
|
|
return HttpResponse(run.export_html, content_type="text/html; charset=utf-8")
|
|
|
|
# Fallback: render the export template which shows run.result.
|
|
return render(request, "rota/rota_run_export.html", {"run": run})
|
|
|
|
|
|
@require_POST
|
|
def rota_options_update(request, rota_id):
|
|
"""Update a rota's JSON `options` from a posted textarea `options_json`.
|
|
|
|
This endpoint expects a POST with `options_json` containing a JSON object.
|
|
It validates JSON and saves it into `RotaSchedule.options`.
|
|
"""
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
options_json = request.POST.get("options_json", "")
|
|
try:
|
|
parsed = json.loads(options_json) if options_json.strip() else {}
|
|
except Exception as exc:
|
|
return HttpResponseBadRequest(f"Invalid JSON: {exc}")
|
|
|
|
rota.options = parsed
|
|
rota.save(update_fields=["options"])
|
|
return redirect("rota:rota_detail", rota_id=rota.id)
|
|
|
|
|
|
@require_http_methods(["GET", "POST"])
|
|
def request_leave(request, token=None):
|
|
"""Allow a worker to request leave for themselves.
|
|
|
|
Resolution strategy:
|
|
- If the user is authenticated, try to find a `Worker` with matching
|
|
email (first match). If found, use that worker.
|
|
- Otherwise, fall back to `worker_id` provided via GET/POST.
|
|
|
|
Supports HTMX modal flow when `HX-Request` header is present.
|
|
"""
|
|
is_hx = request.headers.get("HX-Request", "false").lower() == "true"
|
|
|
|
worker = None
|
|
# 1) token link (preferred for unauthenticated users)
|
|
if token:
|
|
try:
|
|
worker = get_object_or_404(Worker, request_token=token)
|
|
except Exception:
|
|
worker = None
|
|
|
|
# 2) try authenticated user -> worker by email
|
|
if worker is None:
|
|
try:
|
|
if request.user and request.user.is_authenticated and getattr(request.user, "email", None):
|
|
worker = Worker.objects.filter(email=request.user.email).first()
|
|
except Exception:
|
|
worker = None
|
|
|
|
# 3) fallback to explicit worker_id param
|
|
if worker is None:
|
|
worker_id = request.GET.get("worker_id") or request.POST.get("worker_id")
|
|
if worker_id:
|
|
try:
|
|
worker = get_object_or_404(Worker, pk=int(worker_id))
|
|
except Exception:
|
|
worker = None
|
|
|
|
if request.method == "POST":
|
|
form = LeaveForm(request.POST)
|
|
if form.is_valid():
|
|
leave = form.save(commit=False)
|
|
if worker is None:
|
|
return HttpResponseBadRequest("Worker not specified or not found")
|
|
leave.worker = worker
|
|
leave.save()
|
|
# On HTMX modal post, return an OOB swap to close modal and optionally
|
|
# update worker detail partial if present.
|
|
if is_hx:
|
|
# Render updated leave list and return OOB swaps: update leave-list and clear modal.
|
|
leave_list_html = render_to_string(
|
|
"rota/partials/leave_list.html",
|
|
{"leaves": worker.leaves.all(), "token": token},
|
|
request=request,
|
|
)
|
|
resp = f'<div id="leave-list" hx-swap-oob="true">{leave_list_html}</div>'
|
|
response = HttpResponse(resp)
|
|
# Trigger modal close and a leaveUpdated event so client can refresh calendar highlights
|
|
response["HX-Trigger"] = "closeModal,leaveUpdated"
|
|
return response
|
|
# For full-page submissions: if this request was made using a tokenized
|
|
# URL (public leave request page), keep the user on that token page so
|
|
# they can see confirmation and existing requests. Otherwise go to the
|
|
# worker detail page.
|
|
if token:
|
|
try:
|
|
return redirect(reverse('rota:request_leave_token', args=[token]))
|
|
except Exception:
|
|
# fallback to worker detail
|
|
pass
|
|
return redirect("rota:worker_detail", worker_id=worker.id)
|
|
else:
|
|
# Prefill dates when provided via query params for both full page and HTMX modal
|
|
date_q = request.GET.get("date") or request.POST.get("date")
|
|
start_q = request.GET.get("start_date") or request.POST.get("start_date")
|
|
end_q = request.GET.get("end_date") or request.POST.get("end_date")
|
|
if date_q:
|
|
try:
|
|
import datetime
|
|
|
|
d = datetime.date.fromisoformat(date_q)
|
|
form = LeaveForm(initial={"start_date": d, "end_date": d})
|
|
except Exception:
|
|
form = LeaveForm()
|
|
elif start_q or end_q:
|
|
try:
|
|
import datetime
|
|
|
|
sd = datetime.date.fromisoformat(start_q) if start_q else None
|
|
ed = datetime.date.fromisoformat(end_q) if end_q else None
|
|
init = {}
|
|
if sd:
|
|
init["start_date"] = sd
|
|
if ed:
|
|
init["end_date"] = ed
|
|
form = LeaveForm(initial=init)
|
|
except Exception:
|
|
form = LeaveForm()
|
|
else:
|
|
form = LeaveForm()
|
|
|
|
# Render modal or full page
|
|
if is_hx:
|
|
modal_html = render_to_string(
|
|
"rota/partials/leave_request_modal.html",
|
|
{"form": form, "form_action": request.get_full_path(), "worker": worker},
|
|
request=request,
|
|
)
|
|
return HttpResponse(modal_html)
|
|
|
|
return render(request, "rota/leave_request.html", {"form": form, "worker": worker, "token": token})
|
|
|
|
|
|
@require_http_methods(["GET", "POST"])
|
|
def worker_settings_token(request, token):
|
|
"""Allow a worker identified by a request token to update simple settings
|
|
such as FTE and non-working days without requiring authentication.
|
|
"""
|
|
worker = None
|
|
if token:
|
|
try:
|
|
worker = get_object_or_404(Worker, request_token=token)
|
|
except Exception:
|
|
worker = None
|
|
|
|
if worker is None:
|
|
return HttpResponseBadRequest("Invalid token")
|
|
|
|
is_hx = request.headers.get("HX-Request", "false").lower() == "true"
|
|
|
|
if request.method == "POST":
|
|
form = WorkerSelfServiceForm(request.POST, instance=worker)
|
|
if form.is_valid():
|
|
form.save()
|
|
if is_hx:
|
|
# Return minimal fragment confirming success and trigger a leaveUpdated
|
|
# so any calendars can refresh (if they rely on worker settings).
|
|
resp = render_to_string("rota/partials/worker_settings_saved.html", {"worker": worker}, request=request)
|
|
response = HttpResponse(resp)
|
|
response["HX-Trigger"] = "closeModal,leaveUpdated"
|
|
return response
|
|
# Full page: redirect back to tokenized request page
|
|
return redirect(reverse('rota:request_leave_token', args=[token]))
|
|
else:
|
|
form = WorkerSelfServiceForm(instance=worker)
|
|
|
|
# HTMX modal request: return partial modal
|
|
if is_hx:
|
|
modal_html = render_to_string(
|
|
"rota/partials/worker_selfservice_modal.html",
|
|
{"form": form, "form_action": request.get_full_path(), "worker": worker},
|
|
request=request,
|
|
)
|
|
return HttpResponse(modal_html)
|
|
|
|
return render(request, "rota/worker_settings.html", {"form": form, "worker": worker, "token": token})
|
|
|
|
|
|
def rota_options(request, rota_id):
|
|
"""HTMX-aware modal edit for rota.options JSON.
|
|
|
|
GET: return modal partial containing the JSON textarea form.
|
|
POST: validate and save JSON; if HTMX, return OOB swap updating the options display and clear the modal.
|
|
"""
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
is_hx = request.headers.get("HX-Request", "false").lower() == "true"
|
|
|
|
if request.method == "POST":
|
|
# Support two POST modes: legacy `options_json` textarea, or the typed
|
|
# form generated by `RotaScheduleForm` (fields prefixed with `opt__`).
|
|
if "options_json" in request.POST:
|
|
options_json = request.POST.get("options_json", "")
|
|
try:
|
|
parsed = json.loads(options_json) if options_json.strip() else {}
|
|
except Exception as exc:
|
|
if is_hx:
|
|
return HttpResponseBadRequest(f"Invalid JSON: {exc}")
|
|
return HttpResponseBadRequest(f"Invalid JSON: {exc}")
|
|
|
|
rota.options = parsed
|
|
rota.save(update_fields=["options"])
|
|
if is_hx:
|
|
return _oob_update_options_display(request, rota)
|
|
return redirect(reverse("rota:rota_detail", args=(rota.id,)))
|
|
|
|
# Typed form submission for constraint options only
|
|
from .forms import RotaConstraintOptionsForm
|
|
|
|
form = RotaConstraintOptionsForm(request.POST, initial_options=rota.options)
|
|
if not form.is_valid():
|
|
if is_hx:
|
|
# Re-render modal with errors
|
|
modal_html = render_to_string(
|
|
"rota/partials/rota_options_modal.html",
|
|
{"form_action": reverse("rota:rota_options", args=(rota.id,)), "form": form, "rota": rota},
|
|
request=request,
|
|
)
|
|
return HttpResponse(modal_html)
|
|
return HttpResponseBadRequest("Invalid form submission")
|
|
|
|
# Persist only constraint keys into rota.options
|
|
form.save_for_rota(rota)
|
|
if is_hx:
|
|
return _oob_update_options_display(request, rota)
|
|
return redirect(reverse("rota:rota_detail", args=(rota.id,)))
|
|
|
|
# GET: render modal with typed constraint form populated from the rota instance
|
|
from .forms import RotaConstraintOptionsForm
|
|
form = RotaConstraintOptionsForm(initial_options=rota.options)
|
|
modal_html = render_to_string(
|
|
"rota/partials/rota_options_modal.html",
|
|
{"form_action": reverse("rota:rota_options", args=(rota.id,)), "form": form, "rota": rota},
|
|
request=request,
|
|
)
|
|
return HttpResponse(modal_html)
|
|
|
|
|
|
def _oob_update_options_display(request, rota):
|
|
"""Helper: render options display and return HTMX OOB swap response."""
|
|
options_pretty = json.dumps(rota.options or {}, indent=4)
|
|
options_html = render_to_string(
|
|
"rota/partials/rota_options_display.html",
|
|
{"rota": rota, "options_pretty": options_pretty},
|
|
request=request,
|
|
)
|
|
resp = '<div id="rota-options-display" hx-swap-oob="true">' + options_html + '</div>'
|
|
response = HttpResponse(resp)
|
|
response["HX-Trigger"] = "closeModal"
|
|
return response
|
|
|
|
|
|
def rota_option_add(request, rota_id):
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
is_hx = request.headers.get("HX-Request", "false").lower() == "true"
|
|
|
|
if request.method == "POST":
|
|
key = request.POST.get("key", "").strip()
|
|
val = request.POST.get("value", "")
|
|
vtype = request.POST.get("type", "string")
|
|
if not key:
|
|
return HttpResponseBadRequest("Key is required")
|
|
# parse value according to type
|
|
try:
|
|
if vtype == "int":
|
|
parsed = int(val)
|
|
elif vtype == "float":
|
|
parsed = float(val)
|
|
elif vtype == "bool":
|
|
parsed = val.lower() in ("1", "true", "yes", "on")
|
|
elif vtype == "json":
|
|
parsed = json.loads(val) if val.strip() else None
|
|
else:
|
|
parsed = val
|
|
except Exception as exc:
|
|
return HttpResponseBadRequest(f"Invalid value: {exc}")
|
|
|
|
opts = rota.options or {}
|
|
opts[key] = parsed
|
|
rota.options = opts
|
|
rota.save(update_fields=["options"])
|
|
|
|
if is_hx:
|
|
return _oob_update_options_display(request, rota)
|
|
|
|
return redirect("rota:rota_detail", rota_id=rota.id)
|
|
|
|
# GET: show modal partial. If a `key` query param is supplied, try to
|
|
# prefill the value and type from the builder defaults to make adding
|
|
# standard options easy.
|
|
key_q = request.GET.get("key")
|
|
initial = {"key": "", "value": "", "type": "string"}
|
|
if key_q:
|
|
try:
|
|
import importlib
|
|
mod = importlib.import_module("rota_generator.shifts")
|
|
RotaBuilder = getattr(mod, "RotaBuilder")
|
|
rb = RotaBuilder()
|
|
defaults = rb.constraint_options_model.model_dump()
|
|
if key_q in defaults:
|
|
d = defaults[key_q]
|
|
initial["key"] = key_q
|
|
if isinstance(d, bool):
|
|
initial["type"] = "bool"
|
|
initial["value"] = "1" if d else "0"
|
|
elif isinstance(d, int):
|
|
initial["type"] = "int"
|
|
initial["value"] = str(d)
|
|
elif isinstance(d, float):
|
|
initial["type"] = "float"
|
|
initial["value"] = str(d)
|
|
elif isinstance(d, (list, dict)):
|
|
initial["type"] = "json"
|
|
try:
|
|
initial["value"] = json.dumps(d)
|
|
except Exception:
|
|
initial["value"] = ""
|
|
else:
|
|
initial["type"] = "string"
|
|
initial["value"] = str(d)
|
|
except Exception:
|
|
pass
|
|
|
|
# Compute description for the suggested key (if any) and pass into modal
|
|
desc = _get_constraint_description(key_q) if key_q else None
|
|
|
|
modal_html = render_to_string(
|
|
"rota/partials/rota_option_modal_single.html",
|
|
{"form_action": reverse("rota:rota_option_add", args=(rota.id,)), "initial": initial, "rota": rota, "description": desc},
|
|
request=request,
|
|
)
|
|
return HttpResponse(modal_html)
|
|
|
|
|
|
def rota_option_inline(request, rota_id):
|
|
"""Return an inline editor for a single option (or save it).
|
|
|
|
GET: return a small form fragment suitable for in-place replacement of a
|
|
constraint row. POST: save the key/value and return the rendered row
|
|
fragment to replace the editor.
|
|
"""
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
|
|
if request.method == "POST":
|
|
key = request.POST.get("key", "").strip()
|
|
val = request.POST.get("value", "")
|
|
vtype = request.POST.get("type", "string")
|
|
if not key:
|
|
# Re-render the inline form with an error message
|
|
initial = {"key": key, "value": val, "type": vtype}
|
|
inline_html = render_to_string(
|
|
"rota/partials/rota_option_inline.html",
|
|
{"form_action": reverse("rota:rota_option_inline", args=(rota.id,)), "initial": initial, "error": "Key is required", "rota": rota},
|
|
request=request,
|
|
)
|
|
return HttpResponse(inline_html)
|
|
|
|
# Try to parse according to type; on failure re-render form with error
|
|
parsed = None
|
|
parse_error = None
|
|
try:
|
|
if vtype == "int":
|
|
parsed = int(val)
|
|
elif vtype == "float":
|
|
parsed = float(val)
|
|
elif vtype == "bool":
|
|
parsed = val.lower() in ("1", "true", "yes", "on")
|
|
elif vtype == "json":
|
|
parsed = json.loads(val) if val.strip() else None
|
|
else:
|
|
parsed = val
|
|
except Exception as exc:
|
|
parse_error = str(exc)
|
|
|
|
if parse_error:
|
|
initial = {"key": key, "value": val, "type": vtype}
|
|
inline_html = render_to_string(
|
|
"rota/partials/rota_option_inline.html",
|
|
{"form_action": reverse("rota:rota_option_inline", args=(rota.id,)), "initial": initial, "error": f"Invalid value: {parse_error}", "rota": rota},
|
|
request=request,
|
|
)
|
|
return HttpResponse(inline_html)
|
|
|
|
opts = rota.options or {}
|
|
opts[key] = parsed
|
|
rota.options = opts
|
|
rota.save(update_fields=["options"])
|
|
|
|
# Render and return the updated row fragment
|
|
try:
|
|
pretty = json.dumps(parsed, indent=2, default=str)
|
|
except Exception:
|
|
pretty = str(parsed)
|
|
# Include description for the saved key so the updated row has a tooltip
|
|
desc = _get_constraint_description(key)
|
|
row_html = render_to_string(
|
|
"rota/partials/rota_option_row.html",
|
|
{"rota": rota, "key": key, "value": parsed, "pretty": pretty, "description": desc},
|
|
request=request,
|
|
)
|
|
return HttpResponse(row_html)
|
|
|
|
# GET: return inline form fragment prefilled from defaults if present
|
|
key_q = request.GET.get("key")
|
|
initial = {"key": "", "value": "", "type": "string"}
|
|
if key_q:
|
|
try:
|
|
import importlib as _importlib
|
|
|
|
mod = _importlib.import_module("rota_generator.shifts")
|
|
RotaBuilder = getattr(mod, "RotaBuilder")
|
|
rb = RotaBuilder()
|
|
defaults = rb.constraint_options_model.model_dump()
|
|
if key_q in defaults:
|
|
d = defaults[key_q]
|
|
initial["key"] = key_q
|
|
if isinstance(d, bool):
|
|
initial["type"] = "bool"
|
|
initial["value"] = "1" if d else "0"
|
|
elif isinstance(d, int):
|
|
initial["type"] = "int"
|
|
initial["value"] = str(d)
|
|
elif isinstance(d, float):
|
|
initial["type"] = "float"
|
|
initial["value"] = str(d)
|
|
elif isinstance(d, (list, dict)):
|
|
initial["type"] = "json"
|
|
try:
|
|
initial["value"] = json.dumps(d)
|
|
except Exception:
|
|
initial["value"] = ""
|
|
else:
|
|
initial["type"] = "string"
|
|
initial["value"] = str(d)
|
|
except Exception:
|
|
pass
|
|
|
|
inline_html = render_to_string(
|
|
"rota/partials/rota_option_inline.html",
|
|
{"form_action": reverse("rota:rota_option_inline", args=(rota.id,)), "initial": initial, "rota": rota},
|
|
request=request,
|
|
)
|
|
return HttpResponse(inline_html)
|
|
|
|
|
|
def rota_option_edit(request, rota_id):
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
is_hx = request.headers.get("HX-Request", "false").lower() == "true"
|
|
|
|
if request.method == "POST":
|
|
key = request.POST.get("key", "").strip()
|
|
val = request.POST.get("value", "")
|
|
vtype = request.POST.get("type", "string")
|
|
if not key:
|
|
return HttpResponseBadRequest("Key is required")
|
|
try:
|
|
if vtype == "int":
|
|
parsed = int(val)
|
|
elif vtype == "float":
|
|
parsed = float(val)
|
|
elif vtype == "bool":
|
|
parsed = val.lower() in ("1", "true", "yes", "on")
|
|
elif vtype == "json":
|
|
parsed = json.loads(val) if val.strip() else None
|
|
else:
|
|
parsed = val
|
|
except Exception as exc:
|
|
return HttpResponseBadRequest(f"Invalid value: {exc}")
|
|
|
|
opts = rota.options or {}
|
|
opts[key] = parsed
|
|
rota.options = opts
|
|
rota.save(update_fields=["options"])
|
|
|
|
if is_hx:
|
|
return _oob_update_options_display(request, rota)
|
|
|
|
return redirect("rota:rota_detail", rota_id=rota.id)
|
|
|
|
# GET: prepare edit modal with existing value
|
|
key = request.GET.get("key")
|
|
if not key:
|
|
return HttpResponseBadRequest("Missing key")
|
|
current = rota.options or {}
|
|
cur_val = current.get(key, "")
|
|
# choose type hint
|
|
if isinstance(cur_val, bool):
|
|
vtype = "bool"
|
|
value = "1" if cur_val else "0"
|
|
elif isinstance(cur_val, int):
|
|
vtype = "int"
|
|
value = str(cur_val)
|
|
elif isinstance(cur_val, float):
|
|
vtype = "float"
|
|
value = str(cur_val)
|
|
elif isinstance(cur_val, (list, dict)):
|
|
vtype = "json"
|
|
value = json.dumps(cur_val)
|
|
else:
|
|
vtype = "string"
|
|
value = str(cur_val)
|
|
|
|
# Include description for the key in the edit modal
|
|
desc = _get_constraint_description(key)
|
|
modal_html = render_to_string(
|
|
"rota/partials/rota_option_modal_single.html",
|
|
{"form_action": reverse("rota:rota_option_edit", args=(rota.id,)), "initial": {"key": key, "value": value, "type": vtype}, "rota": rota, "description": desc},
|
|
request=request,
|
|
)
|
|
return HttpResponse(modal_html)
|
|
|
|
|
|
@require_POST
|
|
def rota_option_delete(request, rota_id):
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
key = request.POST.get("key")
|
|
if not key:
|
|
return HttpResponseBadRequest("Missing key")
|
|
opts = rota.options or {}
|
|
if key in opts:
|
|
opts.pop(key)
|
|
rota.options = opts
|
|
rota.save(update_fields=["options"])
|
|
|
|
# If this was an inline request, return an empty fragment so the
|
|
# inline editor row can be removed client-side. Otherwise return the
|
|
# standard HTMX OOB update to refresh the options display.
|
|
if request.POST.get("inline") == "1":
|
|
return HttpResponse("")
|
|
|
|
if request.headers.get("HX-Request", "false").lower() == "true":
|
|
return _oob_update_options_display(request, rota)
|
|
|
|
return redirect("rota:rota_detail", rota_id=rota.id)
|
|
|
|
|
|
def rota_run_export_download(request, run_id):
|
|
from .models import RotaRun
|
|
|
|
run = get_object_or_404(RotaRun, pk=run_id)
|
|
# Prefer persisted export HTML if available
|
|
if getattr(run, "export_html", None):
|
|
html = run.export_html
|
|
else:
|
|
html = render_to_string("rota/rota_run_export.html", {"run": run}, request=request)
|
|
|
|
response = HttpResponse(html, content_type="text/html; charset=utf-8")
|
|
filename = f"rota_run_{run.id}_export.html"
|
|
response["Content-Disposition"] = f'attachment; filename="{filename}"'
|
|
return response
|
|
|
|
|
|
def leaves_json(request, worker_id):
|
|
"""Return JSON list of leave ranges for a given worker."""
|
|
worker = get_object_or_404(Worker, pk=worker_id)
|
|
data = []
|
|
for leave in worker.leaves.all().order_by('start_date'):
|
|
data.append({
|
|
'start': leave.start_date.isoformat(),
|
|
'end': leave.end_date.isoformat(),
|
|
'reason': leave.reason,
|
|
})
|
|
return JsonResponse({'leaves': data})
|
|
|
|
|
|
def leaves_global_json(request, rota_id=None):
|
|
"""Return JSON list of all leave ranges across all workers."""
|
|
from .models import Leave
|
|
# Optionally filter by rota_id (provided as query param) so the global
|
|
# calendar can be scoped to a particular rota's workers.
|
|
rota_id = request.GET.get('rota_id') if rota_id is None else rota_id
|
|
qs = Leave.objects.select_related('worker')
|
|
if rota_id:
|
|
try:
|
|
rota_id_i = int(rota_id)
|
|
qs = qs.filter(worker__rotas=rota_id_i)
|
|
except Exception:
|
|
pass
|
|
|
|
# Build a worker->color map for workers present in the query so the
|
|
# client can color-code leaves by worker.
|
|
worker_colors = {}
|
|
def color_for_worker(pk):
|
|
# deterministic hue based on id
|
|
h = (int(pk) * 37) % 360
|
|
return f'hsl({h} 65% 85%)'
|
|
|
|
data = []
|
|
for leave in qs.all().order_by('start_date'):
|
|
worker = getattr(leave, 'worker', None)
|
|
wid = getattr(worker, 'id', None)
|
|
wname = getattr(worker, 'name', '') if worker else ''
|
|
if wid not in worker_colors and wid is not None:
|
|
worker_colors[wid] = color_for_worker(wid)
|
|
data.append({
|
|
'start': leave.start_date.isoformat(),
|
|
'end': leave.end_date.isoformat(),
|
|
'reason': wname or (leave.reason or ''),
|
|
'worker_id': wid,
|
|
'color': worker_colors.get(wid),
|
|
})
|
|
return JsonResponse({'leaves': data})
|
|
|
|
|
|
def leaves_global(request, rota_id=None):
|
|
"""Page showing a global calendar with all leave requests."""
|
|
# Provide an empty JSON payload initially; the calendar will fetch the authoritative list
|
|
leaves_json = '[]'
|
|
selected = rota_id if rota_id is not None else request.GET.get('rota_id')
|
|
return render(request, 'rota/leaves_global.html', {'leaves_json': leaves_json, 'rotas': RotaSchedule.objects.all().order_by('name'), 'selected_rota_id': selected})
|
|
|
|
|
|
@require_http_methods(["GET"])
|
|
def select_workers_modal(request, rota_id=None):
|
|
"""Return an HTMX modal partial that lists workers to apply a selected leave range to.
|
|
|
|
Query params: start_date, end_date, rota_id (optional)
|
|
"""
|
|
start_date = request.GET.get('start_date')
|
|
end_date = request.GET.get('end_date')
|
|
rota_id = request.GET.get('rota_id') if rota_id is None else rota_id
|
|
workers = Worker.objects.all()
|
|
if rota_id:
|
|
try:
|
|
workers = workers.filter(rotas__id=int(rota_id))
|
|
except Exception:
|
|
pass
|
|
workers = workers.order_by('name')
|
|
modal_html = render_to_string(
|
|
'rota/partials/select_workers_modal.html',
|
|
{'workers': workers, 'start_date': start_date, 'end_date': end_date},
|
|
request=request,
|
|
)
|
|
return HttpResponse(modal_html)
|
|
|
|
|
|
@require_http_methods(["POST"])
|
|
def apply_leave_multi(request):
|
|
"""Apply a leave range to multiple workers. Expects POST with worker_ids (list), start_date and end_date."""
|
|
from .models import Leave
|
|
|
|
worker_ids = request.POST.getlist('worker_id')
|
|
start_date = request.POST.get('start_date')
|
|
end_date = request.POST.get('end_date')
|
|
if not worker_ids or not start_date:
|
|
return HttpResponseBadRequest('Missing parameters')
|
|
created = []
|
|
for wid in worker_ids:
|
|
try:
|
|
w = Worker.objects.get(pk=int(wid))
|
|
except Exception:
|
|
continue
|
|
try:
|
|
leave = Leave(worker=w, start_date=start_date, end_date=end_date)
|
|
leave.save()
|
|
created.append(leave)
|
|
except Exception:
|
|
continue
|
|
|
|
# Return simple HTMX response: close modal and trigger calendar refresh
|
|
response = HttpResponse('')
|
|
response['HX-Trigger'] = 'closeModal,leaveUpdated'
|
|
return response
|
|
|
|
|
|
@require_POST
|
|
def leave_delete(request, leave_id):
|
|
"""Delete a Leave record. Allowed if the requester is staff, or if a valid token matches the worker, or the authenticated user's email matches the worker email."""
|
|
from .models import Leave
|
|
|
|
leave = get_object_or_404(Leave, pk=leave_id)
|
|
worker = leave.worker
|
|
|
|
# permission checks
|
|
allowed = False
|
|
token = request.POST.get('token') or request.GET.get('token')
|
|
try:
|
|
if request.user and request.user.is_authenticated and request.user.is_staff:
|
|
allowed = True
|
|
elif request.user and request.user.is_authenticated and getattr(request.user, 'email', None) and request.user.email == (worker.email or ''):
|
|
allowed = True
|
|
elif token:
|
|
try:
|
|
import uuid
|
|
token_uuid = uuid.UUID(token)
|
|
if worker.request_token == token_uuid:
|
|
allowed = True
|
|
except Exception:
|
|
allowed = False
|
|
except Exception:
|
|
allowed = False
|
|
|
|
if not allowed:
|
|
return HttpResponseBadRequest('Permission denied')
|
|
|
|
# perform delete
|
|
leave.delete()
|
|
|
|
# render updated leave list for the worker
|
|
leave_list_html = render_to_string(
|
|
"rota/partials/leave_list.html",
|
|
{"leaves": worker.leaves.all(), "token": token},
|
|
request=request,
|
|
)
|
|
resp = f'<div id="leave-list" hx-swap-oob="true">{leave_list_html}</div>'
|
|
response = HttpResponse(resp)
|
|
response["HX-Trigger"] = "leaveUpdated"
|
|
return response
|
|
|
|
|
|
@require_POST
|
|
def leave_delete_all(request, worker_id):
|
|
"""Delete all Leave records for a worker. Permissions same as `leave_delete`.
|
|
|
|
Returns an OOB swap updating `#leave-list` and triggers `leaveUpdated`.
|
|
"""
|
|
worker = get_object_or_404(Worker, pk=worker_id)
|
|
|
|
# permission checks (same as leave_delete)
|
|
allowed = False
|
|
token = request.POST.get('token') or request.GET.get('token')
|
|
try:
|
|
if request.user and request.user.is_authenticated and request.user.is_staff:
|
|
allowed = True
|
|
elif request.user and request.user.is_authenticated and getattr(request.user, 'email', None) and request.user.email == (worker.email or ''):
|
|
allowed = True
|
|
elif token:
|
|
try:
|
|
import uuid
|
|
|
|
token_uuid = uuid.UUID(token)
|
|
if worker.request_token == token_uuid:
|
|
allowed = True
|
|
except Exception:
|
|
allowed = False
|
|
except Exception:
|
|
allowed = False
|
|
|
|
if not allowed:
|
|
return HttpResponseBadRequest('Permission denied')
|
|
|
|
# delete all leaves for this worker
|
|
worker.leaves.all().delete()
|
|
|
|
# render updated leave list for the worker
|
|
leave_list_html = render_to_string(
|
|
"rota/partials/leave_list.html",
|
|
{"leaves": worker.leaves.all(), "token": token},
|
|
request=request,
|
|
)
|
|
resp = f'<div id="leave-list" hx-swap-oob="true">{leave_list_html}</div>'
|
|
response = HttpResponse(resp)
|
|
response["HX-Trigger"] = "leaveUpdated"
|
|
return response
|
|
|
|
|
|
@require_POST
|
|
def rota_run_export_regenerate(request, run_id):
|
|
"""Regenerate the HTML export for a completed run and persist it on the RotaRun.
|
|
|
|
This is a POST endpoint — regenerating may be slow if the solver is invoked.
|
|
"""
|
|
from .models import RotaRun
|
|
|
|
run = get_object_or_404(RotaRun, pk=run_id)
|
|
rota = run.rota
|
|
solve = request.POST.get("solve") == "1"
|
|
|
|
try:
|
|
rb = rota.to_rota_builder()
|
|
rb.build_and_solve(solve=solve)
|
|
html = rb.get_worker_timetable_html(include_html_tag=True)
|
|
run.export_html = html
|
|
run.save(update_fields=["export_html"])
|
|
except Exception as exc:
|
|
tb = traceback.format_exc()
|
|
run.log = (run.log or "") + "\n" + tb
|
|
run.save(update_fields=["log"])
|
|
return HttpResponse(f"Failed to regenerate export: {exc}", status=500)
|
|
|
|
# Redirect to the HTML view so user sees the generated page
|
|
return HttpResponseRedirect(reverse("rota:rota_run_export_html", args=(run.id,)))
|
|
|
|
|
|
def rota_export_builder(request, rota_id):
|
|
"""Build a `RotaBuilder` for the given `RotaSchedule` and return its HTML export.
|
|
|
|
Query parameters:
|
|
- solve=1 : attempt to solve the model (may require solver binaries and can be slow)
|
|
- solver=<name> : solver name passed to `build_and_solve`
|
|
"""
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
|
|
try:
|
|
rb = rota.to_rota_builder()
|
|
except Exception as exc:
|
|
return HttpResponseBadRequest(f"Unable to construct RotaBuilder: {exc}")
|
|
|
|
solve = request.GET.get("solve") == "1"
|
|
solver = request.GET.get("solver", "appsi_highs")
|
|
|
|
try:
|
|
# Build model (and optionally solve) - keep defaults safe
|
|
rb.build_and_solve(solve=solve, solver=solver)
|
|
except Exception as exc:
|
|
# Return error message in the page so it's visible in browser
|
|
content = f"<html><body><h1>Error running builder</h1><pre>{exc}</pre></body></html>"
|
|
return HttpResponse(content, status=500, content_type="text/html")
|
|
|
|
html = rb.get_worker_timetable_html(include_html_tag=True)
|
|
return HttpResponse(html, content_type="text/html; charset=utf-8")
|
|
|
|
|
|
@require_POST
|
|
def rota_runs_clear(request, rota_id):
|
|
"""Delete all RotaRun records for the given rota and redirect back.
|
|
|
|
This is a POST-only endpoint; the template uses an inline form with a
|
|
JavaScript confirmation to avoid accidental removal.
|
|
"""
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
# Delete all runs for this rota
|
|
from .models import RotaRun
|
|
|
|
RotaRun.objects.filter(rota=rota).delete()
|
|
return redirect("rota:rota_detail", rota_id=rota.id)
|
|
|
|
|
|
@require_http_methods(["GET", "POST"])
|
|
def shift_add(request, rota_id):
|
|
"""HTMX endpoint to render a shift form or accept submission to append a shift to the rota."""
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
|
|
if request.method == "POST":
|
|
form = ShiftForm(request.POST)
|
|
if form.is_valid():
|
|
data = form.cleaned_data
|
|
shift_dict = {
|
|
"sites": data["sites"],
|
|
"name": data["name"],
|
|
"length": float(data["length"]),
|
|
"days": data["days"],
|
|
"workers_required": int(data["workers_required"]),
|
|
"assign_as_block": bool(data["assign_as_block"]),
|
|
"balance_offset": data.get("balance_offset"),
|
|
"constraints": data.get("constraints", []),
|
|
}
|
|
|
|
current = rota.shifts or []
|
|
current.append(shift_dict)
|
|
rota.shifts = current
|
|
rota.save()
|
|
|
|
# Return out-of-band swap: update shift-list and clear modal
|
|
shift_list_html = render_to_string("rota/partials/shift_list.html", {"rota": rota}, request=request)
|
|
# wrap shift list for OOB swap and clear form/modal containers
|
|
resp = (
|
|
f'<div id="shift-list" hx-swap-oob="true">{shift_list_html}</div>'
|
|
+ '<div id="shift-form" hx-swap-oob="true"></div>'
|
|
)
|
|
response = HttpResponse(resp)
|
|
response["HX-Trigger"] = "closeModal"
|
|
return response
|
|
|
|
# invalid: return modal with form and errors
|
|
modal_html = render_to_string("rota/partials/shift_form_modal.html", {"form": form, "rota": rota, "form_action": request.path, "focus": request.GET.get('focus') or request.POST.get('focus')}, request=request)
|
|
return HttpResponse(modal_html)
|
|
|
|
else:
|
|
form = ShiftForm()
|
|
modal_html = render_to_string("rota/partials/shift_form_modal.html", {"form": form, "rota": rota, "form_action": request.path}, request=request)
|
|
return HttpResponse(modal_html)
|
|
|
|
|
|
@require_http_methods(["GET", "POST"])
|
|
def shift_edit(request, rota_id, idx):
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
try:
|
|
shift = rota.shifts[idx]
|
|
except Exception:
|
|
return HttpResponseBadRequest("Shift not found")
|
|
|
|
if request.method == "POST":
|
|
form = ShiftForm(request.POST)
|
|
if form.is_valid():
|
|
data = form.cleaned_data
|
|
shift_dict = {
|
|
"sites": data["sites"],
|
|
"name": data["name"],
|
|
"length": float(data["length"]),
|
|
"days": data["days"],
|
|
"workers_required": int(data["workers_required"]),
|
|
"assign_as_block": bool(data["assign_as_block"]),
|
|
"constraints": data.get("constraints", []),
|
|
}
|
|
current = rota.shifts or []
|
|
current[idx] = shift_dict
|
|
rota.shifts = current
|
|
rota.save()
|
|
|
|
shift_list_html = render_to_string("rota/partials/shift_list.html", {"rota": rota}, request=request)
|
|
resp = (
|
|
f'<div id="shift-list" hx-swap-oob="true">{shift_list_html}</div>'
|
|
+ '<div id="shift-form" hx-swap-oob="true"></div>'
|
|
)
|
|
response = HttpResponse(resp)
|
|
response["HX-Trigger"] = "closeModal"
|
|
return response
|
|
|
|
modal_html = render_to_string("rota/partials/shift_form_modal.html", {"form": form, "rota": rota, "form_action": request.path}, request=request)
|
|
return HttpResponse(modal_html)
|
|
|
|
# GET: populate form with existing shift
|
|
initial = {
|
|
"name": shift.get("name"),
|
|
"sites": ",".join(shift.get("sites") or []),
|
|
"length": shift.get("length"),
|
|
"days": shift.get("days"),
|
|
"workers_required": shift.get("workers_required"),
|
|
"assign_as_block": shift.get("assign_as_block", False),
|
|
"balance_offset": shift.get("balance_offset", None),
|
|
"constraints": json.dumps(shift.get("constraints", [])) if shift.get("constraints") is not None else "",
|
|
}
|
|
form = ShiftForm(initial=initial)
|
|
modal_html = render_to_string("rota/partials/shift_form_modal.html", {"form": form, "rota": rota, "form_action": request.path, "focus": request.GET.get('focus')}, request=request)
|
|
return HttpResponse(modal_html)
|
|
|
|
|
|
@require_http_methods(["GET", "POST"])
|
|
def shift_delete(request, rota_id, idx):
|
|
rota = get_object_or_404(RotaSchedule, pk=rota_id)
|
|
try:
|
|
shift = rota.shifts[idx]
|
|
except Exception:
|
|
return HttpResponseBadRequest("Shift not found")
|
|
|
|
if request.method == "POST":
|
|
current = rota.shifts or []
|
|
try:
|
|
current.pop(idx)
|
|
except Exception:
|
|
return HttpResponseBadRequest("Invalid index")
|
|
rota.shifts = current
|
|
rota.save()
|
|
shift_list_html = render_to_string("rota/partials/shift_list.html", {"rota": rota}, request=request)
|
|
resp = (
|
|
f'<div id="shift-list" hx-swap-oob="true">{shift_list_html}</div>'
|
|
+ '<div id="shift-form" hx-swap-oob="true"></div>'
|
|
)
|
|
response = HttpResponse(resp)
|
|
response["HX-Trigger"] = "closeModal"
|
|
return response
|
|
|
|
# GET: confirmation modal
|
|
modal_html = render_to_string("rota/partials/shift_confirm_delete.html", {"rota": rota, "idx": idx, "shift": shift}, request=request)
|
|
return HttpResponse(modal_html)
|
|
|