our-claude-skills/custom-skills/35-seo-signal-validation/code/scripts/gsc_signal_delta.py

#!/usr/bin/env python3
"""Day-normalized GSC query delta + mover ranking for signal validation.

Reads two GSC query exports (recent, prior) — JSON list or TSV with a header row
containing query / clicks / impressions / position — and reports day-normalized
site totals, top gainers/decliners, and whether a claimed term is a real mover.
This is the deterministic L1/L4 core of the 35-seo-signal-validation skill.
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path


def _norm_row(r: dict) -> dict:
    def num(*keys, default=0.0):
        for k in keys:
            if k in r and r[k] not in (None, ""):
                try:
                    return float(str(r[k]).replace(",", ""))
                except ValueError:
                    pass
        return default
    query = (r.get("query") or r.get("term") or "")
    if isinstance(r.get("keys"), list) and r["keys"]:
        query = str(r["keys"][0])
    return {
        "query": str(query).strip(),
        "clicks": num("clicks"),
        "impressions": num("impressions", "impr"),
        "position": num("position", "pos", default=0.0),
    }


def load_gsc(path: str) -> list[dict]:
    """Parse a GSC export (JSON list/{rows:[...]} or TSV-with-header)."""
    text = Path(path).read_text(encoding="utf-8").strip()
    if not text:
        return []
    if text[0] in "[{":
        data = json.loads(text)
        if isinstance(data, dict):
            data = data.get("rows", [])
        return [_norm_row(r) for r in data]
    lines = text.splitlines()
    header = [h.strip().lower() for h in lines[0].split("\t")]
    rows = []
    for line in lines[1:]:
        if line.strip():
            rows.append(_norm_row(dict(zip(header, line.split("\t")))))
    return rows


def _by_query(rows: list[dict]) -> dict:
    return {r["query"]: r for r in rows if r["query"]}


def compute_delta(recent, prior, recent_days, prior_days,
                  claim_term=None, top_n=10) -> dict:
    if recent_days <= 0 or prior_days <= 0:
        raise ValueError("recent_days and prior_days must be positive")
    r_by, p_by = _by_query(recent), _by_query(prior)

    def totals(rows):
        return {"clicks": sum(r["clicks"] for r in rows),
                "impressions": sum(r["impressions"] for r in rows)}
    rt, pt = totals(recent), totals(prior)

    def per_day(total, days):
        return round(total / days, 2)

    def pct(new, old):
        return round((new - old) / old * 100, 1) if old else None

    r_cpd, p_cpd = per_day(rt["clicks"], recent_days), per_day(pt["clicks"], prior_days)
    r_ipd, p_ipd = per_day(rt["impressions"], recent_days), per_day(pt["impressions"], prior_days)

    deltas = []
    for q in set(r_by) | set(p_by):
        rc = r_by.get(q, {}).get("clicks", 0.0)
        pc = p_by.get(q, {}).get("clicks", 0.0)
        deltas.append({"query": q, "delta_clicks": rc - pc,
                       "recent_clicks": rc, "prior_clicks": pc})
    deltas.sort(key=lambda d: d["delta_clicks"], reverse=True)
    gainers = [d for d in deltas if d["delta_clicks"] > 0][:top_n]
    decliners = sorted([d for d in deltas if d["delta_clicks"] < 0],
                       key=lambda d: d["delta_clicks"])[:top_n]

    out = {
        "site_totals": {
            "recent": {**rt, "clicks_per_day": r_cpd,
                       "impressions_per_day": r_ipd, "days": recent_days},
            "prior": {**pt, "clicks_per_day": p_cpd,
                      "impressions_per_day": p_ipd, "days": prior_days},
            "clicks_per_day_pct": pct(r_cpd, p_cpd),
            "impressions_per_day_pct": pct(r_ipd, p_ipd),
        },
        "top_gainers": gainers,
        "top_decliners": decliners,
        "claim_term": None,
        "verdict_hint": None,
    }

    if claim_term:
        gainer_terms = {g["query"] for g in gainers}
        rc, pc = r_by.get(claim_term, {}), p_by.get(claim_term, {})
        in_movers = claim_term in gainer_terms
        found = bool(rc or pc)
        share = (rc.get("clicks", 0.0) / rt["clicks"] * 100) if rt["clicks"] else 0.0
        out["claim_term"] = {
            "term": claim_term, "found": found,
            "recent": {"clicks": rc.get("clicks", 0.0),
                       "impressions": rc.get("impressions", 0.0),
                       "position": rc.get("position")},
            "prior": {"clicks": pc.get("clicks", 0.0),
                      "impressions": pc.get("impressions", 0.0),
                      "position": pc.get("position")},
            "in_top_movers": in_movers,
            "click_share_pct": round(share, 2),
        }
        if not found:
            out["verdict_hint"] = (
                f"'{claim_term}' is absent from both GSC windows (no impressions / "
                f"likely anonymized) -> INCONCLUSIVE, not refuted; confirm via live "
                f"SERP + entity layer.")
        elif not in_movers and share < 1.0:
            out["verdict_hint"] = (
                f"'{claim_term}' contributes {share:.2f}% of recent clicks and is "
                f"absent from top movers -> claimed impact likely ARTIFACT; real "
                f"movement is elsewhere (see top_gainers).")
        elif in_movers:
            out["verdict_hint"] = (
                f"'{claim_term}' is among top movers -> claim plausibly CONFIRMED/"
                f"PARTIAL; corroborate with live SERP + entity layer.")
        else:
            out["verdict_hint"] = (
                f"'{claim_term}' has non-trivial share ({share:.2f}%) but is not a "
                f"top mover -> PARTIAL; inspect attribution.")
    return out


def main(argv=None):
    ap = argparse.ArgumentParser(description="GSC signal delta for signal validation")
    ap.add_argument("--recent", required=True)
    ap.add_argument("--prior", required=True)
    ap.add_argument("--recent-days", type=int, required=True)
    ap.add_argument("--prior-days", type=int, required=True)
    ap.add_argument("--claim-term", default=None)
    ap.add_argument("--top-n", type=int, default=10)
    a = ap.parse_args(argv)
    out = compute_delta(load_gsc(a.recent), load_gsc(a.prior),
                        a.recent_days, a.prior_days, a.claim_term, a.top_n)
    json.dump(out, sys.stdout, ensure_ascii=False, indent=2)
    sys.stdout.write("\n")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())