`scrape_details.py`

86 lines
#!/usr/bin/env python3
"""
Stroheim detail-spec enrichment — feed-first, $0 local.
Reads data/listing.jsonl, fetches each public detail page, extracts
width / repeat / content(material) / book(collection) / type.
Writes data/details.jsonl  (append-safe: skips skus already done).
"""
import re, json, sys, os, time
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")

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:
            if t == tries - 1:
                return ""
            time.sleep(1.2 * (t + 1))
    return ""

def spec(html, *labels):
    """Find a spec value following any of the label words in a tech block."""
    for lab in labels:
        m = re.search(rf'{lab}\s*[:<][^>]*>?\s*</?[^>]*>?\s*([^<]{{1,60}})', html, re.I)
        if m:
            v = re.sub(r'\s+', ' ', m.group(1)).strip(" : ")
            if v and v.lower() not in ('', 'n/a'):
                return v
    return None

def parse_detail(html):
    # narrow to product tech / specs region if present
    tech = ''
    i = html.find('product_tech')
    if i != -1:
        tech = html[i:i+4000]
    region = tech or html
    d = {}
    d['width'] = spec(region, 'Width')
    d['repeat_v'] = spec(region, 'Vertical Repeat', 'Repeat')
    d['repeat_h'] = spec(region, 'Horizontal Repeat')
    d['material'] = spec(region, 'Content', 'Composition', 'Material')
    d['match_type'] = spec(region, 'Match')
    d['book'] = spec(region, 'Book', 'Collection')
    # product type from breadcrumb/type label
    tm = re.search(r'<div class="type">\s*([^<]+?)\s*</div>', html)
    d['type_label'] = re.sub(r'\s+', ' ', tm.group(1)).strip() if tm else None
    return d

def main():
    listing = [json.loads(l) for l in open('data/listing.jsonl')]
    done = set()
    if os.path.exists('data/details.jsonl'):
        for l in open('data/details.jsonl'):
            try:
                done.add(json.loads(l)['mfr_sku'])
            except (json.JSONDecodeError, KeyError):
                pass
    todo = [r for r in listing if r['mfr_sku'] not in done]
    sys.stderr.write(f"details: {len(todo)} to fetch ({len(done)} already done)\n")
    n = 0
    with open('data/details.jsonl', 'a') as out:
        with ThreadPoolExecutor(max_workers=10) as ex:
            futs = {ex.submit(fetch, r['product_url']): r for r in todo}
            for f in as_completed(futs):
                r = futs[f]
                html = f.result()
                rec = {"mfr_sku": r['mfr_sku']}
                if html:
                    rec.update(parse_detail(html))
                out.write(json.dumps(rec) + "\n")
                out.flush()
                n += 1
                if n % 200 == 0:
                    sys.stderr.write(f"  {n}/{len(todo)}\n")
    print(f"DONE details: {n} fetched")

if __name__ == "__main__":
    main()