← back to Reid Witlin Onboarding
create2.py
161 lines
#!/usr/bin/env python3
"""
Reid Witlin onboarding v2 — GATED live create (Steve runs this).
Targets the CORRECTED pool (Pool A: 623 reconfirmed-old, DWKR- reused as-is;
Pool B: 382 new-gap, DWKR- minted 191394-191775) built by build_batch2.py.
Supersedes create.py (stale 1,071-row/DWDQ plan, quarantined).
Reads targets_ready_v2.csv and creates each as a DRAFT product on the LIVE DW
Shopify store via productSet (product + sample variant + primary image + tags +
core spec metafields), then writes the assigned DWKR sku (continuing live band for Pool B; reused as-is for Pool A) + shopify id back to
rwltd_catalog. (DWDQ = dedicated Reid Witlin prefix, no DWRW/Rebel Walls overlap.) DRAFT status means nothing is customer-facing on create — the
existing dw-rotation-activator drips drafts live per Steve's cadence gate.
SAFETY:
* DRY_RUN=1 by default — prints what it WOULD create, writes nothing.
* Run a 1-product smoke test first: DRY_RUN=0 LIMIT=1 python3 create.py
Verify the draft in Shopify admin, THEN run the full batch.
* status=DRAFT always; activation is a separate gated step (NOT this script).
* Only reads targets_ready_v2.csv — the 156 image-held rows are NEVER created here.
"""
import os, csv, json, time, urllib.request, subprocess
HERE = os.path.dirname(os.path.abspath(__file__))
DRY_RUN = os.environ.get("DRY_RUN", "1") != "0"
LIMIT = int(os.environ.get("LIMIT", "0")) or None
def _tok():
p = os.path.expanduser("~/Projects/secrets-manager/.env")
for line in open(p):
if line.startswith("SHOPIFY_ADMIN_TOKEN="):
return line.split("=", 1)[1].strip().strip('"')
raise SystemExit("SHOPIFY_ADMIN_TOKEN not found")
TOKEN = os.environ.get("AT") or _tok()
URL = "https://designer-laboratory-sandbox.myshopify.com/admin/api/2024-10/graphql.json"
def gql(q, v=None):
body = json.dumps({"query": q, "variables": v or {}}).encode()
req = urllib.request.Request(URL, body,
{"X-Shopify-Access-Token": TOKEN, "Content-Type": "application/json"})
for a in range(8):
try:
d = json.load(urllib.request.urlopen(req, timeout=90))
if "errors" in d and any("THROTTLED" in str(e) for e in d["errors"]):
time.sleep(2 * (a + 1)); continue
return d
except Exception:
time.sleep(2 * (a + 1))
raise RuntimeError("gql failed")
# productSet: create product + sample variant + primary media + tags in one call.
PRODUCT_SET = """
mutation($input: ProductSetInput!) {
productSet(synchronous: true, input: $input) {
product { id handle }
userErrors { field message }
}
}"""
def build_input(row):
# Verified against the LIVE metafield definitions (2026-07-30): custom.content,
# custom.finish, custom.country_of_origin are PRODUCT_REFERENCE — writing text
# to them 422s (the custom.use trap). Route to text-typed custom.* (PDP reads
# these first via custom->specs->global fallback) + the specs.* namespace.
# Keys verified against the live PDP read-chain (product-description-meta.liquid):
# care -> specs.care (custom.cleaning is NOT read; chain is custom.care->specs.care->global.Cleaning)
# finish -> specs.finish (custom.finish is product_reference); width/material/repeat/fire_rating
# -> custom.* (first in chain); style -> specs.style. Dropped: specs.type (Type row removed
# 2026-07-16) and specs.end_use (no render path + duplicates the "Commercial" tag/segment).
spec_map = [
("custom", "width", "single_line_text_field", row.get("width")),
("custom", "material", "multi_line_text_field", row.get("content")),
("specs", "care", "single_line_text_field", row.get("care")),
("custom", "repeat", "single_line_text_field", row.get("repeat")),
("custom", "fire_rating", "single_line_text_field", row.get("fire_rating")),
("specs", "finish", "single_line_text_field", row.get("finish")),
("specs", "abrasion", "single_line_text_field", row.get("abrasion")),
("specs", "style", "single_line_text_field", row.get("style")),
]
def clean(v, t):
# collapse newlines/whitespace for single_line fields so they validate
return v.strip() if t.startswith("multi") else " ".join(v.split())
metafields = [{"namespace": ns, "key": k, "type": t, "value": clean(v, t)}
for ns, k, t, v in spec_map if v and v.strip()]
inp = {
"title": row["title"],
"handle": row["handle"],
"vendor": row["vendor"],
"productType": row["product_type"],
"status": "DRAFT",
"tags": [t.strip() for t in row["tags"].split("|") if t.strip()],
# FIX (TK-11256, 2026-09-04): body_html was never sent -- this is the #2 cause
# (with the sample-only-variant shape being by-design/quote-only) of the
# dw-five-field-canary FAIL on the 2026-09-03 1,005-item batch. build_batch2.py
# now carries description/ai_description through the CSV as row["description"];
# send it as body_html when present. Still empty for rows where the scraper never
# captured a description at all (rwltd_catalog gap, not this script's fault).
"productOptions": [{"name": "Title", "values": [{"name": "Sample"}]}],
"variants": [{
"optionValues": [{"optionName": "Title", "name": "Sample"}],
"price": row["price"],
"sku": row["sku"] + "-Sample",
"inventoryItem": {"tracked": False},
}],
}
if row.get("description") and row["description"].strip():
inp["descriptionHtml"] = row["description"].strip()
if metafields:
inp["metafields"] = metafields
if row.get("image_url"):
inp["files"] = [{"originalSource": row["image_url"], "contentType": "IMAGE"}]
return inp
def main():
rows = list(csv.DictReader(open(os.path.join(HERE, "targets_ready_v2.csv"))))
if LIMIT:
rows = rows[:LIMIT]
print(f"{'DRY-RUN' if DRY_RUN else 'LIVE'}: {len(rows)} DRAFT products "
f"({'no writes' if DRY_RUN else 'creating on LIVE store'})")
created, errs, results = 0, [], []
for i, row in enumerate(rows):
if DRY_RUN:
if i < 3:
print(f" would create {row['sku']} {row['title']!r} draft, "
f"img={'Y' if row.get('image_url') else 'N'}, "
f"tags={len(row['tags'].split('|'))}")
created += 1
continue
d = gql(PRODUCT_SET, {"input": build_input(row)})
r = (d.get("data") or {}).get("productSet") or {}
ue = r.get("userErrors") or []
if ue:
errs.append({"sku": row["sku"], "errors": ue[:2]})
else:
pid = (r.get("product") or {}).get("id")
results.append({"sku": row["sku"], "mfr_sku": row["mfr_sku"],
"handle": row["handle"], "product_id": pid})
created += 1
# IDEMPOTENCY: write id + sku back so build_batch's orphan guard
# (WHERE shopify_product_id IS NULL) never re-creates this product.
subprocess.run(["psql", "host=/tmp dbname=dw_unified", "-c",
"UPDATE rwltd_catalog SET shopify_product_id=%s, dw_sku=COALESCE(NULLIF(dw_sku,''), %s) "
"WHERE mfr_sku=%s;" % (
"'" + (pid or "").replace("'", "") + "'",
"'" + row["sku"].replace("'", "") + "'",
"'" + row["mfr_sku"].replace("'", "''") + "'")],
capture_output=True, text=True)
if i % 25 == 0:
print(f" ...{i}/{len(rows)} created={created} errs={len(errs)}")
time.sleep(0.3)
out = {"created": created, "errors": errs[:20], "results_sample": results[:5],
"total_results": len(results)}
json.dump(out, open(os.path.join(HERE, "create-results.json"), "w"), indent=2)
print(f"\nDONE. created={created} userErrors={len(errs)} "
f"({'DRY-RUN — nothing written' if DRY_RUN else 'DRAFTS on live store'})")
if errs:
print("sample errors:", json.dumps(errs[:3]))
if __name__ == "__main__":
main()