`apply_live.py`

96 lines
#!/usr/bin/env python3
"""
LIVE collection-banner writer for the DW Shopify store (Admin GraphQL collectionUpdate).
Usage:
  python3 apply_live.py canary        # apply the 8 highest-traffic READY collections
  python3 apply_live.py rest          # apply all remaining READY not yet applied
  python3 apply_live.py <handle> ...  # apply specific handles
Records every write to rollback.jsonl (handle, old_src, new_src, ok). Reversible.
Reads token from env SHOPIFY_ADMIN_TOKEN (never printed).
"""
import json, os, sys, time, urllib.request

HERE = os.path.dirname(os.path.abspath(__file__))
STORE = "designer-laboratory-sandbox.myshopify.com"
API = "2024-10"
TOKEN = os.environ.get("SHOPIFY_ADMIN_TOKEN", "")
assert TOKEN, "SHOPIFY_ADMIN_TOKEN not in env"

cols = {c["handle"]: c for c in json.load(open(os.path.join(HERE, "dw_collections.json")))}
proposal = {r["handle"]: r for r in json.load(open(os.path.join(HERE, "proposal.json")))}
ready = [h for h, r in proposal.items() if r["status"] == "READY"]

CANARY = ["primary-suite-1", "grey-silver", "paper", "modern", "warm-taupe",
          "metallic-wallpaper-collections", "botanicals-and-florals", "florals-botanicals"]

MUT = """mutation($input: CollectionInput!) {
  collectionUpdate(input: $input) {
    collection { id handle image { url width height } }
    userErrors { field message }
  }
}"""

def gql(query, variables):
    body = json.dumps({"query": query, "variables": variables}).encode()
    req = urllib.request.Request(
        f"https://{STORE}/admin/api/{API}/graphql.json", data=body,
        headers={"X-Shopify-Access-Token": TOKEN, "Content-Type": "application/json"})
    return json.load(urllib.request.urlopen(req, timeout=40))

def already_applied():
    done = set()
    p = os.path.join(HERE, "rollback.jsonl")
    if os.path.exists(p):
        for line in open(p):
            try:
                d = json.loads(line)
                if d.get("ok"): done.add(d["handle"])
            except: pass
    return done

def apply(handles):
    log = open(os.path.join(HERE, "rollback.jsonl"), "a")
    ok = fail = 0
    for i, h in enumerate(handles, 1):
        r = proposal.get(h); c = cols.get(h)
        if not r or not c or r["status"] != "READY":
            print(f"  SKIP {h} (not a READY proposal)"); continue
        gid = f"gid://shopify/Collection/{c['id']}"
        new_src = r["new_src"]; old_src = r.get("current_src")
        try:
            res = gql(MUT, {"input": {"id": gid, "image": {"src": new_src}}})
            errs = res.get("data", {}).get("collectionUpdate", {}).get("userErrors", [])
            top = res.get("errors")
            if top:
                print(f"  FAIL {h}: {top}"); fail += 1
                log.write(json.dumps({"handle": h, "ok": False, "err": str(top)}) + "\n")
            elif errs:
                print(f"  FAIL {h}: {errs}"); fail += 1
                log.write(json.dumps({"handle": h, "ok": False, "err": str(errs)}) + "\n")
            else:
                img = res["data"]["collectionUpdate"]["collection"]["image"]
                dim = f"{img['width']}x{img['height']}" if img else "?"
                print(f"  OK   {h}  -> {dim}")
                ok += 1
                log.write(json.dumps({"handle": h, "ok": True, "old_src": old_src,
                                      "new_src": new_src, "new_url": img["url"] if img else None,
                                      "new_dim": dim}) + "\n")
        except Exception as e:
            print(f"  ERR  {h}: {str(e)[:120]}"); fail += 1
            log.write(json.dumps({"handle": h, "ok": False, "err": str(e)[:200]}) + "\n")
        time.sleep(0.6)  # gentle on the 2 req/s REST-equiv; GraphQL cost-based but stay polite
    log.close()
    print(f"\napplied ok={ok} fail={fail}")

if __name__ == "__main__":
    arg = sys.argv[1] if len(sys.argv) > 1 else "canary"
    if arg == "canary":
        target = CANARY
    elif arg == "rest":
        done = already_applied()
        target = [h for h in ready if h not in done]
        print(f"remaining READY to apply: {len(target)}")
    else:
        target = sys.argv[1:]
    apply(target)