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 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", )