proc-rota/leave.py

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-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=',')

        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()