`fetch_stage.py`

129 lines
#!/usr/bin/env python3
"""Christian Fischbacher feed-first onboarder (Magento GraphQL, $0).
Pulls the CF fabric line + Wall Coverings from fischbacher1819.com's open GraphQL
endpoint, classifies wallcovering-vs-fabric, and stages into dw_unified.fischbacher_catalog.
SCRAPE + STAGE ONLY — no Shopify push.
"""
import csv, io, json, time, urllib.request, subprocess, sys

GQL = "https://fischbacher1819.com/graphql"
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36"

# (category_uid, product_type, line, priority) — all vendor "Christian Fischbacher".
# Higher priority wins when a SKU is cross-listed (dedup on mfr_sku, not brand):
# most-specific sub-line beats the main CF fabric bucket.
CATS = [
    ("MTA=", "Fabric",       "Benu Recycled",         3),
    ("OQ==", "Fabric",       "Collezione Italia",     2),
    ("NDI=", "Wallcovering", "Wall Coverings",        2),
    ("MzA=", "Fabric",       "Christian Fischbacher", 1),
]

def gql(query):
    req = urllib.request.Request(GQL, data=json.dumps({"query": query}).encode(),
                                 headers={"Content-Type": "application/json", "User-Agent": UA})
    with urllib.request.urlopen(req, timeout=45) as r:
        return json.load(r)

def fetch_category(uid):
    items, page = [], 1
    while True:
        q = ('{products(filter:{category_uid:{eq:"%s"}},pageSize:50,currentPage:%d){'
             'total_count items{sku name url_key description{html} short_description{html} '
             'price_range{minimum_price{regular_price{value currency}}} '
             'image{url label} media_gallery{url} categories{name} '
             '...on ConfigurableProduct{variants{product{sku name}}}}}}' % (uid, page))
        d = gql(q)["data"]["products"]
        items.extend(d["items"])
        total = d["total_count"]
        if len(items) >= total or not d["items"]:
            break
        page += 1
        time.sleep(0.4)
    return items

by_sku = {}  # mfr_sku -> (priority, row)
for uid, ptype, line, prio in CATS:
    got = fetch_category(uid)
    print(f"  {line}: {len(got)} patterns", file=sys.stderr)
    for p in got:
        sku = p["sku"]
        if sku in by_sku and by_sku[sku][0] >= prio:
            continue  # keep the more-specific line already recorded
        variants = [v["product"]["sku"] for v in (p.get("variants") or []) if v.get("product")]
        imgs = [g["url"] for g in (p.get("media_gallery") or []) if g.get("url")]
        primary = (p.get("image") or {}).get("url") or (imgs[0] if imgs else None)
        price = ((p.get("price_range") or {}).get("minimum_price") or {}).get("regular_price") or {}
        by_sku[sku] = (prio, {
            "mfr_sku": sku,
            "pattern_name": p["name"],
            "product_type": ptype,
            "source_category": line,
            "pdp_url": f"https://fischbacher1819.com/{p['url_key']}.html" if p.get("url_key") else None,
            "image_url": primary,
            "all_images": imgs,
            "colorways": variants,
            "colorway_count": len(variants),
            "description_html": (p.get("description") or {}).get("html") or "",
            "short_description": (p.get("short_description") or {}).get("html") or "",
            "price_value": price.get("value"),
            "price_currency": price.get("currency"),
        })

rows = [r for _, r in by_sku.values()]
with open("data/fischbacher_staged.jsonl", "w") as f:
    for r in rows:
        f.write(json.dumps(r) + "\n")
print(f"TOTAL rows: {len(rows)}", file=sys.stderr)

# --- stage to dw_unified ---
ddl = """
CREATE TABLE IF NOT EXISTS fischbacher_catalog (
  id             SERIAL PRIMARY KEY,
  vendor         TEXT NOT NULL DEFAULT 'Christian Fischbacher',
  mfr_sku        TEXT UNIQUE NOT NULL,
  pattern_name   TEXT,
  product_type   TEXT,                 -- Wallcovering | Fabric
  source_category TEXT,
  pdp_url        TEXT,
  image_url      TEXT,
  all_images     JSONB,
  colorways      JSONB,
  colorway_count INT,
  description_html TEXT,
  short_description TEXT,
  price_value    NUMERIC,
  price_currency TEXT,
  palette        JSONB,                -- populated by enrich-ai-tags (pending, paid)
  scraped_at     TIMESTAMPTZ DEFAULT now(),
  method         TEXT DEFAULT 'feed-first:magento-graphql'
);
"""
def psql(sql, inp=None):
    return subprocess.run(["psql", "host=/tmp dbname=dw_unified", "-v", "ON_ERROR_STOP=1", "-c", sql],
                          input=inp, capture_output=True, text=True)

r = psql(ddl)
if r.returncode: print("DDL ERROR:", r.stderr, file=sys.stderr); sys.exit(1)

# Build a \copy via COPY FROM STDIN as CSV
buf = io.StringIO()
w = csv.writer(buf)
for row in rows:
    w.writerow([
        row["mfr_sku"], row["pattern_name"], row["product_type"], row["source_category"],
        row["pdp_url"], row["image_url"], json.dumps(row["all_images"]), json.dumps(row["colorways"]),
        row["colorway_count"], row["description_html"], row["short_description"],
        row["price_value"] if row["price_value"] is not None else "",
        row["price_currency"] or "",
    ])
copy_sql = ("BEGIN; TRUNCATE fischbacher_catalog RESTART IDENTITY; "
            "COPY fischbacher_catalog(mfr_sku,pattern_name,product_type,source_category,pdp_url,"
            "image_url,all_images,colorways,colorway_count,description_html,short_description,"
            "price_value,price_currency) FROM STDIN WITH (FORMAT csv); COMMIT;")
p = subprocess.run(["psql", "host=/tmp dbname=dw_unified", "-v", "ON_ERROR_STOP=1", "-c", copy_sql],
                   input=buf.getvalue(), capture_output=True, text=True)
print("COPY stdout:", p.stdout.strip(), file=sys.stderr)
if p.returncode: print("COPY ERROR:", p.stderr, file=sys.stderr); sys.exit(1)
print("STAGED OK", file=sys.stderr)