Ross
394 lines
15 KiB
Python
394 lines
15 KiB
Python
import csv
|
|
from datetime import datetime, timedelta
|
|
import re
|
|
import sys
|
|
from requests import Session
|
|
from loguru import logger
|
|
|
|
from rich.pretty import pprint
|
|
|
|
|
|
from rota_generator.workers import (
|
|
Worker,
|
|
NotAvailableToWork,
|
|
NonWorkingDays,
|
|
WorkRequests,
|
|
PreferenceNotToWork,
|
|
OutOfProgramme,
|
|
)
|
|
|
|
date_re = r"[\d]{1,2}\/[\d]{1,2}\/[\d]{2}"
|
|
|
|
|
|
live_rota = True
|
|
|
|
|
|
def load_leave(Rota):
|
|
with Session() as s:
|
|
shifts = Rota.get_shift_names()
|
|
|
|
workers = {}
|
|
|
|
headers = {"Cache-Control": "no-cache", "Pragma": "no-cache"}
|
|
|
|
if live_rota:
|
|
download = s.get(
|
|
"https://docs.google.com/spreadsheets/d/e/2PACX-1vSh64r9uUbPsCdhcV7zmZz10gVqIy6rhXFUvDgJxRD5W45IYEaZpKMxAqDsD1ph6dob4AiRzJXVVtOu/pub?gid=580824586&single=true&output=csv",
|
|
headers=headers,
|
|
)
|
|
# download = s.get("https://docs.google.com/spreadsheets/d/e/2PACX-1vSRx9VWXSlRubPyA0RhiI-Oqf5eHNYYEc6rFzlraDbR5_8qqr5g13-4uV-gn4u-TjZxiSMv1fBUaESq/pub?gid=814517272&single=true&output=csv", headers=headers)
|
|
decoded_content = download.content.decode("utf-8")
|
|
|
|
reader = csv.reader(decoded_content.splitlines(), delimiter=",")
|
|
|
|
else:
|
|
with open("/home/ross/Downloads/PROC Draft RK version.csv") as f:
|
|
reader = csv.reader(f.read().splitlines(), delimiter=",")
|
|
|
|
n = 0
|
|
leave = {}
|
|
requests = {}
|
|
site_prefs = {}
|
|
|
|
order = []
|
|
|
|
n = 1
|
|
for row in reader:
|
|
if n < 10:
|
|
pass
|
|
#print(row)
|
|
row_title = row[0]
|
|
#print(row_title)
|
|
row_date = row[0]
|
|
row_extra = row[1]
|
|
r = row[2:]
|
|
# header, extract names
|
|
if n == 1:
|
|
names = r
|
|
order = names
|
|
for name in names:
|
|
name = name.strip()
|
|
|
|
workers[name] = {}
|
|
workers[name]["leave"] = []
|
|
workers[name]["requests"] = []
|
|
workers[name]["site_pref"] = []
|
|
workers[name]["shift_balance_extra"] = {}
|
|
workers[name]["bank_holiday_extra"] = 0
|
|
workers[name]["end_date"] = None
|
|
workers[name]["start_date"] = None
|
|
workers[name]["pair"] = None
|
|
|
|
n = n + 1
|
|
continue
|
|
|
|
for i in range(len(r)):
|
|
try:
|
|
a = order[i].strip()
|
|
except IndexError:
|
|
continue
|
|
|
|
worker = workers[a]
|
|
|
|
lower_item = r[i].lower().strip()
|
|
if lower_item == "derriford":
|
|
lower_item = "plymouth"
|
|
|
|
if lower_item == "derriford twilights":
|
|
lower_item = "plymouth_twilights"
|
|
|
|
if (
|
|
"PROC site" in row_title
|
|
or row_title == "PROC nights from"
|
|
or "proc nights" in row_title.lower()
|
|
):
|
|
worker["site_pref"] = lower_item
|
|
|
|
elif row_title in (
|
|
"Rotation",
|
|
"Site",
|
|
"Placement",
|
|
"Placement location",
|
|
"Placement location",
|
|
):
|
|
worker["site"] = lower_item
|
|
|
|
elif (
|
|
row_title in ("Grade", "Grade (ST)")
|
|
or "Year of training" in row_title
|
|
):
|
|
worker["grade"] = lower_item.lstrip("ST")
|
|
|
|
elif (
|
|
row_title
|
|
in (
|
|
"%FTE",
|
|
"%FTE on-call",
|
|
"FTE",
|
|
"FTE on-call",
|
|
"Full-time equivalent (FTE)",
|
|
)
|
|
or "on-call commitment" in row_title
|
|
):
|
|
if lower_item:
|
|
if lower_item.endswith("%"):
|
|
lower_item = float(lower_item.rstrip("%")) / 100
|
|
worker["fte"] = lower_item
|
|
|
|
elif (
|
|
"NWD" in row_title or row_title == "If <100% FTE, specify days off:"
|
|
):
|
|
worker["nwd"] = lower_item
|
|
|
|
elif row_title == "Flexible NWD":
|
|
worker["flexible_nwd"] = lower_item
|
|
|
|
elif row_title in ("End Date", "CCT date", "CCT Date"):
|
|
if lower_item:
|
|
try:
|
|
date = datetime.strptime(lower_item, "%d/%m/%y").date()
|
|
except ValueError:
|
|
try:
|
|
date = datetime.strptime(lower_item, "%d-%m-%y").date()
|
|
except ValueError:
|
|
try:
|
|
date = datetime.strptime(
|
|
lower_item, "%d/%m/%Y"
|
|
).date()
|
|
except ValueError:
|
|
print(
|
|
f"Cannot parse date: '{lower_item}' (n: {n}, row: {row})"
|
|
)
|
|
sys.exit(1)
|
|
|
|
worker["end_date"] = date
|
|
else:
|
|
worker["end_date"] = None
|
|
|
|
elif row_title in ("Start date",):
|
|
if lower_item:
|
|
try:
|
|
date = datetime.strptime(lower_item, "%d/%m/%y").date()
|
|
except ValueError:
|
|
date = datetime.strptime(lower_item, "%d/%m/%Y").date()
|
|
except ValueError:
|
|
logger.warning(
|
|
f"Cannot parse date: {lower_item} (n: {n}, row: {row})"
|
|
)
|
|
raise ValueError(f"Cannot parse date: {lower_item}")
|
|
worker["start_date"] = date
|
|
else:
|
|
worker["start_date"] = None
|
|
|
|
elif "OOP" in row_title or "out of programme" in row_title.lower():
|
|
if lower_item in (
|
|
"OOPT 1 START",
|
|
"OOPT 1 END",
|
|
"OOPT 2 START",
|
|
"OOPT 2 END",
|
|
):
|
|
continue
|
|
worker["oop"] = lower_item
|
|
|
|
elif row_title == "Group":
|
|
worker["group"] = lower_item
|
|
|
|
elif row_title == "Pair":
|
|
if lower_item == "":
|
|
worker["pair"] = None
|
|
else:
|
|
worker["pair"] = int(lower_item)
|
|
|
|
elif row_title.startswith("Extra_"):
|
|
if lower_item:
|
|
shift = row_title[len("Extra_") :]
|
|
if shift == "twilight":
|
|
shift = f"{worker['site']}_{shift}"
|
|
elif shift == "weekend":
|
|
shift = f"weekend_{worker['site']}"
|
|
worker["shift_balance_extra"][shift] = float(lower_item)
|
|
|
|
elif row_title == "bank_holidays_worked":
|
|
if lower_item:
|
|
worker["bank_holiday_extra"] = int(lower_item)
|
|
|
|
elif row_title == "Weekends/Twilights":
|
|
if lower_item:
|
|
worker["site"] = lower_item
|
|
|
|
|
|
#elif row_title == "Academy":
|
|
# pass
|
|
#elif "night specific" in row_title:
|
|
# if lower_item:
|
|
# for shift in lower_item.split(","):
|
|
# shift = shift.strip()
|
|
# if shift == "nights":
|
|
# shift = "night_weekday"
|
|
# elif shift == "nights only":
|
|
# if date.weekday() < 5:
|
|
# shift = "night_weekday"
|
|
# else:
|
|
# shift = "night_weekend"
|
|
# worker["single_nights"].append(
|
|
# WorkRequests(date=date, shift=shift)
|
|
# )
|
|
|
|
|
|
elif re.match(date_re, row_date) is not None:
|
|
if lower_item != "":
|
|
try:
|
|
try:
|
|
date = datetime.strptime(row_date, "%d/%m/%y").date()
|
|
except ValueError:
|
|
date = datetime.strptime(row_date, "%d/%m/%Y").date()
|
|
|
|
# This may be easier to do as a dict
|
|
if lower_item in shifts:
|
|
worker["requests"].append(
|
|
WorkRequests(date=date, shift=lower_item)
|
|
)
|
|
elif lower_item in ("nights only"):
|
|
if date.weekday() < 5:
|
|
worker["requests"].append(
|
|
WorkRequests(date=date, shift="night_weekday")
|
|
)
|
|
else:
|
|
worker["requests"].append(
|
|
WorkRequests(date=date, shift="night_weekend")
|
|
)
|
|
|
|
elif lower_item in (
|
|
"work_request",
|
|
"happy to work",
|
|
"offered",
|
|
"may be",
|
|
"volunteered",
|
|
):
|
|
worker["requests"].append(
|
|
WorkRequests(date=date, shift="*")
|
|
)
|
|
|
|
else:
|
|
worker["leave"].append(
|
|
NotAvailableToWork(date=date, reason=lower_item)
|
|
)
|
|
except ValueError as e:
|
|
print(f"Error with: {lower_item}")
|
|
print(f"{row=}")
|
|
raise e
|
|
|
|
n = n + 1
|
|
|
|
return workers
|
|
|
|
|
|
def load_academy(Rota):
|
|
"""Load academy month blocks from the main published CSV and register avoids.
|
|
|
|
Spreadsheet layout expected:
|
|
- Column A: row title (should be 'Academy' for relevant rows)
|
|
- Column B: start date for the month (e.g. '07/09/2026')
|
|
- Columns C...: worker columns matching the header row (first CSV row's columns C...)
|
|
Cells containing the word 'academy' (case-insensitive) mark that worker as being in academy
|
|
for that month.
|
|
"""
|
|
# Use the same published CSV as load_leave (main sheet)
|
|
url = (
|
|
"https://docs.google.com/spreadsheets/d/e/2PACX-1vSh64r9uUbPsCdhcV7zmZz10gVqIy6rhXFUvDgJxRD5W45IYEaZpKMxAqDsD1ph6dob4AiRzJXVVtOu/pub?gid=580824586&single=true&output=csv"
|
|
)
|
|
|
|
with Session() as s:
|
|
headers = {"Cache-Control": "no-cache", "Pragma": "no-cache"}
|
|
download = s.get(url, headers=headers)
|
|
decoded = download.content.decode("utf-8")
|
|
|
|
reader = csv.reader(decoded.splitlines(), delimiter=",")
|
|
|
|
names = []
|
|
month_starts: list[datetime.date] = []
|
|
month_cells: list[list[str]] = []
|
|
|
|
n = 0
|
|
for row in reader:
|
|
n += 1
|
|
if not row:
|
|
continue
|
|
|
|
# header row: collect worker names from column C onwards
|
|
if n == 1:
|
|
# some files may have two leading columns (like 'Email Address' and blank/extra), so take from index 2
|
|
names = [c.strip() for c in (row[2:] if len(row) > 2 else row[1:])]
|
|
continue
|
|
|
|
row_title = row[0].strip() if len(row) > 0 else ""
|
|
start_cell = row[1].strip() if len(row) > 1 else ""
|
|
cells = row[2:] if len(row) > 2 else []
|
|
|
|
# Only treat rows where the first column mentions 'Academy' (some sheets may include other rows)
|
|
if row_title and "academy" in row_title.lower():
|
|
# parse start date
|
|
try:
|
|
try:
|
|
start_date = datetime.strptime(start_cell, "%d/%m/%Y").date()
|
|
except ValueError:
|
|
start_date = datetime.strptime(start_cell, "%d/%m/%y").date()
|
|
except Exception:
|
|
# skip rows without a valid date
|
|
continue
|
|
|
|
month_starts.append(start_date)
|
|
# normalize cell list to names length
|
|
row_cells = [(cells[i] if i < len(cells) else "").strip() for i in range(len(names))]
|
|
month_cells.append(row_cells)
|
|
|
|
if not month_starts:
|
|
return
|
|
|
|
# determine twilight shifts available
|
|
twilight_shift_names = [s.name for s in Rota.get_shifts() if "twilight" in s.name]
|
|
|
|
# for each month start, compute end date and register avoids
|
|
for idx, start_date in enumerate(month_starts):
|
|
if idx + 1 < len(month_starts):
|
|
end_date = month_starts[idx + 1] - timedelta(days=1)
|
|
else:
|
|
end_date = Rota.rota_end_date
|
|
|
|
cells = month_cells[idx]
|
|
for i, cell in enumerate(cells):
|
|
if i >= len(names):
|
|
break
|
|
v = (cell or "").lower()
|
|
if "academy" in v:
|
|
worker_name = names[i]
|
|
matching_workers = [w for w in Rota.workers if w.name.strip() == worker_name.strip()]
|
|
if not matching_workers:
|
|
logger.warning(f"Academy data: worker '{worker_name}' not found in rota, skipping")
|
|
continue
|
|
worker = matching_workers[0]
|
|
|
|
# choose twilight shifts relevant to the worker's site
|
|
candidate_shifts = []
|
|
for sh_name in twilight_shift_names:
|
|
try:
|
|
sh = Rota.get_shift_by_name(sh_name)
|
|
except Exception:
|
|
continue
|
|
if hasattr(worker, "site") and worker.site in getattr(sh, "sites", []):
|
|
candidate_shifts.append(sh_name)
|
|
elif sh_name.startswith(f"{worker.site}_"):
|
|
candidate_shifts.append(sh_name)
|
|
|
|
if not candidate_shifts:
|
|
candidate_shifts = twilight_shift_names.copy()
|
|
|
|
#print(f"Academy: {worker.name} on academy from {start_date} to {end_date}, avoiding shifts: {candidate_shifts}")
|
|
|
|
Rota.add_avoid_shifts_for_workers_on_dates(
|
|
names=[worker.name],
|
|
shifts=candidate_shifts,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
reason="academy",
|
|
) |