← back to Mr Image Repair
write_shopify_images.py
194 lines
#!/usr/bin/env python3
"""
TK-11638 — GATED Shopify image write for Maya Romanoff logo-hero repair.
Adds the resolved real product image as position-1 (featured) for each MR product.
HARD-GATED: DRY-RUN by default. Requires --apply AND env CONFIRM_MR_WRITE=1 to
touch Shopify (customer-facing live store designer-laboratory-sandbox).
Sibling of pj-image-repair/write_shopify_images.py (TK-10467, proven — 50 PJ
products written 2026-09-13). This is a faithful mirror adapted for the 7 Maya
Romanoff "River Bed" colorways. Steve APPROVED the write 2026-09-14.
Differences from the PJ writer:
- MAP/JRN/env point at MR files (mr_writable.tsv / mr_write_journal.jsonl / CONFIRM_MR_WRITE).
- NO showroom-only handle guard. Maya Romanoff is a normal customer-facing brand
(only Phillip Jeffries is showroom-only), so every row is written — there is no
browse-grid to leak into.
Reads: data/mr_writable.tsv (dw_sku, handle, mfr_sku, real_image, vendor_title, our_title)
Writes rollback journal: data/mr_write_journal.jsonl (old image ids + featured id per product,
appended BEFORE each write so a revert can restore).
--remove-logo is intentionally NOT implemented / NOT wired (no logo deletion authorized).
"""
import csv, os, sys, json, time, urllib.request, urllib.error
HERE = os.path.dirname(os.path.abspath(__file__))
MAP = os.path.join(HERE, "data", "mr_writable.tsv")
JRN = os.path.join(HERE, "data", "mr_write_journal.jsonl")
SHOP = "designer-laboratory-sandbox.myshopify.com"
API = "2024-10"
APPLY = "--apply" in sys.argv
LIMIT = None
for a in sys.argv:
if a.startswith("--limit="): LIMIT = int(a.split("=")[1])
def token():
# Narrow SHOPIFY_ADMIN_TOKEN (...7d19) is verified to carry write_products, which
# is all an image add needs. Fall back to the full-access token only if absent.
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, body=None, _tries=4):
url = f"https://{SHOP}/admin/api/{API}/{path}"
data = json.dumps(body).encode() if body is not None else None
for attempt in range(1, _tries + 1):
req = urllib.request.Request(url, data=data, method=method,
headers={"X-Shopify-Access-Token": TOK, "Content-Type": "application/json"})
try:
with urllib.request.urlopen(req, timeout=60) as r:
return json.loads(r.read())
except urllib.error.HTTPError as e:
# 429 (rate limit) / 5xx (transient) -> backoff + retry; else re-raise
if e.code in (429, 500, 502, 503, 504) and attempt < _tries:
wait = float(e.headers.get("Retry-After", attempt)) or attempt
time.sleep(max(wait, attempt))
continue
raise
def product_for_handle(handle):
"""Look up the product by its REAL handle (not dw_sku)."""
d = api("GET", f"products.json?handle={handle}&fields=id,image,images,status,title")
ps = d.get("products", [])
return ps[0] if ps else None
def img_token(url):
"""Filename token used to detect an already-added real image (idempotency)."""
base = url.split("?", 1)[0].rsplit("/", 1)[-1]
return base
def verify_and_fix_featured(pid, new_id):
"""After the POST, CONFIRM the new image is actually featured (product.image).
POSTing position:1 does NOT reliably shift existing images in Shopify REST 2024-10 —
a duplicate position:1 can tie toward the older (lower-id) logo image, silently
leaving the logo as hero. If so, PUT the new image to position 1 and re-check.
Returns (is_featured_bool, featured_id)."""
p = api("GET", f"products/{pid}.json?fields=id,image").get("product", {})
feat = (p.get("image") or {}).get("id")
if feat == new_id:
return True, feat
# Not featured yet -> force it to position 1, then re-read the source of truth.
api("PUT", f"products/{pid}/images/{new_id}.json",
{"image": {"id": new_id, "position": 1}})
time.sleep(0.4)
p = api("GET", f"products/{pid}.json?fields=id,image").get("product", {})
feat = (p.get("image") or {}).get("id")
return feat == new_id, feat
def main():
if APPLY and os.environ.get("CONFIRM_MR_WRITE") != "1":
print("REFUSING: --apply requires env CONFIRM_MR_WRITE=1 (Steve-gated). Aborting.")
sys.exit(2)
if "--remove-logo" in sys.argv:
print("REFUSING: --remove-logo is not authorized for this run. Aborting.")
sys.exit(2)
rows = list(csv.DictReader(open(MAP), delimiter="\t"))
if LIMIT:
rows = rows[:LIMIT]
print(f"mode={'APPLY' if APPLY else 'DRY-RUN'} rows={len(rows)} store={SHOP} api={API}")
jf = open(JRN, "a") if APPLY else None
ok = err = skipped = 0
for i, r in enumerate(rows, 1):
dw_sku = r["dw_sku"]
handle = r["handle"] # real handle column, not dw_sku
real = r["real_image"]
try:
prod = product_for_handle(handle)
if not prod:
err += 1
print(f"[{i}] ERR {dw_sku} handle={handle} -> product not found")
continue
pid = prod["id"]
existing = prod.get("images", []) or []
existing_tokens = {img_token(im.get("src", "")) for im in existing}
# Idempotency: if the real image is already present, skip (safe re-run)
if img_token(real) in existing_tokens:
pos1 = next((im for im in existing if im.get("position") == 1), None)
already_featured = pos1 and img_token(pos1.get("src", "")) == img_token(real)
skipped += 1
print(f"[{i}] SKIP {dw_sku} pid={pid} real image already present"
+ (" @pos1" if already_featured else " (not pos1)"))
continue
if not APPLY:
logo = existing[0]["src"].rsplit("/", 1)[-1] if existing else "(none)"
print(f"[{i}] DRY {dw_sku} pid={pid} pos1_now={logo} -> add {img_token(real)} @pos1 ({r['our_title']})")
ok += 1
continue
# --- APPLY path (CONFIRM_MR_WRITE=1 verified) ---
# 1) Journal old state BEFORE the write, for rollback.
jf.write(json.dumps({
"ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
"dw_sku": dw_sku, "handle": handle, "product_id": pid,
"old_featured_image_id": (prod.get("image") or {}).get("id"),
"old_image_ids": [im["id"] for im in existing],
"old_images": [{"id": im["id"], "position": im.get("position"),
"src": im.get("src")} for im in existing],
"new_image_src": real,
}) + "\n")
jf.flush()
# 2) Add the real image at position 1 (becomes the featured image).
resp = api("POST", f"products/{pid}/images.json",
{"image": {"src": real, "position": 1}})
new_img = resp.get("image", {})
new_id = new_img.get("id")
new_pos = new_img.get("position")
# 2b) Journal the NEW image id AFTER the POST so rollback is a clean
# delete-by-id (delete-by-src is fragile: Shopify rewrites src to CDN).
jf.write(json.dumps({
"ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
"dw_sku": dw_sku, "product_id": pid,
"added_image_id": new_id, "added_position": new_pos,
"rollback": f"DELETE products/{pid}/images/{new_id}.json",
}) + "\n")
jf.flush()
# 2c) CONFIRM it is actually the featured/hero image (pos1 tie quirk).
featured, feat_id = verify_and_fix_featured(pid, new_id)
if not featured:
err += 1
print(f"[{i}] WARN {dw_sku} pid={pid} added id={new_id} but featured is {feat_id} "
f"(NOT the new image) — logo may still be hero; flagged for review.")
else:
ok += 1
print(f"[{i}] OK {dw_sku} pid={pid} featured=id {new_id} (real image is hero)")
except urllib.error.HTTPError as e:
err += 1
body = e.read().decode(errors="replace")[:300]
print(f"[{i}] HTTP {e.code} {dw_sku} handle={handle}: {body}")
except Exception as e:
err += 1
print(f"[{i}] ERR {dw_sku} handle={handle}: {e}")
# Early-stop: if the write path is broken, do NOT burn through the whole batch.
if APPLY and (err >= 2 or (err >= 1 and ok == 0)):
print(f"[{i}] STOP: {err} error(s) with {ok} success — halting to diagnose "
f"(not burning through the remaining {len(rows)-i} products).")
break
time.sleep(0.6) # gentle pacing on the live store
if jf:
jf.close()
tail = "(DRY-RUN — no Shopify writes)" if not APPLY else "(LIVE writes committed)"
print(f"done ok={ok} skipped={skipped} err={err} {tail}")
if __name__ == "__main__":
main()