rafa
39256c0f9b
ga4_report.py vivia solo en el checkout separado feadulta-git, que apunta al Gitea archivado y cuyo historial no tiene relacion con este repo. La copia canonica pasa a estar aqui. El entorno de ejecucion (.venv/ y .secrets/ con el cliente OAuth y el token) se queda en feadulta-git y nunca ha estado versionado -- por eso la doc sigue usando rutas absolutas alli. Ademas, el filtro que faltaba: la propiedad G-6RT9ZRS4LW recoge varios hostnames a la vez -- el WordPress vivo (www.feadulta.com) y el archivo estatico del Joomla (antiguo.feadulta.com, que lleva el mismo tag dentro del HTML capturado), mas restos (wp-nuevo, bar). Cualquier informe sin filtro los sumaba en una cifra sin significado. - preset "hosts": desglose de trafico por hostName. - --host: filtro exacto, lista separada por comas, se combina con --page-path-regex en un andGroup. - --host-not: negacion. Refs #180, #187 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
386 lines
14 KiB
Python
Executable File
386 lines
14 KiB
Python
Executable File
#!/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())
|