import requests import pandas as pd import datetime import io from uuid import uuid4 import ics from typing import Tuple from utils import split_csv_row doc_uuid = "1iSA5LsNdarYqrZRxE76lE2VkEsoQksIu6rFG7VAfCNQ" planning_gid = "1001381542" creneau_gid = "1884137958" benevole_gid = "2065754071" def build_sheet_url(doc_id, sheet_id): return f"https://docs.google.com/spreadsheets/d/{doc_id}/export?format=csv&gid={sheet_id}" def downloadAndSave(doc_ui, sheet_gid, fname): url = build_sheet_url(doc_ui, sheet_gid) print("Downloading " + fname) rep = requests.get(url) with open(fname, "wb") as f: f.write(rep.content) def getPlanningDataFrame(csv_filename, starting_date, skip_column=3): list_creneau = [] with io.open(csv_filename, "r", encoding="utf-8") as f: datas = [split_csv_row(s.replace("\n", "")) for s in f.readlines()] def getDate(day: str): s = day.lower() if s.startswith("jeudi"): return starting_date elif s.startswith("vendredi"): return starting_date + datetime.timedelta(days=1) elif s.startswith("samedi"): return starting_date + datetime.timedelta(days=2) elif s.startswith("dimanche"): return starting_date + datetime.timedelta(days=3) elif s.startswith("lundi"): return starting_date + datetime.timedelta(days=4) def getTime(time_str: str): l = time_str.split("h") hours = int(l[0]) if hours < 5: hours += 24 if len(l) > 1 and l[1] != "": return datetime.timedelta(hours=hours, minutes=int(l[1])) else: return datetime.timedelta(hours=hours) def getStartEnd(time_str: str): l = time_str.split("-") if len(l) == 2: return getTime(l[0]), getTime(l[1]) else: start = getTime(time_str.split("+")[0]) return start, start + datetime.timedelta(hours=1) # Parse headers headers = datas[skip_column : skip_column + 2] days = headers[0] hours = headers[1] column_to_dates = {} assert len(hours) == len(hours) for i in range(skip_column, len(days)): days[i] = getDate(days[i]) start, end = getStartEnd(hours[i]) column_to_dates[i] = {"start": days[i] + start, "end": days[i] + end} list_creneau = [] for i in range(5, len(datas)): row = datas[i] current_benevole_name = "" current_time = None for j in range(skip_column, len(row)): if (current_benevole_name != "") and ( row[j] == "" or row[j] != current_benevole_name ): new_creneau = { "id": uuid4(), "description_id": row[0], "nom": row[1], "benevole_nom": current_benevole_name, "ligne": i + 1, **current_time, } list_creneau.append(new_creneau) current_benevole_name = "" current_time = None if row[j] != "": current_benevole_name = row[j] if current_time is None: current_time = column_to_dates[j].copy() else: current_time["end"] = column_to_dates[j]["end"] if current_benevole_name != "": new_creneau = { "id": uuid4(), "nom": row[1], "benevole_nom": current_benevole_name, **current_time, "ligne": i + 1, } list_creneau.append(new_creneau) print(f"{len(list_creneau)} créneaux trouvés") return pd.DataFrame.from_dict(list_creneau) def getContactDataFrame(csv_filename: str, skiprows: int = 2) -> pd.DataFrame: df_contact = pd.read_csv(csv_filename, skiprows=skiprows) column_to_drop = [name for name in df_contact.columns if "Unnamed" in name] df_contact.drop(column_to_drop, axis=1, inplace=True) # Filter out empty name df_contact = df_contact[~df_contact.Nom.isnull()] df_contact.reset_index() return df_contact def getCreneauDataFrame(csv_filename: str) -> pd.DataFrame: df_creneau = pd.read_csv(csv_filename) df_creneau.columns = ["nom", "lieu", "description", "responsable"] return df_creneau def getDataFrameFromGsheet( doc_uuid: str, planning_id="1001381542", creneau_id="1884137958", benevole_id="2065754071", suffix="", starting_date=datetime.datetime(2021, 9, 8), ) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: fname_planning = f"./data/planning{suffix}.csv" fname_creneau = f"./data/creneau{suffix}.csv" fname_contact = f"./data/benevole{suffix}.csv" downloadAndSave(doc_uuid, planning_id, fname_planning) downloadAndSave(doc_uuid, creneau_id, fname_creneau) downloadAndSave(doc_uuid, benevole_id, fname_contact) return ( getPlanningDataFrame(fname_planning, starting_date), getCreneauDataFrame(fname_creneau), getContactDataFrame(fname_contact), )