1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
|
import pickle
import os.path
import logging
import re
from os import getenv
from datetime import date
from dataclasses import dataclass
from googleapiclient.discovery import Resource, build
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from data_parser import ProcessedRow
from dataclass_defs import SheetsRow
# If modifying these scopes, delete the file token.pickle.
SCOPES = ["https://www.googleapis.com/auth/spreadsheets"]
def _get_or_default(list, index, default=""):
try:
return list[index]
except IndexError:
return default
def _row_to_dataclass(values):
interval = _get_or_default(values, 0, "")
name = _get_or_default(values, 1, "")
today = date.today()
last_done = _get_or_default(values, 2, today.strftime("%Y-%m-%d"))
return SheetsRow(interval=interval, name=name, last_done=last_done)
def _get_rightmost_column_in_range(range: str) -> str:
# Match range of format Sheet!A1:D4
match = re.search(r"(.+\!)?([A-Z]+?)(\d+)\:([A-Z]+?)(\d+)", range)
sheet_name: str = match[1] or ""
# topleft_col = match[2]
topleft_row = int(match[3])
bottomright_col = match[4]
bottomright_row = int(match[5])
rm_col = bottomright_col
rm_row_start = topleft_row
rm_row_end = bottomright_row
return f"{sheet_name}{rm_col}{rm_row_start}:{rm_col}{rm_row_end}"
def authenticate_sheets() -> Resource:
logging.info("Reading Google credentials and authenticating...")
creds = None
# The file token.pickle stores the user's access and refresh tokens, and is
# created automatically when the authorization flow completes for the first
# time.
if os.path.exists("token.pickle"):
with open("token.pickle", "rb") as token:
creds = pickle.load(token)
# If there are no (valid) credentials available, let the user log in.
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file("credentials.json", SCOPES)
creds = flow.run_local_server(port=0)
# Save the credentials for the next run
with open("token.pickle", "wb") as token:
pickle.dump(creds, token)
return build("sheets", "v4", credentials=creds, cache_discovery=False)
def fetch_sheet_data(service: Resource) -> list[SheetsRow]:
logging.info("Calling Sheets API to fetch data...")
sheet = service.spreadsheets()
result = (
sheet.values()
.get(
spreadsheetId=getenv("SHEETS_SPREADSHEET_ID"),
range=getenv("SHEETS_RANGE"),
)
.execute()
)
rows = result.get("values", [])
logging.info(f"Fetched {len(rows)} rows of data")
return list(map(_row_to_dataclass, rows))
def update_sheet_last_done_column(
service: Resource, processed_rows: list[ProcessedRow]
):
logging.info("Calling Sheets API to update last done column...")
range = getenv("SHEETS_RANGE")
sheet = service.spreadsheets()
last_done_column_range = _get_rightmost_column_in_range(range)
# Update last done date to match next date
body = {"values": list(map(lambda row: [row.next_datestamp], processed_rows))}
sheet.values().update(
spreadsheetId=getenv("SHEETS_SPREADSHEET_ID"),
range=last_done_column_range,
valueInputOption="RAW",
body=body,
).execute()
|
