← back to Dw Kravet Hires
scripts/build-batchC-map.py
351 lines
#!/usr/bin/env python3
"""TK-12097 Batch C — Hollywood Wallcoverings hi-res featured-image swap map.
Hollywood Wallcoverings is a PRIVATE LABEL of Momentum/Versa. Hi-res source =
the momentum_colorways staging table's Cloudinary image_url, which is served with
a downscale transform (.../image/upload/c_scale,w_640/v1/hi_res/<file>). We build
the hi-res candidate by REWRITING c_scale,w_640 -> c_scale,w_2000 (bigger). No leak
risk: apply-hires uses the hi-res URL only as `originalSource`, so Shopify downloads
and RE-HOSTS the image on its own CDN — the momentum cloudinary domain never appears
on the storefront.
JOIN: shopify_products.dw_sku <-> momentum_colorways.dw_sku is EMPTY (Shopify carries
the real X-codes; momentum staging carries fabricated DWHD-* codes). So the join used
is the manufacturer-code join: shopify mfr_sku tail (code after last '_') = momentum
alt_sku OR pattern_sku, requiring a UNIQUE resulting image (the vendor code encodes
pattern+colorway, so a unique code match is colorway-level authoritative). Ambiguous
(>1 distinct image) rows are held to tier2-colorway-unverified.
Input : data/tk12097/batchC-candidates.tsv
shopify_id \t mfr_sku \t dw_sku \t cur_url \t momentum_raw_url \t join_type
Output: data/tk12097/batchC-final-map.json (apply-hires.mjs row schema + counts)
data/tk12097/batchC-measured.json (full per-row measurement evidence)
data/tk12097/batchC-tier2-colorway-unverified.json
data/tk12097/batchC-no-hires-source.json
All $0, network reads only; every dimension MEASURED (ranged-GET + PIL), never claimed.
"""
import json, io, sys, re, os, 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, "batchC-candidates.tsv")
OUT_MAP = os.path.join(D, "batchC-final-map.json")
OUT_MEAS = os.path.join(D, "batchC-measured.json")
OUT_TIER2 = os.path.join(D, "batchC-tier2-colorway-unverified.json")
OUT_NOSRC = os.path.join(D, "batchC-no-hires-source.json")
MAX_LOWRES = int(os.environ.get("MAX_LOWRES", "500")) # Hollywood low-res class (TK-11494: no ACTIVE image <500px)
HIRES_FLOOR = int(os.environ.get("HIRES_FLOOR", "800")) # candidate must be >= this to count as a genuine upgrade
SHOP_PX = 5000
SHOP_BYTES = 20 * 1024 * 1024
UA = "Mozilla/5.0 (dw-preflight TK-12097 batchC)"
def strip_qs(url):
return url.split("?")[0]
def rewrite_w2000(raw):
"""Rewrite the momentum cloudinary downscale transform to w_2000 (bigger)."""
u = strip_qs(raw)
if "c_scale,w_640" in u:
return u.replace("c_scale,w_640", "c_scale,w_2000")
# non-standard transform (e.g. c_fit,w_auto/...): swap the whole transform segment
m = re.match(r"^(https?://[^/]+/image/upload/)(.+?)(/v\d+/hi_res/.+)$", u)
if m:
return m.group(1) + "c_scale,w_2000" + m.group(3)
return None
def strip_transform(raw):
"""Remove the transform segment entirely -> the original asset."""
u = strip_qs(raw)
m = re.match(r"^(https?://[^/]+/image/upload/)(.+?)(/v\d+/hi_res/.+)$", u)
if m:
return m.group(1) + m.group(3).lstrip("/")
return None
def cap_w5000(raw):
"""Cap an oversized candidate to w_5000 (re-measure)."""
u = strip_qs(raw)
if re.search(r"c_scale,w_\d+", u):
return re.sub(r"c_scale,w_\d+", "c_scale,w_5000", u)
m = re.match(r"^(https?://[^/]+/image/upload/)(/v\d+/hi_res/.+)$", u)
if m:
return m.group(1) + "c_scale,w_5000" + m.group(2)
m2 = re.match(r"^(https?://[^/]+/image/upload/)(.+?)(/v\d+/hi_res/.+)$", u)
if m2:
return m2.group(1) + "c_scale,w_5000" + m2.group(3)
return None
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:
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) < 6:
continue
rows.append({"shopify_id": p[0], "mfr_sku": p[1] or None,
"dw_sku": p[2] or None, "current_url": p[3] or None,
"momentum_raw": p[4] or None, "join_type": p[5]})
return rows
def measure_current(r):
m = measure_url(r["current_url"]) if r.get("current_url") else {"ok": False, "status": "NO_CUR_URL"}
r["cur_measure"] = m
r["cur_le"] = m.get("meas_le")
return r
def measure_hires(r):
"""Prefer the TRUE NATIVE original (transform removed) as the genuine hi-res.
EMPIRICAL FINDING: cloudinary c_scale,w_2000 UPSCALES (interpolates) natives < 2000px
to a fake 2000px, so it is not "genuine" resolution. The transform-removed original
delivers the real maximum-resolution asset momentum serves (verified: it is a genuine
upgrade over the sub-500px shopify thumb in every measured case), so it is the correct
proposed_hires_url. The w_2000 rewrite is still MEASURED and recorded as evidence, and
is used as the fallback if the original URL cannot be formed or measured. Oversized
natives (>5000px / >20MB) are capped to c_scale,w_5000 and re-measured.
"""
raw = r.get("momentum_raw")
if not raw:
r["hires_status"] = "NO_MOMENTUM_SOURCE"; r["ok"] = False; return r
cur_le = r.get("cur_le") or 0
def genuine(c):
return c and c["le"] > cur_le and c["le"] >= HIRES_FLOOR
tries = []
# w_2000 rewrite (evidence + fallback)
c2000 = rewrite_w2000(raw)
m2000 = None
if c2000:
mm = measure_url(c2000); tries.append(("w2000", mm))
if mm.get("ok"):
m2000 = {"url": c2000, "le": mm["meas_le"], "bytes": mm.get("meas_bytes"), "variant": "w2000"}
# native original (preferred)
native = None
orig = strip_transform(raw)
if orig:
mo = measure_url(orig); tries.append(("orig", mo))
if mo.get("ok"):
native = {"url": orig, "le": mo["meas_le"], "bytes": mo.get("meas_bytes"), "variant": "orig_native"}
r["hires_tries"] = [{"variant": v, **mm} for v, mm in tries]
# choose: prefer native original; fall back to w_2000
chosen = native if genuine(native) else (m2000 if genuine(m2000) else (native or m2000))
if not chosen or not genuine(chosen):
best = chosen["le"] if chosen else None
r["hires_status"] = f"NO_UPGRADE(cur={cur_le},best={best})"; r["ok"] = False; return r
# cap oversized + re-measure
if chosen["le"] > SHOP_PX or (chosen["bytes"] and chosen["bytes"] > SHOP_BYTES):
capped = cap_w5000(raw)
if capped:
mc = measure_url(capped); r["capped_measure"] = mc
if mc.get("ok") and mc["meas_le"] <= SHOP_PX and not (mc.get("meas_bytes") and mc["meas_bytes"] > SHOP_BYTES) \
and mc["meas_le"] > cur_le and mc["meas_le"] >= HIRES_FLOOR:
r["final_url"] = capped; r["final_le"] = mc["meas_le"]; r["capped"] = True
r["hires_variant"] = "capped_w5000"; r["hires_status"] = "PASS_CAPPED"; r["ok"] = True; return r
r["hires_status"] = "STILL_OVER_AFTER_CAP:" + mc.get("status", "?"); r["ok"] = False; return r
r["final_url"] = chosen["url"]; r["final_le"] = chosen["le"]; r["capped"] = False
r["hires_variant"] = chosen["variant"]
r["hires_status"] = "PASS"; r["ok"] = True
return r
def main():
cands = load_candidates()
N = len(cands)
print(f"loaded {N} active Hollywood candidates", file=sys.stderr)
# Phase 1: measure current featured for ALL
with cf.ThreadPoolExecutor(max_workers=24) as ex:
for i, _ in enumerate(ex.map(measure_current, cands), 1):
if i % 400 == 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 that HAVE a momentum source
to_measure = [r for r in lowres if r.get("momentum_raw") and r["join_type"] == "code_unique"]
M = len(to_measure)
print(f" phase2 candidates (lowres w/ code_unique momentum src): {M}", file=sys.stderr)
with cf.ThreadPoolExecutor(max_workers=24) as ex:
for i, _ in enumerate(ex.map(measure_hires, to_measure), 1):
if i % 200 == 0:
print(f" phase2 measured {i}/{M}", file=sys.stderr)
swappable = [r for r in to_measure if r.get("ok")]
excluded = [r for r in to_measure if not r.get("ok")]
# tier2 (ambiguous colorway) among low-res
tier2_rows = []
for r in lowres:
if r["join_type"] == "ambiguous":
tier2_rows.append({"shopify_id": r["shopify_id"], "vendor": "Hollywood Wallcoverings",
"mfr_sku": r["mfr_sku"], "dw_sku": r["dw_sku"], "cur_width": r["cur_le"],
"rollback_url": r["current_url"], "momentum_raw": r["momentum_raw"],
"reason": "ambiguous_code_multiple_distinct_images"})
# no hires source among low-res
no_src = [r for r in lowres if not r.get("momentum_raw") or r["join_type"] == "none"]
# Build apply-hires.mjs map rows
map_rows = []
for r in swappable:
map_rows.append({
"shopify_id": r["shopify_id"],
"vendor": "Hollywood Wallcoverings",
"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": "momentum_colorways_cloudinary",
"preflight_capped": bool(r.get("capped")),
"hires_variant": r.get("hires_variant", "w2000" if not r.get("capped") else "capped"),
"swappable_from_local_staging": True,
})
# counts block
def band(lo, hi):
return sum(1 for r in map_rows if lo <= r["hires_le"] < hi)
hist = {
"500-800": band(500, 800), "800-1200": band(800, 1200),
"1200-2000": band(1200, 2000), "2000-3500": band(2000, 3500),
"3500+": sum(1 for r in map_rows if r["hires_le"] >= 3500),
}
floors = {}
for f in (500, 600, 800, 1000, 1200):
floors[f">={f}"] = sum(1 for r in map_rows if r["hires_le"] >= f)
excl_break = dict(Counter(r.get("hires_status") for r in excluded))
counts = {
"active_total": N,
"current_unreachable": len(cur_unreachable),
"already_hires_gt500": len(already_hi),
"lowres_total": len(lowres),
"lowres_with_momentum_src": M,
"colorway_safe_upgrades": len(swappable),
"excluded_no_upgrade_or_dead": len(excluded),
"excluded_breakdown": excl_break,
"tier2_held": len(tier2_rows),
"no_hires_source": len(no_src),
"emitted_at_floor>=800": len(map_rows),
"capped_hires": sum(1 for r in map_rows if r.get("preflight_capped")),
"by_vendor": {"Hollywood Wallcoverings": len(map_rows)},
"width_histogram": hist,
"coverage_per_floor": floors,
}
doc = {
"ticket": "TK-12097", "batch": "C",
"parent": "TK-12090 / TK-11494",
"generated_at": datetime.datetime.utcnow().isoformat() + "Z",
"generated_by": "vp-dw-commerce",
"store": "designer-laboratory-sandbox",
"vendor": "Hollywood Wallcoverings",
"engine": "reuse ~/Projects/dw-kravet-hires/scripts/apply-hires.mjs (--max-width 500 --only-swappable)",
"join_used": ("dw_sku exact join EMPTY (Shopify=real X-codes, momentum staging=fabricated DWHD-*). "
"Used manufacturer-code join: shopify mfr_sku tail (code after last '_') = "
"momentum_colorways.alt_sku OR pattern_sku, requiring a UNIQUE resulting image "
"(vendor code encodes pattern+colorway => unique match is colorway-level authoritative). "
"Ambiguous (>1 distinct image) held to tier2."),
"hires_transform": ("momentum_colorways.image_url Cloudinary downscale c_scale,w_640 rewritten to "
"c_scale,w_2000 (querystring stripped); fallback = transform removed (original); "
"oversized capped to c_scale,w_5000; all MEASURED. Genuine-upgrade gate: "
f"measured hires_le > current AND >= {HIRES_FLOOR}px."),
"no_leak": ("apply-hires passes proposed_hires_url as Shopify productCreateMedia originalSource; "
"Shopify downloads + re-hosts on its own CDN, so the momentum cloudinary domain never "
"appears on the storefront."),
"counts": counts,
"rows": map_rows,
}
json.dump(doc, open(OUT_MAP, "w"), indent=1)
json.dump({"ticket": "TK-12097", "batch": "C",
"measured_at_utc": datetime.datetime.utcnow().isoformat() + "Z",
"max_lowres": MAX_LOWRES, "hires_floor": HIRES_FLOOR, "rows": cands},
open(OUT_MEAS, "w"), indent=1)
json.dump({"rows": tier2_rows}, open(OUT_TIER2, "w"), indent=1)
json.dump(no_src, open(OUT_NOSRC, "w"), indent=1)
print(json.dumps(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)
# a couple of measured before/after examples
print("\n--- sample measured before/after ---", file=sys.stderr)
for r in swappable[:3]:
print(f" {r['mfr_sku']}: cur={r['cur_le']}px -> hires={r['final_le']}px ({r.get('hires_variant')}) {r['final_url']}", file=sys.stderr)
if __name__ == "__main__":
main()