#!/usr/bin/env python3
"""Re-pull Search Console and check every published number in the AI-agent post.

WHY THIS EXISTS. The post at /blog/ai-agent-search-queries/ quotes about thirty
figures drawn from two 28-day Search Console windows. Those windows slide: the
API only ever exposes data up to roughly two days ago, so the same query run a
week later covers different days and returns different numbers. A post that was
true when drafted quietly stops being true, and nothing in the repo notices.

This script is the thing that notices. It re-pulls both windows from the live
API, recomputes every published figure from scratch, asserts that the four
sub-buckets sum to the class totals, regenerates the CSV that the Dataset
JSON-LD points at, and then greps the rendered HTML for each figure it just
computed. It exits non-zero if any published number no longer matches.

    python3 scripts/refresh_agent_query_numbers.py            # check only
    python3 scripts/refresh_agent_query_numbers.py --write-csv  # also rewrite the CSV

Run it immediately before publishing, not only at draft time. The window
boundary is resolved from the API rather than hardcoded, so the script tells you
which dates it actually used; if `last_available_date` has moved since the draft
was written, the prose dates need updating too and the script will say so.

Requires google-api-python-client and google-auth, which no system python on the
build machine has. Make a throwaway venv first:

    python3 -m venv /tmp/gscvenv
    /tmp/gscvenv/bin/pip install google-api-python-client google-auth
    /tmp/gscvenv/bin/python scripts/refresh_agent_query_numbers.py

Reads ~/.credentials/vidclean-gsc-key.json, which lives outside the repo. This
file is served publicly at vidclean.net/scripts/ like everything else here, so
it must never contain the key itself.
"""
import argparse
import collections
import datetime as dt
import json
import os
import re
import sys

KEY = os.path.expanduser("~/.credentials/vidclean-gsc-key.json")
SITE = "sc-domain:vidclean.net"
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
POST = os.path.join(ROOT, "blog", "ai-agent-search-queries", "index.html")
CSV = os.path.join(ROOT, "blog", "ai-agent-search-queries",
                   "vidclean-ai-agent-queries.csv")

# The cutoff. Eight words is deliberately conservative: it excludes plenty of
# real machine traffic (see the post's limitations) rather than sweeping in
# human long-tail to make the class look bigger. Do not loosen it to inflate a
# number. If it ever moves, it moves in the post's prose at the same time.
MIN_WORDS = 8

# Interrogative openers. A trailing "?" also qualifies; humans almost never
# type one into a search box, so it is the single strongest cheap tell.
STARTS = ("how ", "what ", "which ", "why ", "is ", "are ", "can ", "do ",
          "does ", "should ", "where ", "when ", "who ")

FACTCHECK_BRAND = "turboscribe"


def words(s):
    return len([t for t in s.split() if t.strip()])


def classify(q):
    """Bucket one query. Ordered and mutually exclusive, so the four buckets
    partition the class exactly and their counts must sum to the total."""
    low = q.lower()
    if "site:" in low:
        return "operator"
    if FACTCHECK_BRAND in low and "official" in low:
        return "factcheck"
    if low.startswith(STARTS) or "?" in q:
        return "question"
    return "other"


def connect():
    from google.oauth2 import service_account
    from googleapiclient.discovery import build
    creds = service_account.Credentials.from_service_account_file(
        KEY, scopes=["https://www.googleapis.com/auth/webmasters.readonly"])
    return build("searchconsole", "v1", credentials=creds,
                 cache_discovery=False)


def query(svc, start, end, dims, limit=25000):
    body = {"startDate": start, "endDate": end, "dimensions": dims,
            "rowLimit": limit, "startRow": 0}
    out, row = [], 0
    while True:
        body["startRow"] = row
        r = svc.searchanalytics().query(siteUrl=SITE, body=body).execute()
        rows = r.get("rows", [])
        out.extend(rows)
        if len(rows) < limit:
            break
        row += limit
        if row > 100000:
            break
    return out


def pull(svc):
    """Resolve the real freshness boundary, then take the two complete 28-day
    windows that end on it. Never hardcode the dates: GSC lags about two days
    and the lag is not constant."""
    today = dt.date.today()
    daily = query(svc, (today - dt.timedelta(days=120)).isoformat(),
                  today.isoformat(), ["date"])
    last = dt.date.fromisoformat(max(r["keys"][0] for r in daily))
    win = {
        "last_available_date": last.isoformat(),
        "pulled_at": dt.datetime.now(dt.timezone.utc).strftime(
            "%Y-%m-%dT%H:%M:%SZ"),
        "current": {"start": (last - dt.timedelta(days=27)).isoformat(),
                    "end": last.isoformat()},
        "previous": {"start": (last - dt.timedelta(days=55)).isoformat(),
                     "end": (last - dt.timedelta(days=28)).isoformat()},
    }
    data = {}
    for label in ("current", "previous"):
        w = win[label]
        data[label] = {
            "queries": query(svc, w["start"], w["end"], ["query"]),
            "daily": query(svc, w["start"], w["end"], ["date"]),
            "query_page": query(svc, w["start"], w["end"], ["query", "page"]),
        }
    return win, data


def figures(win, data):
    f = {"window": win}
    for label in ("current", "previous"):
        rows = data[label]["queries"]
        daily = data[label]["daily"]
        longs = [r for r in rows if words(r["keys"][0]) >= MIN_WORDS]

        site_impr = sum(r["impressions"] for r in daily)
        site_clicks = sum(r["clicks"] for r in daily)
        attr_impr = sum(r["impressions"] for r in rows)
        attr_clicks = sum(r["clicks"] for r in rows)

        b = collections.defaultdict(list)
        for r in longs:
            b[classify(r["keys"][0])].append(r)

        buckets = {}
        for name in ("question", "factcheck", "operator", "other"):
            v = b[name]
            i = sum(r["impressions"] for r in v)
            c = sum(r["clicks"] for r in v)
            buckets[name] = {"queries": len(v), "impressions": i, "clicks": c,
                             "ctr": round(c / i * 100, 2) if i else 0.0}

        # THE ASSERTION THIS SCRIPT EXISTS FOR. The four buckets are built by an
        # ordered if-chain, so a badly ordered edit could double-count or drop a
        # query silently and every downstream percentage would still look
        # plausible. Fail loudly instead.
        sq = sum(v["queries"] for v in buckets.values())
        si = sum(v["impressions"] for v in buckets.values())
        sc = sum(v["clicks"] for v in buckets.values())
        assert sq == len(longs), f"{label}: buckets sum to {sq} queries, class has {len(longs)}"
        assert si == sum(r["impressions"] for r in longs), f"{label}: impressions do not sum"
        assert sc == sum(r["clicks"] for r in longs), f"{label}: clicks do not sum"

        ts = [r for r in rows if FACTCHECK_BRAND in r["keys"][0].lower()]
        ts_impr = sum(r["impressions"] for r in ts)

        f[label] = {
            "site_impressions": site_impr,
            "site_clicks": site_clicks,
            "site_ctr": round(site_clicks / site_impr * 100, 2),
            "attributable_impressions": attr_impr,
            "attributable_share": round(attr_impr / site_impr * 100, 1),
            "withheld_impressions": site_impr - attr_impr,
            "withheld_share": round((site_impr - attr_impr) / site_impr * 100, 1),
            "attributable_clicks": attr_clicks,
            "all_queries": len(rows),
            "class_queries": len(longs),
            "class_impressions": sum(r["impressions"] for r in longs),
            "class_clicks": sum(r["clicks"] for r in longs),
            "class_share_of_site": round(
                sum(r["impressions"] for r in longs) / site_impr * 100, 2),
            "buckets": buckets,
            "factcheck_cluster": {
                "queries": len(ts),
                "impressions": ts_impr,
                "clicks": sum(r["clicks"] for r in ts),
                "zero_click_queries": sum(1 for r in ts if r["clicks"] == 0),
                "in_class": sum(1 for r in ts if words(r["keys"][0]) >= MIN_WORDS),
                "position_min": round(min(r["position"] for r in ts), 1) if ts else 0,
                "position_max": round(max(r["position"] for r in ts), 1) if ts else 0,
                "position_weighted": round(
                    sum(r["position"] * r["impressions"] for r in ts) / ts_impr, 2)
                if ts_impr else 0,
                "in_band_4_8": sum(1 for r in ts if 4 <= r["position"] <= 8),
            },
        }

    g = lambda a, b: round((a - b) / b * 100, 1) if b else None
    c, p = f["current"], f["previous"]
    f["growth"] = {
        "class_queries": g(c["class_queries"], p["class_queries"]),
        "class_impressions": g(c["class_impressions"], p["class_impressions"]),
        "site_impressions": g(c["site_impressions"], p["site_impressions"]),
        "question_queries": g(c["buckets"]["question"]["queries"],
                              p["buckets"]["question"]["queries"]),
        "question_impressions": g(c["buckets"]["question"]["impressions"],
                                  p["buckets"]["question"]["impressions"]),
        "ratio_vs_site": round(
            g(c["class_impressions"], p["class_impressions"])
            / g(c["site_impressions"], p["site_impressions"]), 2),
        "share_multiple": round(
            c["class_share_of_site"] / p["class_share_of_site"], 2),
    }
    f["ctr_ratio_site_vs_question"] = round(
        c["site_ctr"] / c["buckets"]["question"]["ctr"], 0) \
        if c["buckets"]["question"]["ctr"] else None
    return f


def write_csv(win, data, f):
    """One row per query in the class, both windows, plus the fact-check
    cluster. This is what the Dataset JSON-LD's contentUrl resolves to; if the
    filename here ever changes, that JSON-LD changes in the same commit."""
    import csv as csvmod
    rows = []
    for label in ("current", "previous"):
        w = win[label]
        for r in data[label]["queries"]:
            q = r["keys"][0]
            if words(q) < MIN_WORDS:
                continue
            rows.append({
                "window": label,
                "window_start": w["start"],
                "window_end": w["end"],
                "query": q,
                "word_count": words(q),
                "bucket": classify(q),
                "impressions": r["impressions"],
                "clicks": r["clicks"],
                "position": round(r["position"], 1),
            })
    # The fact-check cluster is NOT a subset of the class (most of its members
    # are under eight words), so it is carried as its own flagged block rather
    # than silently merged, which would make the class counts irreproducible.
    for label in ("current", "previous"):
        w = win[label]
        for r in data[label]["queries"]:
            q = r["keys"][0]
            if FACTCHECK_BRAND not in q.lower() or words(q) >= MIN_WORDS:
                continue
            rows.append({
                "window": label,
                "window_start": w["start"],
                "window_end": w["end"],
                "query": q,
                "word_count": words(q),
                "bucket": "factcheck_cluster_below_cutoff",
                "impressions": r["impressions"],
                "clicks": r["clicks"],
                "position": round(r["position"], 1),
            })
    rows.sort(key=lambda r: (r["window"] != "current", -r["impressions"],
                             r["query"]))
    with open(CSV, "w", newline="") as fh:
        wtr = csvmod.DictWriter(fh, fieldnames=[
            "window", "window_start", "window_end", "query", "word_count",
            "bucket", "impressions", "clicks", "position"])
        wtr.writeheader()
        wtr.writerows(rows)
    print(f"  wrote {CSV} ({len(rows)} rows)")
    return len(rows)


def check_html(f):
    """Grep the rendered post for every figure computed above. A published
    number that no longer reproduces is the whole failure mode this guards."""
    if not os.path.exists(POST):
        print(f"  SKIP html check, {POST} does not exist yet")
        return []
    html = open(POST).read()
    c, p, g = f["current"], f["previous"], f["growth"]
    fmt = lambda n: f"{n:,}"
    want = [
        ("class queries, current", fmt(c["class_queries"])),
        ("class queries, previous", fmt(p["class_queries"])),
        ("class impressions, current", fmt(c["class_impressions"])),
        ("class impressions, previous", fmt(p["class_impressions"])),
        ("question queries, current", fmt(c["buckets"]["question"]["queries"])),
        ("question queries, previous", fmt(p["buckets"]["question"]["queries"])),
        ("question impressions, current",
         fmt(c["buckets"]["question"]["impressions"])),
        ("question impressions, previous",
         fmt(p["buckets"]["question"]["impressions"])),
        ("question clicks, current", str(c["buckets"]["question"]["clicks"])),
        ("question CTR", f'{c["buckets"]["question"]["ctr"]:.2f}%'),
        ("other bucket CTR", f'{c["buckets"]["other"]["ctr"]:.2f}%'),
        ("site CTR", f'{c["site_ctr"]:.2f}%'),
        ("site impressions, current", fmt(c["site_impressions"])),
        ("site impressions, previous", fmt(p["site_impressions"])),
        ("factcheck cluster size", str(c["factcheck_cluster"]["queries"])),
        ("factcheck cluster impressions",
         fmt(c["factcheck_cluster"]["impressions"])),
        ("factcheck in class", str(c["factcheck_cluster"]["in_class"])),
        ("withheld share", f'{c["withheld_share"]}%'),
        ("withheld impressions", fmt(c["withheld_impressions"])),
        ("class growth, queries", f'{g["class_queries"]:.0f}%'),
        ("class growth, impressions", f'{g["class_impressions"]:.0f}%'),
        ("site growth", f'{g["site_impressions"]:.0f}%'),
        ("question growth, impressions", f'{g["question_impressions"]:.0f}%'),
        ("class share of site, current", f'{c["class_share_of_site"]:.2f}%'),
        ("last available date", f["window"]["last_available_date"]),
        ("current window start", f["window"]["current"]["start"]),
        ("previous window start", f["window"]["previous"]["start"]),
    ]
    missing = []
    for name, needle in want:
        if needle not in html:
            missing.append((name, needle))
    for name, needle in want:
        mark = "MISSING" if (name, needle) in missing else "ok"
        print(f"    {mark:>7}  {name:<32} {needle}")
    return missing


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--write-csv", action="store_true")
    ap.add_argument("--dump", help="write the computed figures to this path")
    a = ap.parse_args()

    print("pulling Search Console...")
    svc = connect()
    win, data = pull(svc)
    print(f"  last available date: {win['last_available_date']}")
    print(f"  current  {win['current']['start']} to {win['current']['end']}")
    print(f"  previous {win['previous']['start']} to {win['previous']['end']}")
    print(f"  pulled at {win['pulled_at']}")

    f = figures(win, data)
    print("\nsub-bucket assertions: OK (four buckets sum to class totals, "
          "both windows)")

    c, g = f["current"], f["growth"]
    print(f"\n  class: {f['previous']['class_queries']} -> {c['class_queries']} "
          f"queries ({g['class_queries']:+.1f}%), "
          f"{f['previous']['class_impressions']} -> {c['class_impressions']} "
          f"impressions ({g['class_impressions']:+.1f}%)")
    print(f"  site:  {f['previous']['site_impressions']} -> "
          f"{c['site_impressions']} impressions ({g['site_impressions']:+.1f}%)"
          f"   ratio {g['ratio_vs_site']}x")
    for n, v in c["buckets"].items():
        print(f"  {n:>10}: {v['queries']:>4} q  {v['impressions']:>5} impr  "
              f"{v['clicks']:>3} clicks  CTR {v['ctr']:.2f}%")

    if a.write_csv:
        print("\nwriting CSV...")
        write_csv(win, data, f)

    if a.dump:
        json.dump(f, open(a.dump, "w"), indent=2)
        print(f"  dumped figures to {a.dump}")

    print("\nchecking published numbers in the post...")
    missing = check_html(f)
    if missing:
        print(f"\nFAIL: {len(missing)} published figure(s) no longer reproduce.")
        for name, needle in missing:
            print(f"  {name}: expected to find {needle!r}")
        return 1
    print("\nOK: every published figure reproduces from a fresh pull.")
    return 0


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