← back to Pj Image Repair
write_shopify_images.py
211 lines
#!/usr/bin/env python3
"""
TK-10467 — GATED Shopify image write for Phillip Jeffries logo-hero repair.
Adds the resolved real product image as position-1 (featured) for each PJ product.
HARD-GATED: DRY-RUN by default. Requires --apply AND env CONFIRM_PJ_WRITE=1 to
touch Shopify (customer-facing live store designer-laboratory-sandbox).
Justification (2026-09-13): PJ is showroom-only (hidden from Google Merchant +
every discovery/browse surface). This write only improves the DIRECT PDP +
on-site-search experience — a real pattern photo at position 1 instead of the PJ
brand logo. It does NOT re-expose PJ to any discovery surface.
Reads: data/pj_writable.tsv (dw_sku, handle, mfr_sku, real_image, vendor_title, our_title)
Writes rollback journal: data/pj_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, re, urllib.request, urllib.error
HERE = os.path.dirname(os.path.abspath(__file__))
MAP = os.environ.get("PJ_MAP") or os.path.join(HERE, "data", "pj_writable.tsv")
JRN = os.path.join(HERE, "data", "pj_write_journal.jsonl")
SHOP = "designer-laboratory-sandbox.myshopify.com"
API = "2024-10"
APPLY = "--apply" in sys.argv
ALLOW_UNHIDDEN = "--allow-unhidden" in sys.argv
LIMIT = None
OFFSET = 0
for a in sys.argv:
if a.startswith("--limit="): LIMIT = int(a.split("=")[1])
if a.startswith("--offset="): OFFSET = int(a.split("=")[1])
# Showroom-only guard (TK-11089 / TK-11193). PJ is showroom-only; the storefront hide asset
# dw-pj-hide.js hides any PJ Boost product-item whose IMAGE SRC, HANDLE, or CARD TEXT matches
# the vendor. Swapping the logo image (PhillipJeffriesLogo_*.png) to a webdamdb URL removes the
# image-filename branch, so a product protected ONLY by that branch (the ~893 algolia-refresh
# cohort whose handle carries no vendor marker) would silently become VISIBLE in browse grids.
# HANDLE_HIDE_RX is the handle branch: a handle that matches it stays hidden after the swap.
# Rows whose handle does NOT match are SKIPPED (never un-hidden) unless --allow-unhidden is
# explicitly passed, so this fix can never create a showroom leak on its own. All 25 canary
# rows carry the vendor marker (verified 2026-09-13), so this is a no-op for the canary and a
# safety rail for the separate 2,279-product remainder go.
HANDLE_HIDE_RX = re.compile(r"phil+ip-jeffr", re.I)
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_PJ_WRITE") != "1":
print("REFUSING: --apply requires env CONFIRM_PJ_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"))
rows = rows[OFFSET:]
if LIMIT:
rows = rows[:LIMIT]
print(f"mode={'APPLY' if APPLY else 'DRY-RUN'} map={os.path.basename(MAP)} offset={OFFSET} 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"] # FIXED: real handle column, not dw_sku
real = r["real_image"]
# Showroom-hide guard: never un-hide a product protected only by its image filename.
if not HANDLE_HIDE_RX.search(handle or "") and not ALLOW_UNHIDDEN:
skipped += 1
print(f"[{i}] SKIP {dw_sku} handle={handle} -> no vendor marker in handle; "
f"image swap would un-hide it in browse grids (pass --allow-unhidden to override)")
continue
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")
ok += 1
continue
# --- APPLY path (CONFIRM_PJ_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 (Kimi-flagged 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()