#!/usr/bin/env python3
"""Read-only GA4 snapshot collector for the ga4-audit skill (route B).

Usage, from the audit working folder (never write into the skill folder):
  python3 <skill folder>/scripts/ga4_snapshot.py --property 123456789 --as-of 2026-11-01 \
      --quota-project YOUR_CLOUD_PROJECT --out ga4-snapshot.json [--skip-reports] [--max-custom-dims 60] [--dry-run]

Auth: a short-lived token from `gcloud auth application-default print-access-token`. The token stays
in memory and is never printed, logged or written. The user mints read-only credentials first
(references/reading-sources.md).
A raw token carries no quota project, so pass --quota-project (sent as x-goog-user-project).

Safety built in:
- Scope gate: stops with exit 2 when the token carries the bare analytics scope, analytics.edit,
  analytics.manage.users or analytics.provision. Only scope names are printed.
- Allowlist: only the GET calls and Data API reports in ALLOWED can be sent; anything else raises
  before a request leaves the machine.
- Redaction before writing: Google Ads link creator emails dropped, PII probe URLs reduced to host,
  path and query key names (scripts/lib/pii.py; anything that is not a plain parameter name is written
  as REDACTED), custom dimension values and transaction IDs reduced to counts, and the
  rows of R13A, R13B and R16 dropped (only their rowCount and metadata are kept; the engine reads
  nothing else). R17C returns only the (not set) item row, so no item names are written.
Exit code: 0 complete, 3 partial (some endpoints unreadable, file still written), 2 auth or scope stop.
Standard library only, Python 3.9 or later.
"""
import argparse
import json
import os
import re
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import date, datetime, timedelta, timezone

sys.dont_write_bytecode = True  # never leave __pycache__ in the skill folder
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from lib.pii import EMAIL_RE, PHONE_RE, R15_TERMS, reduce_url  # noqa: E402

ADMIN = 'https://analyticsadmin.googleapis.com/'
DATA = 'https://analyticsdata.googleapis.com/v1beta/'
TOKENINFO = 'https://oauth2.googleapis.com/tokeninfo'
AUTH = 'https://www.googleapis.com/auth/'
STOP_SCOPES = {AUTH + 'analytics', AUTH + 'analytics.edit', AUTH + 'analytics.manage.users', AUTH + 'analytics.provision'}
ADMIN_LISTS = ('dataStreams|keyEvents|customDimensions|customMetrics|googleAdsLinks|firebaseLinks|bigQueryLinks|channelGroups|'
               'reportingDataAnnotations|calculatedMetrics|audiences')
ADMIN_SINGLES = 'dataRetentionSettings|attributionSettings|googleSignalsSettings|reportingIdentitySettings|userProvidedDataSettings'
ALLOWED = [
    ('GET', re.compile(rf'^https://analyticsadmin\.googleapis\.com/v1(beta|alpha)/properties/\d+(/({ADMIN_LISTS}|{ADMIN_SINGLES}))?(\?[^/]*)?$')),
    ('GET', re.compile(r'^https://analyticsadmin\.googleapis\.com/v1alpha/properties/\d+/dataStreams/\d+/'
                       r'(enhancedMeasurementSettings|dataRedactionSettings|eventCreateRules|eventEditRules)(\?[^/]*)?$')),
    ('GET', re.compile(r'^https://analyticsadmin\.googleapis\.com/v1beta/accounts/\d+/dataSharingSettings$')),
    ('POST', re.compile(r'^https://analyticsdata\.googleapis\.com/v1beta/properties/\d+:runReport$')),
    ('GET', re.compile(r'^https://analyticsdata\.googleapis\.com/v1beta/properties/\d+/metadata$')),
    ('POST', re.compile(r'^https://oauth2\.googleapis\.com/tokeninfo$')),
]
METADATA_ONLY = ('R13A', 'R13B', 'R16')  # probes whose rows are never needed; page paths can carry personal data
NEVER = ['any create, update, patch, delete, archive, acknowledge, provision or reorder method',
         'measurementProtocolSecrets (returns secret values)', 'accessBindings (needs analytics.manage.users, returns emails)',
         'accounts.searchChangeHistoryEvents (needs analytics.edit)', 'runAccessReport (returns user records)',
         'audienceExports.create']


class Stop(Exception):
    """Authentication or scope problem: stop with exit code 2."""


def exact(field, value):
    return {'filter': {'fieldName': field, 'stringFilter': {'matchType': 'EXACT', 'value': value}}}


def empty(field):
    """Matches (not set) and empty values."""
    return {'filter': {'fieldName': field, 'emptyFilter': {}}}


def report_specs(w):
    """R-report specs from references/data-checks.md: (id, date range, dimensions, metrics, filter, limit)."""
    cur, prev, both = (w['start'], w['end']), (w['previous_start'], w['previous_end']), (w['previous_start'], w['end'])
    week = ((date.fromisoformat(w['end']) - timedelta(days=6)).isoformat(), w['end'])
    chan = ['sessionDefaultChannelGroup', 'sessionPrimaryChannelGroup']
    return [('R01', cur, ['eventName', 'isKeyEvent'], ['eventCount', 'keyEvents', 'eventValue', 'sessions'], None, 1000),
            ('R02', both, ['date', 'streamId'], ['eventCount'], None, 1000),
            ('R03', cur, ['hostName', 'streamId'], ['eventCount', 'keyEvents', 'sessions'], None, 500),
            ('R04', cur, chan, ['sessions', 'keyEvents', 'sessionKeyEventRate'], None, 250),
            ('R04P', prev, chan, ['sessions', 'keyEvents'], None, 250),
            ('R05', cur, ['sessionSource', 'sessionMedium'], ['sessions', 'keyEvents'], exact('sessionMedium', 'referral'), 250),
            ('R06', cur, ['sourceMedium', 'defaultChannelGroup'], ['keyEvents'], None, 250),
            ('R07', cur, ['linkDomain', 'outbound'], ['eventCount'], exact('eventName', 'click'), 250),
            ('R09', cur, ['currencyCode'], ['eventCount', 'eventValue'], exact('eventName', 'purchase'), 50),
            ('R10', cur, ['sessionSourceMedium', 'sessionGoogleAdsCustomerId'], ['sessions'], exact('sessionSourceMedium', 'google / cpc'), 250),
            ('R11', cur, ['testDataFilterName', 'eventName'], ['eventCount'], None, 1000),
            ('R11T', cur, ['testDataFilterName'], ['eventCount'], None, 50),
            ('R12', cur, ['signedInWithUserId', 'hostName'], ['totalUsers', 'eventCount'], None, 250),
            ('R13A', week, ['userAgeBracket'], ['activeUsers'], None, 50),
            ('R13B', cur, ['eventName', 'date'], ['keyEvents'], exact('isKeyEvent', 'true'), 1000),
            ('R16', cur, ['pagePath'], ['sessions'], None, 100),
            ('R17A', cur, [], ['ecommercePurchases', 'purchaseRevenue', 'grossPurchaseRevenue'], None, 10),
            ('R17B', cur, [], ['itemsPurchased', 'grossItemRevenue'], None, 10),
            ('R17C', cur, ['itemName'], ['itemsPurchased'], empty('itemName'), 10)]


def window(as_of):
    end = as_of - timedelta(days=2)
    start = end - timedelta(days=27)
    pend = start - timedelta(days=1)
    return {'start': start.isoformat(), 'end': end.isoformat(),
            'previous_start': (pend - timedelta(days=27)).isoformat(), 'previous_end': pend.isoformat()}


def rows_of(js):
    dims = [h['name'] for h in js.get('dimensionHeaders', [])]
    mets = [h['name'] for h in js.get('metricHeaders', [])]
    for r in js.get('rows', []):
        row = {k: v.get('value', '') for k, v in zip(dims, r.get('dimensionValues', []))}
        row.update({k: float(v.get('value') or 0) for k, v in zip(mets, r.get('metricValues', []))})
        yield row


class Collector:
    def __init__(self, pid, token, quota_project=None):
        self.pid, self.token, self.qp = pid, token, quota_project
        self.not_readable, self.log = [], []

    def call(self, method, url, body=None, form=False):
        if not any(m == method and rx.match(url) for m, rx in ALLOWED):
            raise RuntimeError(f'blocked before sending: {method} {url.split("?")[0]} is not on the read-only allowlist')
        headers = {'Accept': 'application/json'}
        if form:
            data = urllib.parse.urlencode(body).encode()
            headers['Content-Type'] = 'application/x-www-form-urlencoded'
        else:
            headers['Authorization'] = 'Bearer' + ' ' + self.token
            if self.qp:
                headers['x-goog-user-project'] = self.qp
            data = json.dumps(body).encode() if body is not None else None
            if data:
                headers['Content-Type'] = 'application/json'
            self.log.append(f'{method} {url.split("?")[0].replace(ADMIN, "admin/").replace(DATA, "data/v1beta/")}')
        for attempt in (1, 2):
            try:
                with urllib.request.urlopen(urllib.request.Request(url, data=data, method=method, headers=headers), timeout=60) as r:
                    return 200, json.loads(r.read() or b'{}')
            except urllib.error.HTTPError as e:
                if e.code == 401 and not form:
                    raise Stop('401 Unauthenticated: mint fresh read-only credentials and run again')
                if e.code in (429, 500, 502, 503, 504) and attempt == 1:
                    time.sleep(2)
                    continue
                try:
                    msg = json.loads(e.read() or b'{}').get('error', {}).get('message', '')
                except ValueError:
                    msg = ''
                return e.code, {'error': msg[:200]}
            except urllib.error.URLError as e:
                if attempt == 1:
                    time.sleep(2)
                    continue
                return 0, {'error': str(e.reason)[:200]}
        return 0, {'error': 'no response'}

    def missing(self, what, code, js):
        self.not_readable.append({'endpoint': what, 'status': code, 'reason': js.get('error', '')})
        print(f'  not readable: {what} ({code})', file=sys.stderr)

    def admin(self, path, version='v1beta', listing=False):
        url = f'{ADMIN}{version}/properties/{self.pid}{path}'
        if not listing:
            code, js = self.call('GET', url)
            return js if code == 200 else self.missing(f'{version}{path or " property"}', code, js)
        out, page = [], None
        while True:
            q = {'pageSize': 200, **({'pageToken': page} if page else {})}
            code, js = self.call('GET', url + '?' + urllib.parse.urlencode(q))
            if code != 200:
                return self.missing(f'{version}{path}', code, js)
            out += next((v for k, v in js.items() if isinstance(v, list)), [])
            page = js.get('nextPageToken')
            if not page:
                return out

    def report(self, rng, dims, mets, flt=None, limit=1000, offset=0, label=''):
        body = {'dateRanges': [{'startDate': rng[0], 'endDate': rng[1]}], 'dimensions': [{'name': d} for d in dims],
                'metrics': [{'name': m} for m in mets], 'limit': limit, 'offset': offset}
        if flt:
            body['dimensionFilter'] = flt
        code, js = self.call('POST', f'{DATA}properties/{self.pid}:runReport', body)
        return js if code == 200 else self.missing(f'runReport {label}', code, js)


def scope_gate(col):
    code, info = col.call('POST', TOKENINFO, {'access_token': col.token}, form=True)
    if code != 200:
        print('  scope check could not run; continuing with the allowlist only', file=sys.stderr)
        return f'not run (tokeninfo status {code})', []
    scopes = sorted(set((info.get('scope') or '').split()))
    bad = [s for s in scopes if s in STOP_SCOPES]
    if bad:
        raise Stop('the token carries scopes that can change GA4 or read users: ' + ', '.join(bad)
                   + '. Mint credentials with analytics.readonly only; do not paste a token into chat.')
    return 'passed' + (' (cloud-platform noted)' if AUTH + 'cloud-platform' in scopes else ''), scopes


def get_token():
    try:
        out = subprocess.run(['gcloud', 'auth', 'application-default', 'print-access-token'],
                             capture_output=True, text=True, timeout=60)
    except (OSError, subprocess.TimeoutExpired):
        raise Stop('gcloud is not available; install it or use the official GA4 MCP server or interface exports')
    if out.returncode != 0 or not out.stdout.strip():
        raise Stop('gcloud printed no token; run gcloud auth application-default login with the analytics.readonly scope')
    return out.stdout.strip()


def transaction_counts(col, w):
    out, offset = {'purchases': 0, 'missing_ids': 0, 'distinct_ids': 0, 'ids_on_more_than_one_event': 0}, 0
    while True:
        js = col.report((w['start'], w['end']), ['transactionId'], ['eventCount'], exact('eventName', 'purchase'), 100000, offset, 'R08')
        if js is None:
            return None
        rows = list(rows_of(js))
        for r in rows:
            t, n = r.get('transactionId', '').strip(), r['eventCount']
            out['purchases'] += n
            if t in ('', '(not set)'):
                out['missing_ids'] += n
            else:
                out['distinct_ids'] += 1
                out['ids_on_more_than_one_event'] += n > 1
        offset += len(rows)
        total = int(js.get('rowCount', 0))
        if not rows or offset >= total or offset >= 500000:
            return dict(out, rows_read=offset, truncated=offset < total, metadata=js.get('metadata', {}))


def dimension_probes(col, w, dims, cap):
    out = {}
    for d in [x for x in dims if x.get('scope') in ('EVENT', 'USER')][:cap]:
        api = ('customEvent:' if d['scope'] == 'EVENT' else 'customUser:') + d.get('parameterName', '')
        r28 = col.report((w['start'], w['end']), [api], ['eventCount'], None, 1000, 0, f'R14 {api}')
        r1 = col.report((w['end'], w['end']), [api], ['eventCount'], None, 1000, 0, f'R14 {api} one day')
        if r28 is None or r1 is None:
            continue
        rows = list(rows_of(r28))
        vals = [str(r.get(api, '')) for r in rows]
        total = sum(r['eventCount'] for r in rows)
        notset = sum(r['eventCount'] for r in rows if r.get(api) in ('', '(not set)'))
        out[api] = {'rows_28d': len(rows), 'not_set_share_28d': round(notset / total, 4) if total else 1.0,
                    'distinct_1d': int(r1.get('rowCount', 0)), 'email_like': sum(bool(EMAIL_RE.search(v)) for v in vals),
                    'phone_like': sum(bool(PHONE_RE.match(v.strip())) for v in vals)}
    return out


def pii_probe(col, w):
    out = []
    for field in ('pageLocation', 'pageReferrer'):
        exprs = [{'filter': {'fieldName': field, 'stringFilter': {'matchType': 'CONTAINS', 'value': v}}}
                 for v in R15_TERMS]
        js = col.report((w['start'], w['end']), [field], ['eventCount'], {'orGroup': {'expressions': exprs}}, 1000, 0, f'R15 {field}')
        for r in rows_of(js or {}):
            red = reduce_url(r.get(field, ''), field, r['eventCount'])
            if red:
                out.append(red)
    return {'rows': out}


def dry_run(pid, w, cap):
    print(f'Planned read-only calls for properties/{pid}, window {w["start"]} to {w["end"]} (previous {w["previous_start"]} to {w["previous_end"]}):')
    print(f'  POST {TOKENINFO} (scope gate; token in the body, never in the URL)')
    print(f'  GET  {ADMIN}v1beta/properties/{pid}')
    print(f'  GET  {ADMIN}v1beta/accounts/{{account from the property}}/dataSharingSettings (flags only)')
    for p in ADMIN_SINGLES.split('|'):
        print(f'  GET  {ADMIN}{"v1beta" if p == "dataRetentionSettings" else "v1alpha"}/properties/{pid}/{p}')
    for p in ADMIN_LISTS.split('|'):
        v = 'v1beta' if p in ('dataStreams', 'keyEvents', 'customDimensions', 'customMetrics', 'googleAdsLinks', 'firebaseLinks') else 'v1alpha'
        print(f'  GET  {ADMIN}{v}/properties/{pid}/{p}?pageSize=200 (follows nextPageToken)')
    for p in ('enhancedMeasurementSettings', 'dataRedactionSettings', 'eventCreateRules', 'eventEditRules'):
        print(f'  GET  {ADMIN}v1alpha/properties/{pid}/dataStreams/{{web stream}}/{p} (each web stream)')
    print(f'  GET  {DATA}properties/{pid}/metadata')
    for rid, rng, dims, mets, flt, limit in report_specs(w):
        print(f'  POST {DATA}properties/{pid}:runReport {rid} {rng[0]}..{rng[1]} {",".join(dims) or "(no dimensions)"} | {",".join(mets)}'
              + (' | filtered' if flt else ''))
    print(f'  POST runReport R08 transactionId | eventCount (purchase only, paginated; counts kept, IDs dropped)')
    print(f'  POST runReport R14 two calls per event or user custom dimension, first {cap} (counts kept, values dropped)')
    print(f'  POST runReport R15 pageLocation and pageReferrer PII probe (host, path and key names kept, values dropped)')
    print(f'Rows dropped before writing: {", ".join(METADATA_ONLY)} (rowCount and metadata kept)')
    print('Never called: ' + '; '.join(NEVER))


def main(argv=None):
    ap = argparse.ArgumentParser(description='Read-only GA4 snapshot collector (Admin API GET calls and Data API reports).')
    ap.add_argument('--property', required=True, help='numeric GA4 property ID, for example 123456789')
    ap.add_argument('--as-of', default=date.today().isoformat(), help='YYYY-MM-DD; the window ends two days earlier')
    ap.add_argument('--out', default='ga4-snapshot.json')
    ap.add_argument('--quota-project', help='Cloud project billed for quota (sends x-goog-user-project)')
    ap.add_argument('--skip-reports', action='store_true', help='settings only, no Data API reports')
    ap.add_argument('--max-custom-dims', type=int, default=60)
    ap.add_argument('--dry-run', action='store_true', help='print the planned calls and make no network request')
    a = ap.parse_args(argv)
    pid = re.sub(r'\D', '', a.property)
    try:
        w = window(date.fromisoformat(a.as_of))
    except ValueError:
        print('--as-of must be YYYY-MM-DD', file=sys.stderr)
        return 2
    if a.dry_run:
        dry_run(pid, w, a.max_custom_dims)
        return 0
    try:
        col = Collector(pid, get_token(), a.quota_project)
        check, scopes = scope_gate(col)
        print(f'scope check: {check}; scopes: {", ".join(s.replace(AUTH, "") for s in scopes) or "unknown"}', file=sys.stderr)
        snap = {'snapshot_version': '1', 'read_time_utc': datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'),
                'route': 'collector', 'scopes_granted': scopes, 'scope_check': check, 'window': w}
        snap['property'] = col.admin('')
        if snap['property'] is None and col.not_readable and col.not_readable[-1]['status'] == 403 and not a.quota_project:
            print('  hint: pass --quota-project with a Cloud project that has the Analytics Admin and Data APIs enabled',
                  file=sys.stderr)
        acct = (snap['property'] or {}).get('account') or (snap['property'] or {}).get('parent') or ''
        if re.fullmatch(r'accounts/\d+', acct):
            code, js = col.call('GET', f'{ADMIN}v1beta/{acct}/dataSharingSettings')
            snap['data_sharing'] = js if code == 200 else col.missing('v1beta account dataSharingSettings', code, js)
        snap['data_retention'] = col.admin('/dataRetentionSettings')
        for key, path, ver in (('attribution_settings', '/attributionSettings', 'v1alpha'), ('google_signals', '/googleSignalsSettings', 'v1alpha'),
                               ('reporting_identity', '/reportingIdentitySettings', 'v1alpha'),
                               ('user_provided_data', '/userProvidedDataSettings', 'v1alpha')):
            snap[key] = col.admin(path, ver)
        for key, path, ver in (('data_streams', '/dataStreams', 'v1beta'), ('key_events', '/keyEvents', 'v1beta'),
                               ('custom_dimensions', '/customDimensions', 'v1beta'), ('custom_metrics', '/customMetrics', 'v1beta'),
                               ('google_ads_links', '/googleAdsLinks', 'v1beta'), ('firebase_links', '/firebaseLinks', 'v1beta'),
                               ('bigquery_links', '/bigQueryLinks', 'v1alpha'), ('channel_groups', '/channelGroups', 'v1alpha'),
                               ('annotations', '/reportingDataAnnotations', 'v1alpha'),
                               ('calculated_metrics', '/calculatedMetrics', 'v1alpha')):
            snap[key] = col.admin(path, ver, listing=True)
        for link in snap['google_ads_links'] or []:
            link.pop('creatorEmailAddress', None)
        auds = col.admin('/audiences', 'v1alpha', listing=True)
        snap['audiences_count'] = None if auds is None else len(auds)
        snap['audience_names'] = None if auds is None else [x.get('displayName') for x in auds]
        for s in snap['data_streams'] or []:
            if s.get('type') != 'WEB_DATA_STREAM':
                continue
            base = '/dataStreams/' + s['name'].split('/')[-1]
            s['enhanced_measurement'] = col.admin(base + '/enhancedMeasurementSettings', 'v1alpha')
            s['data_redaction'] = col.admin(base + '/dataRedactionSettings', 'v1alpha')
            s['event_create_rules'] = col.admin(base + '/eventCreateRules', 'v1alpha', listing=True)
            s['event_edit_rules'] = col.admin(base + '/eventEditRules', 'v1alpha', listing=True)
        if not a.skip_reports:
            snap['reports'] = {rid: col.report(rng, dims, mets, flt, limit, 0, rid) for rid, rng, dims, mets, flt, limit in report_specs(w)}
            snap['reports']['R08'] = transaction_counts(col, w)
            snap['reports']['R14'] = dimension_probes(col, w, snap['custom_dimensions'] or [], a.max_custom_dims)
            snap['reports']['R15'] = pii_probe(col, w)
            for rid in METADATA_ONLY:  # the engine reads only rowCount and metadata of these probes
                if snap['reports'].get(rid):
                    snap['reports'][rid] = {'rowCount': snap['reports'][rid].get('rowCount'), 'metadata': snap['reports'][rid].get('metadata', {})}
            code, meta = col.call('GET', f'{DATA}properties/{pid}/metadata')
            if code != 200:
                col.missing('data/v1beta metadata', code, meta)
            snap['metadata_api'] = {k: [{'apiName': x.get('apiName'), 'uiName': x.get('uiName')} for x in meta.get(k, [])
                                        if str(x.get('apiName', '')).startswith('custom')] for k in ('dimensions', 'metrics')} if code == 200 else None
            snap['reports'] = {k: v for k, v in snap['reports'].items() if v is not None}
    except Stop as e:
        print(f'Stopped: {e}', file=sys.stderr)
        return 2
    snap['not_readable'] = col.not_readable
    snap['call_log'] = col.log
    with open(a.out, 'w', encoding='utf-8') as fh:
        json.dump(snap, fh, indent=1)
        fh.write('\n')
    print(f'Wrote {a.out}: {len(col.log)} read-only calls, {len(col.not_readable)} endpoints not readable. Nothing in GA4 was changed.')
    return 3 if col.not_readable else 0


if __name__ == '__main__':
    sys.exit(main())