`ops/tk10631/backfill_existing_ga4.py`

71 lines
#!/usr/bin/env python3
"""TK-10631: backfill data/domains.json with GA4 measurement IDs for domains
that ALREADY have a purpose-built GA4 property (stream default_uri host ==
the domain). No Google write, no deploy — local, reversible via git.

Only fills domains that are (a) in the verified live-target list, (b) currently
have an empty `ga4`, and (c) have exactly one existing stream keyed to that host.
"""
import argparse, json
from urllib.parse import urlparse
from google.oauth2 import service_account
from google.analytics.admin import AnalyticsAdminServiceClient

KEY = "/Users/macstudio3/.config/ga-analytics-agent/service-account.json"
ap = argparse.ArgumentParser()
ap.add_argument("--targets", required=True)
ap.add_argument("--apply", action="store_true", help="write the file (default: preview)")
a = ap.parse_args()

creds = service_account.Credentials.from_service_account_file(
    KEY, scopes=["https://www.googleapis.com/auth/analytics.readonly"])
c = AnalyticsAdminServiceClient(credentials=creds)

idx = {}
for s in c.list_account_summaries():
    for p in s.property_summaries:
        try:
            for st in c.list_data_streams(parent=p.property):
                w = st.web_stream_data
                if w and w.default_uri and w.measurement_id:
                    h = urlparse(w.default_uri).netloc.lower()
                    h = h[4:] if h.startswith("www.") else h
                    if h:
                        idx.setdefault(h, (s.account, p.property, w.measurement_id))
        except Exception:
            pass

cfg = json.load(open("data/domains.json"))
targets = [d.strip() for d in open(a.targets) if d.strip()]
fills, missing = {}, []
for d in targets:
    if cfg.get(d, {}).get("ga4"):
        continue
    if d in idx:
        fills[d] = idx[d]
    else:
        missing.append(d)

for d, (acct, prop, mid) in sorted(fills.items()):
    print(f"  {d:34s} <- {mid:15s} {prop} ({acct})")
print(f"\n{len(fills)} to fill, {len(missing)} have no existing property "
      f"(need creation): {missing}")

if not a.apply:
    print("\nPREVIEW ONLY — pass --apply to write data/domains.json")
    raise SystemExit(0)

for d, (_, _, mid) in fills.items():
    cfg[d]["ga4"] = mid
json.dump(cfg, open("data/domains.json", "w"), indent=2)
open("data/domains.json", "a").write("\n")

# read-back verify
rb = json.load(open("data/domains.json"))
bad = [d for d, (_, _, mid) in fills.items() if rb[d].get("ga4") != mid]
assert not bad, f"read-back MISMATCH: {bad}"
assert len(rb) == len(cfg), "domain count changed!"
print(f"\nOK: wrote + read-back-verified {len(fills)} measurement IDs. "
      f"{sum(1 for v in rb.values() if v.get('ga4'))}/{len(rb)} domains now tagged.")
print("NOT deployed — customer-facing deploy remains gated.")