`lib.py`

101 lines
#!/usr/bin/env python3
"""
lib.py — pure-stdlib Google Sheets read/write helpers with light-green fill marking.

No external deps (no gspread / google-api-python-client). Refreshes an access token
from SHEETS_REFRESH_TOKEN + the desktop OAuth client, then talks to the Sheets REST API.

Every cell this library writes is stamped with a LIGHT-GREEN background so Steve can
see at a glance what was auto-filled vs. what was already in the sheet.
"""
import os, re, json, urllib.parse, urllib.request

SID = os.environ.get("SHEET_ID", "1yDmtj9Wxe71SaoQm8xpFsn722B4Ws67hTtPIHeXoN9E")
ENV_PATH = os.path.expanduser("~/Projects/secrets-manager/.env")
API = "https://sheets.googleapis.com/v4/spreadsheets"

# Light green fill for auto-populated cells (≈ #cbe6b4)
GREEN = {"red": 0.796, "green": 0.902, "blue": 0.706}

def _env():
    d = {}
    try:
        for ln in open(ENV_PATH):
            m = re.match(r'^([A-Z0-9_]+)=(.*)$', ln.rstrip("\n"))
            if m: d[m.group(1)] = m.group(2)
    except FileNotFoundError:
        pass
    return d

def access_token():
    e = _env()
    cid = e.get("YT_UPLOAD_CLIENT_ID"); csec = e.get("YT_UPLOAD_CLIENT_SECRET")
    rt = e.get("SHEETS_REFRESH_TOKEN")
    if not rt:
        raise SystemExit("Not connected yet — run connect.py first (SHEETS_REFRESH_TOKEN missing).")
    data = urllib.parse.urlencode({
        "client_id": cid, "client_secret": csec,
        "refresh_token": rt, "grant_type": "refresh_token"}).encode()
    tok = json.load(urllib.request.urlopen(
        urllib.request.Request("https://oauth2.googleapis.com/token", data)))
    return tok["access_token"]

def _req(method, url, tok, body=None):
    data = json.dumps(body).encode() if body is not None else None
    r = urllib.request.Request(url, data=data, method=method,
        headers={"Authorization": f"Bearer {tok}", "Content-Type": "application/json"})
    try:
        return json.load(urllib.request.urlopen(r))
    except urllib.error.HTTPError as ex:
        raise SystemExit(f"Sheets API {method} {url.split('?')[0]} -> {ex.code}: {ex.read().decode()[:500]}")

def get_meta(tok):
    return _req("GET", f"{API}/{SID}?fields=sheets(properties(sheetId,title,gridProperties))", tok)

def read_tab(tok, title):
    """Return the tab's values as a list of rows (list of str), padded ragged."""
    # A1 notation needs the sheet title wrapped in single quotes when it has spaces/slashes
    rng = urllib.parse.quote("'" + title.replace("'", "''") + "'", safe="")
    res = _req("GET", f"{API}/{SID}/values/{rng}?majorDimension=ROWS", tok)
    return res.get("values", [])

def col_letter(idx0):
    s = ""; n = idx0
    while True:
        s = chr(ord('A') + n % 26) + s; n = n // 26 - 1
        if n < 0: break
    return s

def batch_fill(tok, sheet_id, cells):
    """
    cells: list of dicts {row0, col0, value}  (0-based row/col indices)
    Writes each value AND stamps a light-green background, in ONE batchUpdate.
    """
    def mk(c):
        uev = {"formulaValue": str(c["value"])} if c.get("formula") else {"stringValue": str(c["value"])}
        return {"updateCells": {
            "start": {"sheetId": sheet_id, "rowIndex": c["row0"], "columnIndex": c["col0"]},
            "rows": [{"values": [{
                "userEnteredValue": uev,
                "userEnteredFormat": {"backgroundColor": GREEN}}]}],
            "fields": "userEnteredValue,userEnteredFormat.backgroundColor"}}
    if not cells:
        return {"skipped": True}
    return _chunked(tok, sheet_id, [mk(c) for c in cells])

def clear_cells(tok, sheet_id, coords):
    """coords: list of (row0,col0) — blank the value AND reset background to white."""
    reqs = [{"updateCells": {
        "start": {"sheetId": sheet_id, "rowIndex": r, "columnIndex": c},
        "rows": [{"values": [{"userEnteredValue": None,
            "userEnteredFormat": {"backgroundColor": {"red":1,"green":1,"blue":1}}}]}],
        "fields": "userEnteredValue,userEnteredFormat.backgroundColor"}} for r, c in coords]
    return _chunked(tok, sheet_id, reqs)

def _chunked(tok, sheet_id, reqs):
    CHUNK, total = 400, 0
    for i in range(0, len(reqs), CHUNK):
        _req("POST", f"{API}/{SID}:batchUpdate", tok, {"requests": reqs[i:i+CHUNK]})
        total += len(reqs[i:i+CHUNK])
    return {"totalUpdatedCells": total}