← 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())