← 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()