← back to Kravet Sheet Sync 2026 04 20
bulk_push_v1.py
449 lines
#!/usr/bin/env python3
"""
Bulk-push DWKK NEW-bucket wallcoverings to Shopify as active products.
Pipeline per SKU:
1) dw_sku_registry INSERT
2) productCreate (status=DRAFT, vendor, productType, tags, basic description)
3) productVariantsBulkUpdate main variant: DWKK SKU, MAP price, cost
4) productVariantsBulkCreate sample variant: DWKK-Sample at $4.25
5) productCreateMedia with real Brandfolder URL
6) shopify_products INSERT (enables Full Monty)
7) Full Monty API call (ai tags, silas, spin)
8) kravet_finalize (title, SEO, body, sanitized tags)
9) collectionAddProducts → New Arrivals
10) productUpdate status=ACTIVE (if width + image present)
11) kravet_catalog UPSERT
Run:
python3 bulk_push_v1.py --limit 10
"""
import argparse, json, re, subprocess, sys, time, urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
TOKEN = os.environ.get('SHOPIFY_ADMIN_TOKEN','')
SHOP = 'designer-laboratory-sandbox.myshopify.com'
GQL = f'https://{SHOP}/admin/api/2024-10/graphql.json'
SSH = ['ssh', 'root@45.61.58.125']
NEW_ARRIVALS = 'gid://shopify/Collection/167327760435'
AGENT_AUTH = 'Basic ' + __import__('base64').b64encode(b'admin:DWSecure2024!').decode()
SMALL = {"a","an","the","and","or","but","of","in","on","at","for","to","with","by"}
def pg(sql):
import csv, io
r = subprocess.run(SSH + [
f'PGPASSWORD=DW2024! psql -h 127.0.0.1 -U dw_admin -d dw_unified --csv -t -c "{sql}"'],
capture_output=True, text=True, check=True)
return [row for row in csv.reader(io.StringIO(r.stdout)) if row]
def pg_exec(sql):
subprocess.run(SSH + [
f'PGPASSWORD=DW2024! psql -h 127.0.0.1 -U dw_admin -d dw_unified -c "{sql}"'],
capture_output=True, text=True, check=True)
def gql(q, v=None):
body = json.dumps({"query": q, "variables": v or {}}).encode()
req = urllib.request.Request(GQL, data=body, method="POST",
headers={"Content-Type":"application/json","X-Shopify-Access-Token":TOKEN})
d = json.loads(urllib.request.urlopen(req, timeout=30).read())
if "errors" in d: raise RuntimeError(d["errors"])
return d["data"]
def title_case(s):
if not s: return ""
words = re.split(r"(\s+)", s.strip().lower())
out, first = [], True
for w in words:
if not w.strip(): out.append(w); continue
if not first and w in SMALL: out.append(w)
else:
out.append(re.sub(r"(^|[\s&/-])([a-z])", lambda m: m.group(1)+m.group(2).upper(), w))
first = False
return "".join(out).strip()
def clean_wall(s):
if not s: return s
s = re.sub(r"\bWallpapers\b", "Wallcoverings", s, flags=re.I)
s = re.sub(r"\bWallpaper\b", "Wallcovering", s, flags=re.I)
return s
def norm_sku(s):
return re.sub(r"\.0$","", re.sub(r"-",".", s.strip().upper()))
# -----------------------------------------------------------------------
def fetch_batch(limit, skip_existing=True):
cond = ""
if skip_existing:
cond = """AND NOT EXISTS (
SELECT 1 FROM dw_sku_registry r WHERE r.dw_sku = n.new_dw_sku
) AND NOT EXISTS (
SELECT 1 FROM kravet_catalog c WHERE c.dw_sku = n.new_dw_sku
)"""
q = f"""
WITH ranked AS (
SELECT n.new_dw_sku, n.mfr_sku, n.pattern_name, n.color_name, n.brand,
n.collection, n.country_of_origin, n.width, n.unit_of_measure,
n.content, n.type1, n.type2, n.style1, n.style2,
n.display_status, n.inventory_available,
n.wholesale::text AS wholesale, n.map::text AS map,
n.tariff_pct::text AS tariff_pct, n.cost::text AS cost,
p.url AS image_url,
pr.url AS full_repeat_url,
s.vert_repeat, s.horz_repeat, s.prop_65, s.ca_tb117, s.direction,
s.clean_code, s.durability, s.lead_time_days, s.ship_from,
s.ab_1817, s.memo_sample_available, s.weight, s.barcode,
s.minimum_order_qty, s.order_increment_qty, s.notes,
row_number() OVER (PARTITION BY n.brand ORDER BY n.map DESC) AS rn_brand
FROM kravet_diff_new_2026_04_20 n
JOIN v_kravet_sku_primary_image p
ON p.mfr_sku_norm = regexp_replace(regexp_replace(upper(btrim(n.mfr_sku)),'-','.','g'),'\\.0\$','')
AND p.kind='colorway'
LEFT JOIN v_kravet_sku_primary_image pr
ON pr.mfr_sku_norm = p.mfr_sku_norm AND pr.kind='full_repeat'
LEFT JOIN kravet_sheet_fields s
ON s.mfr_sku_norm = p.mfr_sku_norm
WHERE n.has_image AND n.has_width
AND n.pattern_name <> '' AND n.color_name <> ''
AND n.map IS NOT NULL
AND upper(n.stock_status) IN ('IN STOCK','LIMITED STOCK')
{cond}
)
SELECT new_dw_sku, mfr_sku, pattern_name, color_name, brand,
collection, country_of_origin, width, unit_of_measure, content,
type1, type2, style1, style2, display_status, inventory_available,
wholesale, map, tariff_pct, cost, image_url, full_repeat_url,
vert_repeat, horz_repeat, prop_65, ca_tb117, direction, clean_code,
durability, lead_time_days, ship_from, ab_1817, memo_sample_available,
weight, barcode, minimum_order_qty, order_increment_qty, notes
FROM ranked WHERE rn_brand = 1 ORDER BY brand LIMIT {limit}
"""
cols = ["dw_sku","mfr_sku","pattern_name","color_name","brand",
"collection","country_of_origin","width","unit_of_measure","content",
"type1","type2","style1","style2","display_status","inventory_available",
"wholesale","map","tariff_pct","cost","image_url","full_repeat_url",
"vert_repeat","horz_repeat","prop_65","ca_tb117","direction","clean_code",
"durability","lead_time_days","ship_from","ab_1817","memo_sample_available",
"weight","barcode","minimum_order_qty","order_increment_qty","notes"]
return [dict(zip(cols, r)) for r in pg(q)]
# -----------------------------------------------------------------------
def build_title(it):
pat = title_case(it["pattern_name"])
col = title_case(it["color_name"])
brn = title_case(it["brand"] or "Kravet")
t = " ".join(x for x in [pat, col] if x)
t = f"{t} | {brn}" if t else brn
return clean_wall(t)[:255]
def build_tags(it):
tags = [
"Kravet",
title_case(it["brand"] or "Kravet"),
"Wallcovering",
title_case(it["type1"] or ""),
"New Arrival",
f"Origin: {title_case(it['country_of_origin'])}" if it["country_of_origin"] else "",
]
return [t for t in tags if t]
def build_description(it):
# Editorial prose only — NO specs in body. All spec numbers live in metafields.
pat = title_case(it["pattern_name"])
col = title_case(it["color_name"])
brn = title_case(it["brand"])
coll = title_case(it["collection"])
t1 = (title_case(it["type1"]) or "Wallcovering").lower()
origin = title_case(it["country_of_origin"] or "")
style = title_case(it["style1"] or it["style2"] or "")
opening = f"<p><strong>{pat}</strong>"
if col: opening += f" in <strong>{col}</strong>"
opening += f" by {brn}"
if coll: opening += f", from the {coll} collection"
opening += ".</p>"
body = f"<p>A {t1} wallcovering"
if style: body += f" with a {style.lower()} sensibility"
if origin: body += f", crafted in {origin}"
body += ". Designed for interiors that benefit from considered materials and a distinct point of view.</p>"
close = ("<p>Order a complimentary sample to see the colour and scale in your own light, "
"or message our trade team for coordinating colourways and specification guidance.</p>")
return clean_wall(opening + body + close)
def build_seo(it):
# SEO must sell DW service: free samples, trade service, hand-holding.
pat = title_case(it["pattern_name"])
col = title_case(it["color_name"])
brn = title_case(it["brand"] or "Kravet")
seo_t = clean_wall(f"{pat} {col} {brn} — Free Samples")[:70]
seo_d = clean_wall(
f"Order free samples of {pat} in {col} by {brn}. "
f"Free shipping, trade pricing, and expert guidance from Designer Wallcoverings."
)[:160]
return seo_t, seo_d
def build_metafields(it, pid):
out = []
def add(ns, k, v, t="single_line_text_field"):
if v is None or v in ("","NULL"): return
out.append({"ownerId": pid, "namespace": ns, "key": k, "value": str(v).strip(), "type": t})
add("custom","manufacturer_sku", it["mfr_sku"])
add("dwc","manufacturer_sku", it["mfr_sku"])
add("custom","color", title_case(it["color_name"] or ""))
add("custom","real_color_name", title_case(it["color_name"] or ""))
add("dwc","color", title_case(it["color_name"] or ""))
add("dwc","pattern_name", title_case(it["pattern_name"] or ""))
add("dwc","brand", title_case(it["brand"] or ""))
add("dwc","real_vendor", title_case(it["brand"] or ""))
add("global","Color-Way", title_case(it["color_name"] or ""))
add("global","Brand", title_case(it["brand"] or ""))
add("global","Collection", title_case(it["collection"] or ""))
add("global","width", it["width"])
add("global","unit_of_measure", it["unit_of_measure"])
add("global","Country-of-Origin", title_case(it["country_of_origin"] or ""))
add("global","Content", it["content"])
add("global","Type1", title_case(it["type1"] or ""))
add("global","Vert-Rpt", it["vert_repeat"])
add("global","Horz-Rpt", it["horz_repeat"])
add("global","Weight", it["weight"])
add("global","Direction", it["direction"])
add("global","Prop-65", it["prop_65"])
add("global","CA-TB117", it["ca_tb117"])
add("custom","ab_1817", it["ab_1817"])
add("global","Clean-Code", it["clean_code"])
add("global","Durability", it["durability"])
add("global","Lead-Time-in-Days", it["lead_time_days"])
add("global","Ship-From", it["ship_from"])
add("global","Memo-Sample-Available", it["memo_sample_available"])
add("global","v_prods_quantity_order_min", it["minimum_order_qty"])
add("global","v_prods_quantity_order_units", it["order_increment_qty"])
add("global","WHLS-Price", it["wholesale"])
add("global","Display-Status", it["display_status"])
add("global","Inventory-Available", it["inventory_available"])
add("custom","vendor_notes", it["notes"])
return out
# -----------------------------------------------------------------------
PRODUCT_CREATE = """mutation($p: ProductCreateInput!){ productCreate(product:$p){ product{ id handle variants(first:5){nodes{id}} } userErrors{field message} } }"""
VARIANT_UPDATE = """mutation($pid:ID!,$vs:[ProductVariantsBulkInput!]!){ productVariantsBulkUpdate(productId:$pid,variants:$vs){ userErrors{field message} } }"""
VARIANT_CREATE = """mutation($pid:ID!,$vs:[ProductVariantsBulkInput!]!){ productVariantsBulkCreate(productId:$pid,variants:$vs){ productVariants{id sku} userErrors{field message} } }"""
VARIANT_CREATE_STRATEGY = """mutation($pid:ID!,$vs:[ProductVariantsBulkInput!]!,$st:ProductVariantsBulkCreateStrategy){ productVariantsBulkCreate(productId:$pid,variants:$vs,strategy:$st){ productVariants{id sku} userErrors{field message} } }"""
MEDIA_CREATE = """mutation($pid:ID!,$m:[CreateMediaInput!]!){ productCreateMedia(productId:$pid,media:$m){ media{... on MediaImage{id status}} mediaUserErrors{field message} } }"""
METAS_SET = """mutation($m:[MetafieldsSetInput!]!){ metafieldsSet(metafields:$m){ userErrors{field message} } }"""
COLLECTION_ADD = """mutation($id:ID!,$pids:[ID!]!){ collectionAddProducts(id:$id,productIds:$pids){ userErrors{field message} } }"""
PRODUCT_UPDATE = """mutation($p:ProductUpdateInput!){ productUpdate(product:$p){ product{id status} userErrors{field message} } }"""
def push_one(it):
dw_sku = it["dw_sku"]
mfr = it["mfr_sku"]
print(f"\n== {dw_sku} | {mfr} | {it['brand']} | {it['pattern_name']} / {it['color_name']} ==")
# 1) Register SKU
esc_b = (it["brand"] or "KRAVET").replace("'", "''")
esc_m = mfr.replace("'", "''")
pg_exec(f"INSERT INTO dw_sku_registry (dw_sku, vendor_prefix, vendor_name, mfr_sku, created_at) "
f"VALUES ('{dw_sku}','DWKK','{esc_b}','{esc_m}', now()) ON CONFLICT (dw_sku) DO NOTHING")
# 2) Create draft product
title = build_title(it)
tags = build_tags(it)
desc = build_description(it)
seo_t, seo_d = build_seo(it)
pi = {
"title": title,
"vendor": title_case(it["brand"] or "Kravet"),
"productType": "Wallcovering",
"status": "DRAFT",
"tags": tags,
"descriptionHtml": desc,
"seo": {"title": seo_t, "description": seo_d},
}
d = gql(PRODUCT_CREATE, {"p": pi})
ue = d["productCreate"]["userErrors"]
if ue: raise RuntimeError(f"productCreate: {ue}")
prod = d["productCreate"]["product"]
pid = prod["id"]
dvid = prod["variants"]["nodes"][0]["id"]
print(f" created {pid} ({prod['handle']})")
# 3) Main variant (update default)
try:
map_price = float(it["map"])
cost_price = float(it["cost"]) if it["cost"] else None
except (TypeError, ValueError):
map_price, cost_price = 0.0, None
mv = {
"id": dvid,
"price": f"{map_price:.2f}",
"inventoryItem": {"sku": dw_sku, "tracked": True},
}
if cost_price is not None:
mv["inventoryItem"]["cost"] = f"{cost_price:.2f}"
if it.get("barcode"):
mv["barcode"] = str(it["barcode"]).strip()
gql(VARIANT_UPDATE, {"pid": pid, "vs": [mv]})
# 4) Sample variant — strategy:DEFAULT so the main variant stays
sv = {
"price": "4.25",
"inventoryItem": {"sku": f"{dw_sku}-Sample", "tracked": False},
"optionValues": [{"optionName": "Title", "name": "Sample"}],
}
# First, give the main variant a proper option value to leave room for Sample
w = str(it.get("width","")).strip()
main_opt = f"Sold Per Single Roll - {w}In Wide" if w else "Single Roll"
gql(VARIANT_UPDATE, {"pid": pid, "vs": [{
"id": dvid, "optionValues": [{"optionName": "Title", "name": main_opt}]}]})
gql(VARIANT_CREATE_STRATEGY, {"pid": pid, "vs": [sv], "st": "DEFAULT"})
# 5) Attach image
gql(MEDIA_CREATE, {"pid": pid, "m": [{
"originalSource": it["image_url"],
"mediaContentType": "IMAGE",
"alt": title,
}]})
if it.get("full_repeat_url"):
try:
gql(MEDIA_CREATE, {"pid": pid, "m": [{
"originalSource": it["full_repeat_url"].replace(" ", "%20"),
"mediaContentType": "IMAGE",
"alt": f"{title} — full pattern repeat",
}]})
except Exception as e:
print(f" full_repeat skipped: {e}")
print(f" image attached")
# 6) Metafields (chunked at 25)
mfs = build_metafields(it, pid)
for i in range(0, len(mfs), 25):
gql(METAS_SET, {"m": mfs[i:i+25]})
print(f" metafields {len(mfs)} set")
# 7) shopify_products row so Full Monty can find it
pg_exec(f"""
INSERT INTO shopify_products (shopify_id, handle, title, vendor, product_type,
sku, variant_sku, status, dw_sku, mfr_sku, vendor_prefix,
retail_price, cost_price, price, synced_at)
VALUES ('{pid.split("/")[-1]}', '{prod["handle"].replace("'","''")}',
'{title.replace("'","''")}', '{title_case(it["brand"] or "Kravet").replace("'","''")}',
'Wallcovering', '{dw_sku}', '{dw_sku}', 'draft',
'{dw_sku}', '{mfr.replace("'","''")}', 'DWKK',
{map_price:.2f}, {cost_price or "NULL"}, {map_price:.2f}, now())
ON CONFLICT (shopify_id) DO NOTHING
""")
# 8) Add to New Arrivals collection
try:
gql(COLLECTION_ADD, {"id": NEW_ARRIVALS, "pids": [pid]})
except Exception as e:
print(f" collection add skipped: {e}")
# 9) kravet_catalog upsert
pg_exec(f"""
INSERT INTO kravet_catalog (mfr_sku, pattern_name, color_name, brand, dw_sku,
shopify_product_id, on_shopify, price_retail, cost_price, tariff_pct,
status, imported_at, image_url)
VALUES ('{esc_m}', '{it["pattern_name"].replace("'","''")}',
'{it["color_name"].replace("'","''")}', '{esc_b}',
'{dw_sku}', '{pid}', true,
{it['map']}, {it['cost'] or 'NULL'}, {it['tariff_pct'] or 'NULL'},
'draft', now(), '{it['image_url'].replace("'","''")}')
ON CONFLICT (mfr_sku) DO UPDATE SET
dw_sku = EXCLUDED.dw_sku,
shopify_product_id = EXCLUDED.shopify_product_id,
on_shopify = true,
image_url = EXCLUDED.image_url,
imported_at = now()
""")
return {
"dw_sku": dw_sku, "mfr_sku": mfr, "pid": pid, "pid_num": pid.split("/")[-1],
"handle": prod["handle"], "title": title,
}
# -----------------------------------------------------------------------
def run_full_monty(mfr):
try:
body = json.dumps({"mfr_sku": mfr}).encode()
req = urllib.request.Request("http://45.61.58.125:9750/api/update",
data=body, method="POST",
headers={"Authorization": AGENT_AUTH, "Content-Type": "application/json"})
r = json.loads(urllib.request.urlopen(req, timeout=120).read())
return r.get("error") is None
except Exception as e:
print(f" FM err: {e}")
return False
def activate(pid):
gql(PRODUCT_UPDATE, {"p": {"id": pid, "status": "ACTIVE"}})
# -----------------------------------------------------------------------
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--limit", type=int, default=10)
ap.add_argument("--activate", action="store_true", help="flip to ACTIVE after push")
args = ap.parse_args()
items = fetch_batch(args.limit)
print(f"[batch] {len(items)} items")
results = []
for it in items:
try:
res = push_one(it)
results.append(res)
except Exception as e:
print(f"FAIL {it.get('dw_sku')}: {e}")
print(f"\n[phase 1 created] {len(results)}/{len(items)}")
# Phase 2: Full Monty in batch (use the /api/update endpoint per SKU)
print("\n[phase 2] Full Monty ...")
for r in results:
print(f" FM {r['mfr_sku']} ...", end="", flush=True)
ok = run_full_monty(r["mfr_sku"])
print(" ok" if ok else " FAIL")
# Phase 3: finalize (title/SEO/body/tags sanitize)
print("\n[phase 3] finalize ...")
dw_skus = [r["dw_sku"] for r in results]
subprocess.run(["python3", "kravet_finalize.py"] + dw_skus, check=False)
# Phase 4: activate
if args.activate:
print("\n[phase 4] activate ...")
for r in results:
try:
activate(r["pid"])
print(f" ACTIVE {r['dw_sku']}")
except Exception as e:
print(f" activate fail {r['dw_sku']}: {e}")
print("\n=== URLs ===")
for r in results:
print(f" {r['dw_sku']:15s} https://admin.shopify.com/store/designer-laboratory-sandbox/products/{r['pid_num']}")
with open("bulk_push_results.jsonl","a") as f:
for r in results:
f.write(json.dumps(r) + "\n")
if __name__ == "__main__":
main()