`check-bulk.py`

77 lines
#!/usr/bin/env python3
"""WEEKLY exhaustive pass — the durable net the daily check.py can't be.
Runs a Shopify bulk operation exporting EVERY metafield on EVERY product, then
scans every value with the robust substring detector (catches scalar, list, and
double-escaped envelope variants). No curated key list. CNCP alert on FAIL.
READ-ONLY. Born 2026-07-07 from the /yolo contrarian's 'watch all keys' verdict."""
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"
def gql(q):
    b = json.dumps({"query": q}).encode()
    return json.loads(urllib.request.urlopen(urllib.request.Request(
        URL, data=b, headers={"X-Shopify-Access-Token": TOK, "Content-Type": "application/json"}), timeout=60).read())
def is_env(v):
    return isinstance(v, str) and "single_line_text_field" in v
START = 'mutation { bulkOperationRunQuery(query: """{ products { edges { node { id handle metafields { edges { node { namespace key value } } } } } } }""") { bulkOperation { id status } userErrors { message } } }'
POLL = '{ currentBulkOperation { status url objectCount } }'
def main():
    os.makedirs(os.path.join(HERE, "data"), exist_ok=True)
    r = gql(START)
    ue = r["data"]["bulkOperationRunQuery"]["userErrors"]
    if ue:
        print("start err", ue); return 2
    while True:
        op = gql(POLL)["data"]["currentBulkOperation"]
        if op["status"] in ("COMPLETED", "FAILED", "CANCELED"): break
        time.sleep(20)
    if op["status"] != "COMPLETED" or not op.get("url"):
        print("bulk not completed:", op); return 2
    path = os.path.join(HERE, "data", "bulk.jsonl")
    urllib.request.urlretrieve(op["url"], path)
    hits = {}; nprod = 0; nmf = 0
    with open(path) as fh:
        for ln in fh:
            try:
                o = json.loads(ln)
            except (json.JSONDecodeError, ValueError):
                continue
            if "namespace" in o and "key" in o:
                nmf += 1
                if is_env(o.get("value")):
                    field_key = f'{o["namespace"]}.{o["key"]}'
                    hits[field_key] = hits.get(field_key, 0) + 1
            elif "handle" in o:
                nprod += 1
    os.remove(path)  # 800MB — don't keep
    verdict = "FAIL" if hits else "PASS"
    with open(os.path.join(HERE, "data", "latest-bulk.json"), "w") as fh:
        json.dump({"ts": datetime.datetime.now().isoformat(), "products": nprod, "metafields": nmf,
                   "verdict": verdict, "hits": hits}, fh, indent=1)
    print(f"{verdict}: {nprod} products, {nmf} metafields, {sum(hits.values())} enveloped across {len(hits)} keys")
    if hits:
        try:
            card = {"title": f"⚠️ Metafield envelope regression (weekly bulk): {sum(hits.values())} values across {len(hits)} keys",
                    "note": f"dw-material-envelope-canary weekly bulk scan found leaked JSON envelopes: {json.dumps(hits)[:300]}. Re-run the unwrap fix + find the importer.",
                    "source": "dw-material-envelope-canary-bulk"}
            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 hits else 0
if __name__ == "__main__":
    sys.exit(main())