`scrape_carnegie.py`

169 lines
#!/usr/bin/env python3
"""Carnegie Fabrics feed-first scraper — Magento 2 GraphQL ($0, no browser, no AI).
Stages configurable products + colorway variants into dw_unified.carnegie_catalog.
Classifies wallcovering-vs-fabric by URL/SKU type-suffix. NEVER-DUPLICATE dedup on mfr_sku.
"""
import json, re, sys, time, subprocess, urllib.request

GQL = "https://carnegiefabrics.com/graphql"
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"

# type-suffix -> DW class. Wallcovering = wall-surface products; else Fabric/textile.
WALL_SUFFIXES = {"wallcoverings", "upholstered-walls"}

QUERY = """{ products(search:"", pageSize:%d, currentPage:%d) {
  total_count page_info{ total_pages current_page }
  items {
    sku name url_key __typename
    image{ url } media_gallery{ url }
    price_range{ minimum_price{ regular_price{ value currency } } }
    categories{ name }
    short_description{ html } description{ html }
    ... on ConfigurableProduct {
      variants { product { sku name image{ url } } attributes { label value_index } }
    }
  }
} }"""

def gql(page, size=50, retries=4):
    body = json.dumps({"query": QUERY % (size, page)}).encode()
    for a in range(retries):
        try:
            req = urllib.request.Request(GQL, data=body,
                headers={"Content-Type": "application/json", "User-Agent": UA})
            with urllib.request.urlopen(req, timeout=40) as r:
                d = json.loads(r.read())
            if "errors" in d and not d.get("data"):
                raise RuntimeError(d["errors"])
            return d["data"]["products"]
        except Exception:
            if a == retries - 1:
                raise
            time.sleep(2 * (a + 1))

def suffix_of(url_key, sku):
    # url_key like "mood-6602-windows" ; sku like "6602-windows"
    for src in (url_key or "", sku or ""):
        m = re.search(r"-([a-z-]+)$", src)
        if m:
            return m.group(1)
    return ""

def classify(suffix):
    return "Wallcovering" if suffix in WALL_SUFFIXES else "Fabric"

def sqlq(v):
    if v is None:
        return "NULL"
    if isinstance(v, bool):
        return "true" if v else "false"
    if isinstance(v, (int, float)):
        return str(v)
    if isinstance(v, (dict, list)):
        return "'" + json.dumps(v).replace("'", "''") + "'"
    return "'" + str(v).replace("'", "''") + "'"

def pgarr(lst):
    if not lst:
        return "NULL"
    inner = ",".join('"' + str(x).replace('"', '\\"') + '"' for x in lst)
    return "'{" + inner + "}'"

def main():
    first = gql(1)
    total, pages = first["total_count"], first["page_info"]["total_pages"]
    print(f"[feed] total_count={total} pages={pages} (pageSize=50)", flush=True)
    all_items = list(first["items"])
    for p in range(2, pages + 1):
        d = gql(p)
        all_items.extend(d["items"])
        print(f"[feed] page {p}/{pages} cum={len(all_items)}", flush=True)
        time.sleep(0.4)  # polite

    rows = []          # per-colorway (sellable) SKU rows
    prod_count = 0
    split = {}         # suffix -> count (product level)
    for it in all_items:
        prod_count += 1
        suf = suffix_of(it.get("url_key"), it.get("sku"))
        dwc = classify(suf)
        split[suf] = split.get(suf, 0) + 1
        pnum = re.sub(r"-.*$", "", it.get("sku") or "")
        cats = [c["name"] for c in (it.get("categories") or [])]
        pr = (((it.get("price_range") or {}).get("minimum_price") or {}).get("regular_price") or {})
        price, cur = pr.get("value"), pr.get("currency")
        desc_html = (it.get("description") or {}).get("html") or (it.get("short_description") or {}).get("html")
        desc_txt = re.sub(r"<[^>]+>", "", desc_html or "").strip() or None
        gallery = [g["url"] for g in (it.get("media_gallery") or [])]
        purl = "https://carnegiefabrics.com/" + (it.get("url_key") or "")
        variants = it.get("variants") or []

        def base_row(mfr, cnum, cimg, is_var):
            return dict(mfr_sku=mfr, dw_sku=None, parent_sku=it.get("sku"),
                pattern_number=pnum, pattern_name=it.get("name"),
                color_number=cnum, color_name=(f"Color {cnum}" if cnum else None),
                title=(f"{it.get('name')} {cnum}".strip() if cnum else it.get("name")),
                dw_class=dwc, type_suffix=suf, product_type=(cats[0] if cats else None),
                categories=cats, body_html=desc_html, description_text=desc_txt,
                price=price, currency=cur,
                all_images=([cimg] if cimg else []) + gallery,
                image_url=(cimg or (it.get("image") or {}).get("url")),
                product_url=purl, is_variant=is_var,
                raw={"sku": it.get("sku"), "typename": it.get("__typename")})

        if variants:
            for v in variants:
                vp = v["product"]
                cnum_lbl = (v["attributes"][0]["label"] if v.get("attributes") else None)
                rows.append(base_row(vp["sku"], cnum_lbl, (vp.get("image") or {}).get("url"), True))
        else:
            rows.append(base_row(it.get("sku"), None, (it.get("image") or {}).get("url"), False))

    # dedup on mfr_sku (keep first)
    seen, uniq = set(), []
    for r in rows:
        if r["mfr_sku"] in seen:
            continue
        seen.add(r["mfr_sku"])
        uniq.append(r)

    print(f"[stage] product-level={prod_count} sellable-SKU rows(deduped)={len(uniq)}", flush=True)

    cols = ["mfr_sku","dw_sku","parent_sku","pattern_number","pattern_name","color_number",
            "color_name","title","dw_class","type_suffix","product_type","categories",
            "body_html","description_text","price","currency","all_images","image_url",
            "product_url","is_variant","raw"]
    def val(r, c):
        v = r[c]
        if c in ("categories", "all_images"):
            return pgarr(v)
        return sqlq(v)

    # batch inserts with upsert (NEVER-DUPLICATE)
    BATCH = 200
    inserted = 0
    for i in range(0, len(uniq), BATCH):
        chunk = uniq[i:i+BATCH]
        vals = ",\n".join("(" + ",".join(val(r, c) for c in cols) + ")" for r in chunk)
        sql = (f"INSERT INTO carnegie_catalog ({','.join(cols)}) VALUES\n{vals}\n"
               "ON CONFLICT (mfr_sku) DO UPDATE SET "
               + ", ".join(f"{c}=EXCLUDED.{c}" for c in cols if c != "mfr_sku")
               + ", updated_at=now();")
        p = subprocess.run(["psql", "host=/tmp dbname=dw_unified", "-v", "ON_ERROR_STOP=1", "-q"],
                           input=sql, text=True, capture_output=True)
        if p.returncode != 0:
            sys.stderr.write(p.stderr[:2000])
            raise SystemExit(f"psql failed on batch {i}")
        inserted += len(chunk)
        print(f"[stage] upserted {inserted}/{len(uniq)}", flush=True)

    # summary
    wall = sum(1 for r in uniq if r["dw_class"] == "Wallcovering")
    fab = len(uniq) - wall
    print(json.dumps({"products": prod_count, "sellable_skus": len(uniq),
                      "wallcovering_skus": wall, "fabric_skus": fab,
                      "product_split_by_suffix": split}, indent=2))

if __name__ == "__main__":
    main()