← back to Fentucci Naturals
scripts/go-live.py
196 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. Publish to the 12 DW sales channels — Google & YouTube EXCLUDED
(the $4.25 sample would trip GMC price disapproval).
5. status = ACTIVE.
6. 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)
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
# 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]
TOKEN = os.environ.get("SHOPIFY_ADMIN_TOKEN") or sys.exit("SHOPIFY_ADMIN_TOKEN not 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 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}")
# 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()