`scrape_listing.py`

95 lines
#!/usr/bin/env python3
"""
Stroheim listing scraper v3 — feed-first, $0 local.
Source: fabricut.com/stroheim/search?type=wallcovering  (public SSR, no login)

IMPORTANT: the /stroheim/ portal search returns the WHOLE Fabricut-family catalog
(Stroheim + Vervain + S.Harris + Trend + Fabricut). Each card's href path encodes
the TRUE brand: /{brand}/wallcovering/{sku}/{pattern}/{color}. We capture brand per
card and let the loader filter to Stroheim-only.
Writes data/listing.jsonl (all brands, brand-tagged).
"""
import re, json, sys, time
from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed
import urllib.request

UA = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 "
      "(KHTML, like Gecko) Version/17 Safari/605.1.15")
BASE = "https://fabricut.com/stroheim/search?type=wallcovering&page={}"

def fetch(url, tries=3):
    for t in range(tries):
        try:
            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")
        except Exception as e:
            if t == tries - 1:
                sys.stderr.write(f"FAIL {url}: {e}\n"); return ""
            time.sleep(1.5 * (t + 1))
    return ""

def unent(s):
    return (s.replace("&amp;", "&").replace("&#39;", "'").replace("&quot;", '"')
             .replace("&lt;", "<").replace("&gt;", ">"))

HREF = re.compile(r'href="(/([a-z0-9\-]+)/(wallcovering|fabric|trimming)/(\d+)/([^"/]+)/([^"/?]+))"')

def parse_page(html):
    out = []
    for chunk in html.split('<div class="product swatch"')[1:]:
        chunk = chunk[:2500]
        m = re.search(r'product_sku="([^"]+)"', chunk)
        if not m:
            continue
        sku = m.group(1).strip()
        hm = HREF.search(chunk)
        if hm:
            href, brand, ptype = hm.group(1), hm.group(2), hm.group(3)
        else:
            href = brand = ptype = ""
        tm = re.search(r'<div class="type">\s*([^<]+?)\s*</div>', chunk)
        type_label = re.sub(r'\s+', ' ', tm.group(1)).strip() if tm else ""
        am = re.search(r'(?:alt|title)="([^"]*)"', chunk)
        alt = unent((am.group(1) if am else "").strip())
        im = re.search(r'data-original="([^"]+)"', chunk)
        img = im.group(1) if im else ""
        pm = re.search(r'<span class="swatch-pattern">(.*?)</span>', chunk, re.S)
        pattern = unent(re.sub(r'\s+', ' ', pm.group(1)).strip()) if pm else ""
        color = ""
        if alt and pattern and alt.lower().startswith(pattern.lower()):
            color = alt[len(pattern):].strip()
        if not color and href:
            color = href.rstrip('/').split('/')[-1].replace('-', ' ').title()
        full = img.replace('/thumbnails/', '/').replace('@2x', '')
        out.append({
            "mfr_sku": sku, "brand": brand, "brand_label": type_label, "product_type": ptype,
            "pattern_name": pattern, "color_name": color, "alt": alt,
            "product_url": ("https://fabricut.com" + href) if href else "",
            "image_url": full, "thumb_url": img,
        })
    return out

def main():
    last = int(sys.argv[1]) if len(sys.argv) > 1 else 168
    results = {}
    with ThreadPoolExecutor(max_workers=8) as ex:
        futs = {ex.submit(fetch, BASE.format(p)): p for p in range(1, last + 1)}
        done = 0
        for f in as_completed(futs):
            for r in (parse_page(f.result()) or []):
                results[r["mfr_sku"]] = r
            done += 1
            if done % 40 == 0 or done == last:
                sys.stderr.write(f"  pages {done}/{last}  unique={len(results)}\n")
    with open("data/listing.jsonl", "w") as fh:
        for r in results.values():
            fh.write(json.dumps(r) + "\n")
    bc = Counter(r["brand"] or "(no-href)" for r in results.values())
    print(f"DONE: {len(results)} unique SKUs -> data/listing.jsonl")
    print("BY BRAND:", dict(bc))

if __name__ == "__main__":
    main()