`load_staging.py`

101 lines
#!/usr/bin/env python3
"""
Rebuild dw_unified.stroheim_catalog from TRUE Stroheim rows only.
- Filters listing.jsonl to brand=='stroheim' (the /stroheim/ portal search is a shared
  Fabricut-FAMILY index; only /stroheim/ href rows are actually Stroheim-branded).
- Joins full-page specs from details.jsonl (width/repeat/material).
- Dedups vs fabricut_catalog (DWFC) + dw_sku_registry; flags dedup_status.
- Wipes prior (cross-brand-contaminated) rows and reloads clean.
- Does NOT mint canonical dw_skus / touch dw_sku_registry / shopify_products.
"""
import json, subprocess, os, tempfile

PGENV = {**os.environ, "PGHOST": "/tmp", "PGDATABASE": "dw_unified"}

DDL = """
CREATE TABLE IF NOT EXISTS stroheim_catalog (
  id SERIAL PRIMARY KEY,
  mfr_sku TEXT UNIQUE,
  proposed_dw_sku TEXT,
  existing_dw_sku TEXT,
  dedup_status TEXT,
  brand TEXT DEFAULT 'Stroheim',
  collection TEXT DEFAULT 'Stroheim',
  pattern_name TEXT,
  color_name TEXT,
  product_type TEXT DEFAULT 'Wallcovering',
  width TEXT, repeat_v TEXT, repeat_h TEXT, material TEXT, match_type TEXT,
  price_retail NUMERIC(10,2), price_trade NUMERIC(10,2),
  image_url TEXT, thumb_url TEXT, product_url TEXT,
  specs_enriched BOOLEAN DEFAULT false,
  source TEXT DEFAULT 'fabricut.com/stroheim (feed-first SSR, $0)',
  last_scraped TIMESTAMPTZ DEFAULT now(),
  created_at TIMESTAMPTZ DEFAULT now()
);
"""

def run(sql, inp=None):
    return subprocess.run(["psql","-v","ON_ERROR_STOP=1","-q"] + ([] if inp else ["-c",sql]),
                          env=PGENV, text=True, input=(sql if inp else None), capture_output=True)

def q(sql):
    return subprocess.run(["psql","-tAc",sql], env=PGENV, text=True, capture_output=True).stdout.strip()

def main():
    r = subprocess.run(["psql","-v","ON_ERROR_STOP=1","-q","-c",DDL], env=PGENV, text=True, capture_output=True)
    if r.returncode: print("DDL ERR", r.stderr); return

    details = {}
    for l in open("data/details.jsonl"):
        d = json.loads(l); details[d["mfr_sku"]] = d
    rows = [json.loads(l) for l in open("data/listing.jsonl")]
    stro = [r for r in rows if r.get("brand") == "stroheim"]
    print(f"true Stroheim rows: {len(stro)}")

    def c(x):
        return (str(x) if x else "").replace("\t", " ").replace("\n", " ").replace("\r", " ")

    with tempfile.NamedTemporaryFile("w", suffix=".tsv", delete=False) as tf:
        for r_ in stro:
            d = details.get(r_["mfr_sku"], {})
            enriched = "t" if d.get("width") else "f"
            tf.write("\t".join([c(r_["mfr_sku"]), c(r_["pattern_name"]), c(r_["color_name"]),
                                c(r_["product_type"] or "Wallcovering"),
                                c(d.get("width")), c(d.get("repeat_v")), c(d.get("repeat_h")),
                                c(d.get("material")), c(d.get("match_type")),
                                c(r_["image_url"]), c(r_["thumb_url"]), c(r_["product_url"]),
                                enriched]) + "\n")
        tsv = tf.name

    load = f"""
    TRUNCATE stroheim_catalog RESTART IDENTITY;
    CREATE UNLOGGED TABLE _stro_load (mfr_sku text, pattern_name text, color_name text,
      product_type text, width text, repeat_v text, repeat_h text, material text,
      match_type text, image_url text, thumb_url text, product_url text, specs_enriched boolean);
    \\copy _stro_load FROM '{tsv}'
    INSERT INTO stroheim_catalog
      (mfr_sku, pattern_name, color_name, product_type, width, repeat_v, repeat_h, material,
       match_type, image_url, thumb_url, product_url, specs_enriched,
       existing_dw_sku, dedup_status)
    SELECT l.mfr_sku, l.pattern_name, l.color_name,
           initcap(l.product_type), nullif(l.width,''), nullif(l.repeat_v,''),
           nullif(l.repeat_h,''), nullif(l.material,''), nullif(l.match_type,''),
           l.image_url, l.thumb_url, l.product_url, l.specs_enriched,
           f.dw_sku,
           CASE WHEN f.mfr_sku IS NOT NULL THEN 'existing_dwfc' ELSE 'net_new' END
    FROM _stro_load l
    LEFT JOIN fabricut_catalog f ON f.mfr_sku = l.mfr_sku;
    DROP TABLE _stro_load;
    """
    p = subprocess.run(["psql","-v","ON_ERROR_STOP=1","-q"], env=PGENV, text=True, input=load, capture_output=True)
    if p.returncode: print("LOAD ERR", p.stderr)
    os.unlink(tsv)

    print("=== dedup_status ==="); print(q("SELECT dedup_status, count(*) FROM stroheim_catalog GROUP BY 1 ORDER BY 2 DESC"))
    print("=== specs enriched ==="); print(q("SELECT specs_enriched, count(*) FROM stroheim_catalog GROUP BY 1"))
    print("=== product_type ==="); print(q("SELECT product_type, count(*) FROM stroheim_catalog GROUP BY 1"))
    print("=== distinct patterns ==="); print(q("SELECT count(DISTINCT pattern_name) FROM stroheim_catalog"))

if __name__ == "__main__":
    main()