← back to Ga4 Dashboard
etl.py
232 lines
#!/usr/bin/env python3
"""
GA4 Fleet Traffic ETL — reads from the GA4 Data API and writes to cache/data.json.
Usage:
python3 etl.py # full refresh (7d + 30d for all properties)
python3 etl.py --dry-run # list properties only, no data pull
Cost: $0 local — GA4 Data API quota is free up to 200k tokens/day.
This script issues 2 RunReport calls per property (7d + 30d).
With 126 properties that is 252 API calls — within free quota.
Schedule (Steve-gated, do not auto-install):
Hourly via crontab:
0 * * * * cd /Users/macstudio3/Projects/ga4-dashboard && python3 etl.py >> /tmp/ga4-etl.log 2>&1
Or via launchd plist (see launchd/com.steve.ga4-etl.plist).
"""
from __future__ import annotations
import argparse
import json
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
# Auth
SKILLS_DIR = Path.home() / ".claude" / "skills" / "analytics" / "scripts"
sys.path.insert(0, str(SKILLS_DIR))
from _auth import ensure_credentials # noqa: E402
CACHE_DIR = Path(__file__).parent / "cache"
CACHE_FILE = CACHE_DIR / "data.json"
CONFIG_DIR = Path.home() / ".config" / "ga-analytics-agent"
def list_all_accounts(client) -> list[str]:
"""Return every account name the SA can see."""
try:
return [a.name for a in client.list_accounts()]
except Exception as e:
print(f"WARNING: list_accounts failed: {e}", file=sys.stderr)
return []
def list_properties_for_account(client, account_name: str) -> list[dict]:
"""Return property dicts for one account."""
props = []
try:
for p in client.list_properties(request={"filter": f"parent:{account_name}"}):
pid = p.name.split("/")[-1]
entry = {
"property_name": p.name,
"property_id": pid,
"display_name": p.display_name,
"account_name": account_name,
"measurement_id": None,
"domain": None,
"create_time": p.create_time.isoformat() if p.create_time else None,
"update_time": p.update_time.isoformat() if p.update_time else None,
}
# Get web stream measurement ID + domain
try:
for s in client.list_data_streams(parent=p.name):
if s.type_.name == "WEB_DATA_STREAM":
entry["measurement_id"] = s.web_stream_data.measurement_id
entry["domain"] = s.web_stream_data.default_uri
break
except Exception:
pass
props.append(entry)
except Exception as e:
print(f"WARNING: list_properties for {account_name} failed: {e}", file=sys.stderr)
return props
def run_report_totals(data_client, property_id: str, start: str, end: str) -> dict:
"""Return {sessions, activeUsers, screenPageViews} summed over the date range."""
from google.analytics.data_v1beta.types import DateRange, Metric, RunReportRequest
req = RunReportRequest(
property=f"properties/{property_id}",
date_ranges=[DateRange(start_date=start, end_date=end)],
metrics=[
Metric(name="sessions"),
Metric(name="activeUsers"),
Metric(name="screenPageViews"),
],
)
resp = data_client.run_report(req)
if not resp.rows:
return {"sessions": 0, "activeUsers": 0, "screenPageViews": 0}
# No dimensions so there's exactly 1 summary row
row = resp.rows[0]
return {
"sessions": int(row.metric_values[0].value),
"activeUsers": int(row.metric_values[1].value),
"screenPageViews": int(row.metric_values[2].value),
}
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--dry-run", action="store_true", help="List properties only, skip data pull")
args = ap.parse_args()
ensure_credentials()
from google.analytics.admin import AnalyticsAdminServiceClient
from google.analytics.data_v1beta import BetaAnalyticsDataClient
admin_client = AnalyticsAdminServiceClient()
data_client = BetaAnalyticsDataClient()
# --- Step 1: discover all accounts + properties ---
print("Discovering accounts...")
account_names = list_all_accounts(admin_client)
print(f" {len(account_names)} account(s): {account_names}")
all_props: list[dict] = []
for acct in account_names:
props = list_properties_for_account(admin_client, acct)
print(f" {acct}: {len(props)} properties")
all_props.extend(props)
print(f"Total properties: {len(all_props)}")
if args.dry_run:
for p in all_props:
print(f" {p['property_id']} {p['display_name']} {p['domain']}")
return 0
# --- Step 2: fetch 7d + 30d metrics per property ---
CACHE_DIR.mkdir(parents=True, exist_ok=True)
results = []
errors = []
t0 = time.time()
for i, prop in enumerate(all_props, 1):
pid = prop["property_id"]
name = prop["display_name"]
print(f" [{i}/{len(all_props)}] {name} ({pid})", end="", flush=True)
entry = dict(prop)
try:
d7 = run_report_totals(data_client, pid, "7daysAgo", "yesterday")
d30 = run_report_totals(data_client, pid, "30daysAgo", "yesterday")
entry["stats_7d"] = d7
entry["stats_30d"] = d30
print(f" sessions_7d={d7['sessions']} sessions_30d={d30['sessions']}")
except Exception as e:
msg = str(e)
entry["stats_7d"] = {"sessions": 0, "activeUsers": 0, "screenPageViews": 0}
entry["stats_30d"] = {"sessions": 0, "activeUsers": 0, "screenPageViews": 0}
entry["error"] = msg
errors.append({"pid": pid, "name": name, "error": msg})
print(f" ERROR: {msg[:80]}")
results.append(entry)
# Polite pause to avoid rate limits
if i % 50 == 0:
time.sleep(2)
elapsed = time.time() - t0
# --- Duplicate-domain dedup ---
# Normalize a domain to a bare hostname for comparison.
import re as _re
def _norm_domain(raw: str | None) -> str | None:
if not raw or not raw.strip():
return None
d = raw.strip().lower()
d = _re.sub(r'^https?://', '', d) # strip scheme
d = d.split('/')[0] # drop any path component (before www strip)
if d.startswith('www.'): # strip the www. PREFIX (not a char-set!)
d = d[4:]
d = d.rstrip('/') # strip any trailing slash
return d or None
# Group property indices by normalized domain (blank domain = never dup).
from collections import defaultdict as _defaultdict
domain_groups: dict[str, list[int]] = _defaultdict(list)
for idx, entry in enumerate(results):
key = _norm_domain(entry.get('domain'))
if key:
domain_groups[key].append(idx)
# Mark every entry with duplicate_domain and canonical flags.
for entry in results:
entry['duplicate_domain'] = False
entry['canonical'] = True
for key, indices in domain_groups.items():
if len(indices) <= 1:
continue # unique domain — nothing to do
# Pick canonical = highest sessions_7d; tie-break = newest create_time.
def _sort_key(idx: int):
e = results[idx]
sessions = e.get('stats_7d', {}).get('sessions', 0)
ct = e.get('create_time') or ''
return (sessions, ct)
canonical_idx = max(indices, key=_sort_key)
for idx in indices:
if idx == canonical_idx:
results[idx]['duplicate_domain'] = False
results[idx]['canonical'] = True
else:
results[idx]['duplicate_domain'] = True
results[idx]['canonical'] = False
dup_count = sum(1 for e in results if e.get('duplicate_domain'))
print(f"Duplicate-domain entries flagged: {dup_count}")
payload = {
"fetched_at": datetime.now(timezone.utc).isoformat(),
"elapsed_seconds": round(elapsed, 1),
"property_count": len(results),
"error_count": len(errors),
"properties": results,
}
# Atomic write: temp file + rename, so a killed ETL (OOM / launchd timeout /
# disk full) can never leave a truncated cache that crashes the server.
tmp = CACHE_FILE.with_suffix(".json.tmp")
tmp.write_text(json.dumps(payload, indent=2))
tmp.replace(CACHE_FILE)
print(f"\nWrote {CACHE_FILE} ({len(results)} properties, {len(errors)} errors, {elapsed:.0f}s)")
print("Cost: $0 (GA4 Data API reads are free within quota)")
return 0
if __name__ == "__main__":
sys.exit(main())