`scripts/batch_create_properties.py`

173 lines
#!/usr/bin/env python3
"""
Batch-create GA4 properties + web data streams for domains in
cache/_ga4_inject_targets.json. Resilient: one failure never aborts the
batch; rate-limit errors get exponential back-off. Writes results to
cache/_ga4_inject_ids.json as {domain: measurement_id}.

Cost: $0 — Analytics Admin API is free.
"""
from __future__ import annotations

import json
import os
import sys
import time
from pathlib import Path

# ---------------------------------------------------------------------------
# Paths
# ---------------------------------------------------------------------------
PROJECT_ROOT = Path("/Users/macstudio3/Projects/ga-allsites")
TARGETS_FILE = PROJECT_ROOT / "cache" / "_ga4_inject_targets.json"
OUTPUT_FILE  = PROJECT_ROOT / "cache" / "_ga4_inject_ids.json"

# Re-use auth helpers from the analytics skill
SKILL_SCRIPTS = Path.home() / ".claude/skills/analytics/scripts"
sys.path.insert(0, str(SKILL_SCRIPTS))
from _auth import ensure_credentials, get_account_id  # noqa: E402

# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
TIMEZONE = "America/Los_Angeles"
CURRENCY = "USD"
INDUSTRY = "ONLINE_COMMUNITIES"

MAX_RETRIES    = 4
BASE_BACKOFF_S = 5    # seconds; doubles each retry


# ---------------------------------------------------------------------------
# Core helpers
# ---------------------------------------------------------------------------

def create_property_and_stream(client, account_id: str, domain: str):
    """Create property + web stream; return (property_id, measurement_id)."""
    from google.analytics.admin_v1alpha.types import Property, DataStream

    prop = Property(
        parent=account_id,
        display_name=domain,
        time_zone=TIMEZONE,
        currency_code=CURRENCY,
        industry_category=INDUSTRY,
    )
    created = client.create_property(property=prop)
    prop_name = created.name
    prop_id   = prop_name.split("/")[-1]

    clean_domain = domain.replace("https://", "").replace("http://", "").rstrip("/")
    stream = DataStream(
        type_=DataStream.DataStreamType.WEB_DATA_STREAM,
        display_name=domain,
        web_stream_data=DataStream.WebStreamData(
            default_uri=f"https://{clean_domain}",
        ),
    )
    created_stream = client.create_data_stream(parent=prop_name, data_stream=stream)
    measurement_id = created_stream.web_stream_data.measurement_id

    return prop_id, measurement_id


def create_with_retry(client, account_id: str, domain: str):
    """Wrap create_property_and_stream with exponential back-off."""
    import grpc

    last_err = None
    backoff = BASE_BACKOFF_S
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            return create_property_and_stream(client, account_id, domain)
        except Exception as e:
            last_err = e
            err_str = str(e).lower()
            is_rate_limit = (
                "quota" in err_str
                or "resource_exhausted" in err_str
                or "rate" in err_str
                or "429" in err_str
            )
            if is_rate_limit and attempt < MAX_RETRIES:
                print(f"  [rate-limit] attempt {attempt}/{MAX_RETRIES}, sleeping {backoff}s …")
                time.sleep(backoff)
                backoff *= 2
                continue
            # Non-rate-limit or final attempt — give up on this domain
            break

    raise RuntimeError(f"Failed after {MAX_RETRIES} attempts: {last_err}") from last_err


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main() -> int:
    ensure_credentials()
    account_id = get_account_id()
    print(f"Account: {account_id}")

    domains: list[str] = json.loads(TARGETS_FILE.read_text())
    print(f"Targets: {len(domains)} domains\n")

    # Load any previously written results so a re-run is idempotent
    existing: dict[str, str] = {}
    if OUTPUT_FILE.exists():
        try:
            existing = json.loads(OUTPUT_FILE.read_text())
            if existing:
                print(f"Resuming: {len(existing)} already done → {list(existing.keys())[:3]} …\n")
        except json.JSONDecodeError:
            pass

    from google.analytics.admin import AnalyticsAdminServiceClient
    client = AnalyticsAdminServiceClient()

    results: dict[str, str] = dict(existing)
    succeeded = list(existing.keys())
    failed: list[tuple[str, str]] = []

    for i, domain in enumerate(domains, 1):
        if domain in existing:
            print(f"[{i:02d}/{len(domains)}] {domain:<35}  SKIP (already done: {existing[domain]})")
            continue

        print(f"[{i:02d}/{len(domains)}] {domain:<35}  creating …", end="", flush=True)
        try:
            prop_id, mid = create_with_retry(client, account_id, domain)
            results[domain] = mid
            succeeded.append(domain)
            print(f"  OK  property={prop_id}  measurement_id={mid}")

            # Persist after every success so a crash mid-batch doesn't lose work
            OUTPUT_FILE.write_text(json.dumps(results, indent=2))

            # Polite inter-request gap (avoid burst rate limit)
            if i < len(domains):
                time.sleep(1.2)

        except Exception as exc:
            failed.append((domain, str(exc)))
            print(f"  FAIL  {exc}")

    # Final write
    OUTPUT_FILE.write_text(json.dumps(results, indent=2))

    # Summary
    print("\n" + "=" * 72)
    print(f"SUCCEEDED: {len(succeeded)}   FAILED: {len(failed)}")
    if failed:
        print("\nFailed domains:")
        for d, err in failed:
            print(f"  {d}: {err}")
    print(f"\nOutput: {OUTPUT_FILE}")
    print("Cost: $0 (Admin API free)")

    return 0 if not failed else 1


if __name__ == "__main__":
    sys.exit(main())