src/gcs_io.py

"""Tiny GCS I/O helpers. Uses `gcloud storage` (gsutil-free).

Prefers the google-cloud-storage client if installed; otherwise shells out to
`gcloud storage cp`, which is available wherever gcloud is.
"""
from __future__ import annotations
import subprocess, os


def upload_file(local: str, gcs_uri: str):
    try:
        from google.cloud import storage  # type: ignore
        assert gcs_uri.startswith("gs://")
        bucket, _, blob = gcs_uri[5:].partition("/")
        storage.Client().bucket(bucket).blob(blob).upload_from_filename(local)
    except Exception:
        subprocess.run(["gcloud", "storage", "cp", local, gcs_uri], check=True)


def upload_dir(local_dir: str, gcs_prefix: str):
    for root, _, files in os.walk(local_dir):
        for fn in files:
            lp = os.path.join(root, fn)
            rel = os.path.relpath(lp, local_dir)
            upload_file(lp, f"{gcs_prefix}/{rel}")
            print(f"  uploaded {rel} -> {gcs_prefix}/{rel}")