import csv from datetime import datetime import re 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-1vSfcQi5Qs__A8hE2CVvUA_6ULNJALrtZOBHPUZ3xyrxPif9obtQF2IqhioT_4nebjHV1Ac5iqhNtuq4/pub?gid=2024511103&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: print(row) row_title = row[1] print(row_title) row_date = row[0] 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() 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"): 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") or "on-call commitment" in row_title: if lower_item.endswith("%"): lower_item = float(lower_item.rstrip("%")) / 100 worker["fte"] = lower_item elif "NWD" in row_title: worker["nwd"] = lower_item elif row_title == "Flexible NWD": worker["flexible_nwd"] = lower_item elif row_title in ("End 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() 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(): 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 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 # load_leave()