diff --git a/custom-skills/35-seo-signal-validation/code/CLAUDE.md b/custom-skills/35-seo-signal-validation/code/CLAUDE.md new file mode 100644 index 0000000..4501a1b --- /dev/null +++ b/custom-skills/35-seo-signal-validation/code/CLAUDE.md @@ -0,0 +1,23 @@ +# seo-signal-validation — code environment notes + +## Helper: scripts/gsc_signal_delta.py +Deterministic L1/L4 GSC delta. Feed it two saved GSC query exports (recent, +prior) as JSON or TSV (columns: query, clicks, impressions, position). + +```bash +python3 scripts/gsc_signal_delta.py \ + --recent recent.tsv --prior prior.tsv \ + --recent-days 28 --prior-days 30 --claim-term "호텔" +``` +Returns day-normalized site totals, top gainers/decliners, and a `verdict_hint` +(heuristic only — the final verdict is the skill's job, after L2/L3). + +## Getting the exports +`mcp__dda__gsc_fetch_performance` (property pinned per workspace, e.g. JHR +`sc-domain:josunhotel.com`) → save the query-dimension rows to a file → run the +script. GSC anonymizes ~43% of query clicks; the disclosed subset ≠ the whole. + +## Env / access +- `GOOGLE_KG_API_KEY` for `mcp__ourseo__search_knowledge_graph` (L3). +- GSC/GA4 only exist for first-party properties — third-party entities skip L1. +- Never crawl/audit Marriott for JHR (sameAs only). diff --git a/custom-skills/35-seo-signal-validation/code/scripts/gsc_signal_delta.py b/custom-skills/35-seo-signal-validation/code/scripts/gsc_signal_delta.py new file mode 100644 index 0000000..79c718e --- /dev/null +++ b/custom-skills/35-seo-signal-validation/code/scripts/gsc_signal_delta.py @@ -0,0 +1,154 @@ +#!/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 + share = (rc.get("clicks", 0.0) / rt["clicks"] * 100) if rt["clicks"] else 0.0 + out["claim_term"] = { + "term": claim_term, "found": bool(rc or pc), + "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 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()) diff --git a/custom-skills/35-seo-signal-validation/code/scripts/requirements.txt b/custom-skills/35-seo-signal-validation/code/scripts/requirements.txt new file mode 100644 index 0000000..3491c56 --- /dev/null +++ b/custom-skills/35-seo-signal-validation/code/scripts/requirements.txt @@ -0,0 +1 @@ +# gsc_signal_delta.py uses the Python 3 standard library only — no deps. diff --git a/custom-skills/35-seo-signal-validation/code/scripts/test_gsc_signal_delta.py b/custom-skills/35-seo-signal-validation/code/scripts/test_gsc_signal_delta.py new file mode 100644 index 0000000..79d6c9a --- /dev/null +++ b/custom-skills/35-seo-signal-validation/code/scripts/test_gsc_signal_delta.py @@ -0,0 +1,59 @@ +#!/usr/bin/env python3 +"""Tests for gsc_signal_delta. Run: `python3 test_gsc_signal_delta.py` +(also pytest-compatible). Stdlib only.""" +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent)) +from gsc_signal_delta import compute_delta # noqa: E402 + +# Genesis fixture: JHR "호텔" — flat head term, growth all brand (2026-06 case). +RECENT = [ + {"query": "호텔", "clicks": 5, "impressions": 572, "position": 11.6}, + {"query": "grand josun busan", "clicks": 250, "impressions": 4000, "position": 1.2}, + {"query": "조선호텔", "clicks": 300, "impressions": 6000, "position": 1.1}, +] +PRIOR = [ + {"query": "호텔", "clicks": 9, "impressions": 371, "position": 18.1}, + {"query": "grand josun busan", "clicks": 49, "impressions": 1500, "position": 3.4}, + {"query": "조선호텔", "clicks": 150, "impressions": 5000, "position": 1.3}, +] + + +def test_claim_term_flagged_artifact(): + out = compute_delta(RECENT, PRIOR, 28, 30, claim_term="호텔") + ct = out["claim_term"] + assert ct["found"] is True + assert ct["in_top_movers"] is False + assert ct["click_share_pct"] < 1.0 + assert "ARTIFACT" in out["verdict_hint"] + + +def test_top_gainer_is_brand_term(): + out = compute_delta(RECENT, PRIOR, 28, 30, claim_term="호텔") + assert out["top_gainers"][0]["query"] == "grand josun busan" + assert out["top_gainers"][0]["delta_clicks"] == 201 + + +def test_day_normalization(): + out = compute_delta(RECENT, PRIOR, 28, 30) + assert out["site_totals"]["recent"]["clicks_per_day"] == 19.82 # 555/28 + assert out["site_totals"]["prior"]["clicks_per_day"] == 6.93 # 208/30 + + +def test_positive_days_required(): + try: + compute_delta(RECENT, PRIOR, 0, 30) + except ValueError: + return + raise AssertionError("expected ValueError for non-positive days") + + +def _run(): + fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")] + for fn in fns: + fn(); print(f"PASS {fn.__name__}") + print(f"\n{len(fns)} passed") + + +if __name__ == "__main__": + _run()