`rollback_images.py`

64 lines
#!/usr/bin/env python3
"""
TK-10467 — rollback for the PJ logo-hero image canary.
Reads data/pj_write_journal.jsonl 'post' records (product_id + added_image_id) and DELETEs
each image we added. Because the canary left the logo in place, deleting the one added
webdamdb image returns the PhillipJeffriesLogo_*.png to position 1 — a clean, complete undo.

DRY-RUN by default; requires --apply AND env CONFIRM_PJ_WRITE=1 (same gate as the writer).
"""
import os, sys, json, time, urllib.request, urllib.error

HERE = os.path.dirname(os.path.abspath(__file__))
JRN  = os.path.join(HERE, "data", "pj_write_journal.jsonl")
SHOP = "designer-laboratory-sandbox.myshopify.com"
API  = "2024-10"
APPLY = "--apply" in sys.argv

def token():
    envp = os.path.expanduser("~/Projects/secrets-manager/.env")
    vals = {}
    with open(envp) as f:
        for line in f:
            for k in ("SHOPIFY_ADMIN_TOKEN", "SHOPIFY_FULL_ACCESS_TOKEN"):
                if line.startswith(k + "="):
                    vals[k] = line.split("=", 1)[1].strip().strip('"').strip("'")
    tok = vals.get("SHOPIFY_ADMIN_TOKEN") or vals.get("SHOPIFY_FULL_ACCESS_TOKEN")
    if not tok:
        raise SystemExit("no SHOPIFY_ADMIN_TOKEN / SHOPIFY_FULL_ACCESS_TOKEN")
    return tok

TOK = token()

def api(method, path):
    url = f"https://{SHOP}/admin/api/{API}/{path}"
    req = urllib.request.Request(url, method=method,
        headers={"X-Shopify-Access-Token": TOK, "Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=60) as r:
        return r.getcode()

def main():
    if APPLY and os.environ.get("CONFIRM_PJ_WRITE") != "1":
        print("REFUSING: --apply requires env CONFIRM_PJ_WRITE=1. Aborting.")
        sys.exit(2)
    posts = [json.loads(l) for l in open(JRN) if '"added_image_id"' in l]
    print(f"mode={'APPLY' if APPLY else 'DRY-RUN'} rollback_records={len(posts)}")
    ok = err = 0
    for i, d in enumerate(posts, 1):
        pid, iid = d["product_id"], d["added_image_id"]
        if not APPLY:
            print(f"[{i}] DRY would DELETE products/{pid}/images/{iid}.json ({d.get('dw_sku')})")
            continue
        try:
            api("DELETE", f"products/{pid}/images/{iid}.json")
            ok += 1
            print(f"[{i}] OK deleted image {iid} on {pid} ({d.get('dw_sku')})")
        except Exception as e:
            err += 1
            print(f"[{i}] ERR {pid}/{iid}: {e}")
        time.sleep(0.4)
    print(f"done ok={ok} err={err}" + ("" if APPLY else " (DRY-RUN)"))

if __name__ == "__main__":
    main()