← back to Dw Kravet Hires
scripts/p2-reresolve-slug.py
307 lines
#!/usr/bin/env python3
"""
TK-12097 P2 — recover hi-res for the 150 "kravet.com slug-miss" (bucket=reresolve_slug).
READ-ONLY, $0. Reuses phase2-recover.py hero extraction + verify-batchA-tier2.py colorway
verify. Re-resolves the naive-404 slug via corrected/alt slugs + title-name slugs, fetches
the resolved kravet.com page (follow redirects), colorway-verifies, extracts the isMain
Brandfolder hero, measures it, and emits an apply-hires map (NO writes, NO commit).
Key fix vs phase-2: naive_slug did not convert SPACES (Lee Jofa name-coded SKUs). And the
hero is kept as .auto (NOT force-.jpg, the phase-2 /as/ bug); brandfolder .auto only honors
width&height&pad=true for full-res (fit=bound is ignored -> 400px), matching the page's own
'full' URL.
"""
import subprocess, re, hashlib, struct, json, sys, os
from concurrent.futures import ThreadPoolExecutor, as_completed
HERE = os.path.dirname(os.path.abspath(__file__))
PROJ = os.path.dirname(HERE)
D = os.path.join(PROJ, "data/tk12097")
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126 Safari/537.36"
PLACEHOLDER_MD5 = {"ea85ae7f75e4e10c4aa01ebe1ffa9d5c", "318182be0a481fcb8ce87fd79e2299f0", "7b75637761ce6b90552319e511150f03"}
NORM = ".auto?width=2048&height=2048&pad=true" # brandfolder full-res (page's own 'full' style); <=5000px cap
CAP = 5000
MIN_UPGRADE = 800
VMAP = {'GP & J Baker': 'gp_baker_catalog', 'Brunschwig & Fils': 'brunschwig_catalog',
'Lee Jofa': 'lee_jofa_catalog', 'Gaston Y Daniela': 'gaston_daniela_catalog'}
def hy(s):
return re.sub(r'-+', '-', re.sub(r'[^a-z0-9]+', '-', s.lower())).strip('-')
def get_code(u):
r = subprocess.run(["curl", "-sL", "-A", UA, "--max-time", "25", "-o", "/dev/null",
"-w", "%{http_code} %{url_effective}", u], capture_output=True, text=True).stdout.split(" ", 1)
return r[0], (r[1] if len(r) > 1 else "")
def get_html(u):
r = subprocess.run(["curl", "-sL", "-A", UA, "--max-time", "30",
"-w", "\n%{url_effective} %{http_code}", u], capture_output=True)
out = r.stdout
idx = out.rfind(b"\n")
meta = out[idx + 1:].decode("ascii", "replace").strip().split(" ")
return (meta[0] if meta else u), (meta[-1] if len(meta) > 1 else "000"), out[:idx].decode("utf-8", "replace")
def fetch_bin(url):
r = subprocess.run(["curl", "-sL", "-A", UA, "--max-time", "70",
"-w", "\n%{http_code} %{content_type} %{size_download}", url], capture_output=True)
out = r.stdout
idx = out.rfind(b"\n")
meta = out[idx + 1:].decode("ascii", "replace").strip().split(" ")
return out[:idx], meta[0], (meta[1] if len(meta) > 1 else "")
def dims(body):
if body[:8] == b"\x89PNG\r\n\x1a\n":
return struct.unpack(">I", body[16:20])[0], struct.unpack(">I", body[20:24])[0]
if body[:2] == b"\xff\xd8":
i = 2
while i < len(body) - 9:
if body[i] != 0xFF:
i += 1; continue
if body[i + 1] in (0xC0, 0xC1, 0xC2, 0xC3):
return struct.unpack(">H", body[i + 7:i + 9])[0], struct.unpack(">H", body[i + 5:i + 7])[0]
i += 2 + struct.unpack(">H", body[i + 2:i + 4])[0]
return None, None
def extract_main_hero(html):
for m in re.finditer(r'\{"thumb":"(.*?)","img":"(.*?)","full":"(.*?)".*?(?:"isMain":(true|false))?', html):
if m.group(4) == "true":
return m.group(3).replace("\\/", "/")
# fallback: first isMain-less full, else first brandfolder ref
for m in re.finditer(r'"full":"(https:[^"]+brandfolder[^"]+)"', html):
return m.group(1).replace("\\/", "/")
b = re.findall(r'https://cdn\.brandfolder\.io/[A-Za-z0-9/_.-]+\.(?:jpg|jpeg|png|auto)', html.replace("\\/", "/"))
return b[0] if b else None
def base_of(u):
return re.sub(r'\.(png|jpg|jpeg|auto)$', '', u.split("?", 1)[0], flags=re.I)
def page_title(html):
m = re.search(r'<title>(.*?)</title>', html, re.I | re.S)
return (m.group(1) or "").strip() if m else ""
def sku_norm(mfr):
# upper, strip trailing .0, strip leading P/T, alnum-only
s = re.sub(r'\.0$', '', mfr.strip().upper())
s = re.sub(r'^[PT]', '', s)
return re.sub(r'[^A-Z0-9]', '', s)
def candidates(mfr, title):
out = []
segs = re.sub(r'\.0$', '', mfr).split('.')
corrected = re.sub(r'-0$', '', hy(re.sub(r'[ ./_]+', '-', mfr.lower())))
out.append(('corrected', corrected))
out.append(('corrected_keep0', hy(mfr)))
if len(segs) >= 2:
col = segs[-1]; colz = col.lstrip('0') or '0'; patt = '.'.join(segs[:-1])
out.append(('color_nozero', hy(patt) + '-' + hy(colz)))
out.append(('pattern_only', hy(patt)))
pat = title.split(' By ')[0]
pat = re.sub(r'^(Lee Jofa|G ?P ?& ?J Baker|Brunschwig & Fils|Gaston Y Daniela)\s*-\s*', '', pat)
if ' - ' in pat:
pn, cn = pat.split(' - ', 1)
out.append(('name_full', hy(pn) + '-' + hy(cn)))
if cn.split():
out.append(('name_firstcolor', hy(pn) + '-' + hy(cn.split()[0])))
if len(segs) >= 2:
out.append(('name_mfrcolor', hy(pn) + '-' + hy(segs[-1])))
out.append(('name_pattern', hy(pn)))
seen = set(); uniq = []
for tag, s in out:
if s and s not in seen:
seen.add(s); uniq.append((tag, s))
return uniq, (pat.split(' - ')[0] if ' - ' in pat else pat)
def resolve(mfr, title):
"""return (final_url, tag, slug, html) for first good 200, else (None,...)."""
cands, _ = candidates(mfr, title)
for tag, slug in cands:
url = "https://www.kravet.com/" + slug
code, final = get_code(url)
fl = final.lower()
if code == "200" and "/search" not in fl and "/catalogsearch" not in fl:
# guard: reject bare homepage / vendor category landing (path must have a hyphen-code or the slug)
path = fl.replace("https://www.kravet.com/", "").strip("/")
if path and (path == slug or re.search(r'-\d', path) or slug in path):
fu, fc, html = get_html(final)
return fu, tag, slug, html
return None, None, None, None
def process(row):
rec = {"shopify_id": row["shopify_id"], "vendor": row["vendor"],
"mfr_sku": row["mfr_sku"], "dw_sku": row.get("dw_sku")}
cur = row.get("_cur_img") or ""
rec["current_400px_url"] = cur
rec["rollback_url"] = cur
rec["cur_width"] = row.get("_cur_w")
cur_le = row.get("_cur_le") or 0
mfr = row["mfr_sku"]; title = row.get("_title", "")
final, tag, slug, html = resolve(mfr, title)
if not final:
rec.update(status="still_absent", reason="no kravet.com page under corrected/alt/name slug (off-kravet)")
return rec
rec["resolved_url"] = final; rec["slug_tag"] = tag; rec["slug"] = slug
if re.search(r'[Dd]iscontinued', html):
rec["kravet_discontinued"] = True
hero = extract_main_hero(html)
if not hero or "placeholder" in hero.lower():
rec.update(status="still_absent", reason="resolved page has no brandfolder isMain hero / placeholder",
resolved_url=final)
return rec
rec["page_hero"] = hero
hero_fn = base_of(hero).split("/")[-1]
# --- colorway safety ---
# (A) FILENAME gate: hero filename encodes the full vendor SKU code (color included). This is
# the REQUIRED gate for numeric-coded SKUs (pattern part has digits, no space), because
# a name-alias can silently resolve to a DIFFERENT colorway of the same numeric pattern
# (e.g. GDT5548.021 -> gdt5548-018). Exact colorway proof.
hf = re.sub(r'[^A-Z0-9]', '', hero_fn.upper())
skun = sku_norm(mfr)
gate_filename = bool(skun) and len(skun) >= 5 and skun in hf
# (B) NAME+COLOR gate: for NAME-coded SKUs (pattern part is a real name w/ a space, no numeric
# equivalence to kravet's canonical code) the color proof is: slug built from our PATTERN+
# COLOR (color token present in the winning slug, not a pattern-only fallback) AND the color
# is CONFIRMED on the resolved page (a >=3-char alpha color sub-token appears in the HTML).
segs = re.sub(r'\.0$', '', mfr).split('.')
pattern_part = '.'.join(segs[:-1]) if len(segs) >= 2 else segs[0]
color_tok = segs[-1] if len(segs) >= 2 else ''
name_coded = (' ' in pattern_part) or (not re.search(r'\d', pattern_part))
color_slug_toks = [t for t in hy(color_tok).split('-') if len(t) >= 3]
color_in_slug = bool(color_slug_toks) and any(t in slug for t in color_slug_toks)
html_low = html.lower()
color_on_page = bool(color_slug_toks) and any(t in html_low for t in color_slug_toks)
not_pattern_only = tag not in ("pattern_only", "name_pattern")
gate_name = name_coded and color_in_slug and color_on_page and not_pattern_only
colorway_safe = gate_filename or gate_name
rec["colorway_gate"] = ("filename" if gate_filename else ("name+color" if gate_name else "none"))
rec["name_coded"] = name_coded
if not colorway_safe:
rec.update(status="colorway_dropped",
reason=(f"hero '{hero_fn}' fails colorway gate "
f"(numeric_sku_filename_mismatch)" if not name_coded else
f"name-coded but color unconfirmed (color_in_slug={color_in_slug} "
f"color_on_page={color_on_page} tag={tag})"))
return rec
# measure hero (.auto, NOT forced .jpg)
norm = base_of(hero) + NORM
body, c, ct = fetch_bin(norm)
if not body or c != "200":
rec.update(status="still_absent", reason=f"hero render {c}", proposed_hires_url=norm)
return rec
md5 = hashlib.md5(body).hexdigest()
w, h = dims(body)
le = max(w or 0, h or 0)
magic = body[:3] == b"\xff\xd8\xff" or body[:8] == b"\x89PNG\r\n\x1a\n"
if "image" not in ct or not magic:
rec.update(status="still_absent", reason=f"not a valid image (ct={ct})", proposed_hires_url=norm)
return rec
if md5 in PLACEHOLDER_MD5:
rec.update(status="still_absent", reason="resolves to placeholder md5")
return rec
if le > CAP:
rec.update(status="still_absent", reason=f"hero {le}px exceeds {CAP}px cap")
return rec
if not (le > cur_le and le >= MIN_UPGRADE):
rec.update(status="not_upgrade", reason=f"hero {le}px not a genuine upgrade over cur {cur_le}px (>=800 req)",
hires_le=le, proposed_hires_url=norm)
return rec
rec.update(status="recovered", proposed_hires_url=norm, hires_le=le,
hires_source="kravet_com_reresolved", swappable_from_local_staging=True,
hero_dims=f"{w}x{h}", hero_bytes=len(body), hero_md5=md5)
return rec
def measure_current(row):
"""measure the current featured image -> cur_le, cur_width"""
u = row.get("_cur_img")
if not u:
return
body, c, ct = fetch_bin(u)
if body and c == "200":
w, h = dims(body)
row["_cur_w"] = w
row["_cur_le"] = max(w or 0, h or 0)
def psql(sql):
return subprocess.run(["psql", "-h", "/tmp", "dw_unified", "-tAF", "\t", "-c", sql],
capture_output=True, text=True).stdout
def main():
man = json.load(open(os.path.join(D, "rescrape-lane-manifest.json")))
rows = [i for i in man["items"] if i.get("bucket") == "reresolve_slug"]
ids = [r["shopify_id"] for r in rows]
open("/tmp/p2_ids.txt", "w").write("\n".join(ids))
q = psql("SELECT shopify_id,image_url,title FROM shopify_products WHERE shopify_id=ANY(string_to_array(pg_read_file('/tmp/p2_ids.txt'),E'\\n'));")
img = {}; ttl = {}
for line in q.strip().split("\n"):
p = line.split("\t")
if len(p) >= 3:
img[p[0]] = p[1]; ttl[p[0]] = p[2]
for r in rows:
r["_cur_img"] = img.get(r["shopify_id"], "")
r["_title"] = ttl.get(r["shopify_id"], "")
print(f"measuring {len(rows)} current images...", flush=True)
with ThreadPoolExecutor(max_workers=14) as ex:
list(as_completed([ex.submit(measure_current, r) for r in rows]))
print("re-resolving slugs + heroes...", flush=True)
results = []
with ThreadPoolExecutor(max_workers=12) as ex:
futs = {ex.submit(process, r): r for r in rows}
done = 0
for f in as_completed(futs):
results.append(f.result()); done += 1
if done % 30 == 0:
rc = sum(1 for x in results if x["status"] == "recovered")
print(f" {done}/{len(rows)} | recovered {rc}", flush=True)
json.dump(results, open(os.path.join(D, "batchP2-reresolve-results.json"), "w"), indent=1)
recovered = [r for r in results if r["status"] == "recovered"]
import collections
byv = collections.Counter(r["vendor"] for r in recovered)
wh = collections.Counter()
for r in recovered:
le = r["hires_le"]
b = "401-500" if le <= 500 else "500-800" if le <= 800 else "800-1200" if le <= 1200 else "1200-2500" if le <= 2500 else "2500+"
wh[b] += 1
counts = {
"input_150": len(rows),
"recovered": len(recovered),
"still_absent": sum(1 for r in results if r["status"] == "still_absent"),
"colorway_dropped": sum(1 for r in results if r["status"] == "colorway_dropped"),
"not_upgrade": sum(1 for r in results if r["status"] == "not_upgrade"),
"by_vendor": dict(byv),
"by_status": dict(collections.Counter(r["status"] for r in results)),
"width_histogram": dict(wh),
}
map_rows = [{
"shopify_id": r["shopify_id"], "vendor": r["vendor"], "mfr_sku": r["mfr_sku"], "dw_sku": r["dw_sku"],
"cur_width": r.get("cur_width"), "current_400px_url": r["current_400px_url"], "rollback_url": r["rollback_url"],
"proposed_hires_url": r["proposed_hires_url"], "hires_le": r["hires_le"],
"hires_source": "kravet_com_reresolved", "swappable_from_local_staging": True,
} for r in recovered]
out = {"ticket": "TK-12097", "batch": "P2-reresolve_slug", "generated_by": "vp-dw-commerce",
"note": "150 kravet.com slug-miss re-resolved via corrected/alt/name slugs; colorway-gated; .auto hero (no forced .jpg)",
"counts": counts, "rows": map_rows}
json.dump(out, open(os.path.join(D, "batchP2-recovered-map.json"), "w"), indent=1)
print("\n=== COUNTS ===")
print(json.dumps(counts, indent=1))
print(f"\nmap: {os.path.join(D,'batchP2-recovered-map.json')} ({len(map_rows)} rows)")
if __name__ == "__main__":
main()