← back to Kravet Sheet Sync 2026 04 20
test_push_2_items.py
451 lines
#!/usr/bin/env python3
"""
Test push 2 Kravet wallcoverings to Shopify as drafts.
Follows Designer-Wallcoverings CLAUDE.md rules:
- PG first (already done via kravet_sheet_fields)
- Register in dw_sku_registry
- Draft status (no image yet)
- Sample variant at $4.25 with DW-SKU-Sample
- manufacturer_sku metafields (custom + dwc)
- Never use word "wallpaper" — always "wallcovering"
Run: python3 test_push_2_items.py
"""
import json
import os
import sys
import time
import subprocess
import urllib.request
import urllib.error
SHOP = "designer-laboratory-sandbox.myshopify.com"
TOKEN = os.environ.get('SHOPIFY_ADMIN_TOKEN','')
API = "2024-10"
GQL_URL = f"https://{SHOP}/admin/api/{API}/graphql.json"
TEST_SKUS = ["DWKK-141031", "DWKK-141042"] # Acacia Grey&White, Albery Aqua
# -----------------------------------------------------------------------------
def gql(query: str, variables: dict | None = None) -> dict:
body = json.dumps({"query": query, "variables": variables or {}}).encode("utf-8")
req = urllib.request.Request(
GQL_URL,
data=body,
method="POST",
headers={
"Content-Type": "application/json",
"X-Shopify-Access-Token": TOKEN,
},
)
with urllib.request.urlopen(req, timeout=30) as r:
data = json.loads(r.read())
if "errors" in data:
raise RuntimeError(f"GraphQL errors: {json.dumps(data['errors'], indent=2)}")
return data["data"]
def pg(query: str) -> list[list[str]]:
"""Run psql on Kamatera, return rows as list of list-of-cell strings."""
r = subprocess.run(
[
"ssh",
"root@45.61.58.125",
f'PGPASSWORD=DW2024! psql -h 127.0.0.1 -U dw_admin -d dw_unified '
f'-At -F "|" -c "{query}"',
],
capture_output=True,
text=True,
check=True,
)
return [line.split("|") for line in r.stdout.strip().split("\n") if line]
def title_case(s: str) -> str:
"""Simple title case preserving common separators."""
if not s:
return s
s = s.strip().lower()
words = s.split()
small = {"a", "an", "the", "and", "or", "but", "of", "in", "on", "at", "for", "to", "with"}
out = []
for i, w in enumerate(words):
if i > 0 and w in small:
out.append(w)
else:
out.append(w[0].upper() + w[1:] if w else w)
return " ".join(out)
def clean_wallpaper(s: str) -> str:
"""Per CLAUDE.md: banned word, replace with Wallcovering."""
if not s:
return s
import re
s = re.sub(r"\bWallpapers\b", "Wallcoverings", s, flags=re.I)
s = re.sub(r"\bWalls Wallpaper\b", "Wallcoverings", s, flags=re.I)
s = re.sub(r"\bWallpaper\b", "Wallcovering", s, flags=re.I)
return s
# -----------------------------------------------------------------------------
def fetch_items(dw_skus: list[str]) -> list[dict]:
"""Fetch each DWKK candidate with all needed fields joined from sheet_fields + new bucket."""
skus_csv = ",".join(f"'{s}'" for s in dw_skus)
q = f"""
SELECT
n.new_dw_sku,
n.mfr_sku,
COALESCE(NULLIF(n.pattern_name,''), n.mfr_sku) AS 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,
n.map::text,
n.tariff_pct::text,
n.cost::text,
s.vert_repeat,
s.horz_repeat,
s.repeat_uom,
s.width_uom,
s.weight,
s.weight_uom,
s.prop_65,
s.prop_65_chemical,
s.prop_65_effect,
s.ca_tb117,
s.ab_1817,
s.ufac,
s.direction,
s.lead_time_days,
s.minimum_order_qty,
s.order_increment_qty,
s.horizontal_half_drop,
s.barcode,
s.memo_sample_available,
s.ship_from,
s.clean_code,
s.durability
FROM kravet_diff_new_2026_04_20 n
JOIN kravet_sheet_fields s ON s.mfr_sku_norm = regexp_replace(
regexp_replace(upper(btrim(n.mfr_sku)),'-','.','g'),
'\\.0\$',''
)
WHERE n.new_dw_sku IN ({skus_csv})
ORDER BY n.new_dw_sku
"""
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",
"vert_repeat","horz_repeat","repeat_uom","width_uom","weight","weight_uom",
"prop_65","prop_65_chemical","prop_65_effect","ca_tb117","ab_1817","ufac",
"direction","lead_time_days","minimum_order_qty","order_increment_qty",
"horizontal_half_drop","barcode","memo_sample_available","ship_from",
"clean_code","durability",
]
rows = pg(q)
return [dict(zip(cols, r)) for r in rows]
def register_sku(dw_sku: str, mfr_sku: str, brand: str) -> None:
"""INSERT into dw_sku_registry (skip if exists)."""
esc_brand = (brand or "KRAVET").replace("'", "''")
esc_mfr = mfr_sku.replace("'", "''")
q = f"""
INSERT INTO dw_sku_registry (dw_sku, vendor_prefix, vendor_name, mfr_sku, created_at)
VALUES ('{dw_sku}', 'DWKK', '{esc_brand}', '{esc_mfr}', now())
ON CONFLICT (dw_sku) DO NOTHING
"""
pg(q)
def build_title(item: dict) -> str:
pattern = title_case(item["pattern_name"] or "")
color = title_case(item["color_name"] or "")
brand = title_case(item["brand"] or "Kravet")
title = f"{pattern} {color} | {brand}".strip()
return clean_wallpaper(title)
def build_tags(item: dict) -> list[str]:
tags = [
"Kravet",
title_case(item["brand"] or "Kravet"),
"Wallcovering",
"Test-Import-2026-04-20",
]
if item["type1"]:
tags.append(title_case(item["type1"]))
if item["inventory_available"]:
tags.append(title_case(item["inventory_available"]))
if item["country_of_origin"]:
tags.append(f"Origin: {title_case(item['country_of_origin'])}")
return [t for t in tags if t]
def build_metafields(item: dict) -> list[dict]:
"""Build metafield definitions per CLAUDE.md global.* conventions."""
mfs: list[dict] = []
def add(ns: str, key: str, val, mtype="single_line_text_field"):
if val is None or val == "" or val == "NULL":
return
v = str(val).strip()
if not v:
return
mfs.append({"namespace": ns, "key": key, "value": v, "type": mtype})
# MANDATORY per CLAUDE.md
add("custom", "manufacturer_sku", item["mfr_sku"])
add("dwc", "manufacturer_sku", item["mfr_sku"])
add("custom", "color", title_case(item["color_name"] or ""))
add("custom", "real_color_name", title_case(item["color_name"] or ""))
add("dwc", "color", title_case(item["color_name"] or ""))
add("dwc", "pattern_name", title_case(item["pattern_name"] or ""))
add("dwc", "brand", title_case(item["brand"] or ""))
add("dwc", "real_vendor", title_case(item["brand"] or ""))
# Sheet fields → global.* (subset — 20 most relevant)
add("global", "Brand", title_case(item["brand"] or ""))
add("global", "Collection", title_case(item["collection"] or ""))
add("global", "width", item["width"])
add("global", "unit_of_measure", item["unit_of_measure"])
add("global", "Country-of-Origin", title_case(item["country_of_origin"] or ""))
add("global", "Content", item["content"])
add("global", "Type1", title_case(item["type1"] or ""))
add("global", "Vert-Rpt", item["vert_repeat"])
add("global", "Horz-Rpt", item["horz_repeat"])
add("global", "Direction", item["direction"])
add("global", "Prop-65", item["prop_65"])
add("global", "CA-TB117", item["ca_tb117"])
add("global", "UFAC", item["ufac"])
add("global", "Clean-Code", item["clean_code"])
add("global", "Durability", item["durability"])
add("global", "Weight", item["weight"])
add("global", "Lead-Time-in-Days", item["lead_time_days"])
add("global", "Ship-From", item["ship_from"])
add("global", "Memo-Sample-Available", item["memo_sample_available"])
add("global", "Display-Status", item["display_status"])
add("global", "Inventory-Available", item["inventory_available"])
add("global", "WHLS-Price", item["wholesale"])
add("global", "Minimum", item["minimum_order_qty"])
return mfs
# -----------------------------------------------------------------------------
PRODUCT_CREATE = """
mutation productCreate($product: ProductCreateInput!) {
productCreate(product: $product) {
product {
id
handle
title
status
onlineStoreUrl
variants(first: 5) { nodes { id title sku price } }
}
userErrors { field message }
}
}
"""
VARIANT_UPDATE = """
mutation productVariantsBulkUpdate($productId: ID!, $variants: [ProductVariantsBulkInput!]!) {
productVariantsBulkUpdate(productId: $productId, variants: $variants) {
product { id }
productVariants { id sku price inventoryItem { id unitCost { amount } } }
userErrors { field message }
}
}
"""
VARIANT_CREATE = """
mutation productVariantsBulkCreate($productId: ID!, $variants: [ProductVariantsBulkInput!]!) {
productVariantsBulkCreate(productId: $productId, variants: $variants) {
product { id }
productVariants { id sku price }
userErrors { field message }
}
}
"""
METAFIELDS_SET = """
mutation metafieldsSet($metafields: [MetafieldsSetInput!]!) {
metafieldsSet(metafields: $metafields) {
metafields { id namespace key }
userErrors { field message }
}
}
"""
def push_one(item: dict) -> dict:
dw_sku = item["dw_sku"]
title = build_title(item)
tags = build_tags(item)
print(f"\n=== {dw_sku} | {item['mfr_sku']} ===")
print(f" Title: {title}")
print(f" Tags: {tags}")
# 1) Register DW SKU
register_sku(dw_sku, item["mfr_sku"], item["brand"] or "KRAVET")
print(f" registry: DWKK-registered")
# 2) Create product (draft, no variants yet — Shopify creates default)
product_input = {
"title": title,
"vendor": title_case(item["brand"] or "Kravet"),
"productType": "Wallcovering",
"status": "DRAFT",
"tags": tags,
"descriptionHtml": (
f"<p>{title_case(item['pattern_name'] or '')} in "
f"{title_case(item['color_name'] or '')}. "
f"{title_case(item['content'] or '')}. "
f"Width: {item['width']} {item['width_uom'] or 'IN'}.</p>"
),
}
data = gql(PRODUCT_CREATE, {"product": product_input})
errs = data["productCreate"]["userErrors"]
if errs:
raise RuntimeError(f"productCreate errors: {errs}")
prod = data["productCreate"]["product"]
pid = prod["id"]
default_variant_id = prod["variants"]["nodes"][0]["id"]
print(f" created: {pid} ({prod['handle']})")
# 3) Update default variant → main DWKK variant
try:
map_price = float(item["map"]) if item["map"] not in (None, "", "NULL") else 0.0
cost_price = float(item["cost"]) if item["cost"] not in (None, "", "NULL") else None
except ValueError:
map_price = 0.0
cost_price = None
main_variant_input = {
"id": default_variant_id,
"price": f"{map_price:.2f}",
"inventoryItem": {
"sku": dw_sku,
"tracked": True,
},
}
if cost_price is not None:
main_variant_input["inventoryItem"]["cost"] = f"{cost_price:.2f}"
if item.get("barcode"):
main_variant_input["barcode"] = str(item["barcode"]).strip()
vu = gql(VARIANT_UPDATE, {"productId": pid, "variants": [main_variant_input]})
errs = vu["productVariantsBulkUpdate"]["userErrors"]
if errs:
raise RuntimeError(f"variantUpdate errors: {errs}")
print(f" main variant: {dw_sku} @ ${map_price:.2f} (cost ${cost_price:.2f})")
# 4) Create Sample variant
sample_input = {
"price": "4.25",
"inventoryItem": {
"sku": f"{dw_sku}-Sample",
"tracked": False,
},
"optionValues": [{"optionName": "Title", "name": "Sample"}],
}
vc = gql(VARIANT_CREATE, {"productId": pid, "variants": [sample_input]})
errs = vc["productVariantsBulkCreate"]["userErrors"]
if errs:
# Sample variant creation can fail if default variant is "Default Title"
# and Shopify auto-merges; that's ok for the test. Log and continue.
print(f" sample variant skipped: {errs}")
else:
print(f" sample variant: {dw_sku}-Sample @ $4.25")
# 5) Metafields
mfs = build_metafields(item)
payload = [{"ownerId": pid, **m} for m in mfs]
mr = gql(METAFIELDS_SET, {"metafields": payload})
errs = mr["metafieldsSet"]["userErrors"]
if errs:
print(f" metafields userErrors: {errs}")
print(f" metafields set: {len(mr['metafieldsSet']['metafields'])} / {len(mfs)}")
# 6) Persist back to kravet_catalog
shopify_id_numeric = pid.split("/")[-1]
esc_pattern = (item["pattern_name"] or "").replace("'", "''")
esc_color = (item["color_name"] or "").replace("'", "''")
esc_brand = (item["brand"] or "").replace("'", "''")
esc_mfr = item["mfr_sku"].replace("'", "''")
q = 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
) VALUES (
'{esc_mfr}', '{esc_pattern}', '{esc_color}', '{esc_brand}',
'{dw_sku}', '{pid}',
true,
{item['map'] or 'NULL'}, {item['cost'] or 'NULL'}, {item['tariff_pct'] or 'NULL'},
'draft', now()
)
ON CONFLICT (mfr_sku) DO UPDATE SET
dw_sku = EXCLUDED.dw_sku,
shopify_product_id = EXCLUDED.shopify_product_id,
on_shopify = true,
status = 'draft',
imported_at = now()
"""
pg(q)
print(f" kravet_catalog: upserted")
return {
"dw_sku": dw_sku,
"mfr_sku": item["mfr_sku"],
"shopify_id": pid,
"shopify_numeric_id": shopify_id_numeric,
"handle": prod["handle"],
"title": prod["title"],
"admin_url": f"https://admin.shopify.com/store/designer-laboratory-sandbox/products/{shopify_id_numeric}",
}
# -----------------------------------------------------------------------------
def main():
items = fetch_items(TEST_SKUS)
if len(items) != len(TEST_SKUS):
print(f"WARNING: requested {len(TEST_SKUS)} but found {len(items)} in DB")
for i in items:
print(f" - {i['dw_sku']} {i['mfr_sku']}")
results = []
for item in items:
try:
res = push_one(item)
results.append(res)
except Exception as e:
print(f"FAILED: {item.get('dw_sku')}: {e}")
print("\n=== SUMMARY ===")
for r in results:
print(f" {r['dw_sku']}: {r['admin_url']}")
print(f" title: {r['title']}")
print(f" handle: {r['handle']}")
with open("test_push_results.json", "w") as f:
json.dump(results, f, indent=2)
print(f"\nWritten: test_push_results.json")
if __name__ == "__main__":
main()