# gsheets_db.py from __future__ import annotations import json import os from typing import Any, Dict, Optional, List import pandas as pd from google.oauth2.service_account import Credentials from googleapiclient.discovery import build def _load_service_account_info() -> Dict[str, Any]: """ Unterstützte Varianten: 1) GCP_SERVICE_ACCOUNT_JSON: komplettes JSON als String (HF Secret) 2) GCP_SERVICE_ACCOUNT: JSON als String (Alternative Name) 3) GOOGLE_APPLICATION_CREDENTIALS: Pfad zu JSON Datei (falls vorhanden) """ raw = os.getenv("GCP_SERVICE_ACCOUNT_JSON", "").strip() or os.getenv("GCP_SERVICE_ACCOUNT", "").strip() if raw: return json.loads(raw) path = os.getenv("GOOGLE_APPLICATION_CREDENTIALS", "").strip() if path and os.path.exists(path): with open(path, "r", encoding="utf-8") as f: return json.load(f) raise RuntimeError( "No service account credentials found. " "Set GCP_SERVICE_ACCOUNT_JSON (recommended) or GOOGLE_APPLICATION_CREDENTIALS." ) def _sheets_service(): info = _load_service_account_info() creds = Credentials.from_service_account_info( info, scopes=["https://www.googleapis.com/auth/spreadsheets.readonly"], ) return build("sheets", "v4", credentials=creds, cache_discovery=False) def read_sheet_df(sheet_id: str, tab: str) -> pd.DataFrame: """ Liest ein Sheet-Tab vollständig als DataFrame. Erwartet Header in Zeile 1. """ svc = _sheets_service() rng = f"{tab}!A:ZZ" resp = svc.spreadsheets().values().get(spreadsheetId=sheet_id, range=rng).execute() values: List[List[Any]] = resp.get("values", []) if not values: return pd.DataFrame() header = values[0] rows = values[1:] # rows können kürzer sein als header -> auffüllen norm_rows = [r + [""] * (len(header) - len(r)) for r in rows] return pd.DataFrame(norm_rows, columns=header)