← back to Dw Kravet Hires
scripts/build-batchB-map.py
313 lines
#!/usr/bin/env python3
"""TK-12097 Batch B — build the Jeffrey Stevens hi-res featured-image swap map.
Mirrors scripts/build-batchA-map.py EXACTLY (measure -> upgrade-gate -> colorway-safety),
adapted for Jeffrey Stevens:
Input : data/tk12097/batchB-candidates.csv (all 2,658 ACTIVE Jeffrey Stevens products,
LEFT-joined to jeffrey_stevens_catalog by dw_sku==staging.sku — the ONLY join that
resolves; exact mfr_sku and N2-normalized mfr_sku both resolve 0 rows for JS.
cols: shopify_id,vendor,mfr_sku,dw_sku,image_url(live),stage_sku,stage_url,all_images,join_method)
Method (all $0, network reads only; MEASURED, never claimed):
Phase 1 measure the LIVE current featured image for ALL active products; low-res class
= max-dim <= MAX_LOWRES (500 for JS — vp-engineering verified JS lowres are 410x410,
so a 400 cutoff would skip 100%).
Phase 2 for each low-res survivor WITH a staging join, gather candidate URLs
(staging.image_url + every all_images entry), MEASURE each, and pick the LARGEST.
The memo claimed 600-1800px yorkwall-CDN hi-res; MEASURE the candidate, never trust
the memo — the JS staging table is a mirror of the live Shopify store, so its images
are the SAME Shopify-CDN masters as the live featured.
Phase 3 genuine-upgrade gate: keep only rows where measured hires_le > current AND
hires_le >= HIRES_FLOOR (800). Cap oversized (>5000px or >20MB) with the
Brandfolder/JPEG resize params and RE-MEASURE.
Colorway safety: the dw_sku join is colorway-EXACT (safe). Belt-and-suspenders — the chosen
CANDIDATE image must itself be colorway-verifiable: its filename must contain the
product's mfr SKU code (the swatch file is named <MFR>.jpg, e.g. OG0516.jpg). A
hash-named gallery/room candidate that does NOT contain the SKU is held to tier2.
Output: data/tk12097/batchB-final-map.json (apply-hires.mjs row schema + counts)
data/tk12097/batchB-tier2-colorway-unverified.json (genuine-upgrade but unverifiable candidate)
data/tk12097/batchB-no-hires-source.json (low-res with no genuine upgrade)
data/tk12097/batchB-measured.json (full per-row measurement evidence)
"""
import json, io, sys, re, os, csv, datetime, concurrent.futures as cf
from collections import Counter
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError
from PIL import Image
HERE = os.path.dirname(os.path.abspath(__file__))
PROJ = os.path.dirname(HERE)
D = os.path.join(PROJ, "data/tk12097")
SRC = os.path.join(D, "batchB-candidates.csv")
OUT_MAP = os.path.join(D, "batchB-final-map.json")
OUT_TIER2 = os.path.join(D, "batchB-tier2-colorway-unverified.json")
OUT_NOSRC = os.path.join(D, "batchB-no-hires-source.json")
OUT_MEAS = os.path.join(D, "batchB-measured.json")
MAX_LOWRES = int(os.environ.get("MAX_LOWRES", "500")) # JS low-res class (410x410 verified) — NOT 400
HIRES_FLOOR = int(os.environ.get("HIRES_FLOOR", "800")) # candidate must clear this to count as an upgrade
SHOP_PX = 5000
SHOP_BYTES = 20 * 1024 * 1024
RESIZE_JPEG = "?width=4472&height=4472&fit=bound&quality=90"
RESIZE_PNG = "?width=3000&height=3000&fit=bound"
UA = "Mozilla/5.0 (dw-preflight TK-12097 batchB)"
codeOf = lambda mfr: re.sub(r"[^A-Za-z0-9]", "", re.sub(r"^[PT]", "", re.sub(r"\.0$", "", str(mfr or "")), flags=re.I)).upper()
fnameAlnum = lambda url: re.sub(r"[^A-Za-z0-9]", "", url.split("?")[0].split("/")[-1]).upper()
def resize_for(src):
return RESIZE_PNG if re.search(r"\.png($|\?)", src, re.I) else RESIZE_JPEG
def fetch_head_bytes(url, nbytes):
req = Request(url, headers={"User-Agent": UA, "Range": f"bytes=0-{nbytes-1}"})
with urlopen(req, timeout=45) as r:
code = r.status
cr = r.headers.get("Content-Range", "")
total = None
if "/" in cr:
tail = cr.rsplit("/", 1)[1].strip()
if tail.isdigit():
total = int(tail)
if total is None:
gcl = r.headers.get("x-goog-stored-content-length")
if gcl and gcl.isdigit():
total = int(gcl)
if total is None:
cl = r.headers.get("Content-Length")
if code == 200 and cl and cl.isdigit():
total = int(cl)
data = r.read()
return code, total, data
def measure_url(url):
res = {"url": url}
try:
code, total, data = fetch_head_bytes(url, 65536)
dims = None
for rng in (65536, 262144, 1048576):
try:
if rng > len(data):
code2, total2, data = fetch_head_bytes(url, rng)
if total2:
total = total2
im = Image.open(io.BytesIO(data)); im.draft(None, None)
dims = im.size
break
except Exception:
continue
res.update(http=code, meas_bytes=total,
meas_w=(dims[0] if dims else None), meas_h=(dims[1] if dims else None))
if code not in (200, 206):
res.update(status=f"HTTP_{code}", ok=False); return res
if dims is None:
res.update(status="NO_DIMS", ok=False); return res
res["meas_le"] = max(dims)
res.update(status="OK", ok=True)
except HTTPError as e:
res.update(http=e.code, status=f"HTTP_{e.code}", ok=False)
except (URLError, Exception) as e:
res.update(status="ERR", ok=False, err=str(e)[:140])
return res
def load_candidates():
rows = []
with open(SRC) as f:
for r in csv.DictReader(f):
try:
ai = json.loads(r["all_images"]) if r.get("all_images") else []
except Exception:
ai = []
rows.append({
"shopify_id": r["shopify_id"], "vendor": r["vendor"],
"mfr_sku": r["mfr_sku"] or None, "dw_sku": r["dw_sku"] or None,
"current_url": r["image_url"], "stage_url": r.get("stage_url") or None,
"all_images": [u for u in ai if isinstance(u, str) and u.startswith("http")],
"join_method": r["join_method"],
})
return rows
def measure_current(r):
m = measure_url(r["current_url"]) if r.get("current_url") else {"status": "NO_CURRENT_URL", "ok": False}
r["cur_measure"] = m
r["cur_le"] = m.get("meas_le")
return r
def measure_hires(r):
# gather distinct candidate URLs from staging.image_url + all_images
cands = []
seen = set()
for u in ([r["stage_url"]] if r.get("stage_url") else []) + r.get("all_images", []):
if not u or "placeholder" in u.lower():
continue
base = u.split("?")[0]
if base in seen:
continue
seen.add(base)
cands.append(u)
if not cands:
r["hires_status"] = "NO_CANDIDATE"; r["ok"] = False; return r
measured = []
for u in cands:
m = measure_url(u)
if m.get("ok"):
# cap oversized then re-measure
le = m["meas_le"]
over = (le > SHOP_PX) or (m.get("meas_bytes") and m["meas_bytes"] > SHOP_BYTES)
if over:
rp = resize_for(u)
cm = measure_url(u + rp)
if cm.get("ok") and cm["meas_le"] <= SHOP_PX and not (cm.get("meas_bytes") and cm["meas_bytes"] > SHOP_BYTES):
measured.append({"url": u + rp, "le": cm["meas_le"], "capped": True, "src_base": u.split("?")[0]})
# still-over → drop this candidate
else:
measured.append({"url": u, "le": le, "capped": False, "src_base": u.split("?")[0]})
r["cand_measured"] = [{"url": m["url"], "le": m["le"]} for m in measured]
if not measured:
r["hires_status"] = "NO_REACHABLE_CANDIDATE"; r["ok"] = False; return r
# genuine-upgrade gate: strictly bigger than current AND clears the floor
cur = r.get("cur_le") or 0
genuine = [m for m in measured if m["le"] > cur and m["le"] >= HIRES_FLOOR]
if not genuine:
best_le = max(m["le"] for m in measured)
r["hires_status"] = f"NO_UPGRADE(cur={cur},best_cand={best_le})"; r["ok"] = False; return r
# colorway safety: prefer a candidate whose filename contains the mfr SKU code
code = codeOf(r.get("mfr_sku"))
safe = [m for m in genuine if code and code in fnameAlnum(m["url"])]
if safe:
best = max(safe, key=lambda m: m["le"])
r["final_url"] = best["url"]; r["final_le"] = best["le"]; r["capped"] = best["capped"]
r["hires_status"] = "PASS_CAPPED" if best["capped"] else "PASS"; r["ok"] = True
r["colorway"] = "safe"
return r
# genuine upgrade but candidate filename does not verify the colorway -> tier2
best2 = max(genuine, key=lambda m: m["le"])
r["final_url"] = best2["url"]; r["final_le"] = best2["le"]; r["capped"] = best2["capped"]
r["hires_status"] = "TIER2_COLORWAY_UNVERIFIED"; r["ok"] = False; r["colorway"] = "unverified"
return r
def main():
cands = load_candidates()
N = len(cands)
joined = [c for c in cands if c["join_method"] == "dwsku_eq_stagingsku"]
print(f"loaded {N} ACTIVE Jeffrey Stevens products ({len(joined)} with staging join)", file=sys.stderr)
# Phase 1: measure current featured for ALL active products
with cf.ThreadPoolExecutor(max_workers=24) as ex:
for i, _ in enumerate(ex.map(measure_current, cands), 1):
if i % 300 == 0:
print(f" phase1 measured {i}/{N}", file=sys.stderr)
cur_unreachable = [r for r in cands if r["cur_le"] is None]
lowres = [r for r in cands if r["cur_le"] is not None and r["cur_le"] <= MAX_LOWRES]
already_hi = [r for r in cands if r["cur_le"] is not None and r["cur_le"] > MAX_LOWRES]
print(f" phase1: lowres(<= {MAX_LOWRES}px)={len(lowres)} already-hi={len(already_hi)} cur-unreachable={len(cur_unreachable)}", file=sys.stderr)
# Phase 2/3: only low-res survivors WITH a staging join can have a candidate
lowres_with_src = [r for r in lowres if r["join_method"] == "dwsku_eq_stagingsku"]
lowres_no_src = [r for r in lowres if r["join_method"] != "dwsku_eq_stagingsku"]
M = len(lowres_with_src)
with cf.ThreadPoolExecutor(max_workers=24) as ex:
for i, _ in enumerate(ex.map(measure_hires, lowres_with_src), 1):
if i % 100 == 0:
print(f" phase2 measured {i}/{M}", file=sys.stderr)
swappable = [r for r in lowres_with_src if r.get("ok")]
tier2 = [r for r in lowres_with_src if r.get("colorway") == "unverified"]
no_hires = lowres_no_src + [r for r in lowres_with_src if not r.get("ok") and r.get("colorway") != "unverified"]
# Build apply-hires.mjs map rows (colorway-safe genuine upgrades)
map_rows = []
for r in swappable:
map_rows.append({
"shopify_id": r["shopify_id"],
"vendor": r["vendor"],
"mfr_sku": r["mfr_sku"],
"dw_sku": r["dw_sku"],
"cur_width": r["cur_le"],
"current_400px_url": r["current_url"],
"rollback_url": r["current_url"],
"proposed_hires_url": r["final_url"],
"hires_le": r["final_le"],
"hires_source": "jeffrey_stevens_catalog",
"preflight_capped": bool(r.get("capped")),
"swappable_from_local_staging": True,
})
# histogram + per-floor coverage over the colorway-safe genuine-upgrade set
bands = [[501, 600], [600, 800], [800, 1200], [1200, 2500], [2500, 99999]]
hist = {}
for lo, hi in bands:
hist[f"{lo}-{'+' if hi == 99999 else hi}"] = sum(1 for r in map_rows if lo <= r["hires_le"] < hi)
floors = [500, 600, 800, 1000, 1200]
per_floor = {f">={f}": sum(1 for r in map_rows if r["hires_le"] >= f) for f in floors}
by_vendor = dict(Counter(r["vendor"] for r in map_rows))
excl_break = dict(Counter(r.get("hires_status") for r in lowres_with_src if not r.get("ok")))
doc = {
"ticket": "TK-12097", "batch": "B",
"parent": "TK-12090 / TK-11494",
"generated_at": datetime.datetime.utcnow().isoformat() + "Z",
"generated_by": "vp-dw-commerce",
"store": "designer-laboratory-sandbox",
"engine": "reuse ~/Projects/dw-kravet-hires/scripts/apply-hires.mjs (--max-width 500)",
"floor_applied": HIRES_FLOOR,
"scope": (
"Batch B = Jeffrey Stevens (2,658 ACTIVE). Colorway-EXACT join dw_sku==staging.sku "
"(exact mfr_sku and N2-normalized mfr_sku both resolve 0 for JS). Live featured "
f"MEASURED; low-res class max-dim <= {MAX_LOWRES}px. Candidate resolved from "
"jeffrey_stevens_catalog.image_url + all_images, MEASURED (never trusting the memo), "
f"kept only as a genuine upgrade (> current AND >= {HIRES_FLOOR}px) within Shopify "
"5000px/20MB ceilings. Candidate filename must contain the mfr SKU code (colorway "
"belt-and-suspenders) else held to tier2. Old low-res media retained per product."
),
"cap_params": {"jpeg": RESIZE_JPEG, "png": RESIZE_PNG},
"counts": {
"active_jeffrey_stevens": N,
"with_staging_join": len(joined),
"current_unreachable": len(cur_unreachable),
"already_hires_gt_maxlowres": len(already_hi),
"lowres_total": len(lowres),
"lowres_with_staging_source": len(lowres_with_src),
"lowres_no_staging_source": len(lowres_no_src),
"colorway_safe_upgrades": len(map_rows),
"tier2_held": len(tier2),
"no_hires_source": len(no_hires),
"emitted_at_floor": len(map_rows),
"by_vendor": by_vendor,
"width_histogram": hist,
"coverage_per_floor": per_floor,
"excluded_breakdown": excl_break,
},
"rows": map_rows,
}
json.dump(doc, open(OUT_MAP, "w"), indent=1)
json.dump({"rows": [{"shopify_id": r["shopify_id"], "vendor": r["vendor"], "mfr_sku": r["mfr_sku"],
"dw_sku": r["dw_sku"], "cur_width": r["cur_le"], "rollback_url": r["current_url"],
"proposed_hires_url": r.get("final_url"), "hires_le": r.get("final_le"),
"hires_source": "jeffrey_stevens_catalog",
"reason": "colorway_unverified_filename_no_sku"} for r in tier2]},
open(OUT_TIER2, "w"), indent=1)
json.dump([{"shopify_id": r["shopify_id"], "dw_sku": r["dw_sku"], "mfr_sku": r["mfr_sku"],
"cur_le": r["cur_le"], "join_method": r["join_method"],
"hires_status": r.get("hires_status", "NO_STAGING_SOURCE")} for r in no_hires],
open(OUT_NOSRC, "w"), indent=1)
json.dump({"ticket": "TK-12097", "batch": "B", "measured_at_utc": datetime.datetime.utcnow().isoformat() + "Z",
"max_lowres": MAX_LOWRES, "hires_floor": HIRES_FLOOR, "rows": cands}, open(OUT_MEAS, "w"), indent=1)
print(json.dumps(doc["counts"], indent=2))
if map_rows:
les = [r["hires_le"] for r in map_rows]
print(f"hires long-edge: min={min(les)} max={max(les)} px", file=sys.stderr)
if __name__ == "__main__":
main()