`scripts/extract.py`

111 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)')

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 = sorted(set(BASE + "/" + m for m in IMG.findall(h)))
    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")