← back to Kravet Sheet Sync 2026 04 20
test_push_metafields_only.py
165 lines
#!/usr/bin/env python3
"""Chunked metafields retry for the 2 products already created by test_push_2_items.py."""
import json, urllib.request, subprocess
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"
# From test_push_2_items.py output
PRODUCTS = [
{"dw_sku": "DWKK-141031", "pid": "gid://shopify/Product/7821427441715"},
{"dw_sku": "DWKK-141042", "pid": "gid://shopify/Product/7821427474483"},
]
def gql(query, variables=None):
body = json.dumps({"query": query, "variables": variables or {}}).encode()
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: {data['errors']}")
return data["data"]
def pg_rows(q):
r = subprocess.run(["ssh","root@45.61.58.125",
f'PGPASSWORD=DW2024! psql -h 127.0.0.1 -U dw_admin -d dw_unified -At -F "|" -c "{q}"'],
capture_output=True, text=True, check=True)
return [line.split("|") for line in r.stdout.strip().split("\n") if line]
def title_case(s):
if not s: return s
small = {"a","an","the","and","or","but","of","in","on","at","for","to","with"}
words = s.strip().lower().split()
return " ".join(w if i>0 and w in small else (w[0].upper()+w[1:] if w else w)
for i,w in enumerate(words))
def fetch(dw_sku):
q = f"""
SELECT 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.display_status, n.inventory_available,
n.wholesale::text, n.map::text, n.tariff_pct::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, s.notes, s.wallcovering_area,
s.qty_on_hand, s.memo_sample_qty
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 = '{dw_sku}'
"""
cols = ["mfr_sku","pattern_name","color_name","brand","collection","country_of_origin",
"width","unit_of_measure","content","type1","display_status","inventory_available",
"wholesale","map","tariff_pct","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","notes","wallcovering_area","qty_on_hand","memo_sample_qty"]
rows = pg_rows(q)
return dict(zip(cols, rows[0])) if rows else None
def build_all_metafields(item, pid):
out = []
def add(ns,k,v):
if v is None or v in ("","NULL"): return
out.append({"ownerId": pid, "namespace": ns, "key": k,
"value": str(v).strip(), "type": "single_line_text_field"})
# MANDATORY manufacturer_sku (CLAUDE.md)
add("custom","manufacturer_sku", item["mfr_sku"])
add("dwc", "manufacturer_sku", item["mfr_sku"])
# Real color / pattern / brand (CLAUDE.md)
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 ""))
add("global","Color-Way", title_case(item["color_name"] or ""))
# Core specs
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","Weight", item["weight"])
# Compliance / safety
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("custom","prop_65_chemical", item["prop_65_chemical"])
add("custom","prop_65_effect", item["prop_65_effect"])
add("custom","ab_1817", item["ab_1817"])
# Ordering / fulfillment
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","Minimum", item["minimum_order_qty"])
add("global","v_prods_quantity_order_min", item["minimum_order_qty"])
add("global","v_prods_quantity_order_units", item["order_increment_qty"])
add("global","HORIZONTAL_HALFDROP", item["horizontal_half_drop"])
# Internal / display
add("global","Display-Status", item["display_status"])
add("global","Inventory-Available", item["inventory_available"])
add("global","WHLS-Price", item["wholesale"])
add("custom","vendor_notes", item["notes"])
add("custom","qty_on_hand", item["qty_on_hand"])
add("custom","memo_sample_qty", item["memo_sample_qty"])
add("global","wallcover_length_yd", item["wallcovering_area"])
return out
METAFIELDS_SET = """
mutation metafieldsSet($metafields: [MetafieldsSetInput!]!) {
metafieldsSet(metafields: $metafields) {
metafields { id namespace key }
userErrors { field message }
}
}
"""
for p in PRODUCTS:
item = fetch(p["dw_sku"])
if not item:
print(f" {p['dw_sku']}: not found in PG")
continue
mfs = build_all_metafields(item, p["pid"])
print(f"\n{p['dw_sku']}: {len(mfs)} metafields to set, chunked at 25")
total_ok = 0
errs = []
for i in range(0, len(mfs), 25):
chunk = mfs[i:i+25]
r = gql(METAFIELDS_SET, {"metafields": chunk})
saved = r["metafieldsSet"]["metafields"] or []
ue = r["metafieldsSet"]["userErrors"] or []
total_ok += len(saved)
if ue:
errs.extend(ue)
print(f" saved {total_ok}/{len(mfs)}")
if errs:
print(f" errors: {errs[:3]}")