`load_catalog_to_pg.py`

91 lines
#!/usr/bin/env python3
"""Load kravet_image_catalog.jsonl into dw_unified.kravet_sku_images."""
import json, os, subprocess, csv, tempfile, re

SSH = ["ssh", "root@45.61.58.125"]
JSONL = "kravet_image_catalog.jsonl"


def pg_exec(sql):
    subprocess.run(SSH + [
        f'PGPASSWORD=DW2024! psql -h 127.0.0.1 -U dw_admin -d dw_unified -c "{sql}"'],
        capture_output=True, text=True, check=True)


def norm_sku(s):
    return re.sub(r"\.0$", "", re.sub(r"-", ".", s.strip().upper()))


def canonical_url(u):
    """Strip size params for dedup key."""
    return re.sub(r"[?&](width|height|pad)=\w+", "", u).rstrip("?&")


# Setup schema
pg_exec("""
  CREATE TABLE IF NOT EXISTS kravet_sku_images (
    id               serial PRIMARY KEY,
    mfr_sku_norm     text NOT NULL,
    mfr_sku_raw      text NOT NULL,
    url              text NOT NULL,
    url_canonical    text NOT NULL,
    kind             text NOT NULL,
    source           text NOT NULL,
    product_url      text,
    product_name     text,
    created_at       timestamptz DEFAULT now()
  );
  CREATE INDEX IF NOT EXISTS idx_kravet_sku_images_norm ON kravet_sku_images(mfr_sku_norm);
  CREATE INDEX IF NOT EXISTS idx_kravet_sku_images_kind ON kravet_sku_images(kind);
  CREATE UNIQUE INDEX IF NOT EXISTS ux_kravet_sku_images_norm_canon
    ON kravet_sku_images(mfr_sku_norm, url_canonical);
""")

# Write rows to CSV for COPY
rows = 0
skipped_dupes = 0
seen = set()
with tempfile.NamedTemporaryFile("w", suffix=".csv", delete=False, newline="") as tf:
    w = csv.writer(tf)
    with open(JSONL) as f:
        for line in f:
            r = json.loads(line)
            if r["status"] != "ok":
                continue
            sku = r["sku"]
            norm = norm_sku(sku)
            alg = r.get("algolia") or {}
            purl = alg.get("url")
            pname = alg.get("name")
            for u in r["urls"]:
                cu = canonical_url(u["url"])
                key = (norm, cu)
                if key in seen:
                    skipped_dupes += 1
                    continue
                seen.add(key)
                w.writerow([norm, sku, u["url"], cu, u["kind"], u["source"], purl or "", pname or ""])
                rows += 1
    tmp = tf.name

print(f"csv: {rows:,} rows ({skipped_dupes:,} in-file dupes)")

# Ship to Kamatera and COPY in
subprocess.run(["scp", tmp, "root@45.61.58.125:/tmp/kravet_images_load.csv"], check=True)
pg_exec("TRUNCATE kravet_sku_images")
subprocess.run(SSH + ["""PGPASSWORD=DW2024! psql -h 127.0.0.1 -U dw_admin -d dw_unified <<'SQL'
\\copy kravet_sku_images (mfr_sku_norm, mfr_sku_raw, url, url_canonical, kind, source, product_url, product_name) FROM '/tmp/kravet_images_load.csv' WITH (FORMAT csv);
SELECT COUNT(*) AS rows, COUNT(DISTINCT mfr_sku_norm) AS skus FROM kravet_sku_images;
SELECT kind, COUNT(*) FROM kravet_sku_images GROUP BY kind ORDER BY COUNT(*) DESC;
SELECT 'coverage' AS metric, COUNT(DISTINCT s.mfr_sku_norm) AS sheet_wallcovers,
       COUNT(DISTINCT i.mfr_sku_norm) AS skus_with_at_least_one_image,
       COUNT(DISTINCT s.mfr_sku_norm) - COUNT(DISTINCT i.mfr_sku_norm) AS missing
  FROM kravet_sheet_fields s
  LEFT JOIN kravet_sku_images i ON i.mfr_sku_norm = s.mfr_sku_norm
 WHERE upper(COALESCE(s.use,'')) LIKE '%WALL%';
SQL
"""], check=True)

os.unlink(tmp)
print("done")