#!/usr/bin/env python3 """Query GA4 via the Google Analytics Data API and Admin API. This script is intended for practical editorial analysis: - resolve a GA4 property from a measurement ID - run a few reusable reports - export the result to CSV """ from __future__ import annotations import argparse import csv import json import os import sys from dataclasses import dataclass from pathlib import Path from typing import Any import requests from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow SCOPES = ["https://www.googleapis.com/auth/analytics.readonly"] DATA_API_BASE = "https://analyticsdata.googleapis.com/v1beta" ADMIN_API_BASE = "https://analyticsadmin.googleapis.com/v1beta" @dataclass class Config: client_secrets_path: Path token_path: Path property_id: str | None measurement_id: str | None no_browser: bool PRESETS: dict[str, dict[str, Any]] = { "summary": { "dimensions": [], "metrics": ["screenPageViews", "totalUsers", "sessions", "engagedSessions", "engagementRate"], "order_bys": [], }, "traffic": { "dimensions": ["date"], "metrics": ["sessions", "totalUsers", "engagedSessions", "engagementRate", "screenPageViews"], "order_bys": [{"dimension": {"dimensionName": "date"}}], }, "content": { "dimensions": ["pageTitle", "pagePath"], "metrics": ["screenPageViews", "totalUsers", "engagedSessions", "engagementRate", "averageSessionDuration"], "order_bys": [{"metric": {"metricName": "screenPageViews"}, "desc": True}], }, "landing-pages": { "dimensions": ["landingPagePlusQueryString"], "metrics": ["sessions", "totalUsers", "engagedSessions", "engagementRate", "screenPageViews"], "order_bys": [{"metric": {"metricName": "sessions"}, "desc": True}], }, "source-medium": { "dimensions": ["sessionSourceMedium"], "metrics": ["sessions", "totalUsers", "engagedSessions", "engagementRate", "screenPageViews"], "order_bys": [{"metric": {"metricName": "sessions"}, "desc": True}], }, "device": { "dimensions": ["deviceCategory"], "metrics": ["sessions", "totalUsers", "engagedSessions", "engagementRate", "screenPageViews"], "order_bys": [{"metric": {"metricName": "sessions"}, "desc": True}], }, "hosts": { "dimensions": ["hostName"], "metrics": ["sessions", "totalUsers", "engagedSessions", "engagementRate", "screenPageViews"], "order_bys": [{"metric": {"metricName": "sessions"}, "desc": True}], }, } def load_config(args: argparse.Namespace) -> Config: client_secrets = args.client_secrets_path or os.getenv("GA4_CLIENT_SECRETS_PATH") token_path = args.token_path or os.getenv("GA4_TOKEN_PATH") or ".secrets/ga4-token.json" property_id = args.property_id or os.getenv("GA4_PROPERTY_ID") measurement_id = args.measurement_id or os.getenv("GA4_MEASUREMENT_ID") if not client_secrets: raise SystemExit( "Missing OAuth client secrets path. Set --client-secrets-path or GA4_CLIENT_SECRETS_PATH." ) return Config( client_secrets_path=Path(client_secrets), token_path=Path(token_path), property_id=property_id, measurement_id=measurement_id, no_browser=bool(args.no_browser), ) def get_credentials(config: Config) -> Credentials: creds: Credentials | None = None if config.token_path.exists(): creds = Credentials.from_authorized_user_file(str(config.token_path), SCOPES) if creds and creds.valid: return creds if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) config.token_path.parent.mkdir(parents=True, exist_ok=True) config.token_path.write_text(creds.to_json(), encoding="utf-8") return creds if not config.client_secrets_path.exists(): raise SystemExit(f"Client secrets file not found: {config.client_secrets_path}") flow = InstalledAppFlow.from_client_secrets_file(str(config.client_secrets_path), SCOPES) prompt_message = "Please visit this URL to authorize this application: {url}" creds = flow.run_local_server( port=0, open_browser=not config.no_browser, authorization_prompt_message=prompt_message, ) config.token_path.parent.mkdir(parents=True, exist_ok=True) config.token_path.write_text(creds.to_json(), encoding="utf-8") return creds def auth_headers(creds: Credentials) -> dict[str, str]: if not creds.valid: creds.refresh(Request()) return { "Authorization": f"Bearer {creds.token}", "Content-Type": "application/json", } def admin_get(creds: Credentials, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]: url = f"{ADMIN_API_BASE}/{path.lstrip('/')}" response = requests.get(url, headers=auth_headers(creds), params=params, timeout=60) response.raise_for_status() return response.json() def data_post(creds: Credentials, path: str, payload: dict[str, Any]) -> dict[str, Any]: url = f"{DATA_API_BASE}/{path.lstrip('/')}" response = requests.post(url, headers=auth_headers(creds), json=payload, timeout=60) response.raise_for_status() return response.json() def iterate_account_summaries(creds: Credentials) -> list[dict[str, Any]]: results: list[dict[str, Any]] = [] page_token: str | None = None while True: params = {"pageSize": 200} if page_token: params["pageToken"] = page_token payload = admin_get(creds, "accountSummaries", params=params) results.extend(payload.get("accountSummaries", [])) page_token = payload.get("nextPageToken") if not page_token: return results def resolve_property_id(creds: Credentials, measurement_id: str) -> dict[str, str]: summaries = iterate_account_summaries(creds) for summary in summaries: for prop in summary.get("propertySummaries", []): prop_resource = prop.get("property", "") if not prop_resource.startswith("properties/"): continue prop_id = prop_resource.split("/", 1)[1] streams = admin_get(creds, f"properties/{prop_id}/dataStreams") for stream in streams.get("dataStreams", []): web_stream = stream.get("webStreamData", {}) if web_stream.get("measurementId") == measurement_id: return { "property_id": prop_id, "property_display_name": prop.get("displayName", ""), "account_display_name": summary.get("displayName", ""), "stream_display_name": stream.get("displayName", ""), } raise SystemExit(f"No accessible GA4 property matched measurement ID {measurement_id}.") def build_report_payload(args: argparse.Namespace) -> dict[str, Any]: preset = PRESETS[args.preset] start_date = args.start_date or f"{args.days}daysAgo" end_date = args.end_date or "yesterday" payload: dict[str, Any] = { "metrics": [{"name": m} for m in preset["metrics"]], "dateRanges": [{"startDate": start_date, "endDate": end_date}], "limit": str(args.limit), "keepEmptyRows": False, "returnPropertyQuota": True, } if preset["dimensions"]: payload["dimensions"] = [{"name": d} for d in preset["dimensions"]] if preset["order_bys"]: payload["orderBys"] = preset["order_bys"] filters: list[dict[str, Any]] = [] if args.page_path_regex: expression: dict[str, Any] = { "filter": { "fieldName": "pagePath", "stringFilter": { "matchType": "FULL_REGEXP", "value": args.page_path_regex, }, } } if args.page_path_regex_not: expression = {"notExpression": expression} filters.append(expression) # La propiedad G-6RT9ZRS4LW mide varios hostnames a la vez (www.feadulta.com # vivo y antiguo.feadulta.com, el archivo estatico). Sin este filtro los # informes los mezclan y no significan nada. host_filter = getattr(args, "host", None) if host_filter: hosts = [h.strip() for h in host_filter.split(",") if h.strip()] host_expression: dict[str, Any] = { "filter": { "fieldName": "hostName", "inListFilter": {"values": hosts, "caseSensitive": False}, } } if getattr(args, "host_not", False): host_expression = {"notExpression": host_expression} filters.append(host_expression) if len(filters) == 1: payload["dimensionFilter"] = filters[0] elif len(filters) > 1: payload["dimensionFilter"] = {"andGroup": {"expressions": filters}} return payload def rows_from_response(response: dict[str, Any]) -> tuple[list[str], list[list[str]]]: dimensions = [h["name"] for h in response.get("dimensionHeaders", [])] metrics = [h["name"] for h in response.get("metricHeaders", [])] headers = dimensions + metrics rows: list[list[str]] = [] for row in response.get("rows", []): dimension_values = [v.get("value", "") for v in row.get("dimensionValues", [])] metric_values = [v.get("value", "") for v in row.get("metricValues", [])] rows.append(dimension_values + metric_values) return headers, rows def write_csv(path: str, headers: list[str], rows: list[list[str]]) -> None: out_path = Path(path) out_path.parent.mkdir(parents=True, exist_ok=True) with out_path.open("w", newline="", encoding="utf-8") as handle: writer = csv.writer(handle) writer.writerow(headers) writer.writerows(rows) def print_table(headers: list[str], rows: list[list[str]]) -> None: widths = [len(h) for h in headers] for row in rows: for idx, value in enumerate(row): widths[idx] = max(widths[idx], len(value)) fmt = " | ".join(f"{{:{w}}}" for w in widths) print(fmt.format(*headers)) print("-+-".join("-" * w for w in widths)) for row in rows: print(fmt.format(*row)) def cmd_resolve_property(args: argparse.Namespace) -> int: config = load_config(args) if not config.measurement_id: raise SystemExit("Missing measurement ID. Set --measurement-id or GA4_MEASUREMENT_ID.") creds = get_credentials(config) result = resolve_property_id(creds, config.measurement_id) print(json.dumps(result, indent=2, ensure_ascii=True)) return 0 def cmd_report(args: argparse.Namespace) -> int: config = load_config(args) creds = get_credentials(config) property_id = config.property_id if not property_id: if not config.measurement_id: raise SystemExit( "Missing property ID. Set --property-id / GA4_PROPERTY_ID or provide --measurement-id / GA4_MEASUREMENT_ID." ) resolved = resolve_property_id(creds, config.measurement_id) property_id = resolved["property_id"] print( f"Resolved measurement ID {config.measurement_id} to property {property_id} " f"({resolved['property_display_name']})", file=sys.stderr, ) payload = build_report_payload(args) response = data_post(creds, f"properties/{property_id}:runReport", payload) headers, rows = rows_from_response(response) if args.csv: write_csv(args.csv, headers, rows) print(f"Wrote CSV to {args.csv}", file=sys.stderr) print_table(headers, rows) return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Query GA4 via OAuth.") parser.add_argument("--client-secrets-path", help="Path to OAuth desktop client secrets JSON.") parser.add_argument("--token-path", help="Path to cached OAuth token JSON.") parser.add_argument("--property-id", help="GA4 property ID.") parser.add_argument("--measurement-id", help="GA4 measurement ID (G-...).") parser.add_argument( "--no-browser", action="store_true", help="Print the OAuth URL instead of trying to open a browser automatically.", ) subparsers = parser.add_subparsers(dest="command", required=True) resolve_parser = subparsers.add_parser("resolve-property", help="Resolve GA4 property from measurement ID.") resolve_parser.set_defaults(func=cmd_resolve_property) report_parser = subparsers.add_parser("report", help="Run a preset GA4 report.") report_parser.add_argument( "--preset", choices=sorted(PRESETS.keys()), default="content", help="Which report shape to run.", ) report_parser.add_argument("--days", type=int, default=28, help="Lookback window in days.") report_parser.add_argument("--start-date", help="Explicit GA4 start date, e.g. 2026-06-18.") report_parser.add_argument("--end-date", help="Explicit GA4 end date, e.g. 2026-06-20.") report_parser.add_argument("--limit", type=int, default=25, help="Max rows to request.") report_parser.add_argument("--csv", help="Optional CSV output path.") report_parser.add_argument( "--page-path-regex", help="Optional GA4 FULL_REGEXP filter applied to pagePath.", ) report_parser.add_argument( "--page-path-regex-not", action="store_true", help="Negate --page-path-regex.", ) report_parser.add_argument( "--host", help=( "Filtra por hostName (exacto, varios separados por coma). " "Ej: www.feadulta.com o antiguo.feadulta.com. " "Sin esto, la propiedad mezcla el sitio vivo y el archivo estatico." ), ) report_parser.add_argument( "--host-not", action="store_true", help="Negate --host (todo MENOS esos hostnames).", ) report_parser.set_defaults(func=cmd_report) return parser def main() -> int: parser = build_parser() args = parser.parse_args() return args.func(args) if __name__ == "__main__": raise SystemExit(main())