← back to Dw Kravet Hires
scripts/build-batchA-map.py
241 lines
#!/usr/bin/env python3
"""TK-12097 Batch A — build the Kravet-family hi-res featured-image swap map.
Input : data/tk12097/batchA-candidates.tsv (colorway-EXACT, vendor-scoped join of
ACTIVE Donghia/Brunschwig & Fils/Lee Jofa/GP & J Baker products to their
vendor _catalog Brandfolder hi-res URL; cols:
shopify_id \\t vendor \\t mfr_sku \\t dw_sku \\t cat_mfr \\t current_url \\t hires_url)
Method (all $0, network reads only; MEASURED, never claimed):
Phase 1 measure the LIVE current featured image; keep only max-dim <= MAX_LOWRES.
Phase 2 for survivors measure the Brandfolder candidate; a row is SWAPPABLE iff
the candidate is a genuine upgrade (max-dim > current AND >= HIRES_FLOOR),
reachable (HTTP 200/206), parseable, and within Shopify ceilings
(<=5000px long-edge, <=20MB) — oversized candidates are capped with the
TK-11740-verified Brandfolder resize param and RE-MEASURED.
Placeholder / dead / no-upgrade / still-over-after-cap candidates are EXCLUDED
(never swap a real low-res image for a placeholder or a not-bigger image).
Output: data/tk12097/batchA-swap-map.json (apply-hires.mjs row schema)
data/tk12097/batchA-measured.json (full per-row measurement evidence)
"""
import json, io, sys, re, os, datetime, concurrent.futures as cf
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)
SRC = os.path.join(PROJ, "data/tk12097/batchA-candidates.tsv")
OUT_MAP = os.path.join(PROJ, "data/tk12097/batchA-swap-map.json")
OUT_MEAS = os.path.join(PROJ, "data/tk12097/batchA-measured.json")
MAX_LOWRES = int(os.environ.get("MAX_LOWRES", "400")) # Kravet-family low-res class (300/400px CDN thumbs)
HIRES_FLOOR = int(os.environ.get("HIRES_FLOOR", "800")) # candidate must be at least this big 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 batchA)"
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 line in f:
line = line.rstrip("\n")
if not line:
continue
p = line.split("\t")
if len(p) < 7:
continue
rows.append({"shopify_id": p[0], "vendor": p[1], "mfr_sku": p[2],
"dw_sku": p[3] or None, "cat_mfr": p[4],
"current_url": p[5], "hires_url": p[6]})
return rows
def measure_current(r):
m = measure_url(r["current_url"])
r["cur_measure"] = m
r["cur_le"] = m.get("meas_le")
return r
def measure_hires(r):
raw = r["hires_url"] or ""
if not raw or "placeholder" in raw.lower():
r["hires_status"] = "BAD_SOURCE_PLACEHOLDER"; r["ok"] = False; return r
if "?" in raw:
r["hires_status"] = "SOURCE_HAS_QUERYSTRING"; r["ok"] = False; return r
m = measure_url(raw)
r["hires_measure"] = m
if not m.get("ok"):
r["hires_status"] = m["status"]; r["ok"] = False; return r
le = m["meas_le"]
over = (le > SHOP_PX) or (m.get("meas_bytes") and m["meas_bytes"] > SHOP_BYTES)
if over:
rp = resize_for(raw)
cm = measure_url(raw + rp)
r["cap_param"] = rp; r["capped_measure"] = cm
if not cm.get("ok") or cm["meas_le"] > SHOP_PX or (cm.get("meas_bytes") and cm["meas_bytes"] > SHOP_BYTES):
r["hires_status"] = "STILL_OVER_AFTER_CAP:" + cm.get("status", "?"); r["ok"] = False; return r
r["final_url"] = raw + rp; r["final_le"] = cm["meas_le"]; r["capped"] = True
else:
r["final_url"] = raw; r["final_le"] = le; r["capped"] = False
# genuine-upgrade gate: must be strictly bigger than current AND clear the hi-res floor
if r["final_le"] <= (r.get("cur_le") or 0) or r["final_le"] < HIRES_FLOOR:
r["hires_status"] = f"NO_UPGRADE(cur={r.get('cur_le')},hires={r['final_le']})"; r["ok"] = False; return r
r["hires_status"] = "PASS_CAPPED" if r.get("capped") else "PASS"; r["ok"] = True
return r
def main():
cands = load_candidates()
N = len(cands)
print(f"loaded {N} colorway-exact candidates", file=sys.stderr)
# Phase 1: measure current featured; keep <= MAX_LOWRES
with cf.ThreadPoolExecutor(max_workers=24) as ex:
for i, _ in enumerate(ex.map(measure_current, cands), 1):
if i % 200 == 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: measure hires candidate for low-res survivors only
M = len(lowres)
with cf.ThreadPoolExecutor(max_workers=24) as ex:
for i, _ in enumerate(ex.map(measure_hires, lowres), 1):
if i % 100 == 0:
print(f" phase2 measured {i}/{M}", file=sys.stderr)
swappable = [r for r in lowres if r.get("ok")]
excluded = [r for r in lowres if not r.get("ok")]
# Build apply-hires.mjs map rows
map_rows = []
for r in swappable:
map_rows.append({
"shopify_id": r["shopify_id"],
"vendor": r["vendor"],
"dw_sku": r["dw_sku"],
"mfr_sku": r["mfr_sku"],
"cat_mfr": r["cat_mfr"],
"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": "vendor_catalog_brandfolder",
"preflight_capped": bool(r.get("capped")),
"swappable_from_local_staging": True,
})
from collections import Counter
excl_break = dict(Counter(r.get("hires_status") for r in excluded))
by_vendor = Counter(r["vendor"] for r in swappable)
map_doc = {
"ticket": "TK-12097", "batch": "A",
"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 400)",
"scope": ("Batch A = Kravet-family (Donghia, Brunschwig & Fils, Lee Jofa, GP & J Baker). "
"Colorway-EXACT vendor-scoped join (N2 norm) to vendor _catalog Brandfolder hi-res. "
f"current featured MEASURED <= {MAX_LOWRES}px; candidate MEASURED as a genuine upgrade "
f"(> current AND >= {HIRES_FLOOR}px) within Shopify 5000px/20MB ceilings (oversized capped+re-measured). "
"Old low-res media retained per product (rollback anchor)."),
"cap_params": {"jpeg": RESIZE_JPEG, "png": RESIZE_PNG},
"counts": {
"candidates_colorway_exact": N,
"current_unreachable": len(cur_unreachable),
"already_hires": len(already_hi),
"lowres_survivors": len(lowres),
"swappable": len(swappable),
"excluded": len(excluded),
"swappable_by_vendor": dict(by_vendor),
"excluded_breakdown": excl_break,
"capped_hires": sum(1 for r in swappable if r.get("capped")),
},
"rows": map_rows,
}
json.dump(map_doc, open(OUT_MAP, "w"), indent=1)
meas = {"ticket": "TK-12097", "batch": "A",
"measured_at_utc": datetime.datetime.utcnow().isoformat() + "Z",
"max_lowres": MAX_LOWRES, "hires_floor": HIRES_FLOOR,
"rows": cands}
json.dump(meas, open(OUT_MEAS, "w"), indent=1)
print(json.dumps(map_doc["counts"], indent=2))
if swappable:
les = [r["final_le"] for r in swappable]
print(f"hires long-edge: min={min(les)} max={max(les)} px", file=sys.stderr)
if __name__ == "__main__":
main()