`jd-combined-cost-load.py`

169 lines
#!/usr/bin/env python3
"""
JD COMBINED-COST LOADER — load the 1,934 Justin-David private-label costs DURABLY
+ make them priceable + queue them. Steve-authorized write (officers/vp-dw-commerce,
2026-06-21). Source = out/jd-combined-cost-diff.json (will_load=true rows only).

WHY THIS IS DIFFERENT FROM batch-1 (cost-relink-load.py):
  Every JD product (Shopify vendors "Los Angeles Fabrics" / "Phillipe Romano" /
  "Architectural Wallcoverings") has dw_sku = NULL everywhere — shopify_products,
  active_five_field_gaps, active_five_field_status. So the batch-1 path
  (product_map keyed by the REAL dw_sku) is unavailable. DTD verdict C (3/3,
  2026-06-21) chose: reuse product_map, key each row by a DETERMINISTIC SYNTHETIC
  dw_sku = the mfr_sku string (JD mfr_skus are unique-by-pattern-color, satisfy
  the PK, replication-safe upsert), populate product_map.mfr_sku, and add ONE
  mfr_sku LEFT JOIN to the two audit read paths (priceability.sql +
  resolve-priceability.js) so the 5-field audit resolves these by mfr_sku after a
  rebuild. Those read-path edits are committed alongside this loader.

KEYS:
  - product_map (durable retail): 1,562 DISTINCT mfr_skus (367 mfr_skus map to
    >1 Shopify product; cost is pattern-level so de-dupe to one map_price each).
    dw_sku = mfr_sku (synthetic), map_price = computed retail, mfr_sku populated,
    map_source = 'jd-pointe-pricelist-2026-06', map_unit='ROLL'. UPSERT on PK.
  - bulk_fivefield_worklist (queue): all 1,934 DISTINCT shopify_ids
    (status='pending', fix_type='build-roll', computed_price = retail, dw_sku
    left blank since none exists). INSERT-only, replication-safe, idempotent by
    shopify_id.

HOLD (untouched): the 114 discontinued/MTO + 47 unmatched = 161 will_load=false rows.
NO Shopify writes. NO variant builds. NO activation. The drain handles that on its
own Steve-gated schedule. NEVER expose Pointe / Justin David / Command54 customer-
facing (private label).

Run with --apply to write; default is dry-run.
"""
import json, os, re, subprocess, sys, datetime

ROOT = os.path.dirname(os.path.abspath(__file__))
DIFF = os.path.join(ROOT, "out", "jd-combined-cost-diff.json")
REPORT = os.path.join(ROOT, "out", "jd-combined-cost-load-report.json")
APPLY = "--apply" in sys.argv
SOURCE_TAG = "jd-pointe-pricelist-2026-06"


def psql(sql, want_rows=True):
    env = dict(os.environ); env.pop("PGPASSWORD", None); env.pop("DATABASE_URL", None)
    r = subprocess.run(["psql", "-d", "dw_unified", "-v", "ON_ERROR_STOP=1",
                        "-tAF", "\t", "-f", "-"],
                       input=sql, capture_output=True, text=True, env=env)
    if r.returncode != 0:
        raise RuntimeError(f"psql failed: {r.stderr.strip()[:800]}")
    if not want_rows:
        return r.stdout
    return [ln.split("\t") for ln in r.stdout.split("\n") if ln != ""]


def q(v):
    return "'" + str(v).replace("'", "''") + "'"


def main():
    d = json.load(open(DIFF))
    rows = d["rows"]
    will = [r for r in rows if r.get("will_load")]
    held = [r for r in rows if not r.get("will_load")]
    assert len(will) == 1934, f"expected 1934 will_load, got {len(will)}"
    assert len(held) == 161, f"expected 161 held, got {len(held)}"

    # ---- product_map upsert set: ONE row per distinct mfr_sku (synthetic dw_sku) ----
    # cost is pattern-level so all dupes share map_price; de-dupe, assert no conflict.
    pm = {}
    for r in will:
        mfr = r["mfr_sku"]
        price = r["proposed_retail"]
        if mfr in pm and abs(pm[mfr]["map_price"] - price) > 1e-9:
            raise RuntimeError(f"conflicting retail for {mfr}: {pm[mfr]['map_price']} vs {price}")
        pm.setdefault(mfr, {
            "dw_sku": mfr,                       # synthetic PK = mfr_sku (DTD-C)
            "sku": mfr,
            "vendor": r["vendor"],
            "pattern": (r.get("matched_pattern") or ""),
            "color": "",
            "map_price": price,
            "mfr_sku": mfr,
        })
    pm_rows = list(pm.values())

    # ---- worklist queue set: one row per distinct shopify_id ----
    queue = [{
        "shopify_id": r["shopify_id"], "dw_sku": (r.get("dw_sku") or ""),
        "mfr_sku": r["mfr_sku"], "vendor": r["vendor"],
        "computed_price": r["proposed_retail"],
    } for r in will]

    print(f"will_load rows:               {len(will)}", file=sys.stderr)
    print(f"  -> product_map upserts:     {len(pm_rows)} (distinct mfr_sku)", file=sys.stderr)
    print(f"  -> worklist queue rows:     {len(queue)} (distinct shopify_id)", file=sys.stderr)
    print(f"held (untouched):             {len(held)}", file=sys.stderr)

    report = {
        "generatedAt": datetime.datetime.utcnow().isoformat() + "Z",
        "apply": APPLY, "source_tag": SOURCE_TAG,
        "durable_cost_location": "product_map.map_price (synthetic dw_sku=mfr_sku, "
                                 "mfr_sku populated; audit reads via new mfr_sku LEFT JOIN) "
                                 "[DTD-C 2026-06-21]",
        "will_load": len(will),
        "product_map_upserts": len(pm_rows),
        "worklist_queue": len(queue),
        "held_untouched": len(held),
        "held_breakdown": {"discontinued_or_mto": 114, "unmatched": 47},
        "retail_range": [min(x["map_price"] for x in pm_rows), max(x["map_price"] for x in pm_rows)],
    }

    if not APPLY:
        report["mode"] = "DRY-RUN (no writes). Re-run with --apply."
        json.dump(report, open(REPORT, "w"), indent=2)
        print(json.dumps(report, indent=2))
        return

    # ---- WRITE 1: UPSERT into product_map (durable, replication-safe via PK) ----
    vals = []
    for x in pm_rows:
        vals.append("(" + ",".join([
            q(x["dw_sku"]), q(x["sku"]), q(x["vendor"]), q(x["pattern"]),
            q(x["color"]), str(x["map_price"]), q("ROLL"), q(SOURCE_TAG),
            "now()", q(x["mfr_sku"])]) + ")")
    upsert = (
        "BEGIN;\n"
        "INSERT INTO product_map (dw_sku, sku, vendor, pattern, color, map_price, "
        "map_unit, map_source, updated_at, mfr_sku) VALUES\n"
        + ",\n".join(vals) +
        "\nON CONFLICT (dw_sku) DO UPDATE SET "
        "map_price=EXCLUDED.map_price, map_source=EXCLUDED.map_source, "
        "map_unit=EXCLUDED.map_unit, updated_at=EXCLUDED.updated_at, "
        "vendor=EXCLUDED.vendor, pattern=EXCLUDED.pattern, "
        "mfr_sku=COALESCE(NULLIF(product_map.mfr_sku,''),EXCLUDED.mfr_sku);\n"
        "COMMIT;")
    psql(upsert, want_rows=False)
    print(f"product_map UPSERT done: {len(pm_rows)} rows", file=sys.stderr)

    # ---- WRITE 2: INSERT into bulk_fivefield_worklist (INSERT-only, repl-safe) ----
    existing = set(r[0] for r in psql(
        "SELECT DISTINCT shopify_id FROM bulk_fivefield_worklist;"))
    base = int(psql("SELECT COALESCE(max(order_rank),0) FROM bulk_fivefield_worklist;")[0][0])
    qvals = []; rank = base; seen = set()
    for x in queue:
        if x["shopify_id"] in existing or x["shopify_id"] in seen:
            continue
        seen.add(x["shopify_id"]); rank += 1
        qvals.append("(" + ",".join([
            str(rank), q(x["shopify_id"]), q(x["dw_sku"]), q(x["mfr_sku"]),
            q(x["vendor"]), q("build-roll"), "false",
            str(x["computed_price"]), "NULL", q("pending")]) + ")")
    if qvals:
        ins = ("BEGIN;\nINSERT INTO bulk_fivefield_worklist "
               "(order_rank, shopify_id, dw_sku, mfr_sku, vendor, fix_type, "
               "is_kravet_fam, computed_price, kravet_price, status) VALUES\n"
               + ",\n".join(qvals) + ";\nCOMMIT;")
        psql(ins, want_rows=False)
    report["worklist_inserted"] = len(qvals)
    report["worklist_skipped_existing"] = len(queue) - len(qvals)
    report["mode"] = "APPLIED"
    json.dump(report, open(REPORT, "w"), indent=2)
    print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()