`scripts/create_inject_properties.py`

141 lines
#!/usr/bin/env python3
"""
Batch-create GA4 properties for the 21 inject targets.
Account: accounts/15714274 (SAND LLC / DesignerWallcoverings.com)
Output: cache/_ga4_inject_ids.json
Idempotent: skips domains already in the output file.
"""
from __future__ import annotations

import json
import os
import sys
import time
from pathlib import Path

SCRIPT_DIR = Path(__file__).parent
SKILLS_DIR = Path.home() / ".claude" / "skills" / "analytics" / "scripts"
sys.path.insert(0, str(SKILLS_DIR))
sys.path.insert(0, str(SCRIPT_DIR))   # so the fail-closed guard can import ga4_absence_detect

from _auth import ensure_credentials

ACCOUNT_ID = "accounts/15714274"   # SAND LLC — SA has Editor here
TARGETS_FILE = SCRIPT_DIR.parent / "cache" / "_ga4_inject_targets.json"
OUTPUT_FILE  = SCRIPT_DIR.parent / "cache" / "_ga4_inject_ids.json"


def load_existing() -> dict:
    if OUTPUT_FILE.exists():
        try:
            return json.loads(OUTPUT_FILE.read_text())
        except Exception:
            return {}
    return {}


def save(data: dict) -> None:
    OUTPUT_FILE.write_text(json.dumps(data, indent=2))


def create_property(client, domain: str) -> str:
    """Create a GA4 property + web data stream; return measurement ID."""
    from google.analytics.admin_v1alpha.types import DataStream, Property

    prop = Property(
        parent=ACCOUNT_ID,
        display_name=domain,
        time_zone="America/Los_Angeles",
        currency_code="USD",
        industry_category="ONLINE_COMMUNITIES",
    )
    created = client.create_property(property=prop)
    prop_name = created.name
    print(f"  property: {prop_name}", flush=True)

    stream = DataStream(
        type_=DataStream.DataStreamType.WEB_DATA_STREAM,
        display_name=domain,
        web_stream_data=DataStream.WebStreamData(
            default_uri=f"https://{domain}",
        ),
    )
    stream = client.create_data_stream(parent=prop_name, data_stream=stream)
    mid = stream.web_stream_data.measurement_id
    print(f"  stream:   {stream.name}  →  {mid}", flush=True)
    return mid


def main() -> None:
    ensure_credentials()

    # Force the correct account ID regardless of on-disk config
    os.environ["GA_ACCOUNT_ID"] = ACCOUNT_ID

    from google.analytics.admin import AnalyticsAdminServiceClient
    client = AnalyticsAdminServiceClient()

    targets: list[str] = json.loads(TARGETS_FILE.read_text())
    existing = load_existing()

    succeeded = 0
    failed = 0
    failed_domains: list[str] = []

    # Fail-closed dup/park guard: re-verify each domain is genuinely injectable at mint time,
    # even if the targets file is stale. Prevents the duplicate-property (already-wired) and
    # orphan-property (parked) mints that happened 2026-08-18. See ga4_absence_detect.py.
    try:
        from ga4_absence_detect import probe as _absence_probe
    except Exception:
        _absence_probe = None

    for domain in targets:
        if domain in existing:
            print(f"[SKIP] {domain} already has {existing[domain]}", flush=True)
            continue

        if _absence_probe is not None:
            _d, bucket, info = _absence_probe(domain)
            if bucket != "NEEDS_INJECT":
                print(f"[GUARD-SKIP] {domain} — {bucket} ({info}); not minting", flush=True)
                continue

        print(f"[CREATE] {domain}", flush=True)
        for attempt in range(3):
            try:
                mid = create_property(client, domain)
                existing[domain] = mid
                save(existing)
                print(f"[OK] {domain} -> {mid}", flush=True)
                succeeded += 1
                time.sleep(1.2)   # polite rate-limit spacing
                break
            except Exception as exc:
                msg = str(exc)
                if "429" in msg or "RESOURCE_EXHAUSTED" in msg:
                    wait = 30 * (attempt + 1)
                    print(f"  rate-limited, waiting {wait}s …", flush=True)
                    time.sleep(wait)
                else:
                    print(f"  [FAIL attempt {attempt+1}] {exc}", flush=True)
                    if attempt == 2:
                        print(f"[FAIL] {domain} — giving up", flush=True)
                        failed += 1
                        failed_domains.append(domain)
                    else:
                        time.sleep(3)

    print()
    print(f"Account used : {ACCOUNT_ID}")
    print(f"Succeeded    : {succeeded}")
    print(f"Skipped      : {len([d for d in targets if d in existing]) - succeeded}")
    print(f"Failed       : {failed}  {failed_domains if failed_domains else ''}")
    print(f"Output file  : {OUTPUT_FILE}")
    print()
    print(json.dumps(existing, indent=2))


if __name__ == "__main__":
    main()