← back to Dw Five Field Step0
bulk-fivefield-exec.py
540 lines
#!/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
# Sentinel for Shopify's DAILY variant-creation 429. Inherits BaseException (NOT Exception) so the
# per-SKU `except Exception` inside process() can't swallow it — it propagates straight to main()'s
# handler, which stops the run cleanly + refunds. (2026-06-23 DTD-C 429-back-off.)
class DailyVariantLimit(BaseException):
pass
DOMAIN = "designer-laboratory-sandbox.myshopify.com"
API = "2024-10"
OUTDIR = "/Users/macstudio3/Projects/dw-five-field-step0/out"
RESULT = os.path.join(OUTDIR, "bulk-fivefield-result.json")
SAMPLE_PRICE = "4.25"
# --- display_variant legacy tag backfill (Option A, DTD 3/3 2026-06-24) -------
# Flag-gated, default OFF: the next scheduled drain run will NOT auto-tag unless
# DISPLAY_VARIANT_BACKFILL=1 is exported. tagsAdd is NOT a variant create -> it
# does NOT debit the variant budget.cjs ledger and does NOT hit the daily 1k
# variant cap. Paced with its OWN per-run cap + the SAME inter-call sleep as the
# variant drain, and runs AFTER the variant work each hourly slot.
DV_TAG = "display_variant"
DV_RESULT = os.path.join(OUTDIR, "display-variant-backfill-result.json")
BUDGET = os.path.expanduser("~/Projects/designerwallcoverings/scripts/variant-budget/budget.cjs")
BUDGET_CAT = "backlog" # 2026-06-21 EVEN 50/50: own reserved 500/day category (was 'upload', which the cadence starved to ~140/day)
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"}
def gql(query, variables=None, retries=6):
"""Admin GraphQL POST with simple 429/5xx back-off. Used only by the
display_variant tag backfill (Option A, DTD 3/3 2026-06-24) — tagsAdd is a
tag mutation, NOT a variant create, so it never touches the variant cap or
the budget.cjs ledger."""
url = f"https://{DOMAIN}/admin/api/{API}/graphql.json"
body = json.dumps({"query": query, "variables": variables or {}}).encode()
for attempt in range(retries):
r = urllib.request.Request(url, data=body, headers=HDR, method="POST")
try:
with urllib.request.urlopen(r, timeout=120) as resp:
return json.loads(resp.read().decode())
except urllib.error.HTTPError as e:
if e.code in (429, 502, 503):
time.sleep(2 * (attempt + 1)); continue
try:
return {"errors": [{"message": f"http_{e.code}", "body": e.read().decode()[:300]}]}
except Exception:
return {"errors": [{"message": f"http_{e.code}"}]}
return {"errors": [{"message": "gql_retries_exhausted"}]}
# ── 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'"
# Schumacher is internal-only (never launched/published to the storefront),
# so building sellable Sample/roll variants for it only burns the scarce
# daily variant budget on products that stay archived. Skip it at the source.
" AND lower(coalesce(vendor,'')) <> 'schumacher'"
") 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
# skip-guard (2026-06-21): rows with no resolvable base SKU would create a
# "None-Sample" / null-sku variant. Skip them (never name-guess) — covers the
# 514 unresolvable add-sample rows and the build-roll rows with dw_sku=None.
if dw is None or str(dw).strip() in ("", "None", "null"):
rec["status"] = "skipped"; rec["reason"] = "no_resolvable_base_sku"
results.append(rec); print(f"SKIP {dw} [{fix}]: no resolvable base sku"); 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}
existing_opts = {(v.get("option1") or "").strip().lower() 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
if "sample" in existing_opts: # product already has a Sample option (different sku) -> 422 if we POST
rec["status"] = "skipped"; rec["reason"] = "sample_option_already_present"
rec["after_variants"] = rec["before_variants"]
results.append(rec); print(f"SKIP {dw}: sample option present"); 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
if "roll" in existing_opts: # product already has a Roll option (different sku) -> 422 if we POST
rec["status"] = "skipped"; rec["reason"] = "roll_option_already_present"
rec["after_variants"] = rec["before_variants"]
results.append(rec); print(f"SKIP {dw}: roll option present"); 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):
# 2026-06-23 (DTD-C / Steve): on Shopify's DAILY variant-creation 429, STOP the whole
# run cleanly instead of grinding the rest of the granted slice into thousands of failed
# 429s (11,754 in one day → it monopolized the cap creating ~0). The cap resets in ~24h;
# the unused grant is refunded by main()'s refund-rule path. Raise a sentinel the loop
# catches → records this SKU as errored:daily_limit, refunds, exits.
body_txt = json.dumps(r2)
if st2 == 429 and "Daily variant creation limit" in body_txt:
rec["status"] = "errored"; rec["reason"] = "daily_variant_limit_429_backoff"; rec["api"] = r2
results.append(rec); print(f"BACKOFF {dw}: Daily variant creation limit 429 — stopping run cleanly (resets ~24h)")
raise DailyVariantLimit(dw)
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 backfill_display_variant(slot_max, dry_run=False):
"""Option A (DTD 3/3 2026-06-24): idempotent backfill of the legacy
'display_variant' tag on ACTIVE products that lack it.
- Pages products(query:"status:active AND -tag:display_variant") so the
population is self-correcting — anything already tagged drops out of the
query, so a re-run never re-tags. Belt-and-braces: also skips if the live
tag list already contains DV_TAG.
- Caps this run at `slot_max` tag-adds (own lane; see DISPLAY_VARIANT_SLOT_MAX),
mirroring the variant drain's per-slot batching so the daily total spreads
across the 24 hourly slots instead of bursting.
- tagsAdd is a tag mutation, NOT a variant create: NO variant-budget debit,
no daily variant-cap exposure.
- Appends an audit ledger to DV_RESULT. DELETEs nothing. Resumable (the
query naturally excludes already-tagged products on the next run).
Returns the number of products tagged this run.
"""
results = []
if os.path.exists(DV_RESULT):
try:
results = json.load(open(DV_RESULT))
except Exception:
results = []
Q = ('query($c:String){ products(first:50, after:$c, '
'query:"status:active AND -tag:display_variant"){ '
'pageInfo{hasNextPage endCursor} '
'edges{ node{ id handle tags } } } }')
M = ('mutation($id:ID!, $tags:[String!]!){ tagsAdd(id:$id, tags:$tags){ '
'node{ id } userErrors{ field message } } }')
print(f"\n=== display_variant backfill === cap_this_run={slot_max} dry_run={dry_run}")
tagged = 0
cursor = None
while tagged < slot_max:
d = gql(Q, {"c": cursor})
if "errors" in d:
print(f"[dv] GQL error: {json.dumps(d['errors'])[:300]} — stopping run")
break
conn = d["data"]["products"]
edges = conn["edges"]
if not edges:
print("[dv] no more untagged active products — backfill complete for now")
break
for e in edges:
if tagged >= slot_max:
break
node = e["node"]
pid = node["id"]
handle = node.get("handle")
live_tags = node.get("tags") or []
rec = {"product_id": pid, "handle": handle, "status": None, "reason": None}
# belt-and-braces idempotency (query already filters, but double-check)
if DV_TAG in live_tags:
rec["status"] = "skipped"; rec["reason"] = "already_tagged"
results.append(rec)
continue
if dry_run:
rec["status"] = "dryrun"; rec["reason"] = "would_tag"
results.append(rec)
print(f"[dv] DRYRUN would tag {handle} ({pid})")
tagged += 1
continue
md = gql(M, {"id": pid, "tags": [DV_TAG]})
if "errors" in md:
rec["status"] = "errored"; rec["reason"] = json.dumps(md["errors"])[:200]
results.append(rec); print(f"[dv] ERR {handle}: {rec['reason']}")
continue
ue = md.get("data", {}).get("tagsAdd", {}).get("userErrors", [])
if ue:
rec["status"] = "errored"; rec["reason"] = json.dumps(ue)[:200]
results.append(rec); print(f"[dv] ERR {handle}: {rec['reason']}")
continue
rec["status"] = "tagged"; rec["reason"] = "ok"
results.append(rec); tagged += 1
print(f"[dv] TAGGED {handle} ({pid}) [{tagged}/{slot_max}]")
json.dump(results, open(DV_RESULT, "w"), indent=1)
time.sleep(0.4) # SAME inter-call sleep as the variant drain (shared pacing)
# advance cursor only if we tagged some but didn't exit; since tagged rows
# leave the query, re-query from the top (cursor=None) to pick up the next
# untagged batch deterministically.
if dry_run:
pi = conn["pageInfo"]
if not pi["hasNextPage"]:
break
cursor = pi["endCursor"]
else:
cursor = None # tagged products fall out of the query -> fresh top page
time.sleep(0.25)
if not dry_run:
json.dump(results, open(DV_RESULT, "w"), indent=1)
total_tagged = sum(1 for r in results if r["status"] == "tagged")
print(f"[dv] run done: this_run={tagged} all_time_tagged={total_tagged} audit={DV_RESULT}")
return tagged
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")
ap.add_argument("--display-variant-only", action="store_true",
help="run ONLY the display_variant tag backfill (skip the variant drain). "
"Still gated by env DISPLAY_VARIANT_BACKFILL=1.")
args = ap.parse_args()
os.makedirs(OUTDIR, exist_ok=True)
# --- display_variant tag backfill (Option A, DTD 3/3 2026-06-24) ----------
# GATED OFF by default: only runs when env DISPLAY_VARIANT_BACKFILL=1. Own
# per-run cap (DISPLAY_VARIANT_SLOT_MAX, default 30) + same inter-call sleep;
# NO variant-budget debit. Runs AFTER the variant drain unless
# --display-variant-only is passed.
dv_enabled = os.environ.get("DISPLAY_VARIANT_BACKFILL", "0") == "1"
dv_slot_max = int(os.environ.get("DISPLAY_VARIANT_SLOT_MAX", "30"))
if args.display_variant_only:
if not dv_enabled:
print("[dv] --display-variant-only requested but DISPLAY_VARIANT_BACKFILL!=1 — "
"gated OFF, nothing to do. Export DISPLAY_VARIANT_BACKFILL=1 to enable.")
return
backfill_display_variant(dv_slot_max, dry_run=args.dry_run)
return
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
try:
made = process(w, results, dry_run=args.dry_run)
except DailyVariantLimit:
# Shopify's daily variant cap is hit — stop NOW (no point grinding the rest of the
# grant into 429s). The refund path below returns the whole unused grant to budget.cjs.
json.dump(results, open(RESULT, "w"), indent=1)
print(f"\n[backoff] Daily variant creation limit reached — stopping run cleanly after "
f"{created} create(s) this run. Resumable next budget window (~24h reset).")
break
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} {BUDGET_CAT} 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)
# --- display_variant tag backfill (runs AFTER the variant drain) ----------
# Option A (DTD 3/3 2026-06-24). GATED OFF by default: only fires when env
# DISPLAY_VARIANT_BACKFILL=1. Own cap + same sleep; NO variant-budget debit.
if dv_enabled:
backfill_display_variant(dv_slot_max, dry_run=args.dry_run)
else:
print("[dv] display_variant backfill GATED OFF (set DISPLAY_VARIANT_BACKFILL=1 to enable).")
if __name__ == "__main__":
main()