← back to Kravet Sheet Sync 2026 04 20
kravet_image_catalog.py
243 lines
#!/usr/bin/env python3
"""
Build per-SKU image catalog for all Kravet wallcoverings.
For each SKU:
1) Query Algolia API → get canonical URL + primary/full_repeat Brandfolder URLs
2) Fetch product page HTML → extract additional sku-scoped brandfolder URLs
(room settings, detail shots, larger sizes)
3) Append one JSONL line per URL to kravet_image_catalog.jsonl
Features:
- Resumable: tracks completed SKUs in .done set
- Concurrent: 12 workers (tunable)
- Algolia API key is scraped from the search page; re-fetched if 401
- Rate-aware: retries with backoff on 429/5xx
After completion, rows are bulk-COPYd into dw_unified.kravet_sku_images.
"""
import argparse, json, os, re, subprocess, sys, time, threading, codecs
import urllib.request, urllib.parse, urllib.error
from concurrent.futures import ThreadPoolExecutor, as_completed
UA = {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_0)"}
ALGOLIA_APP = "M9TBUM1WAE"
ALGOLIA_INDEX = "kravet_production_kravet_us_products"
SSH = ["ssh", "root@45.61.58.125"]
_lock = threading.Lock()
_algolia_key = None
# ------------------------------------------------------------------
def fetch_algolia_key():
"""Scrape a fresh Algolia search API key from the search page."""
url = "https://www.kravet.com/catalogsearch/result/?q=test"
html = urllib.request.urlopen(urllib.request.Request(url, headers=UA), timeout=20).read().decode("utf-8","ignore")
m = re.search(r"window\.algoliaConfig\s*=\s*JSON\.parse\('(.*?)'\);", html, re.S)
if not m: return None
raw = codecs.decode(m.group(1), "unicode_escape")
k = re.search(r'"apiKey"\s*:\s*"([^"]+)"', raw)
return k.group(1) if k else None
def get_algolia_key():
global _algolia_key
with _lock:
if _algolia_key is None:
_algolia_key = fetch_algolia_key()
print(f"[algolia] got key ending …{_algolia_key[-10:]}", flush=True)
return _algolia_key
def algolia_search(sku, retries=2):
"""Return the first hit or None."""
for attempt in range(retries + 1):
key = get_algolia_key()
body = json.dumps({"query": sku, "hitsPerPage": 3, "restrictSearchableAttributes": ["sku","name"]}).encode()
req = urllib.request.Request(
f"https://{ALGOLIA_APP.lower()}-dsn.algolia.net/1/indexes/{ALGOLIA_INDEX}/query",
data=body, method="POST",
headers={"X-Algolia-Application-Id": ALGOLIA_APP, "X-Algolia-API-Key": key,
"Content-Type": "application/json"})
try:
data = json.loads(urllib.request.urlopen(req, timeout=15).read())
# Find exact SKU match
for h in data.get("hits", []):
if h.get("sku") == sku:
return h
# Else first hit (may be partial)
hits = data.get("hits", [])
return hits[0] if hits else None
except urllib.error.HTTPError as e:
if e.code == 401 and attempt < retries:
with _lock:
globals()["_algolia_key"] = None # force re-fetch
continue
if e.code == 429 and attempt < retries:
time.sleep(2 ** attempt); continue
return None
except Exception:
return None
return None
def fetch_product_html(url, retries=2):
for attempt in range(retries + 1):
try:
return urllib.request.urlopen(urllib.request.Request(url, headers=UA), timeout=20).read().decode("utf-8","ignore")
except urllib.error.HTTPError as e:
if e.code == 429 and attempt < retries:
time.sleep(2 ** attempt); continue
return None
except Exception:
return None
return None
def extract_brandfolder_urls(html, sku_key_lower):
"""Return [(url, kind)] — kind ∈ colorway | full_repeat | room | sibling | other."""
if not html: return []
urls = sorted(set(re.findall(r'https://cdn\.brandfolder\.io/[^"\'\s<>]+', html)))
out = []
for u in urls:
u_clean = u.replace("&", "&")
# Identify the filename segment (last path component before ?)
tail = u_clean.split("?", 1)[0].rsplit("/", 1)[-1].lower()
kind = "other"
if "full_repeat" in tail:
kind = "full_repeat"
elif "_room" in tail or "/room" in u_clean.lower() or "room_setting" in u_clean.lower():
kind = "room"
elif sku_key_lower in tail or sku_key_lower.replace("-","_") in tail:
kind = "colorway"
else:
# could be a sibling colorway (different last digits) — useful but not primary
kind = "sibling"
out.append((u_clean, kind))
return out
def canonicalize(url):
"""Prefer jpg with width=2000, dedupe by filename-without-size."""
base = re.sub(r"[?&](width|height|pad)=\w+", "", url).rstrip("?&")
return base
# ------------------------------------------------------------------
def process_sku(sku):
"""Return dict with Algolia + HTML results for one SKU."""
result = {"sku": sku, "status": "ok", "algolia": None, "urls": []}
hit = algolia_search(sku)
if not hit:
result["status"] = "algolia_miss"
return result
result["algolia"] = {
"name": hit.get("name"),
"url": hit.get("url") or hit.get("url_key"),
"thumbnail": hit.get("thumbnail_url"),
"image": hit.get("image_url"),
"full": hit.get("external_image_url_1"),
"full_repeat": hit.get("external_image_url_full_repeat"),
}
# Collect URLs from Algolia fields
seen = set()
def add(u, kind):
if not u: return
cu = canonicalize(u)
if cu in seen: return
seen.add(cu)
result["urls"].append({"url": u, "kind": kind, "source": "algolia"})
add(hit.get("external_image_url_1"), "colorway")
add(hit.get("external_image_url_full_repeat"), "full_repeat")
add(hit.get("image_url"), "colorway")
# Fetch product page for additional SKU-specific images
purl = hit.get("url") or hit.get("url_key")
if purl and purl.startswith("http"):
html = fetch_product_html(purl)
if html:
# SKU key lowercased, dashes preserved: "94/4021.CS.0" → "94-4021"
sku_key = re.sub(r"\W+", "-", sku.split(".")[0]).lower().strip("-")
for u, kind in extract_brandfolder_urls(html, sku_key):
# Only keep colorway matches for THIS sku or room/full_repeat/other
if kind in ("sibling",):
continue
add(u, kind)
else:
result["status"] = "html_err"
return result
# ------------------------------------------------------------------
def pg_query(sql):
r = subprocess.run(SSH + [
f'PGPASSWORD=DW2024! psql -h 127.0.0.1 -U dw_admin -d dw_unified --csv -t -c "{sql}"'],
capture_output=True, text=True, check=True)
import csv, io
return [row for row in csv.reader(io.StringIO(r.stdout)) if row]
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--limit", type=int, default=0, help="limit rows (0=all)")
ap.add_argument("--threads", type=int, default=12)
ap.add_argument("--output", default="kravet_image_catalog.jsonl")
ap.add_argument("--scope", choices=["wallcover","all","new-bucket"], default="wallcover")
ap.add_argument("--test-skus", nargs="*")
args = ap.parse_args()
# Which SKUs to process?
if args.test_skus:
skus = args.test_skus
elif args.scope == "wallcover":
q = "SELECT DISTINCT mfr_sku_raw FROM kravet_sheet_fields WHERE upper(COALESCE(use,'')) LIKE '%WALL%'"
if args.limit: q += f" LIMIT {args.limit}"
skus = [r[0] for r in pg_query(q)]
elif args.scope == "new-bucket":
q = "SELECT mfr_sku FROM kravet_diff_new_2026_04_20"
if args.limit: q += f" LIMIT {args.limit}"
skus = [r[0] for r in pg_query(q)]
else:
q = "SELECT DISTINCT mfr_sku_raw FROM kravet_sheet_fields"
if args.limit: q += f" LIMIT {args.limit}"
skus = [r[0] for r in pg_query(q)]
print(f"[scope] {len(skus):,} SKUs to process → {args.output}")
# Resume: read existing JSONL, skip SKUs we already have
done = set()
if os.path.exists(args.output):
with open(args.output) as f:
for line in f:
try: done.add(json.loads(line)["sku"])
except Exception: pass
print(f"[resume] {len(done):,} already processed")
todo = [s for s in skus if s not in done]
print(f"[todo] {len(todo):,}")
out_f = open(args.output, "a", buffering=1)
t0 = time.time()
ok = miss = err = 0
with ThreadPoolExecutor(max_workers=args.threads) as ex:
futs = [ex.submit(process_sku, s) for s in todo]
for i, f in enumerate(as_completed(futs), 1):
r = f.result()
out_f.write(json.dumps(r) + "\n")
if r["status"] == "ok": ok += 1
elif r["status"] == "algolia_miss": miss += 1
else: err += 1
if i % 200 == 0 or i == len(todo):
rate = i / max(1, time.time() - t0)
eta = (len(todo) - i) / max(1, rate)
print(f" [{i:,}/{len(todo):,}] ok={ok:,} miss={miss:,} err={err:,} "
f"{rate:.1f}/s eta {eta/60:.0f}m", flush=True)
out_f.close()
print(f"[done] ok={ok:,} miss={miss:,} err={err:,} {time.time()-t0:.0f}s")
if __name__ == "__main__":
main()