← back to Ga Allsites
etl.py
359 lines
#!/usr/bin/env python3
"""ETL: pull 1d + 7d + 30d + 365d traffic for EVERY GA4 property the service
account can see, write a single cache/data.json the dashboard reads. Run on a
schedule — the dashboard NEVER calls the GA API on page load (DTD verdict,
2026-08-04: Option 1 live-local-server + contrarian-adopted cache-ETL).
Also writes cache/keywords.json — Search Console query stats per window (only
populates once the SA is granted access in GSC) + GA4 channel-group stats
(Organic Search etc.) which work with GA4 access alone.
Auth: reuses the analytics skill's service account key.
"""
from __future__ import annotations
import json
import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from pathlib import Path
KEY_PATH = Path.home() / ".config" / "ga-analytics-agent" / "service-account.json"
PROP_IDS_JSON = Path.home() / ".config" / "ga-analytics-agent" / "property-ids.json"
CACHE = Path(__file__).parent / "cache" / "data.json"
WORKERS = 8
METRICS = ["sessions", "activeUsers", "screenPageViews", "conversions",
"averageSessionDuration"]
# GA4 allows max 4 date ranges per request — exactly our 4 windows, so the
# day/week/month/year pull costs the same one request per property as before.
WINDOWS = {"d1": "yesterday", "d7": "7daysAgo", "d30": "30daysAgo", "d365": "365daysAgo"}
RANGE_INDEX = {f"date_range_{i}": k for i, k in enumerate(WINDOWS)}
def _win_of(tag: str) -> str:
if tag in WINDOWS:
return tag
return RANGE_INDEX.get(tag, "d30")
def gid_map() -> dict[str, str]:
"""property_id -> G-XXXX measurement id, from the stored map (best-effort)."""
out: dict[str, str] = {}
try:
d = json.loads(PROP_IDS_JSON.read_text())
for k, v in d.items():
if k.startswith("G-"):
out[str(v)] = k
except Exception:
pass
return out
def main() -> int:
if not KEY_PATH.exists():
raise SystemExit(f"ERROR: SA key not found at {KEY_PATH}")
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = str(KEY_PATH)
from google.analytics.admin import AnalyticsAdminServiceClient
from google.analytics.data_v1beta import BetaAnalyticsDataClient
from google.analytics.data_v1beta.types import DateRange, Metric, RunReportRequest
admin = AnalyticsAdminServiceClient()
gids = gid_map()
# Enumerate every property the SA can access, across all accounts.
props: list[dict] = []
for s in admin.list_account_summaries():
acct = s.display_name or s.account.split("/")[-1]
for ps in s.property_summaries:
pid = ps.property.split("/")[-1]
props.append({
"account": acct,
"property": ps.display_name,
"property_id": pid,
"gid": gids.get(pid, ""),
})
print(f"Enumerated {len(props)} properties. Pulling 1d + 7d + 30d + 365d ...")
data = BetaAnalyticsDataClient()
ranges = [DateRange(start_date=s, end_date="yesterday", name=k)
for k, s in WINDOWS.items()]
def pull(p: dict) -> dict:
req = RunReportRequest(
property=f"properties/{p['property_id']}",
date_ranges=ranges,
metrics=[Metric(name=m) for m in METRICS],
)
empty = {w: {m: 0 for m in METRICS} for w in WINDOWS}
try:
resp = data.run_report(req)
# With N date ranges + no dimensions, GA returns one row per range,
# tagged by a trailing 'dateRange' dimension value (the range name).
buckets = {w: {m: 0 for m in METRICS} for w in WINDOWS}
for row in resp.rows:
tag = row.dimension_values[-1].value if row.dimension_values else "d30"
key = _win_of(tag)
for i, m in enumerate(METRICS):
try:
buckets[key][m] = float(row.metric_values[i].value or 0)
except (IndexError, ValueError):
buckets[key][m] = 0
return {**p, **buckets, "error": ""}
except Exception as e:
return {**p, **empty, "error": str(e)[:120]}
rows: list[dict] = []
t0 = time.time()
with ThreadPoolExecutor(max_workers=WORKERS) as ex:
futs = {ex.submit(pull, p): p for p in props}
done = 0
for f in as_completed(futs):
rows.append(f.result())
done += 1
if done % 20 == 0:
print(f" {done}/{len(props)} ...")
# sort by 30d sessions desc
rows.sort(key=lambda r: r["d30"]["sessions"], reverse=True)
payload = {
"generated_at": datetime.now(timezone.utc).isoformat(),
"elapsed_s": round(time.time() - t0, 1),
"property_count": len(rows),
"metrics": METRICS,
"rows": rows,
}
# ---- Country pass: fleet geography + per-site top country (30d) ----
from google.analytics.data_v1beta.types import Dimension
countries: dict[str, dict] = {}
prop_top: dict[str, str] = {}
def pull_country(p: dict):
try:
resp = data.run_report(RunReportRequest(
property=f"properties/{p['property_id']}",
date_ranges=[DateRange(start_date="30daysAgo", end_date="yesterday")],
dimensions=[Dimension(name="country")],
metrics=[Metric(name="sessions"), Metric(name="activeUsers")],
limit=250,
))
out = []
for row in resp.rows:
c = row.dimension_values[0].value or "(not set)"
out.append((c, float(row.metric_values[0].value or 0),
float(row.metric_values[1].value or 0)))
return p["property_id"], out
except Exception:
return p["property_id"], []
with ThreadPoolExecutor(max_workers=WORKERS) as ex:
for pid, res in ex.map(pull_country, props):
for c, s, u in res:
d = countries.setdefault(c, {"sessions": 0, "users": 0})
d["sessions"] += s
d["users"] += u
if res:
prop_top[pid] = max(res, key=lambda x: x[1])[0]
for r in rows:
r["top_country"] = prop_top.get(r["property_id"], "")
# write main data.json (now carries top_country + averageSessionDuration)
CACHE.parent.mkdir(parents=True, exist_ok=True)
CACHE.write_text(json.dumps(payload, indent=2))
errs = sum(1 for r in rows if r["error"])
print(f"Wrote {CACHE} · {len(rows)} props · {errs} errors · {payload['elapsed_s']}s")
crows = [{"country": c, "sessions": v["sessions"], "users": v["users"]}
for c, v in countries.items()]
crows.sort(key=lambda r: r["sessions"], reverse=True)
ctot = sum(r["sessions"] for r in crows) or 1
for r in crows:
r["share"] = round(100 * r["sessions"] / ctot, 1)
cpath = CACHE.parent / "countries.json"
cpath.write_text(json.dumps({
"generated_at": datetime.now(timezone.utc).isoformat(),
"window": "30d",
"total_sessions": sum(r["sessions"] for r in crows),
"country_count": len(crows),
"rows": crows,
}, indent=2))
print(f"Wrote {cpath} · {len(crows)} countries")
# ---- Keywords pass: GA4 channel groups (works now) + GSC queries ----
# Channels: one extra request per property, all 4 windows tagged in one call.
channels: dict[str, dict[str, float]] = {w: {} for w in WINDOWS}
organic_sites: dict[str, dict[str, float]] = {w: {} for w in WINDOWS}
def pull_channels(p: dict):
try:
resp = data.run_report(RunReportRequest(
property=f"properties/{p['property_id']}",
date_ranges=ranges,
dimensions=[Dimension(name="sessionDefaultChannelGroup")],
metrics=[Metric(name="sessions")],
limit=100,
))
out = []
for row in resp.rows:
ch = row.dimension_values[0].value or "(other)"
win = _win_of(row.dimension_values[-1].value or "d30")
out.append((win, ch, float(row.metric_values[0].value or 0)))
return p, out
except Exception:
return p, []
with ThreadPoolExecutor(max_workers=WORKERS) as ex:
for p, res in ex.map(pull_channels, props):
for win, ch, s in res:
channels[win][ch] = channels[win].get(ch, 0) + s
if ch == "Organic Search" and s > 0:
organic_sites[win][p["property"]] = \
organic_sites[win].get(p["property"], 0) + s
org_top = {w: sorted(({"property": k, "sessions": v} for k, v in d.items()),
key=lambda r: r["sessions"], reverse=True)[:25]
for w, d in organic_sites.items()}
# GSC: real organic queries. Needs the SA added as a user in Search Console;
# until then sites.list returns empty and the dashboard shows the how-to.
gsc_sites: list[dict] = []
gsc_windows: dict[str, dict] = {w: {"queries": [], "totals": {}} for w in WINDOWS}
sa_email = ""
try:
import urllib.parse
import urllib.request
from datetime import date, timedelta
import google.auth.transport.requests as _gareq
from google.oauth2 import service_account as _sa
sa_email = json.loads(KEY_PATH.read_text()).get("client_email", "")
creds = _sa.Credentials.from_service_account_file(
str(KEY_PATH),
scopes=["https://www.googleapis.com/auth/webmasters.readonly"])
creds.refresh(_gareq.Request())
hdrs = {"Authorization": f"Bearer {creds.token}",
"Content-Type": "application/json"}
def gsc(url, payload=None):
req = urllib.request.Request(
url, headers=hdrs,
data=json.dumps(payload).encode() if payload else None,
method="POST" if payload else "GET")
with urllib.request.urlopen(req, timeout=30) as r:
return json.loads(r.read().decode() or "{}")
base = "https://searchconsole.googleapis.com/webmasters/v3"
gsc_sites = [{"url": s["siteUrl"], "permission": s["permissionLevel"]}
for s in gsc(f"{base}/sites").get("siteEntry", [])
if s.get("permissionLevel") != "siteUnverifiedUser"]
# GSC data lags ~2 days; end every window at today-2 so numbers are final.
end = date.today() - timedelta(days=2)
spans = {"d1": 0, "d7": 6, "d30": 29, "d365": 364}
for w, back in spans.items():
agg: dict[str, dict] = {}
for site in gsc_sites:
try:
q = urllib.parse.quote(site["url"], safe="")
resp = gsc(f"{base}/sites/{q}/searchAnalytics/query", {
"startDate": str(end - timedelta(days=back)),
"endDate": str(end),
"dimensions": ["query"], "rowLimit": 250,
})
for row in resp.get("rows", []):
k = row["keys"][0]
d = agg.setdefault(k, {"query": k, "clicks": 0,
"impressions": 0, "pos_w": 0.0})
d["clicks"] += row.get("clicks", 0)
d["impressions"] += row.get("impressions", 0)
d["pos_w"] += row.get("position", 0) * row.get("impressions", 0)
except Exception:
continue
qrows = []
for d in agg.values():
impr = d["impressions"] or 1
qrows.append({"query": d["query"], "clicks": round(d["clicks"]),
"impressions": round(d["impressions"]),
"ctr": round(100 * d["clicks"] / impr, 1),
"position": round(d["pos_w"] / impr, 1)})
qrows.sort(key=lambda r: (r["clicks"], r["impressions"]), reverse=True)
gsc_windows[w] = {"queries": qrows[:300], "totals": {
"clicks": sum(r["clicks"] for r in qrows),
"impressions": sum(r["impressions"] for r in qrows),
"query_count": len(qrows)}}
except Exception as e:
print(f"GSC pass skipped: {e}")
# ---- Grant-priority queue: which properties are worth a GSC grant ----
# GSC value tracks ORGANIC-SEARCH traffic, not total sessions, so rank on
# 365d organic sessions (floored) and flag domain-verifiability. This is the
# ready-to-click grant queue the dashboard + humans read, best-first.
ORGANIC_FLOOR = 10 # <10 organic/yr → GSC has nothing to show
NONSITE = {"chargeandexplore"} # GA4 props with no standalone storefront
total_365 = {r["property"]: float((r.get("d365") or {}).get("sessions") or 0)
for r in rows}
granted_slugs = ["".join(c for c in s["url"].split(":")[-1].lower()
if c.isalnum())
for s in gsc_sites]
def _slug(name: str) -> str:
s = name.lower().replace(" - ga4", "").replace("&", "and")
return "".join(c for c in s if c.isalnum())
grant_priority = []
for name, org in organic_sites["d365"].items():
slug = _slug(name)
granted = bool(slug) and any(slug in gs for gs in granted_slugs)
verifiable = slug not in NONSITE
tot = total_365.get(name, 0.0)
grant_priority.append({
"property": name,
"domain": f"{slug}.com" if slug else "",
"organic_365d": round(org),
"total_365d": round(tot),
"organic_share": round(100 * org / tot, 1) if tot else 0.0,
"granted": granted,
"verifiable": verifiable,
"meets_floor": org >= ORGANIC_FLOOR,
"recommend": org >= ORGANIC_FLOOR and verifiable and not granted,
})
grant_priority.sort(key=lambda r: r["organic_365d"], reverse=True)
kpath = CACHE.parent / "keywords.json"
kpath.write_text(json.dumps({
"generated_at": datetime.now(timezone.utc).isoformat(),
"sa_email": sa_email,
"gsc_site_count": len(gsc_sites),
"gsc_sites": gsc_sites,
"windows": gsc_windows,
"channels": channels,
"organic_sites": org_top,
"grant_priority": grant_priority,
}, indent=2))
print(f"Wrote {kpath} · {len(gsc_sites)} GSC sites · "
f"{sum(len(v['queries']) for v in gsc_windows.values())} query rows")
# Freshness push: if the flag file exists, rsync the fresh caches to the public
# Kamatera instance (analytics.agentabrams.com). The server reads cache files per
# request, so no restart is needed. Flag-gated so local dev runs never push.
if (Path(__file__).parent / ".push-kamatera").exists():
import subprocess
try:
r = subprocess.run(
["rsync", "-az", str(CACHE), str(cpath), str(kpath),
"root@45.61.58.125:/root/Projects/ga-allsites/cache/"],
capture_output=True, timeout=90)
print("Pushed cache → Kamatera" if r.returncode == 0
else f"Kamatera push FAILED: {r.stderr.decode()[:120]}")
except Exception as e:
print(f"Kamatera push error: {e}")
return 0
if __name__ == "__main__":
raise SystemExit(main())