#!/usr/bin/env python3 """CrUX + GSC fetch → normalized JSON. Third-party imports are LAZY so mock and degraded paths run stdlib-only (no venv, no network).""" import argparse, json, os, sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) def _mock(name): d = os.environ.get("SEO_DATA_MOCK_DIR") if not d: return None path = os.path.join(d, name) if not os.path.exists(path): return None with open(path, encoding="utf-8") as f: return json.load(f) def _norm_crux(raw): m = raw["record"]["metrics"] def p75(metric): return m.get(metric, {}).get("percentiles", {}).get("p75") out = {"status": "ok", "source": "crux"} lcp = p75("largest_contentful_paint") inp = p75("interaction_to_next_paint") cls = p75("cumulative_layout_shift") # Low-traffic origins often miss a metric (INP notably) — omit, don't crash. if lcp is not None: out["lcp_p75_ms"] = int(lcp) if inp is not None: out["inp_p75_ms"] = int(inp) if cls is not None: out["cls_p75"] = float(cls) if len(out) == 2: # no metric at all return {"status": "degraded", "reason": "no_field_data"} return out def _crux_query(key, body): import requests # lazy return requests.post( "https://chromeuxreport.googleapis.com/v1/records:queryRecord?key=" + key, json=body, timeout=20) def _origin(url): from urllib.parse import urlparse # stdlib p = urlparse(url) return "%s://%s" % (p.scheme, p.netloc) # strip path — CrUX origin = scheme+host only def crux(url, strategy="mobile"): raw = _mock("crux_%s.json" % strategy) if raw is None: key = os.environ.get("CRUX_API_KEY") if not key: return {"status": "degraded", "reason": "no_crux_key"} ff = "PHONE" if strategy == "mobile" else "DESKTOP" r = _crux_query(key, {"url": url, "formFactor": ff}) if r.status_code == 404: # no page-level data → try origin-level r = _crux_query(key, {"origin": _origin(url), "formFactor": ff}) if r.status_code == 404: return {"status": "degraded", "reason": "no_field_data"} if r.status_code == 429: return {"status": "degraded", "reason": "rate_limited"} r.raise_for_status() raw = r.json() return _norm_crux(raw) def _gsc_session(store_path, account): """Return an authorized requests.Session or a degrade dict. Lazy imports.""" rt = None if store_path and account: import tokenstore # local module, stdlib rt = tokenstore.get_refresh_token(store_path, account) cid = os.environ.get("GOOGLE_OAUTH_CLIENT_ID") csec = os.environ.get("GOOGLE_OAUTH_CLIENT_SECRET") if not (rt and cid and csec): return {"status": "degraded", "reason": "no_credentials"} from google.oauth2.credentials import Credentials # lazy from google.auth.transport.requests import AuthorizedSession, Request creds = Credentials(None, refresh_token=rt, client_id=cid, client_secret=csec, token_uri="https://oauth2.googleapis.com/token", scopes=["https://www.googleapis.com/auth/webmasters.readonly"]) try: creds.refresh(Request()) except Exception as e: # Only a real RefreshError means re-consent; a network blip must NOT # send the user back through OAuth. from google.auth.exceptions import RefreshError # lazy reason = "token_revoked" if isinstance(e, RefreshError) else "network_error" return {"status": "degraded", "reason": reason} return AuthorizedSession(creds) def _norm_queries(raw, dim): return {"status": "ok", "source": "gsc", "dimension": dim, "rows": [ {"key": r["keys"][0], "clicks": r.get("clicks", 0), "impressions": r.get("impressions", 0), "ctr": r.get("ctr", 0), "position": r.get("position")} for r in raw.get("rows", [])]} def queries(store_path, account, property, days=90, dim="query"): raw = _mock("gsc_queries.json") if raw is None: sess = _gsc_session(store_path, account) if isinstance(sess, dict): return sess import datetime as _dt end = _dt.date.today(); start = end - _dt.timedelta(days=days) import urllib.parse url = ("https://searchconsole.googleapis.com/webmasters/v3/sites/" + urllib.parse.quote(property, safe="") + "/searchAnalytics/query") r = sess.post(url, json={"startDate": start.isoformat(), "endDate": end.isoformat(), "dimensions": [dim], "rowLimit": 100}, timeout=30) if r.status_code == 429: return {"status": "degraded", "reason": "rate_limited"} r.raise_for_status() raw = r.json() return _norm_queries(raw, dim) def _rollup_issues(items): """Count issue instances by severity; dedupe messages (they repeat per item).""" errors = warnings = 0 msgs = [] for item in items: for iss in item.get("issues", []): sev = iss.get("severity") if sev == "ERROR": errors += 1 elif sev == "WARNING": warnings += 1 msg = iss.get("issueMessage") if msg and msg not in msgs: msgs.append(msg) return errors, warnings, msgs def _norm_rich(ir): """richResultsResult → verdict + per-type rollup. Google OMITS the key when it detects no rich results, so absence is data, not an error: surfaced as the synthetic verdict ABSENT (not a Google enum) rather than a missing key, which a caller cannot tell apart from a check that never ran. PARTIAL is never emitted — the API reserves it as unused.""" rr = ir.get("richResultsResult") if rr is None: return {"verdict": "ABSENT", "types": []} types = [] for det in rr.get("detectedItems", []): errors, warnings, msgs = _rollup_issues(det.get("items", [])) types.append({"type": det.get("richResultType"), "items": len(det.get("items", [])), "errors": errors, "warnings": warnings, "issues": msgs}) return {"verdict": rr.get("verdict"), "types": types} def inspect(store_path, account, property, url): raw = _mock("gsc_inspect.json") if raw is None: sess = _gsc_session(store_path, account) if isinstance(sess, dict): return sess r = sess.post("https://searchconsole.googleapis.com/v1/urlInspection/index:inspect", json={"inspectionUrl": url, "siteUrl": property}, timeout=30) if r.status_code == 429: return {"status": "degraded", "reason": "rate_limited"} r.raise_for_status() raw = r.json() ir = raw["inspectionResult"] isr = ir["indexStatusResult"] # rich_results rides the SAME response — Google already sent it and this # function used to discard it. No extra call, no extra quota, no new scope. return {"status": "ok", "source": "gsc", "indexed": isr.get("verdict") == "PASS", "coverage": isr.get("coverageState"), "last_crawl": isr.get("lastCrawlTime"), "rich_results": _norm_rich(ir)} def _cli(): try: p = argparse.ArgumentParser() sub = p.add_subparsers(dest="cmd", required=True) pc = sub.add_parser("crux") pc.add_argument("--url", required=True) pc.add_argument("--strategy", default="mobile", choices=["mobile", "desktop"]) pc.add_argument("--store", default=None) # accepted+ignored: uniform fetch.sh dispatch pq = sub.add_parser("queries") pq.add_argument("--store", required=True) pq.add_argument("--account", required=True) pq.add_argument("--property", required=True) pq.add_argument("--days", type=int, default=90) pq.add_argument("--dim", default="query") pi = sub.add_parser("inspect") pi.add_argument("--store", required=True) pi.add_argument("--account", required=True) pi.add_argument("--property", required=True) pi.add_argument("--url", required=True) args = p.parse_args() if args.cmd == "crux": print(json.dumps(crux(args.url, args.strategy), indent=2)) elif args.cmd == "queries": print(json.dumps(queries(args.store, args.account, args.property, args.days, args.dim), indent=2)) elif args.cmd == "inspect": print(json.dumps(inspect(args.store, args.account, args.property, args.url), indent=2)) except SystemExit as e: # argparse usage error if e.code not in (0, None): print(json.dumps({"status": "error", "reason": "bad_usage"})) raise # preserve argparse's exit code except Exception: # Fail-open data contract: ANY unexpected error (HTTP 403/5xx, DNS, # timeout) degrades with exit 0 — never a traceback, never empty stdout. print(json.dumps({"status": "degraded", "reason": "unexpected_error"})) if __name__ == "__main__": _cli()