`exec/ws1_reprice.py`

198 lines
#!/usr/bin/env python3
"""WS1 — Elitis reprice: rename option Title->Size, add ONE sellable variant per product.
Recipe (Steve-confirmed): Size option; keep Sample untouched; add sellable
{DW}-{Roll|Yard|Panel} at retail, CONTINUE, tracked, min/step=1 (NOT double-roll).
Neutralize the known-bad blanket qty metafields (min/units 2 -> 1) on touched products.
Idempotent + resumable: skip any product already carrying a non-sample variant.
$0 (Shopify Admin API is free)."""
import csv, sys, os, json, time, datetime
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "lib"))
from shopify import gql, find_product_by_title

LEDGER = os.path.join(os.path.dirname(__file__), "ledger", "ws1_priced.csv")
CSV = os.path.join(os.path.dirname(__file__), "..", "OUT_223_priced.csv")

UNIT_LABEL = {"Roll": "Sold Per Roll", "Yard": "Sold Per Yard", "Panel": "Sold Per Panel"}
UNIT_SKU = {"Roll": "Roll", "Yard": "Yard", "Panel": "Panel"}

Q_PROD = """query($id:ID!){ product(id:$id){
  id legacyResourceId title status handle
  options{ id name position values }
  variants(first:30){ edges{ node{ id title sku price inventoryPolicy
    inventoryItem{ id tracked } selectedOptions{name value} } } }
  metafields(first:60){ edges{ node{ id namespace key value type } } }
} }"""

M_OPT_RENAME = """mutation($pid:ID!,$oid:ID!,$name:String!){
  productOptionUpdate(productId:$pid, option:{id:$oid, name:$name}){
    userErrors{ field message } product{ options{ id name values } } } }"""

M_VAR_CREATE = """mutation($pid:ID!,$variants:[ProductVariantsBulkInput!]!){
  productVariantsBulkCreate(productId:$pid, variants:$variants){
    userErrors{ field message }
    productVariants{ id title sku price inventoryPolicy
      inventoryItem{ tracked } selectedOptions{name value} } } }"""

M_MF_SET = """mutation($mf:[MetafieldsSetInput!]!){
  metafieldsSet(metafields:$mf){ userErrors{ field message } metafields{ key value } } }"""


def prod_by_gid(gid):
    return gql(Q_PROD, {"id": gid})["product"]


def dw_sku_from_sample(node):
    for e in node["variants"]["edges"]:
        n = e["node"]
        sk = n["sku"] or ""
        if sk.endswith("-Sample") or n["title"] == "Sample":
            if sk.endswith("-Sample"):
                return sk[:-len("-Sample")]
    return None


def has_nonsample_variant(node):
    for e in node["variants"]["edges"]:
        n = e["node"]
        sk = n["sku"] or ""
        if n["title"] != "Sample" and not sk.endswith("-Sample"):
            return True
    return False


def ledger_seen():
    seen = set()
    if os.path.exists(LEDGER):
        for r in csv.DictReader(open(LEDGER)):
            if r.get("product_id") and r.get("status", "").startswith("OK"):
                seen.add(r["product_id"])
    return seen


def append_ledger(row):
    exists = os.path.exists(LEDGER) and os.path.getsize(LEDGER) > 0
    with open(LEDGER, "a", newline="") as f:
        w = csv.writer(f)
        if not exists:
            w.writerow(["ts", "mfr_sku", "title", "product_id", "dw_sku", "unit",
                        "retail", "sellable_variant_id", "status"])
        w.writerow(row)


def process(r, gid_override=None, verbose=False):
    ts = datetime.datetime.now().isoformat(timespec="seconds")
    title = r["title"]; unit = r["unit"]; mfr = r["mfr_sku"]
    try:
        retail = round(float(r["retail"]), 2)
    except Exception:
        append_ledger([ts, mfr, title, "", "", unit, r.get("retail"), "", "SKIP:bad-retail"])
        return "SKIP:bad-retail"

    if gid_override:
        node = prod_by_gid(gid_override)
    else:
        found = find_product_by_title(title)
        if not found:
            append_ledger([ts, mfr, title, "", "", unit, retail, "", "SKIP:not-found"])
            return "SKIP:not-found"
        # re-fetch full node by gid so option ids + full metafields are present
        node = prod_by_gid(found["id"])
    if not node:
        append_ledger([ts, mfr, title, "", "", unit, retail, "", "SKIP:not-found"])
        return "SKIP:not-found"
    gid = node["id"]; pid = str(node["legacyResourceId"])

    if has_nonsample_variant(node):
        append_ledger([ts, mfr, title, pid, "", unit, retail, "", "SKIP:already-has-sellable"])
        return "SKIP:already-has-sellable"

    dw = dw_sku_from_sample(node)
    if not dw:
        append_ledger([ts, mfr, title, pid, "", unit, retail, "", "SKIP:no-dw-sku"])
        return "SKIP:no-dw-sku"

    label = UNIT_LABEL.get(unit)
    if not label:
        append_ledger([ts, mfr, title, pid, dw, unit, retail, "", f"SKIP:bad-unit:{unit}"])
        return f"SKIP:bad-unit:{unit}"

    # 1) rename option Title -> Size (idempotent)
    opt = node["options"][0]
    if opt["name"] != "Size":
        d = gql(M_OPT_RENAME, {"pid": gid, "oid": opt["id"], "name": "Size"})
        ue = d["productOptionUpdate"]["userErrors"]
        if ue:
            append_ledger([ts, mfr, title, pid, dw, unit, retail, "", "FAIL:opt-rename:" + json.dumps(ue)[:120]])
            return "FAIL:opt-rename"

    # 2) add sellable variant
    sellable_sku = f"{dw}-{UNIT_SKU[unit]}"
    variants = [{
        "price": f"{retail:.2f}",
        "optionValues": [{"optionName": "Size", "name": label}],
        "inventoryPolicy": "CONTINUE",
        "inventoryItem": {"sku": sellable_sku, "tracked": True},
    }]
    d = gql(M_VAR_CREATE, {"pid": gid, "variants": variants})
    res = d["productVariantsBulkCreate"]
    ue = res["userErrors"]
    if ue:
        append_ledger([ts, mfr, title, pid, dw, unit, retail, "", "FAIL:var-create:" + json.dumps(ue)[:160]])
        return "FAIL:var-create"
    new_var = res["productVariants"][0]
    vid = new_var["id"]

    # 3) neutralize known-bad blanket qty metafields (min/units 2 -> 1) so PDP min=1
    mf = [
        {"ownerId": gid, "namespace": "global", "key": "v_prods_quantity_order_min",
         "type": "single_line_text_field", "value": "1"},
        {"ownerId": gid, "namespace": "global", "key": "v_prods_quantity_order_units",
         "type": "single_line_text_field", "value": "1"},
    ]
    dm = gql(M_MF_SET, {"mf": mf})
    mue = dm["metafieldsSet"]["userErrors"]
    mstat = "OK" if not mue else "OK(mf-warn)"

    append_ledger([ts, mfr, title, pid, dw, unit, retail, vid, mstat])
    if verbose:
        print(json.dumps({"gid": gid, "dw": dw, "sellable_sku": sellable_sku,
                          "new_variant": new_var}, indent=2, ensure_ascii=False))
    return mstat


if __name__ == "__main__":
    args = sys.argv[1:]
    canary_gid = None
    only_title = None
    if "--canary" in args:
        canary_gid = "gid://shopify/Product/7789643792435"
        only_title = "Lin Expérience Minimaliste | Elitis"

    rows = list(csv.DictReader(open(CSV)))
    seen = ledger_seen()

    if canary_gid:
        r = next(x for x in rows if x["mfr_sku"] == "ELIT-VP-953-22")
        print("=== WS1 CANARY:", r["title"], "===")
        st = process(r, gid_override=canary_gid, verbose=True)
        print("RESULT:", st)
        sys.exit(0)

    # full run (skips canary already done via ledger)
    BAD_MFR = {"ELIT-VP-1036-44"}  # duplicate mfr_sku -> handled separately/flagged
    n_ok = n_skip = n_fail = 0
    for i, r in enumerate(rows):
        # dedupe: skip if this exact title already landed in this run's seen set
        node_pid_key = r["title"]
        st = process(r)
        if st.startswith("OK"):
            n_ok += 1
        elif st.startswith("FAIL"):
            n_fail += 1
        else:
            n_skip += 1
        if (i + 1) % 20 == 0:
            print(f"  ...{i+1}/{len(rows)}  ok={n_ok} skip={n_skip} fail={n_fail}")
        time.sleep(0.15)
    print(f"DONE: ok={n_ok} skip={n_skip} fail={n_fail} total={len(rows)}")