From b98c28401f20f508ee77c742d112c168f14d7617 Mon Sep 17 00:00:00 2001 From: Ross Date: Mon, 5 Jan 2026 20:01:41 +0000 Subject: [PATCH] Enhance error handling in leave loading and update date parsing logic; add new site preference for Exeter shifts. --- gen_proc.py | 30 +++++++----- leave.py | 139 ++++++++++++++++++++++++++++++++++++++-------------- 2 files changed, 119 insertions(+), 50 deletions(-) diff --git a/gen_proc.py b/gen_proc.py index 7e52da6..e5c276d 100644 --- a/gen_proc.py +++ b/gen_proc.py @@ -138,6 +138,7 @@ def main( "exeter twilights and weekends", "exeter no nights", "exeter weekends", + "exeter twilights", ), name="weekend_exeter", length=12.5, @@ -303,13 +304,18 @@ def main( site = w["site"] except KeyError: print(f"Worker {worker} has no site") - raise KeyError + print(w) + sys.exit(1) grade = w["grade"] try: fte = float(w["fte"]) * 100 except ValueError: print(f"{worker} has invalid fte: {fte}") raise ValueError + except KeyError: + print(f"Worker {worker} has no FTE") + print(w) + sys.exit(1) nwd = w["nwd"] end_date = w["end_date"] @@ -374,10 +380,10 @@ def main( formatted_oops.append( { "start_date": datetime.datetime.strptime( - s.strip(), "%d/%m/%y" + s.strip(), "%d/%m/%Y" ).date(), "end_date": datetime.datetime.strptime( - e.strip(), "%d/%m/%y" + e.strip(), "%d/%m/%Y" ).date(), } ) @@ -425,21 +431,19 @@ def main( shift_fte_overrides = {} - #if worker_name == "Ben Kemp": - # shift_fte_overrides = { - # "plymouth_twilight": 100, - # "weekend_plymouth1": 50, - # "weekend_plymouth2": 50, - # } + if worker_name == "Hadi Mohamed": + shift_fte_overrides = { + "weekend_exeter": 50, + } #elif worker_name == "Joel Lim": # shift_fte_overrides = { # "plymouth_twilight": 100, # "weekend_exeter": 50, # } - if worker_name == "Nang Thiriphoo": - shift_fte_overrides = { - "weekend_exeter": 40, - } + #if worker_name == "Nang Thiriphoo": + # shift_fte_overrides = { + # "weekend_exeter": 40, + # } w = Worker( name=worker_name, diff --git a/leave.py b/leave.py index 41f5eed..748da45 100644 --- a/leave.py +++ b/leave.py @@ -1,41 +1,49 @@ import csv from datetime import datetime 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 +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): +def load_leave(Rota): with Session() as s: shifts = Rota.get_shift_names() workers = {} - headers = { - "Cache-Control": "no-cache", - "Pragma": "no-cache" - } + headers = {"Cache-Control": "no-cache", "Pragma": "no-cache"} if live_rota: - download = s.get("https://docs.google.com/spreadsheets/d/e/2PACX-1vQtNrp9F1fjtAh8vFX-W_R3RlD9H-2P_vCLMoHrVKR0e24yabbsdrD65VjwxoHlV1Qnatmw1NeghQHG/pub?gid=1199196967&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') + download = s.get( + "https://docs.google.com/spreadsheets/d/e/2PACX-1vQtNrp9F1fjtAh8vFX-W_R3RlD9H-2P_vCLMoHrVKR0e24yabbsdrD65VjwxoHlV1Qnatmw1NeghQHG/pub?gid=1199196967&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=',') + 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=',') + reader = csv.reader(f.read().splitlines(), delimiter=",") n = 0 leave = {} @@ -48,7 +56,7 @@ def load_leave(Rota): for row in reader: if n < 10: print(row) - row_title = row[1] + row_title = row[0] print(row_title) row_date = row[0] r = row[2:] @@ -80,44 +88,81 @@ def load_leave(Rota): worker = workers[a] - lower_item = r[i].lower() + 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(): + 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"): + 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: + 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") or "on-call commitment" in row_title: - if lower_item.endswith("%"): - lower_item = float(lower_item.rstrip("%")) / 100 - worker["fte"] = lower_item + 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: + 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"): + 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: - date = datetime.strptime(lower_item, "%d/%m/%Y").date() + 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: @@ -125,13 +170,22 @@ def load_leave(Rota): 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})") + 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": @@ -157,39 +211,50 @@ def load_leave(Rota): worker["bank_holiday_extra"] = int(lower_item) 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)) + 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")) + 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="*")) + 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)) + 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