← back to Dw Kravet Hires
scripts/phase2-recover.py
144 lines
#!/usr/bin/env python3
"""
Phase-2 hi-res recovery for TK-11658 (1,593 non-swappable) + TK-11742 (28 broken-source).
Feed-first, $0: fetch kravet.com product page (naive SKU->slug), extract the gallery hero
(isMain 'full' entry), normalize to a bounded JPEG, fetch-and-VERIFY (200 + jpeg magic +
NOT placeholder md5 + long-edge > 1200px). Records recovered vs genuinely-absent.
"""
import subprocess, re, hashlib, struct, json, sys
from concurrent.futures import ThreadPoolExecutor, as_completed
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 = ".jpg?width=2048&height=2048&fit=bound" # bounded JPEG: >1200px, <=2048px (<<5000px/20MP), ~1-2MB
MINPX = 1200
def naive_slug(mfr):
b = re.sub(r'[./_]', '-', mfr.strip().lower())
b = re.sub(r'-+', '-', b).strip('-')
return re.sub(r'-0$', '', b) # drop trailing finish '.0'
def get_code(u):
return subprocess.run(["curl", "-sL", "-A", UA, "--max-time", "30", "-o", "/dev/null",
"-w", "%{http_code}", u], capture_output=True).stdout.decode()
def get_html(u):
return subprocess.run(["curl", "-sL", "-A", UA, "--max-time", "30", u],
capture_output=True).stdout.decode("utf-8", "replace")
def fetch_bin(url):
r = subprocess.run(["curl", "-sL", "-A", UA, "--max-time", "50",
"-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()
body = out[:idx]
p = meta.split(" ")
return body, p[0], (p[1] if len(p) > 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_hero(html):
imgs = []
for m in re.finditer(r'\{"thumb":"(.*?)","img":"(.*?)","full":"(.*?)".*?(?:"isMain":(true|false))?', html):
imgs.append((m.group(3).replace("\\/", "/"), m.group(4)))
for full, ismain in imgs:
if ismain == "true":
return full
if imgs:
return imgs[0][0]
b = re.findall(r'https://cdn\.brandfolder\.io/[A-Za-z0-9/_.-]+\.(?:jpg|jpeg|png)', html.replace("\\/", "/"))
return b[0] if b else None
def base_of(u):
u = u.split("?", 1)[0]
return re.sub(r'\.(png|jpg|jpeg|auto)$', '', u, flags=re.I)
def process(row):
rec = {k: row.get(k) for k in ("job", "vendor", "dw_sku", "mfr_sku", "shopify_id", "rollback_url")}
mfr = (row.get("mfr_sku") or "").strip()
if not mfr:
rec.update(status="absent", reason="no mfr_sku (cannot build vendor URL)")
return rec
slug = naive_slug(mfr)
url = "https://www.kravet.com/" + slug
rec["product_url"] = url
code = get_code(url)
if code != "200":
rec.update(status="absent", reason=f"kravet.com product page {code} (no vendor page)")
return rec
html = get_html(url)
rec["discontinued"] = bool(re.search(r'[Dd]iscontinued', html))
hero = extract_hero(html)
if not hero or "placeholder" in hero.lower():
rec.update(status="absent", reason="vendor page serves placeholder / no brandfolder hero")
return rec
norm = base_of(hero) + NORM
body, c, ct = fetch_bin(norm)
md5 = hashlib.md5(body).hexdigest() if body else ""
w, h = dims(body)
magic = body[:3] == b"\xff\xd8\xff"
longedge = max(w or 0, h or 0)
if c != "200":
rec.update(status="absent", reason=f"hero render {c} (asset unprocessable)", hero_asset=hero)
elif "image" not in ct or not magic:
rec.update(status="absent", reason=f"not a valid image (ct={ct} magic={magic})", hero_asset=hero)
elif md5 in PLACEHOLDER_MD5:
rec.update(status="absent", reason="resolves to placeholder md5")
elif longedge <= 400:
# not even an upgrade over the current 400px featured — treat as absent
rec.update(status="absent", reason=f"asset only {longedge}px (no upgrade over 400px)")
elif longedge <= MINPX:
# real vendor asset, a genuine upgrade over 400px, but under the 1200px quality bar
rec.update(status="below_threshold", recovered_hires_url=norm, http=c, content_type=ct,
bytes=len(body), dims=f"{w}x{h}", long_edge=longedge, md5=md5)
else:
rec.update(status="recovered", recovered_hires_url=norm, http=c, content_type=ct,
bytes=len(body), dims=f"{w}x{h}", long_edge=longedge, md5=md5)
return rec
def main():
targets = json.load(open(sys.argv[1]))
out_path = sys.argv[2]
# resume: keep already-processed rows, only work the remainder
results = []
seen = set()
try:
results = json.load(open(out_path))
seen = {r.get("shopify_id") for r in results}
except Exception:
pass
remaining = [r for r in targets if r.get("shopify_id") not in seen]
print(f"resume: {len(results)} already done, {len(remaining)} remaining", flush=True)
done = len(results)
total = len(targets)
with ThreadPoolExecutor(max_workers=12) as ex:
futs = {ex.submit(process, r): r for r in remaining}
for f in as_completed(futs):
results.append(f.result())
done += 1
if done % 50 == 0:
rec = sum(1 for r in results if r["status"] == "recovered")
bt = sum(1 for r in results if r["status"] == "below_threshold")
print(f" {done}/{total} | recovered {rec} | below1200 {bt}", flush=True)
json.dump(results, open(out_path, "w"), indent=1)
json.dump(results, open(out_path, "w"), indent=1)
rec = sum(1 for r in results if r["status"] == "recovered")
bt = sum(1 for r in results if r["status"] == "below_threshold")
ab = sum(1 for r in results if r["status"] == "absent")
print(f"\nDONE {out_path}: recovered(>1200) {rec} | below_threshold(401-1200) {bt} | absent {ab} / {len(results)}")
if __name__ == "__main__":
main()