breakpointsoftware/document-parser

1from __future__ import annotations2 3import json4import math5import os6from typing import Any7 8import gspread9from dotenv import load_dotenv10from gspread.exceptions import WorksheetNotFound11from google.oauth2.service_account import Credentials12 13 14SCOPES = [15    "https://www.googleapis.com/auth/spreadsheets",16    "https://www.googleapis.com/auth/drive",17]18 19 20load_dotenv()21 22 23class GoogleSheetsConfigError(RuntimeError):24    """Raised when required Google Sheets environment configuration is missing or invalid."""25 26 27def _build_credentials() -> Credentials:28    service_account_json = (os.getenv("GOOGLE_SERVICE_ACCOUNT_JSON") or "").strip()29    service_account_file = (os.getenv("GOOGLE_SERVICE_ACCOUNT_FILE") or "").strip()30 31    if service_account_json:32        try:33            info = json.loads(service_account_json)34        except json.JSONDecodeError as exc:35            raise GoogleSheetsConfigError("GOOGLE_SERVICE_ACCOUNT_JSON is not valid JSON.") from exc36        return Credentials.from_service_account_info(info, scopes=SCOPES)37 38    if service_account_file:39        return Credentials.from_service_account_file(service_account_file, scopes=SCOPES)40 41    raise GoogleSheetsConfigError(42        "Missing Google credentials. Set GOOGLE_SERVICE_ACCOUNT_JSON or GOOGLE_SERVICE_ACCOUNT_FILE."43    )44 45 46def _open_worksheet(client: gspread.Client, worksheet_name: str | None = None, column_count: int = 0):47    spreadsheet_id = (os.getenv("GOOGLE_SHEETS_SPREADSHEET_ID") or "").strip()48    resolved_worksheet_name = (worksheet_name or os.getenv("GOOGLE_SHEETS_WORKSHEET") or "Sheet1").strip()49 50    if not spreadsheet_id:51        raise GoogleSheetsConfigError("Missing GOOGLE_SHEETS_SPREADSHEET_ID.")52 53    spreadsheet = client.open_by_key(spreadsheet_id)54    try:55        return spreadsheet.worksheet(resolved_worksheet_name)56    except WorksheetNotFound:57        return spreadsheet.add_worksheet(58            title=resolved_worksheet_name,59            rows=1000,60            cols=max(column_count, 1),61        )62 63 64def _normalize_cell_value(value: Any) -> Any:65    if value is None:66        return ""67 68    # Pandas uses a dedicated NA sentinel that is not JSON serializable.69    if type(value).__name__ == "NAType":70        return ""71 72    # Convert numpy scalar wrappers into plain Python values when possible.73    if hasattr(value, "item"):74        try:75            value = value.item()76        except Exception:77            pass78 79    if isinstance(value, float) and not math.isfinite(value):80        return ""81 82    # Fallback for NaN-like values that are not plain floats.83    try:84        if value != value:85            return ""86    except Exception:87        pass88 89    return value90 91 92def _column_index_to_letter(index: int) -> str:93    if index < 1:94        raise ValueError("Column index must be greater than zero.")95 96    letters = []97    while index > 0:98        index, remainder = divmod(index - 1, 26)99        letters.append(chr(ord("A") + remainder))100    return "".join(reversed(letters))101 102 103def append_row_to_google_sheet(104    row: dict[str, Any],105    ordered_columns: list[str],106    worksheet_name: str | None = None,107) -> None:108    credentials = _build_credentials()109    client = gspread.authorize(credentials)110    worksheet = _open_worksheet(client, worksheet_name=worksheet_name, column_count=len(ordered_columns))111 112    values = [_normalize_cell_value(row.get(column, "")) for column in ordered_columns]113 114    # Insert a new row at the end of the table anchored at column A.115    end_column = _column_index_to_letter(len(values))116    table_values = worksheet.get(f"A:{end_column}")117    next_row = len(table_values) + 1118    worksheet.insert_row(values, index=next_row, value_input_option="USER_ENTERED")119