← back to Carnegie Import
publish_shopify.py
137 lines
#!/usr/bin/env python3
"""Carnegie -> LIVE Shopify publisher. One product per pattern-application,
colorways as variants + a $4.25 memo Sample variant. Tag-scoped so the smart
collections (carnegie-wallcovering / -wall-panel / -textile) pick each up.
SAFE BY DEFAULT: dry-run unless --apply. Metered cadence (delay between writes),
canary-first (--limit N). Only publishes activation_pass=TRUE products.
Usage:
python3 publish_shopify.py --class Wallcovering # dry-run, all wallcovering
python3 publish_shopify.py --class Wallcovering --limit 3 --apply # canary: 3 live
python3 publish_shopify.py --class Wallcovering --apply --delay 1.5 # metered rollout
"""
import argparse, base64, json, os, subprocess, sys, time, urllib.request
DSN = "host=/tmp dbname=dw_unified"
UA = "Mozilla/5.0"
def env(k):
with open(os.path.expanduser("~/Projects/secrets-manager/.env")) as f:
for line in f:
if line.startswith(k + "="): return line.split("=", 1)[1].strip()
return None
TOKEN = env("SHOPIFY_ADMIN_TOKEN")
STORE = "designer-laboratory-sandbox.myshopify.com"
API = f"https://{STORE}/admin/api/2024-10"
def api(method, path, payload=None):
data = json.dumps(payload).encode() if payload is not None else None
req = urllib.request.Request(f"{API}/{path}", data=data, method=method,
headers={"X-Shopify-Access-Token":TOKEN,"Content-Type":"application/json","User-Agent":UA})
with urllib.request.urlopen(req, timeout=60) as r:
return json.loads(r.read())
def q(sql):
out = subprocess.run(["psql",DSN,"-tAF","\x1f","-c",sql],text=True,capture_output=True)
if out.returncode: sys.stderr.write(out.stderr); raise SystemExit("psql fail")
return [l.split("\x1f") for l in out.stdout.splitlines() if l]
def products_for(cls):
# group gate-passing sellable SKUs by parent pattern-application
rows = q(f"""SELECT parent_sku, pattern_name, product_type, type_suffix, collection_tag,
max(regexp_replace(coalesce(body_html,''), E'[\\n\\r\\t]+', ' ', 'g')),
max(product_url), array_agg(dw_sku ORDER BY color_number),
array_agg(color_number ORDER BY color_number), array_agg(price ORDER BY color_number),
array_agg(coalesce(color_tags[1],'') ORDER BY color_number),
array_agg(coalesce(local_image,'') ORDER BY color_number),
max(array_to_string(style_tags,'||'))
FROM carnegie_catalog
WHERE dw_class='{cls}' AND activation_pass IS TRUE
GROUP BY parent_sku, pattern_name, product_type, type_suffix, collection_tag
ORDER BY parent_sku""")
def parr(s): return [x for x in s.strip('{}').replace('"','').split(',') if x] if s and s!='{}' else []
out=[]
for r in rows:
out.append(dict(parent=r[0],name=r[1],ptype=r[2],suffix=r[3],ctag=r[4],
body=r[5],url=r[6],dwskus=parr(r[7]),colors=parr(r[8]),
prices=parr(r[9]),colortags=parr(r[10]),imgs=parr(r[11]),
styles=[x for x in (r[12] or '').split('||') if x]))
return out
def build_payload(p):
tags = sorted(set([p["ctag"],"Carnegie",p["ptype"]] + p["styles"] + p["colortags"]))
tags = [t for t in tags if t]
variants=[]
for dwsku,color,price in zip(p["dwskus"],p["colors"],p["prices"]):
variants.append({"option1":f"Color {color}","sku":dwsku,"price":str(price),
"inventory_management":None,"taxable":True})
variants.append({"option1":"Memo Sample","sku":p["parent"]+"-SAMPLE","price":"4.25"})
body = p["body"] or f"<p>{p['name']} by Carnegie.</p>"
return {"product":{
"title":f"Carnegie {p['name']}","body_html":body,"vendor":"Carnegie",
"product_type":p["ptype"] or "Wallcovering","tags":", ".join(tags),
"status":"active","options":[{"name":"Color"}],"variants":variants}}
def add_image(pid, local):
if not local: return
path=os.path.expanduser(f"~/Projects/carnegie-import/images/{local}")
if not os.path.exists(path): return
b=base64.b64encode(open(path,"rb").read()).decode()
api("POST",f"products/{pid}/images.json",{"image":{"attachment":b}})
def main():
ap=argparse.ArgumentParser()
ap.add_argument("--class",dest="cls",required=True)
ap.add_argument("--limit",type=int,default=0)
ap.add_argument("--delay",type=float,default=1.5)
ap.add_argument("--apply",action="store_true")
a=ap.parse_args()
prods=products_for(a.cls)
# Dedup guard: skip parent-applications already live (idempotent re-runs, canary overlap).
existing=set()
try:
url="products.json?vendor=Carnegie&limit=250&fields=id,variants"
while url:
req=urllib.request.Request(f"{API}/{url}",headers={"X-Shopify-Access-Token":TOKEN,"User-Agent":UA})
with urllib.request.urlopen(req,timeout=60) as r:
body=json.loads(r.read()); link=r.headers.get("Link","")
for pr in body.get("products",[]):
for v in pr.get("variants",[]):
sku=v.get("sku","")
if sku.endswith("-SAMPLE"): existing.add(sku[:-7])
url=None
for part in link.split(","):
if 'rel="next"' in part:
url=part[part.find("<")+1:part.find(">")].split("/admin/api/2024-10/")[-1]
except Exception as e:
print(f"[publish] warn: could not read existing products for dedup: {e}")
before=len(prods)
prods=[p for p in prods if p["parent"] not in existing]
skipped=before-len(prods)
if skipped: print(f"[publish] dedup: skipping {skipped} already-live parent-applications")
if a.limit: prods=prods[:a.limit]
print(f"[publish] class={a.cls} products={len(prods)} apply={a.apply} delay={a.delay}s")
tot_var=sum(len(p['dwskus'])+1 for p in prods)
print(f"[publish] variants(incl sample)={tot_var}")
if not a.apply:
for p in prods[:5]:
pl=build_payload(p)["product"]
print(f" DRY {p['parent']:24} {pl['title'][:34]:34} variants={len(pl['variants'])} tags=[{pl['tags'][:60]}]")
print(f" ... ({len(prods)} total) — dry-run, no writes. Add --apply to publish.")
return
created=0
for p in prods:
try:
res=api("POST","products.json",build_payload(p))
pid=res["product"]["id"]
# one product image from first colorway swatch
add_image(pid, p["imgs"][0] if p["imgs"] else None)
created+=1
print(f"[publish] {created}/{len(prods)} created {p['parent']} -> product {pid}")
time.sleep(a.delay)
except Exception as e:
print(f"[publish] FAIL {p['parent']}: {e}")
print(json.dumps({"created":created,"attempted":len(prods)}))
if __name__=="__main__": main()