← back to Dw Five Field Step0
5-field drain: vendor round-robin order — activate across as many vendors as possible each day
be75fa3146e099124aff52d0e503bbece3fede16 · 2026-06-21 18:06:17 -0700 · Steve
Files touched
A bulk-fivefield-exec.pyA drain.sh
Diff
commit be75fa3146e099124aff52d0e503bbece3fede16
Author: Steve <steve@designerwallcoverings.com>
Date: Sun Jun 21 18:06:17 2026 -0700
5-field drain: vendor round-robin order — activate across as many vendors as possible each day
---
bulk-fivefield-exec.py | 338 +++++++++++++++++++++++++++++++++++++++++++++++++
drain.sh | 41 ++++++
2 files changed, 379 insertions(+)
diff --git a/bulk-fivefield-exec.py b/bulk-fivefield-exec.py
new file mode 100644
index 0000000..22605c5
--- /dev/null
+++ b/bulk-fivefield-exec.py
@@ -0,0 +1,338 @@
+#!/usr/bin/env python3
+"""
+DW five-field BULK executor — releases the ~17,317 AUTO-FIXABLE priceable cohort.
+
+Built on the proven 20/20 canary (/tmp/threads-canary-exec.py). Same shape + guards:
+ PG-before-Shopify, fetch-live-then-create, idempotent skip-if-exists,
+ MAP + $0 guards BEFORE and AFTER create, vendor sanity, DELETE nothing,
+ per-SKU try/catch, audit JSON, FULLY RESUMABLE from the audit file.
+
+Source of truth = dw_unified.bulk_fivefield_worklist (status='pending'), ordered by order_rank:
+ rank 1..14,425 add-sample -> create {DW_SKU}-Sample @ $4.25, inventory tracking OFF
+ rank 14,426..17,198 build-roll (Kravet-fam) -> create {DW_SKU} @ MAP (computed_price; >= kravet_price)
+ rank 17,199..17,317 build-roll (non-Kravet priceable) -> create {DW_SKU} @ computed_price
+The 33 reprice-zero rows are status='held-no-cost' in the table and are NEVER loaded here
+(no cost -> cannot MAP-price -> no variant-create path). The 30,107 NO-COST HOLD cohort is
+excluded at the SQL source: the worklist was built WHERE priceable=true (Steve's hard gate).
+
+Store = designer-laboratory-sandbox.myshopify.com (Steve-authorized, token last4 75b9). API = 2024-10.
+
+SCALE ADDITIONS over the canary:
+ 1) DAILY BUDGET CAP via the shared ledger budget.cjs ('upload' category, DTD verdict A 2026-06-21):
+ - default --max = live `budget.cjs remaining upload` (today's leftover global headroom)
+ - take('upload', N) UP FRONT to size the batch; refund the unused remainder at the end
+ (refund rule: we debit the whole grant to size the batch, but consume 1 variant per CREATE;
+ skips/errors create nothing -> refund grant-minus-created). Never starves the live cadence;
+ both honor the 884/day ceiling first-come-first-served.
+ - process AT MOST `granted` variant-CREATES this run, then stop. Resumable next run.
+ 2) Per-batch summary line + running total-done / total-remaining (printed live + at end).
+ 3) Idempotency: a re-run after a partial batch skips everything already created (status in
+ audit JSON = fixed/skipped) AND re-checks the live variant list (skip-if-exists) belt-and-braces.
+
+DOES NOT mutate the worklist table. Audit JSON is the resume ledger.
+"""
+import json, urllib.request, urllib.error, subprocess, time, os, sys, argparse
+
+DOMAIN = "designer-laboratory-sandbox.myshopify.com"
+API = "2024-10"
+OUTDIR = "/Users/stevestudio2/Projects/dw-five-field-step0/out"
+RESULT = os.path.join(OUTDIR, "bulk-fivefield-result.json")
+SAMPLE_PRICE = "4.25"
+BUDGET = os.path.expanduser("~/Projects/designerwallcoverings/scripts/variant-budget/budget.cjs")
+BUDGET_CAT = "upload" # DTD verdict A: share the live cadence ledger category
+NODE = "/opt/homebrew/bin/node" if os.path.exists("/opt/homebrew/bin/node") else "node"
+
+tok = subprocess.check_output(
+ "grep -E '^SHOPIFY_ADMIN_TOKEN=' ~/Projects/secrets-manager/.env | head -1 | cut -d= -f2- | tr -d '\"'\"'\"' '",
+ shell=True, executable="/bin/bash").decode().strip()
+HDR = {"X-Shopify-Access-Token": tok, "Content-Type": "application/json"}
+
+
+def shop(method, path, body=None):
+ url = f"https://{DOMAIN}/admin/api/{API}/{path}"
+ data = json.dumps(body).encode() if body is not None else None
+ r = urllib.request.Request(url, data=data, headers=HDR, method=method)
+ try:
+ with urllib.request.urlopen(r, timeout=60) as resp:
+ return resp.status, json.loads(resp.read().decode())
+ except urllib.error.HTTPError as e:
+ try:
+ return e.code, json.loads(e.read().decode())
+ except Exception:
+ return e.code, {"error": "non-json"}
+
+
+# ── budget ledger helpers (fail-open like budget.cjs itself) ──────────────────
+def budget_remaining():
+ try:
+ out = subprocess.check_output([NODE, BUDGET, "remaining", BUDGET_CAT],
+ stderr=subprocess.DEVNULL, timeout=30).decode().strip()
+ n = int(float(out))
+ return max(0, n)
+ except Exception as e:
+ print(f"[budget] WARN remaining() failed ({e}) -> 0 (conservative)")
+ return 0
+
+
+def budget_take(n):
+ if n <= 0:
+ return 0
+ try:
+ out = subprocess.check_output([NODE, BUDGET, "take", BUDGET_CAT, str(n)],
+ stderr=subprocess.DEVNULL, timeout=30).decode().strip()
+ return max(0, int(float(out)))
+ except Exception as e:
+ print(f"[budget] WARN take() failed ({e}) -> granting {n} (fail-open, matches budget.cjs)")
+ return n
+
+
+def budget_refund(n):
+ if n <= 0:
+ return 0
+ try:
+ out = subprocess.check_output([NODE, BUDGET, "refund", BUDGET_CAT, str(n)],
+ stderr=subprocess.DEVNULL, timeout=30).decode().strip()
+ return max(0, int(float(out)))
+ except Exception as e:
+ print(f"[budget] WARN refund() failed ({e}) -> no-op")
+ return 0
+
+
+# ── worklist load (PG source of truth) ────────────────────────────────────────
+def load_worklist():
+ # Emit JSON from psql so embedded |, newlines, quotes in vendor/sku text can't break parsing.
+ # rr = VENDOR ROUND-ROBIN order (Steve: "activate as many as possible across as
+ # many vendors every day"). Processing rr-ascending takes ONE sku from every
+ # vendor before a 2nd from any — so each daily budget-capped batch spreads
+ # across all vendors. Samples-first within each vendor (Steve's sequencing).
+ q = (
+ "SELECT coalesce(json_agg(row_to_json(t) ORDER BY t.rr), '[]'::json) FROM ("
+ " SELECT (row_number() OVER (PARTITION BY vendor ORDER BY (fix_type='add-sample') DESC, dw_sku))*100000"
+ " + dense_rank() OVER (ORDER BY vendor) AS rr, "
+ " shopify_id, dw_sku, mfr_sku, vendor, fix_type, "
+ " is_kravet_fam, computed_price::float8 AS computed_price, "
+ " kravet_price::float8 AS kravet_price "
+ " FROM bulk_fivefield_worklist WHERE status='pending'"
+ ") t;"
+ )
+ raw = subprocess.check_output(
+ ["psql", "-d", "dw_unified", "-t", "-A", "-c", q]).decode()
+ data = json.loads(raw)
+ rows = []
+ for d in data:
+ rows.append({
+ "rr": int(d["rr"]), "shopify_id": d["shopify_id"],
+ "dw_sku": d["dw_sku"], "mfr_sku": d["mfr_sku"], "vendor": d["vendor"],
+ "fix_type": d["fix_type"], "is_kravet_fam": bool(d["is_kravet_fam"]),
+ "computed_price": d["computed_price"], "kravet_price": d["kravet_price"],
+ })
+ rows.sort(key=lambda r: r["rr"])
+ return rows
+
+
+def pid_of(shopify_id):
+ return shopify_id.rsplit("/", 1)[-1]
+
+
+def process(item, results, dry_run=False):
+ """Returns 1 if a variant was CREATED (consumes 1 budget unit), else 0."""
+ pid = pid_of(item["shopify_id"])
+ dw = item["dw_sku"]
+ fix = item["fix_type"]
+ rec = {"rr": item["rr"], "dw_sku": dw, "mfr_sku": item["mfr_sku"],
+ "vendor": item["vendor"], "product_id": pid, "handle": None, "fix": fix,
+ "target_sku": None, "price": None, "before_variants": None,
+ "after_variants": None, "new_variant_id": None, "status": None, "reason": None}
+ try:
+ # --- pre-flight integrity (PG source of truth) ---
+ if fix not in ("add-sample", "build-roll"):
+ rec["status"] = "errored"; rec["reason"] = f"unknown_fix_type:{fix}"
+ results.append(rec); print(f"ERR {dw}: unknown fix {fix}"); return 0
+
+ # --- fetch live ---
+ st, r = shop("GET", f"products/{pid}.json?fields=id,handle,title,status,vendor,variants")
+ if st != 200:
+ rec["status"] = "errored"; rec["reason"] = f"fetch_{st}"; rec["api"] = r
+ results.append(rec); print(f"ERR {dw}: fetch {st}"); return 0
+ p = r.get("product", {})
+ rec["handle"] = p.get("handle")
+ # vendor sanity: live Shopify vendor must match the worklist vendor (not a hardcoded string)
+ live_vendor = (p.get("vendor") or "").strip()
+ wl_vendor = (item["vendor"] or "").strip()
+ if wl_vendor and live_vendor and live_vendor != wl_vendor:
+ rec["status"] = "errored"
+ rec["reason"] = f"vendor_mismatch:live={live_vendor!r}!=worklist={wl_vendor!r}"
+ results.append(rec); print(f"ERR {dw}: vendor {live_vendor!r} != {wl_vendor!r}"); return 0
+
+ variants = p.get("variants", [])
+ rec["before_variants"] = [
+ {"id": v["id"], "sku": v.get("sku"), "price": v.get("price"),
+ "option1": v.get("option1"), "inv_mgmt": v.get("inventory_management")}
+ for v in variants]
+ existing_skus = {(v.get("sku") or "") for v in variants}
+
+ if fix == "add-sample":
+ target = f"{dw}-Sample"
+ price = SAMPLE_PRICE
+ rec["target_sku"] = target; rec["price"] = price
+ if target in existing_skus: # idempotent belt-and-braces
+ rec["status"] = "skipped"; rec["reason"] = "sample_variant_already_exists"
+ rec["after_variants"] = rec["before_variants"]
+ results.append(rec); print(f"SKIP {dw}: sample exists"); return 0
+ vbody = {"variant": {
+ "option1": "Sample", "sku": target, "price": price,
+ "inventory_management": None, "inventory_policy": "continue",
+ "requires_shipping": True, "taxable": True}}
+
+ else: # build-roll
+ target = dw
+ map_price = item["computed_price"]
+ rec["target_sku"] = target; rec["price"] = map_price
+ # $0 / null guard (BEFORE create)
+ if map_price is None or map_price <= 0:
+ rec["status"] = "errored"; rec["reason"] = f"zero_or_null_price:{map_price}"
+ results.append(rec); print(f"ERR {dw}: bad price {map_price}"); return 0
+ # MAP-floor guard (BEFORE create): never below the Kravet MAP
+ if item["kravet_price"] is not None and map_price + 1e-6 < item["kravet_price"]:
+ rec["status"] = "errored"
+ rec["reason"] = f"MAP_breach:{map_price}<kravet_price{item['kravet_price']}"
+ results.append(rec); print(f"ERR {dw}: MAP breach"); return 0
+ if target in existing_skus: # idempotent belt-and-braces
+ rec["status"] = "skipped"; rec["reason"] = "roll_variant_already_exists"
+ rec["after_variants"] = rec["before_variants"]
+ results.append(rec); print(f"SKIP {dw}: roll exists"); return 0
+ vbody = {"variant": {
+ "option1": "Roll", "sku": target, "price": f"{map_price:.2f}",
+ "inventory_management": "shopify", "inventory_policy": "continue",
+ "requires_shipping": True, "taxable": True}}
+
+ if dry_run:
+ rec["status"] = "dryrun"; rec["reason"] = "would_create"
+ results.append(rec)
+ print(f"DRYRUN {dw} [{fix}] -> {rec['target_sku']} @ {rec['price']} (product {pid}, vendor {wl_vendor})")
+ return 0
+
+ # --- create the missing variant ---
+ st2, r2 = shop("POST", f"products/{pid}/variants.json", vbody)
+ if st2 not in (200, 201):
+ rec["status"] = "errored"; rec["reason"] = f"create_{st2}"; rec["api"] = r2
+ results.append(rec); print(f"ERR {dw}: create {st2} {r2}"); return 0
+ nv = r2.get("variant", {})
+ rec["new_variant_id"] = nv.get("id")
+ # AFTER-create price guards
+ created_price = float(nv.get("price") or 0)
+ if created_price <= 0:
+ rec["status"] = "errored"; rec["reason"] = f"created_price_zero:{created_price}"
+ results.append(rec); print(f"ERR {dw}: created price 0"); return 1 # variant WAS created -> consumed budget
+ if fix == "build-roll" and item["kravet_price"] is not None and created_price + 1e-6 < item["kravet_price"]:
+ rec["status"] = "errored"; rec["reason"] = f"post_create_MAP_breach:{created_price}<{item['kravet_price']}"
+ results.append(rec); print(f"ERR {dw}: post-create MAP breach"); return 1
+ time.sleep(0.6)
+
+ # --- re-fetch to confirm ---
+ st3, r3 = shop("GET", f"products/{pid}.json?fields=id,variants")
+ after = r3.get("product", {}).get("variants", [])
+ rec["after_variants"] = [
+ {"id": v["id"], "sku": v.get("sku"), "price": v.get("price"),
+ "option1": v.get("option1"), "inv_mgmt": v.get("inventory_management")}
+ for v in after]
+ rec["status"] = "fixed"; rec["reason"] = "ok"
+ results.append(rec)
+ print(f"FIXED {dw} [{fix}] -> {rec['target_sku']} @ {nv.get('price')} (variant {nv.get('id')})")
+ return 1
+ except Exception as e:
+ rec["status"] = "errored"; rec["reason"] = f"exception:{e}"
+ results.append(rec); print(f"ERR {dw}: {e}")
+ return 0
+
+
+def main():
+ ap = argparse.ArgumentParser()
+ ap.add_argument("--max", type=int, default=None,
+ help="max variant-CREATES this run (default = live budget.cjs remaining upload)")
+ ap.add_argument("--dry-run", action="store_true",
+ help="print the day-1 batch (first --max creatable SKUs) WITHOUT writing or debiting budget")
+ args = ap.parse_args()
+
+ os.makedirs(OUTDIR, exist_ok=True)
+ work = load_worklist()
+ total_work = len(work)
+ print(f"Loaded {total_work} pending SKUs from dw_unified.bulk_fivefield_worklist")
+
+ # --- resume from audit file ---
+ results = []
+ done = set()
+ if os.path.exists(RESULT):
+ try:
+ results = json.load(open(RESULT))
+ done = {r["dw_sku"] for r in results if r.get("status") in ("fixed", "skipped")}
+ except Exception:
+ results = []
+ todo = [w for w in work if w["dw_sku"] not in done]
+ print(f"=== bulk five-field: {total_work} pending, {len(done)} already done (audit), {len(todo)} remaining ===")
+
+ # --- budget sizing ---
+ if args.dry_run:
+ cap = args.max if args.max is not None else 25
+ print(f"[DRY RUN] no Shopify writes, no budget debit. Showing first {cap} creatable SKUs.")
+ granted = cap
+ else:
+ want = args.max if args.max is not None else budget_remaining()
+ if want <= 0:
+ print(f"[budget] 0 upload headroom remaining today (cadence has the day's slice). "
+ f"Nothing to do this run — resume next budget window. total_remaining={len(todo)}")
+ return
+ granted = budget_take(want)
+ if granted <= 0:
+ print(f"[budget] take() granted 0 (requested {want}). Cadence owns the day's slice. "
+ f"Resume next window. total_remaining={len(todo)}")
+ return
+ print(f"[budget] requested {want}, GRANTED {granted} upload variants this run "
+ f"(shared 884/day ceiling, category={BUDGET_CAT}).")
+
+ print(f"store={DOMAIN} api={API} cap_this_run={granted}\n")
+
+ created = 0 # real variant-creates this run (consumes budget)
+ processed = 0
+ for w in todo:
+ if not args.dry_run and created >= granted:
+ print(f"\n[budget] hit run cap ({granted} creates) — stopping. Resumable next run.")
+ break
+ if args.dry_run and processed >= granted:
+ break
+ made = process(w, results, dry_run=args.dry_run)
+ created += made
+ processed += 1
+ if not args.dry_run:
+ json.dump(results, open(RESULT, "w"), indent=1)
+ if created and created % 25 == 0:
+ fixed_so_far = sum(1 for r in results if r["status"] == "fixed")
+ print(f" … batch progress: {created}/{granted} creates this run · "
+ f"{fixed_so_far} total fixed all-time · {total_work - fixed_so_far} remaining")
+ time.sleep(0.4)
+
+ # --- refund unused grant (refund rule) ---
+ if not args.dry_run:
+ json.dump(results, open(RESULT, "w"), indent=1)
+ unused = granted - created
+ if unused > 0:
+ got = budget_refund(unused)
+ print(f"[budget] refund: granted={granted} used={created} -> refunded {got} upload variants.")
+
+ # --- summary ---
+ fixed = sum(1 for r in results if r["status"] == "fixed")
+ skipped = sum(1 for r in results if r["status"] == "skipped")
+ errored = sum(1 for r in results if r["status"] == "errored")
+ dryruns = sum(1 for r in results if r["status"] == "dryrun")
+ remaining_work = total_work - fixed - skipped
+ print(f"\n=== RUN DONE ===")
+ print(f"this run: creates={created} processed={processed}")
+ print(f"all-time: fixed={fixed} skipped={skipped} errored={errored} dryrun={dryruns}")
+ print(f"TOTAL DONE (fixed+skipped) = {fixed + skipped} / {total_work} · TOTAL REMAINING = {remaining_work}")
+ print(f"audit: {RESULT}")
+ if not args.dry_run:
+ json.dump(results, open(RESULT, "w"), indent=1)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/drain.sh b/drain.sh
new file mode 100644
index 0000000..d03bba1
--- /dev/null
+++ b/drain.sh
@@ -0,0 +1,41 @@
+#!/bin/zsh
+# Daily auto-drain of the 5-field auto-fixable worklist.
+# Budget-aware (claims the day's leftover Shopify variant headroom via budget.cjs),
+# idempotent + resumable (skips anything already in the result JSON), DELETE-nothing.
+# Self-disables when the worklist is fully drained. Singleton-guarded.
+set -u
+export PATH="/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin"
+SK="$HOME/Projects/dw-five-field-step0"
+LOG="$SK/drain.log"
+LOCKDIR="$SK/.drain.lock"
+EXE="$SK/bulk-fivefield-exec.py"
+RESULT="$SK/out/bulk-fivefield-result.json"
+LABEL="com.steve.dw-fivefield-drain"
+ts(){ date '+%Y-%m-%d %H:%M:%S'; }
+
+# singleton (macOS has no flock) — atomic mkdir lock, stale after 2h
+if ! mkdir "$LOCKDIR" 2>/dev/null; then
+ if [ -d "$LOCKDIR" ] && [ "$(find "$LOCKDIR" -maxdepth 0 -mmin +120 2>/dev/null)" ]; then
+ rmdir "$LOCKDIR" 2>/dev/null; mkdir "$LOCKDIR" 2>/dev/null || { echo "[$(ts)] locked, skip" >>"$LOG"; exit 0; }
+ else
+ echo "[$(ts)] already running, skip" >>"$LOG"; exit 0
+ fi
+fi
+trap 'rmdir "$LOCKDIR" 2>/dev/null' EXIT
+
+# executable total (worklist minus the held-no-cost reprice-zeros)
+TOTAL=$(psql -d dw_unified -tAc "select count(*) from bulk_fivefield_worklist where status='pending' and fix_type <> 'reprice-zero'" 2>/dev/null || echo 0)
+DONE=$(node -e 'try{const r=require(process.argv[1]);console.log(r.filter(x=>x.status==="fixed"||x.status==="skipped").length)}catch(e){console.log(0)}' "$RESULT" 2>/dev/null || echo 0)
+echo "[$(ts)] drain start — done=$DONE / total=$TOTAL" >>"$LOG"
+
+if [ "$TOTAL" -gt 0 ] && [ "$DONE" -ge "$TOTAL" ]; then
+ echo "[$(ts)] DRAIN COMPLETE ($DONE/$TOTAL) — self-disabling launchd job" >>"$LOG"
+ launchctl bootout "gui/$(id -u)/$LABEL" 2>/dev/null
+ exit 0
+fi
+
+# run one budget-capped batch (default --max = today's leftover upload headroom)
+python3 "$EXE" >>"$LOG" 2>&1
+RC=$?
+DONE2=$(node -e 'try{const r=require(process.argv[1]);console.log(r.filter(x=>x.status==="fixed"||x.status==="skipped").length)}catch(e){console.log(0)}' "$RESULT" 2>/dev/null || echo 0)
+echo "[$(ts)] drain end rc=$RC — now done=$DONE2 / total=$TOTAL (this run +$((DONE2-DONE)))" >>"$LOG"
← 568fee0 Step-0 definitive priceable-vs-hold split: active_five_field
·
back to Dw Five Field Step0
·
drain: own 'backlog' budget category (500/day) + hourly even 2359edc →