← back to Fentucci Naturals
scripts/go-live.py
279 lines
#!/usr/bin/env python3
"""GATED go-live (daily TRICKLE) for the Fentucci Naturals (Tokiwa) drafts.
Steve override 2026-07-26 (AskUserQuestion): activate as QUOTE-ONLY despite no
mill width and no per-yard price. Modeled on the CMO Paris quote-only precedent
+ the stout-onboard/go-live.mjs machinery.
Per product (data/created.jsonl):
1. SETTLEMENT re-gate — tokiwa_catalog.settlement_status must be 'clear'.
2. validate-lite (width check OVERRIDDEN per Steve; the other guards stay):
title has no "Unknown"/"Wallpaper", description present, >=2 tags,
>=1 image, a {SKU}-Sample variant. FAIL -> stay DRAFT, record held, skip.
3. CMO-Paris quote-only shape: DELETE the $0.00 "Per Yard" variant (a $0
orderable variant is a free-checkout loss vector), keep ONLY {SKU}-Sample
@ $4.25 as the orderable item: inventory tracked, policy=DENY, on_hand=2026
at Ventura Blvd. Add tag 'contact-for-price'.
4. WEIGHT GATE (TK-11414 hard rule, wired TK-11471): re-read every surviving
variant's weight through a query that ACTUALLY returns it, write the approved
default (sample 0.25 lb / Wallcovering 3.0 lb) onto any that are zero or unset,
RE-VERIFY, and if any variant is STILL zero-weight -> record HELD ('weight>0')
and do NOT activate. Zero weight collapses the order into the lowest weight
tier / free-shipping band and mis-costs DW freight.
5. Publish to the 12 DW sales channels — Google & YouTube EXCLUDED
(the $4.25 sample would trip GMC price disapproval).
6. status = ACTIVE.
7. PG-first write-back: tokiwa_catalog.status='active' (on_shopify already true).
Idempotent: skips products already ACTIVE or in data/golive-done.jsonl.
Trickle: --limit N per run. NO same-day bulk-dump.
python3 scripts/go-live.py # dry-run summary
python3 scripts/go-live.py --apply --limit 2 # canary
python3 scripts/go-live.py --apply --limit 40 # daily trickle tranche
"""
import argparse, json, os, subprocess, sys, time, urllib.request, urllib.error
HERE = os.path.dirname(os.path.abspath(__file__)); ROOT = os.path.dirname(HERE)
sys.path.insert(0, os.path.join(HERE, "lib"))
from weight_guard import ( # noqa: E402 (TK-11471 weight gate)
WEIGHT_REQUERY, heal_and_verify_weights, response_carries_weight_field)
CREATED = os.path.join(ROOT, "data", "created.jsonl")
DONE = os.path.join(ROOT, "data", "golive-done.jsonl")
HELD = os.path.join(ROOT, "data", "golive-held.jsonl")
SHOP = "designer-laboratory-sandbox.myshopify.com"; API = "2024-10"
LOCATION_GID = "gid://shopify/Location/5795643504" # 15442 Ventura Blvd.
TARGET_QTY = 2026
# ─── CODIFIED CARVE-OUT (TK-11664, Steve-APPROVED 2026-09-14 — Option B) ──────────
# The Sample variant of this QUOTE-ONLY line is DELIBERATELY set (step 3b below) to
# price=$4.25 · inventory_management=shopify (TRACKED) · inventory_policy=deny ·
# on_hand=TARGET_QTY(=2026) @ Location 5795643504.
# This is the line's INTENDED quote-only mechanism per Steve's AskUserQuestion ruling
# (2026-07-26, CMO-Paris precedent): the $4.25 Sample is the SOLE orderable item, and
# it is made finite/tracked-DENY so a customer buys a sample, not a $0 roll.
#
# DO NOT re-flag this as a defect. Specifically it is NOT:
# • the TK-11357 "$0-orderable re-stamper" fingerprint — that stamps qty=2026 onto a
# $0 SELLABLE variant; here qty=2026 sits on a $4.25 SAMPLE (price>0), so the
# restamp/orderable detectors' price>0 guard correctly skips it.
# • a violation of the DW "samples get no inventory tracking" convention — this vendor
# is a documented carve-out from that rule (see fentucci-scraper-manager SKILL.md
# and data/sample-qty-carveout.json for the machine-readable record + its boundary).
# The ONLY armed defect on this line is a $0-priced variant that is checkout-ORDERABLE;
# that stays fully in scope for every canary. This carve-out narrows nothing there.
# ─────────────────────────────────────────────────────────────────────────────────
# Google & YouTube (29646651457) EXCLUDED per Steve's rule + GMC $4.25-leak guard.
PUBLICATIONS = [22208643184, 22497296496, 29739483201, 29776969793, 37904089153,
43657658419, 44234276915, 44317474867, 44317507635, 71898464307, 115856375859, 140027723827]
# Prefer the FULL-access token: the weight heal (inventoryItemUpdate) and the
# inventory_levels writes below both need write_inventory, which the narrow
# SHOPIFY_ADMIN_TOKEN does not carry (dw-golive-token-guard-canary).
TOKEN = (os.environ.get("SHOPIFY_FULL_ACCESS_TOKEN") or os.environ.get("SHOPIFY_ADMIN_TOKEN")
or sys.exit("no Shopify token set — source ~/Projects/secrets-manager/.env"))
def rest(path, method="GET", body=None):
req = urllib.request.Request(f"https://{SHOP}/admin/api/{API}/{path}", method=method,
data=json.dumps(body).encode() if body else None,
headers={"X-Shopify-Access-Token": TOKEN, "Content-Type": "application/json"})
for attempt in range(5):
try:
with urllib.request.urlopen(req, timeout=60) as r:
return json.loads(r.read() or "{}")
except urllib.error.HTTPError as e:
if e.code == 429: time.sleep(4 * (attempt + 1)); continue
raise
raise RuntimeError(f"429 storm on {path}")
def graphql(query, variables):
return rest("graphql.json", "POST", {"query": query, "variables": variables})
def settlement_map(skus):
env = {**os.environ, "PGHOST": "/tmp"}
out = subprocess.run(["psql", "-d", "dw_unified", "-tAc",
"SELECT dw_sku, settlement_status FROM tokiwa_catalog"], env=env, capture_output=True, text=True)
m = {}
for line in out.stdout.splitlines():
if "|" in line:
k, v = line.split("|", 1); m[k] = v
return m
def pg_activate(sku):
env = {**os.environ, "PGHOST": "/tmp"}
q = sku.replace("'", "''")
subprocess.run(["psql", "-d", "dw_unified", "-c",
f"UPDATE tokiwa_catalog SET status='active' WHERE dw_sku='{q}';"],
env=env, capture_output=True, text=True) # non-fatal; Shopify is authoritative
def validate_lite(p):
"""5 guards; width intentionally skipped per Steve override."""
reasons = []
t = (p.get("title") or "").lower()
if "unknown" in t: reasons.append("title has Unknown")
if "wallpaper" in t: reasons.append('title has "Wallpaper"')
if not (p.get("body_html") or "").strip(): reasons.append("no description")
if len([x for x in (p.get("tags") or "").split(",") if x.strip()]) < 2: reasons.append("<2 tags")
if not p.get("images"): reasons.append("no image")
if not any((v.get("sku") or "").endswith("-Sample") for v in p.get("variants", [])):
reasons.append("no sample variant")
return reasons
def weight_gate(pid, gid_product):
"""TK-11414 hard rule: NO product goes ACTIVE with a missing/zero variant weight.
SELF-HEAL then VERIFY (the pattern Steve approved in sanderson-onboard
create_sdg.mjs 8d09eed) — stranding product is worse than assigning the
already-approved default, but a heal that silently fails must never activate.
Returns [] to proceed, or a list of `weight>0` hold reasons. Idempotent and a
no-op (zero extra calls beyond the one re-read) when weights are already positive.
The re-read runs WEIGHT_REQUERY, which explicitly asks for
inventoryItem{measurement{weight{value unit}}} — and the response is checked for
that field before any verdict, so a response that never carried weight can never
be mistaken for a clean reading (fails CLOSED, never a false green).
"""
try:
resp = graphql(WEIGHT_REQUERY, {"id": gid_product})
except Exception as e: # noqa: BLE001 - fail closed
return ["weight>0: weight re-read failed: %s" % str(e)[:80]]
if (resp or {}).get("errors"):
return ["weight>0: weight re-read errored: %s" % str(resp["errors"])[:100]]
fresh = ((resp or {}).get("data") or {}).get("product")
if not fresh:
return ["weight>0: product not readable for weights"]
if not response_carries_weight_field(fresh):
return ["weight>0: NOT MEASURED — re-read response carried no weight field"]
def rest_heal(variant, lb):
"""Fallback when the GraphQL inventoryItemUpdate path is unavailable (a token
without write_inventory). REST variant update needs only write_products."""
vid = str(variant.get("id") or "").rsplit("/", 1)[-1]
if not vid.isdigit():
return False
r = rest("variants/%s.json" % vid, "PUT",
{"variant": {"id": int(vid), "weight": lb, "weight_unit": "lb"}})
return bool((r or {}).get("variant"))
res = heal_and_verify_weights(graphql, gid_product, fresh, rest_heal=rest_heal)
if res["ok"]:
return []
reasons = ["weight>0: still zero after heal: %s" % ", ".join(res["stillZero"])] if res["stillZero"] \
else ["weight>0: heal could not be confirmed"]
reasons += ["weight>0: " + e for e in res["errs"][:3]]
return reasons
def go_live_one(pid, sku):
gidP = f"gid://shopify/Product/{pid}"
p = rest(f"products/{pid}.json").get("product")
if not p: return ("missing", [])
if p["status"] == "active":
pg_activate(sku); return ("skipped", [])
bad = validate_lite(p)
if bad: return ("held", bad)
errs = []
variants = p["variants"]
sample = next((v for v in variants if (v.get("sku") or "").endswith("-Sample")), None)
peryard = next((v for v in variants if v.get("sku") == sku), None) # $0 per-yard == bare dw_sku
# 3a) CMO precedent — drop the $0 per-yard variant (only if a sample remains)
if peryard and sample:
try: rest(f"products/{pid}/variants/{peryard['id']}.json", "DELETE")
except urllib.error.HTTPError as e: errs.append(f"del-peryard:{e.code}")
# 3b) sample -> tracked + policy=deny + on_hand=2026 at Ventura Blvd
if sample:
try:
rest(f"variants/{sample['id']}.json", "PUT", {"variant": {
"id": sample["id"], "inventory_management": "shopify", "inventory_policy": "deny"}})
except urllib.error.HTTPError as e: errs.append(f"policy:{e.code}")
iid = sample.get("inventory_item_id")
if iid:
try: rest("inventory_levels/connect.json", "POST",
{"location_id": 5795643504, "inventory_item_id": iid})
except urllib.error.HTTPError: pass # already connected
try: rest("inventory_levels/set.json", "POST",
{"location_id": 5795643504, "inventory_item_id": iid, "available": TARGET_QTY})
except urllib.error.HTTPError as e: errs.append(f"setqty:{e.code}")
# 3c) tag contact-for-price (mirror CMO)
tags = [x.strip() for x in (p.get("tags") or "").split(",") if x.strip()]
if "contact-for-price" not in tags:
tags.append("contact-for-price")
try: rest(f"products/{pid}.json", "PUT", {"product": {"id": pid, "tags": ", ".join(tags)}})
except urllib.error.HTTPError as e: errs.append(f"tag:{e.code}")
# 3d) WEIGHT GATE — heal then verify. A product that is STILL zero-weight is
# HELD as a draft; it is never published and never flipped ACTIVE.
wreasons = weight_gate(pid, gidP)
if wreasons:
return ("held", wreasons)
# 4) publish to 12 channels (Google excluded)
r = graphql("mutation($id:ID!,$pubs:[PublicationInput!]!){ publishablePublish(id:$id, input:$pubs){ userErrors{message} } }",
{"id": gidP, "pubs": [{"publicationId": f"gid://shopify/Publication/{n}"} for n in PUBLICATIONS]})
for e in (r.get("data", {}).get("publishablePublish", {}) or {}).get("userErrors", []):
errs.append("publish:" + e["message"])
# 5) status ACTIVE
up = rest(f"products/{pid}.json", "PUT", {"product": {"id": pid, "status": "active"}})
status = up.get("product", {}).get("status")
if status == "active":
pg_activate(sku) # 6) PG-first writeback
else:
errs.append(f"status={status}")
return (status, errs)
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--apply", action="store_true")
ap.add_argument("--limit", type=int, default=0)
args = ap.parse_args()
done = {json.loads(l)["sku"] for l in open(DONE)} if os.path.exists(DONE) else set()
rows = [json.loads(l) for l in open(CREATED) if l.strip()]
todo = [r for r in rows if r["sku"] not in done]
if args.limit: todo = todo[: args.limit]
sett = settlement_map([r["sku"] for r in todo])
print(f"go-live: {len(todo)} to process · {len(PUBLICATIONS)} channels (Google excluded) · "
f"quote-only (drop $0 per-yard, keep $4.25 sample) · {'APPLY' if args.apply else 'DRY-RUN'}")
if not args.apply:
print("first 3:", ", ".join(r["sku"] for r in todo[:3]))
print("DRY-RUN — pass --apply --limit N to execute a trickle tranche.")
return
fd_done = open(DONE, "a"); fd_held = open(HELD, "a")
ok = skip = held = err = miss = 0
for i, r in enumerate(todo):
sku, pid = r["sku"], r["product_id"]
if sett.get(sku) != "clear":
fd_held.write(json.dumps({"sku": sku, "reason": f"settlement:{sett.get(sku)}"}) + "\n"); fd_held.flush()
held += 1; print(f" ⏸ HOLD {sku}: settlement {sett.get(sku)}"); continue
try:
status, errs = go_live_one(pid, sku)
if status == "missing": miss += 1; print(f" ? {sku}: not found")
elif status == "held": fd_held.write(json.dumps({"sku": sku, "reason": errs}) + "\n"); fd_held.flush(); held += 1; print(f" ⏸ HOLD {sku}: {errs}")
elif status == "skipped": skip += 1; fd_done.write(json.dumps({"sku": sku, "product_id": pid}) + "\n"); fd_done.flush()
elif errs: err += 1; print(f" ⚠ {sku} {status}: {errs[:3]}")
if status == "active" or status == "skipped":
if status == "active": ok += 1; fd_done.write(json.dumps({"sku": sku, "product_id": pid}) + "\n"); fd_done.flush(); print(f" ✓ {sku} ACTIVE")
except Exception as e:
err += 1; print(f" ERR {sku}: {str(e)[:160]}"); time.sleep(1)
time.sleep(0.4)
if (i + 1) % 20 == 0: print(f" ...{i+1}/{len(todo)} (active={ok} skip={skip} held={held} err={err})"); time.sleep(5)
fd_done.close(); fd_held.close()
print(f"\nDONE. active={ok} skipped={skip} held={held} errors={err} missing={miss} of {len(todo)}")
if __name__ == "__main__":
main()