← back to Ga Global Viewer
fetch_stats.py
346 lines
#!/usr/bin/env python3
"""
fetch_stats.py — Pull GA4 stats across all of Steve's properties.
Writes JSON to stdout (or cache/stats.json if --cache flag given).
On credential failure, writes a structured error object so the
Node server can show a clear "creds needed" state.
Usage:
python3 fetch_stats.py # stdout
python3 fetch_stats.py --cache # writes cache/stats.json
python3 fetch_stats.py --cache --max 5 # limit to 5 properties (fast test)
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import time
from datetime import datetime, timedelta
from pathlib import Path
PROPERTIES_JSON = Path.home() / ".config" / "ga-analytics-agent" / "properties.json"
SA_KEY_PATH = Path.home() / ".config" / "ga-analytics-agent" / "service-account.json"
# Properties on the do-not-touch list (per CLAUDE.md standing rules)
SKIP_DOMAINS = {"designerwallcoverings.com", "studentdebtcrisis.org"}
CACHE_DIR = Path(__file__).parent / "cache"
def load_property_map() -> dict[str, str]:
"""Returns {domain: measurement_id} from the cached properties.json."""
if not PROPERTIES_JSON.exists():
return {}
with open(PROPERTIES_JSON) as f:
d = json.load(f)
props = d.get("properties", {})
# Filter out do-not-touch domains
return {
domain: mid
for domain, mid in props.items()
if not any(skip in domain for skip in SKIP_DOMAINS)
}
def load_property_id_map() -> dict[str, str]:
"""Returns {measurement_id: numeric_property_id} from property-ids.json."""
pid_path = Path.home() / ".config" / "ga-analytics-agent" / "property-ids.json"
if not pid_path.exists():
return {}
with open(pid_path) as f:
d = json.load(f)
# Keys include both domain and G-XXXXX forms; pick G-XXXXX -> numeric
return {
k: v for k, v in d.items()
if k.startswith("G-")
}
def check_auth() -> tuple[bool, str]:
"""Returns (ok, error_message)."""
if not SA_KEY_PATH.exists():
return False, f"Service account key not found at {SA_KEY_PATH}. Run setup.sh."
try:
from google.oauth2 import service_account
from google.auth.transport.requests import Request
creds = service_account.Credentials.from_service_account_file(
str(SA_KEY_PATH),
scopes=["https://www.googleapis.com/auth/analytics.readonly"],
)
creds.refresh(Request())
if creds.valid:
return True, ""
return False, "Credentials refreshed but marked invalid."
except Exception as e:
err = str(e)
if "account not found" in err.lower() or "invalid_grant" in err.lower():
return False, (
"Service account key is revoked or the GCP project 'sheets-updater-485823' "
"has been deleted/disabled. A new service account key is needed."
)
return False, f"Auth error: {err}"
def run_report_for_property(
client,
property_id: str,
start: str = "30daysAgo",
end: str = "yesterday",
) -> dict | None:
"""Run a GA4 Data API report for one property. Returns dict or None on error."""
from google.analytics.data_v1beta.types import (
DateRange, Dimension, Metric, RunReportRequest,
OrderBy,
)
prop_path = f"properties/{property_id}"
try:
# Time series: sessions + users + pageviews by date
ts_req = RunReportRequest(
property=prop_path,
date_ranges=[DateRange(start_date=start, end_date=end)],
dimensions=[Dimension(name="date")],
metrics=[
Metric(name="sessions"),
Metric(name="activeUsers"),
Metric(name="screenPageViews"),
],
order_bys=[OrderBy(dimension=OrderBy.DimensionOrderBy(dimension_name="date"))],
limit=60,
)
ts_resp = client.run_report(ts_req)
timeseries = []
total_sessions = 0
total_users = 0
total_pageviews = 0
for row in ts_resp.rows:
date = row.dimension_values[0].value
s = int(row.metric_values[0].value)
u = int(row.metric_values[1].value)
pv = int(row.metric_values[2].value)
timeseries.append({"date": date, "sessions": s, "users": u, "pageviews": pv})
total_sessions += s
total_users += u
total_pageviews += pv
# Top pages
pages_req = RunReportRequest(
property=prop_path,
date_ranges=[DateRange(start_date=start, end_date=end)],
dimensions=[Dimension(name="pagePath")],
metrics=[Metric(name="screenPageViews")],
order_bys=[OrderBy(
metric=OrderBy.MetricOrderBy(metric_name="screenPageViews"),
desc=True,
)],
limit=10,
)
pages_resp = client.run_report(pages_req)
top_pages = [
{"path": r.dimension_values[0].value, "pageviews": int(r.metric_values[0].value)}
for r in pages_resp.rows
]
# Top sources
sources_req = RunReportRequest(
property=prop_path,
date_ranges=[DateRange(start_date=start, end_date=end)],
dimensions=[Dimension(name="sessionSource")],
metrics=[Metric(name="sessions")],
order_bys=[OrderBy(
metric=OrderBy.MetricOrderBy(metric_name="sessions"),
desc=True,
)],
limit=10,
)
src_resp = client.run_report(sources_req)
top_sources = [
{"source": r.dimension_values[0].value, "sessions": int(r.metric_values[0].value)}
for r in src_resp.rows
]
return {
"timeseries": timeseries,
"totals": {
"sessions": total_sessions,
"users": total_users,
"pageviews": total_pageviews,
},
"top_pages": top_pages,
"top_sources": top_sources,
}
except Exception as e:
err_str = str(e)
if "403" in err_str or "PERMISSION_DENIED" in err_str:
return {"error": "permission_denied", "detail": err_str[:200]}
if "404" in err_str or "NOT_FOUND" in err_str:
return {"error": "property_not_found", "detail": err_str[:200]}
return {"error": "api_error", "detail": err_str[:200]}
def merge_timeseries(all_series: list[list[dict]]) -> list[dict]:
"""Merge per-property timeseries into a combined global timeseries."""
combined: dict[str, dict] = {}
for series in all_series:
for row in series:
d = row["date"]
if d not in combined:
combined[d] = {"date": d, "sessions": 0, "users": 0, "pageviews": 0}
combined[d]["sessions"] += row["sessions"]
combined[d]["users"] += row["users"]
combined[d]["pageviews"] += row["pageviews"]
return sorted(combined.values(), key=lambda r: r["date"])
def merge_top_list(all_lists: list[list[dict]], key: str, value: str, top_n: int = 15) -> list[dict]:
"""Merge per-property top-N lists into a global top-N."""
combined: dict[str, int] = {}
for lst in all_lists:
for item in lst:
k = item[key]
combined[k] = combined.get(k, 0) + item[value]
sorted_items = sorted(combined.items(), key=lambda x: x[1], reverse=True)
return [{key: k, value: v} for k, v in sorted_items[:top_n]]
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--cache", action="store_true", help="Write to cache/stats.json")
ap.add_argument("--max", type=int, default=0, help="Limit number of properties queried")
ap.add_argument("--start", default="30daysAgo")
ap.add_argument("--end", default="yesterday")
args = ap.parse_args()
result = {
"generated_at": datetime.utcnow().isoformat() + "Z",
"date_range": {"start": args.start, "end": args.end},
"auth_ok": False,
"auth_error": None,
"properties_count": 0,
"properties_queried": 0,
"properties_with_data": 0,
"global": {
"timeseries": [],
"totals": {"sessions": 0, "users": 0, "pageviews": 0},
"top_pages": [],
"top_sources": [],
},
"per_property": [],
}
# Load property map from cache
prop_map = load_property_map() # domain -> G-ID
pid_map = load_property_id_map() # G-ID -> numeric property_id
result["properties_count"] = len(prop_map)
# Auth check
auth_ok, auth_error = check_auth()
result["auth_ok"] = auth_ok
result["auth_error"] = auth_error
if not auth_ok:
if args.cache:
CACHE_DIR.mkdir(exist_ok=True)
(CACHE_DIR / "stats.json").write_text(json.dumps(result, indent=2))
else:
print(json.dumps(result, indent=2))
return 0
# Auth is good — run reports
from google.oauth2 import service_account
from google.analytics.data_v1beta import BetaAnalyticsDataClient
creds = service_account.Credentials.from_service_account_file(
str(SA_KEY_PATH),
scopes=["https://www.googleapis.com/auth/analytics.readonly"],
)
client = BetaAnalyticsDataClient(credentials=creds)
per_property_timeseries = []
per_property_pages = []
per_property_sources = []
global_totals = {"sessions": 0, "users": 0, "pageviews": 0}
items = list(prop_map.items())
if args.max:
items = items[: args.max]
result["properties_queried"] = len(items)
queried_count = 0
for domain, mid in items:
numeric_id = pid_map.get(mid)
if not numeric_id:
result["per_property"].append({
"domain": domain,
"measurement_id": mid,
"numeric_id": None,
"error": "no_numeric_id",
})
continue
queried_count += 1
prop_result = run_report_for_property(client, numeric_id, args.start, args.end)
if prop_result is None or "error" in prop_result:
result["per_property"].append({
"domain": domain,
"measurement_id": mid,
"numeric_id": numeric_id,
"error": prop_result.get("error") if prop_result else "null_response",
"detail": prop_result.get("detail", "") if prop_result else "",
})
continue
# Has real data
result["properties_with_data"] += 1
totals = prop_result["totals"]
global_totals["sessions"] += totals["sessions"]
global_totals["users"] += totals["users"]
global_totals["pageviews"] += totals["pageviews"]
per_property_timeseries.append(prop_result["timeseries"])
per_property_pages.append(prop_result["top_pages"])
per_property_sources.append(prop_result["top_sources"])
result["per_property"].append({
"domain": domain,
"measurement_id": mid,
"numeric_id": numeric_id,
"totals": totals,
"timeseries": prop_result["timeseries"],
"top_pages": prop_result["top_pages"],
"top_sources": prop_result["top_sources"],
})
# Rate-limit pause
if queried_count % 10 == 0:
time.sleep(1)
# Build global aggregates
result["global"]["totals"] = global_totals
result["global"]["timeseries"] = merge_timeseries(per_property_timeseries)
result["global"]["top_pages"] = merge_top_list(per_property_pages, "path", "pageviews")
result["global"]["top_sources"] = merge_top_list(per_property_sources, "source", "sessions")
if args.cache:
CACHE_DIR.mkdir(exist_ok=True)
out_path = CACHE_DIR / "stats.json"
out_path.write_text(json.dumps(result, indent=2))
print(f"Wrote {out_path}", file=sys.stderr)
else:
print(json.dumps(result, indent=2))
return 0
if __name__ == "__main__":
sys.exit(main())