Ross
1190 lines
39 KiB
Python
1190 lines
39 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 .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():
|
|
# 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
|
|
configured_constraints[k] = {"value": v, "pretty": pretty, "typed": (k in available_constraints_rich), "description": desc}
|
|
except Exception:
|
|
configured_constraints = {}
|
|
|
|
# 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},
|
|
)
|
|
|
|
|
|
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})
|
|
|
|
|
|
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()
|
|
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()
|
|
|
|
return render(request, "rota/worker_detail.html", {"worker": worker, "form": form})
|
|
|
|
|
|
@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},
|
|
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 ''},
|
|
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)
|
|
return render(request, "rota/rota_run_detail.html", {"run": run})
|
|
|
|
|
|
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
|
|
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})
|
|
|
|
|
|
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})
|
|
|
|
|
|
@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 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}, 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}, 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)
|
|
|