gsheets_db.py

# 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)