517 lines
16 KiB
Python
517 lines
16 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 (
|
|
ALIAS_TO_FIELD,
|
|
DateFormat,
|
|
DateTimeSeparator,
|
|
FormDataField,
|
|
MonthNormalisation,
|
|
)
|
|
|
|
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.
|
|
"""
|
|
return [record for _,record in rows_to_indexed_records(rows)]
|
|
|
|
|
|
def rows_to_indexed_records(rows):
|
|
"""Sheet rows -> (1-based sheet row number, record) pairs.
|
|
|
|
Blank interior rows are skipped but do not shift later row numbers — the index
|
|
is the true sheet row (header is row 1), which is half of the unique key.
|
|
"""
|
|
if not rows:
|
|
return []
|
|
headers=normalise_headers(rows[0])
|
|
indexed=[]
|
|
for offset,row in enumerate(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))
|
|
indexed.append((offset+2,dict(zip(headers,values[:len(headers)]))))
|
|
return indexed
|
|
|
|
|
|
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 ------------------------------------------------------
|
|
|
|
_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+")
|
|
_SCORE_RE=re.compile(r"\d+")
|
|
_SALARY_UNIT_RE=re.compile(
|
|
r"(?P<num>\d+(?:[.,]\d+)?)\s*(?P<unit>k|lac|lakh|lacs|lakhs|crore|crores)?\b",
|
|
re.I,
|
|
)
|
|
_CURRENCY_STRIP_RE=re.compile(r"(?:rs\.?|pkr|inr|usd|\$|€|£)",re.I)
|
|
|
|
# Every typed column key the mapper must emit (uniform dicts for bulk insert).
|
|
_FORM_DATA_COLUMN_KEYS=tuple(field.value for field in FormDataField)+(
|
|
"age_raw","current_salary_value","expected_salary_value","job_post_id",
|
|
)
|
|
|
|
|
|
def canonical_header(h):
|
|
"""Lower, collapse whitespace (incl. embedded newlines), strip _N, (tails), trailing punct."""
|
|
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()
|
|
text=text.rstrip("?:.,").strip()
|
|
return text
|
|
|
|
|
|
def match_field(h):
|
|
"""Map a sheet header to a FormDataField via exact alias lookup, or None."""
|
|
canon=canonical_header(h)
|
|
if not canon:
|
|
return None
|
|
return ALIAS_TO_FIELD.get(canon)
|
|
|
|
|
|
def resolve_name(record,headers):
|
|
"""Candidate name: alias match, else first non-meta column (not Timestamp/date)."""
|
|
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()
|
|
# Skip entry/meta columns so Google Form "Timestamp" is never treated as a name.
|
|
_skip={
|
|
FormDataField.ENTRY_DATE,FormDataField.ENTRY_TIME,
|
|
FormDataField.ENTRY_YEAR,FormDataField.ENTRY_MONTH,FormDataField.SERIAL_NO,
|
|
}
|
|
for header in headers:
|
|
if match_field(header) in _skip:
|
|
continue
|
|
value=record.get(header)
|
|
if value is not None and str(value).strip():
|
|
return str(value).strip()
|
|
return None
|
|
|
|
|
|
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 entry_time when the cell carries 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 parse_score(value):
|
|
"""First digit run kept only when 0 <= n <= 10 (communication skills scale)."""
|
|
if value is None:
|
|
return None
|
|
text=str(value).strip()
|
|
if not text:
|
|
return None
|
|
match=_SCORE_RE.search(text)
|
|
if not match:
|
|
return None
|
|
number=int(match.group())
|
|
if 0<=number<=10:
|
|
return number
|
|
return None
|
|
|
|
|
|
def parse_salary(value):
|
|
"""Numeric salary in whole currency units, or None for non-numeric cells.
|
|
|
|
Understands k/K, lac/lakh, crore; on a range takes the first number.
|
|
The raw cell text still goes to *_salary — a None here loses nothing.
|
|
"""
|
|
if value is None:
|
|
return None
|
|
text=str(value).strip()
|
|
if not text:
|
|
return None
|
|
cleaned=_CURRENCY_STRIP_RE.sub(" ",text)
|
|
cleaned=cleaned.replace(",","")
|
|
match=_SALARY_UNIT_RE.search(cleaned)
|
|
if not match:
|
|
return None
|
|
raw_num=match.group("num").replace(",","")
|
|
try:
|
|
amount=float(raw_num)
|
|
except ValueError:
|
|
return None
|
|
unit=(match.group("unit") or "").lower()
|
|
if unit=="k":
|
|
amount*=1000
|
|
elif unit in ("lac","lakh","lacs","lakhs"):
|
|
amount*=100000
|
|
elif unit in ("crore","crores"):
|
|
amount*=10000000
|
|
return int(amount)
|
|
|
|
|
|
def _blank_to_none(value):
|
|
if value is None:
|
|
return None
|
|
text=str(value).strip()
|
|
return text if text else None
|
|
|
|
|
|
def _header_field_map(headers):
|
|
"""header -> FormDataField, first header that claims each field wins."""
|
|
claimed={}
|
|
header_to_field={}
|
|
for header in headers:
|
|
field=match_field(header)
|
|
if field is None or field in claimed:
|
|
continue
|
|
claimed[field]=header
|
|
header_to_field[header]=field
|
|
return header_to_field
|
|
|
|
|
|
def map_record_to_form_data(sheet,record,headers,row_number):
|
|
"""Pure row mapper → kwargs dict for FormData (uniform keys for bulk insert)."""
|
|
mapped={key:None for key in _FORM_DATA_COLUMN_KEYS}
|
|
mapped["sheet"]=sheet
|
|
mapped["row_number"]=row_number
|
|
mapped["raw_record"]=dict(record)
|
|
mapped["name"]=_blank_to_none(resolve_name(record,headers))
|
|
|
|
for header,field in _header_field_map(headers).items():
|
|
value=record.get(header)
|
|
key=field.value
|
|
if field==FormDataField.AGE:
|
|
age,age_raw=parse_age(value)
|
|
mapped["age"]=age
|
|
mapped["age_raw"]=age_raw
|
|
elif field==FormDataField.ENTRY_DATE:
|
|
dt,tm=parse_date_time(value)
|
|
mapped["entry_date"]=dt
|
|
if tm and not mapped.get("entry_time"):
|
|
mapped["entry_time"]=tm
|
|
elif field==FormDataField.DATE_OF_BIRTH:
|
|
mapped["date_of_birth"]=parse_date(value)
|
|
elif field==FormDataField.COMMUNICATION_SKILLS:
|
|
mapped["communication_skills"]=parse_score(value)
|
|
elif field==FormDataField.CURRENT_SALARY:
|
|
mapped["current_salary"]=_blank_to_none(value)
|
|
mapped["current_salary_value"]=parse_salary(value)
|
|
elif field==FormDataField.EXPECTED_SALARY:
|
|
mapped["expected_salary"]=_blank_to_none(value)
|
|
mapped["expected_salary_value"]=parse_salary(value)
|
|
elif field==FormDataField.NAME:
|
|
# resolve_name already set this; keep its column-A fallback behaviour.
|
|
continue
|
|
else:
|
|
mapped[key]=_blank_to_none(value)
|
|
|
|
return mapped
|
|
|
|
|
|
def collect_unmapped_headers(headers):
|
|
"""Headers that do not exact-match any alias."""
|
|
return [header for header in headers if match_field(header) is None]
|
|
|
|
|
|
def import_row_stats(mapped_rows,headers):
|
|
"""Aggregate parse diagnostics for an import report."""
|
|
unmapped=collect_unmapped_headers(headers)
|
|
dates_parsed=0
|
|
dates_unparsed=0
|
|
ages_parsed=0
|
|
salaries_parsed=0
|
|
# Find which raw header feeds entry_date (if any) once, not per row.
|
|
entry_date_header=None
|
|
for header in headers:
|
|
if match_field(header)==FormDataField.ENTRY_DATE:
|
|
entry_date_header=header
|
|
break
|
|
for row in mapped_rows:
|
|
if entry_date_header is not None:
|
|
raw=row.get("raw_record") or {}
|
|
cell=raw.get(entry_date_header)
|
|
if cell is not None and str(cell).strip():
|
|
if row.get("entry_date") 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
|
|
if (
|
|
row.get("current_salary_value") is not None
|
|
or row.get("expected_salary_value") is not None
|
|
):
|
|
salaries_parsed+=1
|
|
return {
|
|
"dates_parsed":dates_parsed,
|
|
"dates_unparsed":dates_unparsed,
|
|
"ages_parsed":ages_parsed,
|
|
"salaries_parsed":salaries_parsed,
|
|
"unmapped_headers":unmapped,
|
|
}
|
|
|