← back to Dw Five Field Step0
cost-relink-load.py
172 lines
#!/usr/bin/env python3
"""
Part A LOADER — load the 1,202 high-confidence sourced costs DURABLY + queue them.
Steve-authorized write (officers/vp-dw-commerce, 2026-06-21). DTD verdict A.
Durable cost location (DTD A, 2/2): product_map.map_price keyed by dw_sku,
map_source='fivefield-relink-2026-06-21'. The 5-field audit (priceability.sql /
resolve-priceability.js) reads product_map.map_price by dw_sku as a FINAL retail
price for non-Kravet rows -> re-running the audit tomorrow keeps these priceable.
product_map has a PRIMARY KEY (dw_sku) so UPSERT is replication-safe under the
all-tables publication dw_unified_pub (the gaps/worklist tables are not).
Then INSERT the 1,202 into bulk_fivefield_worklist (status='pending',
fix_type='build-roll', computed_price set) for the activation drain's rr
round-robin. INSERT is replication-safe even without a PK (no replica identity
needed for INSERT — verified).
Authoritative keys come from active_five_field_gaps (real dw_sku + shopify_id +
is_kravet_fam), NOT the diff's coalesced dw_sku field. Cost/price come from the
diff. Match is on (vendor, mfr_sku) which is unique within the flagged set.
NO Shopify writes. NO activation. NO UPDATE on the gaps table (replica-identity
blocked). 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", "cost-relink-diff.json")
APPLY = "--apply" in sys.argv
SOURCE_TAG = "fivefield-relink-2026-06-21"
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()[:600]}")
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))
hi = [r for r in d["diff"] if r.get("match_confidence") in ("exact", "fuzzy-high")]
assert len(hi) == 1202, f"expected 1202 high-confidence, got {len(hi)}"
# Re-derive authoritative keys from the gaps table by (vendor, mfr_sku).
# Build a lookup of every flagged (vendor, upper(mfr_sku)) -> (dw_sku, shopify_id, is_kravet_fam)
rows = psql("""
SELECT vendor, upper(mfr_sku), COALESCE(NULLIF(dw_sku,''),''),
COALESCE(shopify_id,''), is_kravet_fam
FROM active_five_field_gaps
WHERE priceable IS NOT TRUE
AND vendor IN ('Scalamandre Wallpaper','Rebel Walls','Osborne & Little');""")
gap = {}
for vendor, mfr, dw, sid, kf in rows:
gap[(vendor, mfr)] = (dw, sid, kf == "t")
load = [] # for product_map upsert (needs real dw_sku)
queue = [] # for worklist insert (needs shopify_id + real dw_sku)
no_dw, no_match = [], []
for r in hi:
key = (r["vendor"], (r["mfr_sku"] or "").upper())
g = gap.get(key)
if not g:
no_match.append(r); continue
dw, sid, kf = g
price = r["computed_price"]
if not dw:
# real dw_sku is blank -> cannot key product_map (PK is dw_sku).
# These are EXACT mfr_sku catalog matches (durable via catalog mfr_sku
# join already), so they STILL get queued — just no product_map write.
no_dw.append({**r, "shopify_id": sid})
else:
load.append({"dw_sku": dw, "sku": r["mfr_sku"], "vendor": r["vendor"],
"pattern": r.get("matched_pattern") or r.get("flagged_pattern") or "",
"color": "", "map_price": price, "mfr_sku": r["mfr_sku"]})
queue.append({"shopify_id": sid, "dw_sku": dw, "mfr_sku": r["mfr_sku"],
"vendor": r["vendor"], "is_kravet_fam": kf,
"computed_price": price})
print(f"high-confidence rows: {len(hi)}", file=sys.stderr)
print(f"matched to gaps keys: {len(hi)-len(no_match)}", file=sys.stderr)
print(f" -> loadable (real dw_sku): {len(load)}", file=sys.stderr)
print(f" -> blank real dw_sku: {len(no_dw)}", file=sys.stderr)
print(f" -> no gaps match: {len(no_match)}", file=sys.stderr)
# de-dupe product_map upsert by dw_sku (keep first; should already be unique)
seen = set(); uniq_load = []
for x in load:
if x["dw_sku"] in seen: continue
seen.add(x["dw_sku"]); uniq_load.append(x)
report = {
"generatedAt": datetime.datetime.utcnow().isoformat() + "Z",
"apply": APPLY, "source_tag": SOURCE_TAG,
"high_confidence": len(hi),
"loadable_product_map": len(uniq_load),
"queue_worklist": len(queue),
"blank_real_dwsku": len(no_dw),
"no_gaps_match": len(no_match),
"blank_dwsku_samples": [{"vendor": r["vendor"], "mfr_sku": r["mfr_sku"],
"shopify_id": r["shopify_id"]} for r in no_dw[:20]],
"no_match_samples": [{"vendor": r["vendor"], "mfr_sku": r["mfr_sku"]} for r in no_match[:20]],
}
if not APPLY:
report["mode"] = "DRY-RUN (no writes). Re-run with --apply."
json.dump(report, open(os.path.join(ROOT, "out", "cost-relink-load-report.json"), "w"), indent=2)
print(json.dumps(report, indent=2))
return
# ---- WRITE 1: UPSERT into product_map (durable, replication-safe via PK) ----
# Single transaction. ON CONFLICT(dw_sku) DO UPDATE — idempotent.
vals = []
now = "now()"
for x in uniq_load:
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, "
"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(uniq_load)} rows", file=sys.stderr)
# ---- WRITE 2: INSERT into bulk_fivefield_worklist (INSERT-only, repl-safe) ----
# Skip any shopify_id already in the worklist (idempotent re-run guard).
existing = set(r[0] for r in psql(
"SELECT DISTINCT shopify_id FROM bulk_fivefield_worklist;"))
maxrank = psql("SELECT COALESCE(max(order_rank),0) FROM bulk_fivefield_worklist;")
base = int(maxrank[0][0])
qvals = []
rank = base
inserted_ids = set()
for x in queue:
if x["shopify_id"] in existing or x["shopify_id"] in inserted_ids:
continue
inserted_ids.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"),
"true" if x["is_kravet_fam"] else "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(os.path.join(ROOT, "out", "cost-relink-load-report.json"), "w"), indent=2)
print(json.dumps(report, indent=2))
if __name__ == "__main__":
main()