`scripts/extract.py`

151 lines
#!/usr/bin/env python3
"""Feed-first Sanderson extractor: plain-fetch product pages, parse JSON-LD, emit CSV for PG staging. $0 (no proxy/API)."""
import json, os, re, sys, time, urllib.request, csv, html

BASE = "https://www.sanderson.design"
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120 Safari/537.36"
LIMIT = int(sys.argv[1]) if len(sys.argv) > 1 else 0
URLFILE = sys.argv[2] if len(sys.argv) > 2 else "all_prod_urls.txt"
OUTFILE = sys.argv[3] if len(sys.argv) > 3 else "rows.csv"
RETRY = "--retry" in sys.argv

def load_map(path):
    m = {}
    with open(path) as f:
        for line in f:
            parts = line.rstrip("\n").split("\t")
            if len(parts) == 2:
                m[parts[0].upper()] = parts[1]
    return m

pattern_map = load_map("pattern_map.tsv") if os.path.exists("pattern_map.tsv") else {}
if os.path.exists("settlement_urls.txt"):
    with open("settlement_urls.txt") as _sf: settlement = set(l.strip() for l in _sf if l.strip())
else:
    settlement = set()
if not os.path.exists(URLFILE):
    sys.exit(f"ERROR: url file not found: {URLFILE}")
with open(URLFILE) as _uf: urls = [l.strip() for l in _uf if l.strip()]
urls = [u if u.startswith("/") else "/wallpaper/"+u for u in urls]
if LIMIT:
    urls = urls[:LIMIT]

def fetch(url):
    req = urllib.request.Request(url, headers={"User-Agent": UA})
    with urllib.request.urlopen(req, timeout=30) as r:
        return r.read().decode("utf-8", "replace")

def norm(w):
    return re.sub(r"[^a-z0-9]", "", w.lower())

def strip_code(tok):
    # strip design-code tokens like Dcavad101 / DCAVAE103
    return re.sub(r"\b[A-Za-z]{3,6}\d{2,4}\b", "", tok).strip()

def derive_color(name, pattern):
    n = strip_code(name)
    nwords = n.split()
    if pattern:
        pwords = [norm(w) for w in pattern.split()]
        j = 0
        # strip leading name-words that match the pattern words (order-tolerant, punct/case-insensitive)
        while j < len(nwords) and norm(nwords[j]) in pwords:
            j += 1
        nwords = nwords[j:]
    color = re.sub(r"\s+", " ", " ".join(nwords)).strip(" -/")
    return color or None

rows = []
JSONLD = re.compile(r'<script[^>]*application/ld\+json[^>]*>(.*?)</script>', re.S)
IMG = re.compile(r'(static/media/catalog/product/[A-Za-z0-9/_.\-]+\.jpg)')

# FIX (TK-11256, 2026-09-04): the old gallery extraction (IMG.findall(h) over the WHOLE
# page) swept up EVERY image referenced anywhere on the PDP -- including the "shop other
# colourways" swatch carousel and cross-sell "shop the room" panels, which show OTHER
# products' images at real (non-thumbnail) resolution. That's the dw-image-identity-canary
# foreign_colorway/foreign_pattern contamination (Sanderson foreign_products 6->24).
# Verified live (2026-09-04): Sanderson's own gallery/detail shots consistently render at
# srcset max-width 2560, while "other colourway" swatches cap at 64w and cross-sell/
# "shop the room" images cap at 1080w -- a clean, reproducible separation on every PDP
# checked (SAW0094-01/-03, SAW0224-01). Scope gallery_images to that >=1500w band only.
GALLERY_MIN_W = 1500
SRCSET = re.compile(r'srcset="([^"]*static/media/catalog/product/[A-Za-z0-9/_.\-]+\.jpg[^"]*)"')
FNAME = re.compile(r'([A-Za-z0-9_]+\.jpg)')
PATHFNAME = re.compile(r'(static/media/catalog/product/[A-Za-z0-9/]+/([A-Za-z0-9_]+\.jpg))')

def own_gallery(h):
    best_w = {}
    for m in SRCSET.finditer(h):
        srcset = m.group(1)
        widths = [int(w) for w in re.findall(r'(\d+)w', srcset)]
        if not widths:
            continue
        w = max(widths)
        for fm in FNAME.finditer(srcset):
            fname = fm.group(1)
            best_w[fname] = max(best_w.get(fname, 0), w)
    keep = {f for f, w in best_w.items() if w >= GALLERY_MIN_W}
    if not keep:
        return []
    fname_to_path = {}
    for pm in PATHFNAME.finditer(h):
        fname_to_path.setdefault(pm.group(2), pm.group(1))
    return sorted(fname_to_path[f] for f in keep if f in fname_to_path)

for i, path in enumerate(urls):
    url = BASE + path
    prod = None
    for attempt in range(3):
        try:
            h = fetch(url)
        except Exception as e:
            sys.stderr.write(f"ERR {path}: {e}\n"); time.sleep(1); continue
        for blk in JSONLD.findall(h):
            try:
                d = json.loads(blk.strip())
            except Exception:
                continue
            if isinstance(d, dict) and d.get("@type") == "Product":
                prod = d; break
        if prod: break
        time.sleep(1.2)  # cold-cache shell: pause lets SSR prerender warm
    if not prod:
        sys.stderr.write(f"NOLD {path}\n"); continue
    name = html.unescape(prod.get("name", "").strip())
    sku = prod.get("sku", "").strip().upper()
    pref = (re.match(r"(SAW\d{4})", sku) or [None, None])[1] if sku else None
    pattern = pattern_map.get(pref) or (name.split()[0] if name else None)
    color = derive_color(name, pattern)
    imgs = prod.get("image") or []
    if isinstance(imgs, str): imgs = [imgs]
    image = imgs[0] if imgs else None
    offers = prod.get("offers") or {}
    price = offers.get("price"); cur = offers.get("priceCurrency")
    gallery_paths = own_gallery(h)
    if gallery_paths:
        gallery = sorted(set(BASE + "/" + m for m in gallery_paths))
    else:
        # fallback: no >=1500w own-gallery images found on this PDP shape -- better to
        # carry only the product's own primary image than to fall back to the old
        # whole-page sweep (which is exactly the contamination bug this fixes).
        gallery = [image] if image else []
    settle = "review" if path in settlement else "clear"
    rows.append({
        "mfr_sku": sku, "pattern_name": pattern, "color_name": color,
        "collection": pattern, "image_url": image, "product_url": url,
        "product_type": "Wallcovering", "price_gbp": price, "currency": cur,
        "settlement_flag": settle,
        "gallery_images": "{" + ",".join('"'+g+'"' for g in gallery[:8]) + "}" if gallery else "{}",
    })
    if (i+1) % 50 == 0:
        sys.stderr.write(f"  ...{i+1}/{len(urls)}\n")
    time.sleep(0.12)

cols = ["mfr_sku","pattern_name","color_name","collection","image_url","product_url",
        "product_type","price_gbp","currency","settlement_flag","gallery_images"]
with open(OUTFILE, "w", newline="") as f:
    w = csv.DictWriter(f, fieldnames=cols)
    w.writeheader()
    for r in rows: w.writerow(r)
sys.stderr.write(f"DONE: {len(rows)} rows -> {OUTFILE}\n")