`check.py`

98 lines
#!/usr/bin/env python3
"""dw-material-envelope-canary — READ-ONLY recurrence monitor.

Scans the live DW Shopify catalog for the double-encoded metafield bug where a
value leaked its raw JSON wrapper, e.g. {"type": "single_line_text_field",
"value": "Paper"} — instead of the plain value. Born 2026-07-07 after a Python
importer double-encoded custom.material + dwc.contents on 571 products.

Alerts (CNCP parking-lot card) ONLY on FAIL (>0 envelopes found). PASS is silent.
Always writes data/latest.json so the canary meta-watchdog sees it ran.
Touches nothing — pure GraphQL reads."""
import os, json, sys, time, urllib.request, datetime

HERE = os.path.dirname(os.path.abspath(__file__))
def load_env():
    for p in [os.path.expanduser("~/Projects/Designer-Wallcoverings/.env"),
              os.path.join(HERE, ".env")]:
        if os.path.exists(p):
            for ln in open(p):
                if "=" in ln and not ln.strip().startswith("#"):
                    k, v = ln.strip().split("=", 1)
                    os.environ.setdefault(k, v.strip().strip('"').strip("'"))
load_env()
DOMAIN = os.environ.get("SHOPIFY_STORE_DOMAIN", "designer-laboratory-sandbox.myshopify.com")
TOK = os.environ.get("SHOPIFY_ADMIN_TOKEN") or os.environ.get("SHOPIFY_ADMIN_ACCESS_TOKEN")
if not TOK:
    sys.exit("ERROR: SHOPIFY_ADMIN_TOKEN not set (check ~/Projects/Designer-Wallcoverings/.env)")
URL = f"https://{DOMAIN}/admin/api/2024-01/graphql.json"
CNCP = "http://127.0.0.1:3333/api/parking_lot"

# Daily FAST pass: the 3 keys the bug actually hit (cheap GraphQL, quick signal).
# The WEEKLY check-bulk.py covers ALL metafields exhaustively (the durable net).
CHECK_KEYS = [("custom", "material"), ("dwc", "contents"), ("global", "Material")]
FIELDS = " ".join(f'k{i}: metafield(namespace:"{ns}", key:"{k}"){{ value }}'
                  for i, (ns, k) in enumerate(CHECK_KEYS))
QUERY = "query($c:String){ products(first:200, after:$c){ pageInfo{ hasNextPage endCursor } edges{ node{ handle status " + FIELDS + " } } } }"

def gql(cursor):
    body = json.dumps({"query": QUERY, "variables": {"c": cursor}}).encode()
    req = urllib.request.Request(URL, data=body, headers={
        "X-Shopify-Access-Token": TOK, "Content-Type": "application/json"})
    for attempt in range(6):
        try:
            with urllib.request.urlopen(req, timeout=60) as r:
                d = json.loads(r.read())
            if d.get("errors") and any("throttl" in str(e).lower() for e in d["errors"]):
                time.sleep(2 + attempt); continue
            return d
        except Exception:
            time.sleep(2 + attempt)
    raise SystemExit("gql failed")

def is_env(v):
    # robust: the metafield TYPE NAME never legitimately appears in a spec value,
    # so any value containing it is a leaked envelope — catches scalar, list,
    # and double-escaped (\"single_line_text_field\") variants alike.
    return isinstance(v, str) and "single_line_text_field" in v

def main():
    os.makedirs(os.path.join(HERE, "data"), exist_ok=True)
    total = 0; bad = []
    cursor = None
    while True:
        conn = gql(cursor)["data"]["products"]
        for e in conn["edges"]:
            n = e["node"]; total += 1
            for i, (ns, k) in enumerate(CHECK_KEYS):
                mf = n.get(f"k{i}")
                if mf and is_env(mf.get("value")):
                    bad.append({"handle": n["handle"], "status": n["status"], "field": f"{ns}.{k}"})
                    break
        if not conn["pageInfo"]["hasNextPage"]:
            break
        cursor = conn["pageInfo"]["endCursor"]
    verdict = "FAIL" if bad else "PASS"
    result = {"ts": datetime.datetime.now().isoformat(), "scanned": total,
              "broken": len(bad), "verdict": verdict, "sample": bad[:20]}
    with open(os.path.join(HERE, "data", "latest.json"), "w") as fh:
        json.dump(result, fh, indent=1)
    print(f"{verdict}: scanned {total}, {len(bad)} envelope-corrupted")
    if bad:
        try:
            active = sum(1 for b in bad if b["status"] == "ACTIVE")
            card = {"title": f"⚠️ Material envelope regression: {len(bad)} products ({active} ACTIVE)",
                    "note": f"dw-material-envelope-canary found metafield values leaking raw JSON wrappers again. "
                            f"e.g. {', '.join(b['handle'] for b in bad[:5])}. Re-run /tmp/fix_material.py or check the importer.",
                    "source": "dw-material-envelope-canary"}
            urllib.request.urlopen(urllib.request.Request(
                CNCP, data=json.dumps(card).encode(),
                headers={"Content-Type": "application/json"}), timeout=5)
            print("  posted CNCP alert")
        except Exception as ex:
            print("  CNCP post failed:", ex)
    return 1 if bad else 0

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