537 lines
17 KiB
Python
537 lines
17 KiB
Python
"""Google Sheets helpers — credential loading, retrying API calls, row/record shaping.
|
|
|
|
No FastAPI imports here by house rule: this module raises its own SheetsServiceError
|
|
family and lets g_sheet/views.py translate that into HTTPException.
|
|
|
|
Auth reuses the credentials already on disk (authorized_user ADC + a valid refresh
|
|
token). Nothing here launches a browser, runs InstalledAppFlow, or reads stdin.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import random
|
|
import re
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from dotenv import load_dotenv
|
|
from google.auth import default as google_auth_default
|
|
from google.auth.transport.requests import Request
|
|
from googleapiclient.discovery import build
|
|
from googleapiclient.errors import HttpError
|
|
|
|
from g_sheet.enums import (
|
|
ConductedByAlias,
|
|
DateFormat,
|
|
DateTimeSeparator,
|
|
FIELD_ALIAS_ENUMS,
|
|
FormDataField,
|
|
MonthNormalisation,
|
|
NotesToken,
|
|
RoundByColumn,
|
|
RoundDateColumn,
|
|
RoundNotesColumn,
|
|
RoundOrdinal,
|
|
RoundResultColumn,
|
|
RoundRole,
|
|
RoundStatusColumn,
|
|
RoundTimeColumn,
|
|
)
|
|
|
|
load_dotenv()
|
|
|
|
logger=logging.getLogger("g_sheet.plugins")
|
|
|
|
# backend/ — GOOGLE_APPLICATION_CREDENTIALS is stored relative to it ("credentials/...").
|
|
ROOT=Path(__file__).resolve().parent.parent
|
|
|
|
SCOPES=[
|
|
"https://www.googleapis.com/auth/spreadsheets",
|
|
"https://www.googleapis.com/auth/drive",
|
|
]
|
|
|
|
SPREADSHEET_ID=os.getenv("SPREADSHEET_ID")
|
|
SPREADSHEET_NAME=os.getenv("SPREADSHEET_NAME")
|
|
SPREADSHEET_URL=os.getenv("SPREADSHEET_URL")
|
|
GOOGLE_APPLICATION_CREDENTIALS=os.getenv("GOOGLE_APPLICATION_CREDENTIALS")
|
|
|
|
# 429 and 5xx are transient; every other 4xx is a bad request that a retry repeats.
|
|
RETRY_ATTEMPTS=3
|
|
RETRY_BASE_DELAY=0.5
|
|
RETRY_MAX_DELAY=8.0
|
|
RETRYABLE_STATUSES={429,500,502,503,504}
|
|
|
|
|
|
class SheetsServiceError(Exception):
|
|
"""Base for every failure this domain raises. Carries an HTTP-ish status code."""
|
|
|
|
status_code=500
|
|
|
|
def __init__(self,message,status_code=None):
|
|
super().__init__(message)
|
|
self.message=message
|
|
if status_code is not None:
|
|
self.status_code=status_code
|
|
|
|
|
|
class SheetsAuthError(SheetsServiceError):
|
|
"""Credentials missing, unreadable, or rejected by Google."""
|
|
|
|
status_code=401
|
|
|
|
|
|
class SheetsApiError(SheetsServiceError):
|
|
"""The Sheets API answered with an error. status_code is Google's own."""
|
|
|
|
status_code=502
|
|
|
|
|
|
def resolve_credentials_path(credentials_path=None):
|
|
"""Absolute path to the ADC json. Relative values resolve against backend/.
|
|
|
|
The service may be imported from any working directory, so a bare
|
|
"credentials/application_default_credentials.json" must not depend on cwd.
|
|
"""
|
|
raw=credentials_path or GOOGLE_APPLICATION_CREDENTIALS
|
|
if not raw:
|
|
return None
|
|
path=Path(raw)
|
|
if not path.is_absolute():
|
|
path=ROOT/path
|
|
return path
|
|
|
|
|
|
def load_credentials(credentials_path=None,scopes=None):
|
|
"""Build scoped ADC credentials and refresh them once. Never prompts."""
|
|
path=resolve_credentials_path(credentials_path)
|
|
if path is not None:
|
|
if not path.exists():
|
|
raise SheetsAuthError(f"Google credentials file not found: {path.name}")
|
|
os.environ["GOOGLE_APPLICATION_CREDENTIALS"]=str(path)
|
|
try:
|
|
credentials,_=google_auth_default(scopes=scopes or SCOPES)
|
|
credentials.refresh(Request())
|
|
except SheetsServiceError:
|
|
raise
|
|
except Exception as e:
|
|
raise SheetsAuthError(f"Google credential refresh failed: {e}")
|
|
return credentials
|
|
|
|
|
|
def ensure_fresh(credentials):
|
|
"""Refresh only when the token has actually gone stale — not on every call."""
|
|
if credentials is None:
|
|
raise SheetsAuthError("Google credentials are not initialised")
|
|
if credentials.valid and not credentials.expired:
|
|
return credentials
|
|
try:
|
|
credentials.refresh(Request())
|
|
except Exception as e:
|
|
raise SheetsAuthError(f"Google credential refresh failed: {e}")
|
|
return credentials
|
|
|
|
|
|
def build_sheets_client(credentials):
|
|
"""Sheets v4 client. cache_discovery=False — the file cache warns under threads."""
|
|
try:
|
|
return build("sheets","v4",credentials=credentials,cache_discovery=False)
|
|
except Exception as e:
|
|
raise SheetsApiError(f"Could not build the Sheets client: {e}")
|
|
|
|
|
|
def _status_of(error):
|
|
status=getattr(getattr(error,"resp",None),"status",None)
|
|
if status is None:
|
|
status=getattr(error,"status_code",None)
|
|
try:
|
|
return int(status)
|
|
except (TypeError,ValueError):
|
|
return None
|
|
|
|
|
|
def _reason_of(error):
|
|
"""Google's message without the response body, so nothing sensitive leaks out."""
|
|
try:
|
|
return error._get_reason().strip()
|
|
except Exception:
|
|
return str(error)
|
|
|
|
|
|
def execute(request,description="sheets request"):
|
|
"""Run a googleapiclient request with jittered exponential backoff.
|
|
|
|
Retries 429 and 5xx up to RETRY_ATTEMPTS; every other HttpError raises straight
|
|
away as SheetsApiError carrying Google's status code.
|
|
"""
|
|
delay=RETRY_BASE_DELAY
|
|
last_error=None
|
|
for attempt in range(1,RETRY_ATTEMPTS+1):
|
|
try:
|
|
return request.execute()
|
|
except HttpError as e:
|
|
status=_status_of(e)
|
|
reason=_reason_of(e)
|
|
last_error=SheetsApiError(f"{description} failed: {reason}",status or 502)
|
|
if status not in RETRYABLE_STATUSES or attempt==RETRY_ATTEMPTS:
|
|
raise last_error
|
|
sleep_for=min(delay,RETRY_MAX_DELAY)+random.uniform(0,RETRY_BASE_DELAY)
|
|
logger.warning(
|
|
"%s got %s, retry %s/%s in %.2fs",
|
|
description,status,attempt,RETRY_ATTEMPTS,sleep_for,
|
|
)
|
|
time.sleep(sleep_for)
|
|
delay*=2
|
|
except SheetsServiceError:
|
|
raise
|
|
except Exception as e:
|
|
raise SheetsApiError(f"{description} failed: {e}")
|
|
raise last_error
|
|
|
|
|
|
def quote_tab(tab,cell_range=None):
|
|
"""A1 target for a tab whose name may contain spaces or quotes."""
|
|
safe=str(tab).replace("'","''")
|
|
if cell_range:
|
|
return f"'{safe}'!{cell_range}"
|
|
return f"'{safe}'"
|
|
|
|
|
|
def normalise_headers(header_row):
|
|
"""First row -> unique, non-empty column keys.
|
|
|
|
Blank cells become column_{i}; a repeated header keeps its first spelling and the
|
|
later ones get _1, _2 so no key silently overwrites another.
|
|
"""
|
|
headers=[]
|
|
seen={}
|
|
for index,raw in enumerate(header_row):
|
|
name=str(raw).strip() if raw is not None else ""
|
|
if not name:
|
|
name=f"column_{index}"
|
|
count=seen.get(name,0)
|
|
seen[name]=count+1
|
|
headers.append(name if count==0 else f"{name}_{count}")
|
|
return headers
|
|
|
|
|
|
def rows_to_records(rows):
|
|
"""Sheet rows -> list of dicts keyed by the header row.
|
|
|
|
Sheets truncates trailing empties, so short rows are padded to header width.
|
|
Fully blank rows are dropped rather than emitted as all-empty records.
|
|
"""
|
|
if not rows:
|
|
return []
|
|
headers=normalise_headers(rows[0])
|
|
records=[]
|
|
for row in rows[1:]:
|
|
values=[str(cell) if cell is not None else "" for cell in row]
|
|
if not any(value.strip() for value in values):
|
|
continue
|
|
if len(values)<len(headers):
|
|
values=values+[""]*(len(headers)-len(values))
|
|
records.append(dict(zip(headers,values[:len(headers)])))
|
|
return records
|
|
|
|
|
|
def stringify_rows(rows):
|
|
"""Normalise raw values() output into list[list[str]] with no None holes."""
|
|
return [[str(cell) if cell is not None else "" for cell in row] for row in rows or []]
|
|
|
|
|
|
# -- FormData mapping ------------------------------------------------------
|
|
|
|
BY_FIELDS=RoundByColumn.ordered()
|
|
DATE_FIELDS=RoundDateColumn.ordered()
|
|
TIME_FIELDS=RoundTimeColumn.ordered()
|
|
STATUS_FIELDS=RoundStatusColumn.ordered()
|
|
NOTES_FIELDS=RoundNotesColumn.ordered()
|
|
RESULT_FIELDS=RoundResultColumn.ordered()
|
|
|
|
_ORDINAL_PATTERNS=tuple(
|
|
(member.slot,re.compile(rf"\b{member.pattern}\b",re.I))
|
|
for member in RoundOrdinal
|
|
)
|
|
|
|
_TIME_RE=re.compile(r"(\d{1,2}:\d{2}\s*(?:[AaPp][Mm])?)")
|
|
_DAY_ORDINAL_RE=re.compile(r"\b(\d+)(st|nd|rd|th)\b",re.I)
|
|
_DIGIT_RE=re.compile(r"\d")
|
|
_AGE_RE=re.compile(r"\d+")
|
|
|
|
|
|
def canonical_header(h):
|
|
"""Lower, collapse whitespace (incl. embedded newlines), strip _N and (tails)."""
|
|
text=str(h or "").replace("\n"," ").replace("\r"," ")
|
|
text=re.sub(r"\s+"," ",text).strip().lower()
|
|
text=re.sub(r"_\d+$","",text)
|
|
text=re.sub(r"\s*\([^)]*\)\s*$","",text).strip()
|
|
return text
|
|
|
|
|
|
def match_field(h):
|
|
"""Map a sheet header to a FormDataField, or None.
|
|
|
|
Exact alias first, then startswith. No fuzzy matching — dirty headers mislabel
|
|
more often than they rescue, and a miss is non-fatal (value stays in JSONB).
|
|
"""
|
|
canon=canonical_header(h)
|
|
if not canon:
|
|
return None
|
|
for field,alias_enum in FIELD_ALIAS_ENUMS.items():
|
|
if alias_enum.has(canon):
|
|
return field
|
|
for field,alias_enum in FIELD_ALIAS_ENUMS.items():
|
|
for alias in alias_enum:
|
|
if canon.startswith(alias.value):
|
|
return field
|
|
return None
|
|
|
|
|
|
def resolve_name(record,headers):
|
|
"""Candidate name: alias match, else column A (headers[0]) — always the name."""
|
|
for header in headers:
|
|
if match_field(header)==FormDataField.NAME:
|
|
value=record.get(header)
|
|
if value is not None and str(value).strip():
|
|
return str(value).strip()
|
|
if headers:
|
|
value=record.get(headers[0])
|
|
if value is not None and str(value).strip():
|
|
return str(value).strip()
|
|
return None
|
|
|
|
|
|
def _classify_round_role(canon):
|
|
"""RoundRole for a canonical header, or None for unrecognised headers."""
|
|
if not canon:
|
|
return None
|
|
if ConductedByAlias.contained_in(canon) or ConductedByAlias.has(canon):
|
|
return RoundRole.BY
|
|
if canon.startswith(ConductedByAlias.CONDUCTED.value):
|
|
return RoundRole.BY
|
|
if RoundRole.RESULT.value in canon:
|
|
return RoundRole.RESULT
|
|
if RoundRole.STATUS.value in canon:
|
|
return RoundRole.STATUS
|
|
if NotesToken.contained_in(canon):
|
|
return RoundRole.NOTES
|
|
if RoundRole.DATE.value in canon:
|
|
return RoundRole.DATE
|
|
return None
|
|
|
|
|
|
def _extract_ordinal(canon):
|
|
for slot,pattern in _ORDINAL_PATTERNS:
|
|
if pattern.search(canon):
|
|
return slot
|
|
return None
|
|
|
|
|
|
def resolve_round_columns(headers):
|
|
"""Positional interview-round map: scan left→right into four slots.
|
|
|
|
Ordinal in the header (`2nd`, `second`) pins the slot; otherwise the first free
|
|
slot for that role is taken, never moving backwards. A fifth Results_4 stays
|
|
unmapped (JSONB). Literal-date headers like `19-Feb-2026` classify as nothing.
|
|
"""
|
|
slots=[{role:None for role in RoundRole} for _ in range(4)]
|
|
cursor={role:0 for role in RoundRole}
|
|
|
|
for header in headers:
|
|
canon=canonical_header(header)
|
|
role=_classify_round_role(canon)
|
|
if role is None:
|
|
continue
|
|
ordinal=_extract_ordinal(canon)
|
|
if ordinal is not None:
|
|
if slots[ordinal][role] is None:
|
|
slots[ordinal][role]=header
|
|
continue
|
|
start=cursor[role]
|
|
chosen=None
|
|
for index in range(start,4):
|
|
if slots[index][role] is None:
|
|
chosen=index
|
|
break
|
|
if chosen is None:
|
|
continue
|
|
slots[chosen][role]=header
|
|
cursor[role]=chosen+1
|
|
return slots
|
|
|
|
|
|
def _normalise_month_spellings(text):
|
|
"""strptime %b rejects `Sept`; expand common sheet spellings first."""
|
|
lowered=text.lower()
|
|
for member in MonthNormalisation:
|
|
if member.source in lowered:
|
|
text=re.sub(member.source,member.short,text,flags=re.I)
|
|
lowered=text.lower()
|
|
return text
|
|
|
|
|
|
def parse_date(value):
|
|
"""Tolerant date parse → aware UTC datetime, or None. Never raises."""
|
|
if value is None:
|
|
return None
|
|
text=str(value).strip()
|
|
if not text or not _DIGIT_RE.search(text):
|
|
return None
|
|
|
|
date_part=text
|
|
for sep in DateTimeSeparator:
|
|
if sep.value in text:
|
|
date_part=text.split(sep.value,1)[0].strip()
|
|
break
|
|
# Drop a trailing time when joined without a dash: "6th Nov 2025 7:30 PM"
|
|
time_match=_TIME_RE.search(date_part)
|
|
if time_match and time_match.start()>0:
|
|
date_part=date_part[:time_match.start()].strip(" ,;-")
|
|
|
|
date_part=_DAY_ORDINAL_RE.sub(r"\1",date_part)
|
|
date_part=_normalise_month_spellings(date_part)
|
|
date_part=re.sub(r"\s+"," ",date_part).strip(" ,;")
|
|
|
|
for fmt in DateFormat:
|
|
try:
|
|
return datetime.strptime(date_part,fmt.value).replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def parse_date_time(value):
|
|
"""(datetime|None, time_string|None) — fills *_time for the cells that carry one."""
|
|
parsed=parse_date(value)
|
|
if value is None:
|
|
return parsed,None
|
|
text=str(value).strip()
|
|
match=_TIME_RE.search(text)
|
|
time_str=match.group(1).strip() if match else None
|
|
return parsed,time_str
|
|
|
|
|
|
def parse_age(value):
|
|
"""(int|None, raw|None) — first digit run if 0 < n < 100, always keep the raw."""
|
|
if value is None:
|
|
return None,None
|
|
raw=str(value).strip()
|
|
if not raw:
|
|
return None,None
|
|
match=_AGE_RE.search(raw)
|
|
if not match:
|
|
return None,raw
|
|
number=int(match.group())
|
|
if 0<number<100:
|
|
return number,raw
|
|
return None,raw
|
|
|
|
|
|
def _blank_to_none(value):
|
|
if value is None:
|
|
return None
|
|
text=str(value).strip()
|
|
return text if text else None
|
|
|
|
|
|
def map_record_to_form_data(sheet,record,headers,row_number):
|
|
"""Pure row mapper → kwargs dict for FormData(**...)."""
|
|
rounds=resolve_round_columns(headers)
|
|
mapped={
|
|
"sheet":sheet,
|
|
"row_number":row_number,
|
|
"raw_record":dict(record),
|
|
"name":_blank_to_none(resolve_name(record,headers)),
|
|
"degree":None,
|
|
"experience":None,
|
|
"age":None,
|
|
"age_raw":None,
|
|
"family_details":None,
|
|
}
|
|
for field in BY_FIELDS+TIME_FIELDS+STATUS_FIELDS+NOTES_FIELDS+RESULT_FIELDS:
|
|
mapped[field]=None
|
|
for field in DATE_FIELDS:
|
|
mapped[field]=None
|
|
|
|
for header,value in record.items():
|
|
field=match_field(header)
|
|
if field==FormDataField.DEGREE:
|
|
mapped["degree"]=_blank_to_none(value)
|
|
elif field==FormDataField.EXPERIENCE:
|
|
mapped["experience"]=_blank_to_none(value)
|
|
elif field==FormDataField.AGE:
|
|
age,age_raw=parse_age(value)
|
|
mapped["age"]=age
|
|
mapped["age_raw"]=age_raw
|
|
elif field==FormDataField.FAMILY_DETAILS:
|
|
mapped["family_details"]=_blank_to_none(value)
|
|
|
|
for index,slot in enumerate(rounds):
|
|
if slot.get(RoundRole.DATE):
|
|
dt,tm=parse_date_time(record.get(slot[RoundRole.DATE]))
|
|
mapped[DATE_FIELDS[index]]=dt
|
|
mapped[TIME_FIELDS[index]]=tm
|
|
if slot.get(RoundRole.BY):
|
|
mapped[BY_FIELDS[index]]=_blank_to_none(record.get(slot[RoundRole.BY]))
|
|
if slot.get(RoundRole.STATUS):
|
|
mapped[STATUS_FIELDS[index]]=_blank_to_none(record.get(slot[RoundRole.STATUS]))
|
|
if slot.get(RoundRole.NOTES):
|
|
mapped[NOTES_FIELDS[index]]=_blank_to_none(record.get(slot[RoundRole.NOTES]))
|
|
if slot.get(RoundRole.RESULT):
|
|
mapped[RESULT_FIELDS[index]]=_blank_to_none(record.get(slot[RoundRole.RESULT]))
|
|
|
|
return mapped
|
|
|
|
|
|
def collect_unmapped_headers(headers):
|
|
"""Headers that are neither a typed alias nor claimed by a round slot.
|
|
|
|
`title` aliases are included — they have no FormData column and live in JSONB.
|
|
"""
|
|
rounds=resolve_round_columns(headers)
|
|
claimed=set()
|
|
for slot in rounds:
|
|
for role in RoundRole:
|
|
if slot.get(role):
|
|
claimed.add(slot[role])
|
|
unmapped=[]
|
|
for header in headers:
|
|
if header in claimed:
|
|
continue
|
|
field=match_field(header)
|
|
if field is None or field==FormDataField.TITLE:
|
|
unmapped.append(header)
|
|
return unmapped
|
|
|
|
|
|
def import_row_stats(mapped_rows,headers):
|
|
"""Aggregate parse diagnostics for an import report."""
|
|
dates_parsed=0
|
|
dates_unparsed=0
|
|
ages_parsed=0
|
|
for row in mapped_rows:
|
|
raw=row.get("raw_record") or {}
|
|
rounds=resolve_round_columns(headers)
|
|
for index,slot in enumerate(rounds):
|
|
header=slot.get(RoundRole.DATE)
|
|
if not header:
|
|
continue
|
|
cell=raw.get(header)
|
|
if cell is None or not str(cell).strip():
|
|
continue
|
|
if row.get(DATE_FIELDS[index]) is not None:
|
|
dates_parsed+=1
|
|
elif _DIGIT_RE.search(str(cell)):
|
|
dates_unparsed+=1
|
|
if row.get("age") is not None:
|
|
ages_parsed+=1
|
|
return {
|
|
"dates_parsed":dates_parsed,
|
|
"dates_unparsed":dates_unparsed,
|
|
"ages_parsed":ages_parsed,
|
|
"unmapped_headers":collect_unmapped_headers(headers),
|
|
}
|