← back to Kravet Sheet Sync 2026 04 20
kravet_finalize.py
244 lines
#!/usr/bin/env python3
"""
Kravet finalizer — runs AFTER the Full Monty agent.
Generates the fields the shared agent's onboard_enrich can't:
- Proper title: "Pattern Color | Brand" (CLAUDE.md title format)
- SEO title: "Shop Pattern in Color | Brand"
- SEO desc: 30-160 chars from AI + specs
- body_html: description with specs + compliance notes
Sanitizes banned word "wallpaper" → "wallcovering" everywhere.
Usage:
python3 kravet_finalize.py DWKK-141031 DWKK-141042
python3 kravet_finalize.py --vendor-prefix DWKK --status draft (batch)
"""
import json, re, subprocess, sys, urllib.request, argparse
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']
SMALL = {"a","an","the","and","or","but","of","in","on","at","for","to","with","by"}
def pg(sql):
"""Run psql, return rows as list-of-list-of-cells. Uses CSV format to avoid
delimiter collisions with pipe-containing data (e.g. product titles)."""
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 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:
# keep & and / as-is; cap letters after them
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_wallpaper(s):
if not s: return s
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_context(dw_skus):
"""Return list of dicts joining shopify_products + kravet_sheet_fields."""
skus_csv = ",".join(f"'{s}'" for s in dw_skus)
rows = pg(f"""
SELECT sp.dw_sku, sp.shopify_id, sp.mfr_sku, sp.title AS current_title,
sp.vendor AS sp_vendor, sp.tags AS sp_tags,
sf.pattern AS sf_pattern, sf.color AS sf_color, sf.brand AS sf_brand,
sf.collection, sf.country_of_origin, sf.width, sf.width_uom,
sf.unit_of_measure, sf.content, sf.type1, sf.style1,
sf.vert_repeat, sf.horz_repeat,
sf.prop_65, sf.ca_tb117, sf.direction, sf.clean_code, sf.durability,
sf.lead_time_days, sf.inventory_available, sf.display_status
FROM shopify_products sp
LEFT JOIN kravet_sheet_fields sf
ON sf.mfr_sku_norm = regexp_replace(
regexp_replace(upper(btrim(sp.mfr_sku)),'-','.','g'),
'\\.0\$','')
WHERE sp.dw_sku IN ({skus_csv})
""")
cols = ["dw_sku","shopify_id","mfr_sku","current_title","sp_vendor","sp_tags",
"sf_pattern","sf_color","sf_brand","collection","country_of_origin",
"width","width_uom","unit_of_measure","content","type1","style1",
"vert_repeat","horz_repeat","prop_65","ca_tb117","direction","clean_code",
"durability","lead_time_days","inventory_available","display_status"]
return [dict(zip(cols, r)) for r in rows]
def fetch_ai_tags(pid):
q = """query($id:ID!){ product(id:$id){ tags } }"""
return gql(q, {"id": pid})["product"]["tags"]
def build_title(ctx):
pat = title_case(ctx["sf_pattern"] or "")
col = title_case(ctx["sf_color"] or "")
brn = title_case(ctx["sf_brand"] or ctx["sp_vendor"] or "Kravet")
title = " ".join(x for x in [pat, col] if x)
title = f"{title} | {brn}" if title else brn
return clean_wallpaper(title)
def build_seo_title(ctx):
# SEO must sell DW service — never plain "Shop X | Brand".
pat = title_case(ctx["sf_pattern"] or "")
col = title_case(ctx["sf_color"] or "")
brn = title_case(ctx["sf_brand"] or ctx["sp_vendor"] or "Kravet")
base = f"{pat} {col} {brn}".strip() or brn
st = f"{base} — Free Samples"
return clean_wallpaper(st)[:70]
def build_seo_desc(ctx, ai_tags):
pat = title_case(ctx["sf_pattern"] or "")
col = title_case(ctx["sf_color"] or "")
brn = title_case(ctx["sf_brand"] or "")
desc = (f"Order free samples of {pat} in {col} by {brn}. "
f"Free shipping, trade pricing, and expert guidance from Designer Wallcoverings.")
return clean_wallpaper(desc)[:160]
def build_body_html(ctx):
# Editorial prose only. NO specs in body — every spec lives in custom.* metafields.
pat = title_case(ctx["sf_pattern"] or "")
col = title_case(ctx["sf_color"] or "")
brn = title_case(ctx["sf_brand"] or "")
t1 = (title_case(ctx["type1"]) or "Wallcovering").lower()
coll = title_case(ctx["collection"] or "")
origin = title_case(ctx["country_of_origin"] 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 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_wallpaper(opening + body + close)
def sanitize_tags(tags):
"""Strip banned words from tag list, dedupe."""
cleaned = []
seen = set()
for t in tags:
if not t: continue
if t.lower() == "wallpaper":
t = "wallcovering"
key = t.lower()
if key in seen: continue
seen.add(key); cleaned.append(t)
return cleaned
PRODUCT_UPDATE = """
mutation productUpdate($product: ProductUpdateInput!) {
productUpdate(product: $product) {
product { id title handle seo { title description } }
userErrors { field message }
}
}
"""
def finalize_one(ctx):
pid = f"gid://shopify/Product/{ctx['shopify_id']}"
print(f"\n=== {ctx['dw_sku']} | {ctx['mfr_sku']} ===")
# Get current tags from Shopify (AI tagger may have added some)
q_tags = """query($id:ID!){ product(id:$id){ tags } }"""
curr_tags = gql(q_tags, {"id": pid})["product"]["tags"]
clean_tags = sanitize_tags(curr_tags)
if clean_tags != curr_tags:
print(f" tags sanitized: removed {set(curr_tags) - set(clean_tags)}")
title = build_title(ctx)
seo_t = build_seo_title(ctx)
seo_d = build_seo_desc(ctx, clean_tags)
body = build_body_html(ctx)
print(f" title: {title}")
print(f" seo_title: {seo_t}")
print(f" seo_desc: {seo_d}")
print(f" body: {len(body)} chars")
update_input = {
"id": pid,
"title": title,
"descriptionHtml": body,
"seo": {"title": seo_t, "description": seo_d},
"tags": clean_tags,
}
r = gql(PRODUCT_UPDATE, {"product": update_input})
errs = r["productUpdate"]["userErrors"]
if errs:
print(f" ERROR: {errs}")
return
p = r["productUpdate"]["product"]
print(f" updated: {p['handle']}")
def main():
ap = argparse.ArgumentParser()
ap.add_argument("dw_skus", nargs="*")
ap.add_argument("--vendor-prefix")
ap.add_argument("--status", default=None)
args = ap.parse_args()
if args.vendor_prefix:
q = f"SELECT dw_sku FROM shopify_products WHERE vendor_prefix='{args.vendor_prefix}'"
if args.status: q += f" AND status='{args.status}'"
skus = [r[0] for r in pg(q)]
else:
skus = args.dw_skus
if not skus:
print("no SKUs"); sys.exit(1)
ctxs = fetch_context(skus)
for c in ctxs:
try:
finalize_one(c)
except Exception as e:
print(f"FAIL {c.get('dw_sku')}: {e}")
if __name__ == "__main__":
main()