`job2_vendor.py`

153 lines
#!/usr/bin/env python3
"""JOB 2 — scrub banned word 'Wallpaper' from generic-descriptor vendor field.
DTD verdict A: rename all 5 generic vendors AND update the 3 coupled
smart-collection VENDOR rule conditions in lockstep so membership is preserved.
HOLD proper-noun brand vendors (Scalamandre/Roberto Cavalli/Laura Ashley/
Missoni/MC Escher Wallpaper) — never touched.
PG-first mirror to dw_unified.shopify_products, then Shopify.
ACTIVE products only. Vendor edits create NO variants.
"""
import json, re, sys, time, urllib.request, subprocess
from collections import Counter

TOKEN = subprocess.check_output(
    "grep -E '^SHOPIFY_ADMIN_TOKEN=' ~/Projects/secrets-manager/.env | cut -d= -f2",
    shell=True, text=True).strip()
DOMAIN = "designer-laboratory-sandbox.myshopify.com"
VER = "2024-10"
URL = f"https://{DOMAIN}/admin/api/{VER}/graphql.json"

# generic vendor -> new value (proper-noun brands deliberately ABSENT = HELD)
RENAME = {
    "Malibu Wallpaper": "Malibu Wallcoverings",
    "PS Removable Wallpaper": "PS Removable Wallcoverings",
    "Apartment Wallpaper": "Apartment Wallcoverings",
    "Wallpaper NYC": "Wallcovering NYC",
    "DW Exclusive Wallpaper": "DW Exclusive Wallcoverings",
}
HELD = ["Scalamandre Wallpaper", "Roberto Cavalli Wallpaper", "Laura Ashley Wallpaper",
        "Missoni Wallpaper", "MC Escher Wallpaper"]

def gql(query, variables=None):
    body = {"query": query}
    if variables is not None:
        body["variables"] = variables
    req = urllib.request.Request(URL, data=json.dumps(body).encode(),
        headers={"X-Shopify-Access-Token": TOKEN, "Content-Type": "application/json"})
    for attempt in range(6):
        try:
            d = json.load(urllib.request.urlopen(req, timeout=60))
            cost = d.get("extensions", {}).get("cost", {}).get("throttleStatus", {})
            if cost.get("currentlyAvailable", 1000) < 300:
                time.sleep(2)
            return d
        except urllib.error.HTTPError as e:
            if e.code == 429:
                time.sleep(3 * (attempt + 1)); continue
            time.sleep(2 * (attempt + 1))
        except Exception:
            time.sleep(2 * (attempt + 1))
    raise RuntimeError("gql failed")

PROD_UPDATE = """mutation($input:ProductInput!){ productUpdate(input:$input){ product{id vendor title} userErrors{field message} } }"""

def fetch_vendor_products(vendor):
    """All ACTIVE product gids+titles for a vendor (exact match)."""
    out = []
    cursor = None
    while True:
        after = ', after:"%s"' % cursor if cursor else ''
        q = ('{ products(first:250, query:%s%s){ pageInfo{hasNextPage endCursor} '
             'edges{ node{ id title vendor status } } } }') % (
             json.dumps('status:active AND vendor:"%s"' % vendor), after)
        d = gql(q)["data"]["products"]
        for e in d["edges"]:
            n = e["node"]
            if n["vendor"] == vendor and n["status"] == "ACTIVE":
                out.append({"id": n["id"], "title": n["title"]})
        if not d["pageInfo"]["hasNextPage"]:
            break
        cursor = d["pageInfo"]["endCursor"]
    return out

def mirror_pg(old_vendor, new_vendor):
    subprocess.run(
        ["psql", "-d", "dw_unified", "-v", "ON_ERROR_STOP=1", "-c",
         "UPDATE shopify_products SET vendor=$$%s$$ WHERE status='ACTIVE' AND vendor=$$%s$$;"
         % (new_vendor, old_vendor)],
        check=True, capture_output=True, text=True)

# --- smart collection rule update ---
COLL_RULE_Q = """{ collection(id:"%s"){ id title ruleSet{ appliedDisjunctively rules{ column relation condition } } } }"""
COLL_UPDATE = """mutation($input:CollectionInput!){ collectionUpdate(input:$input){ collection{id title} userErrors{field message} } }"""

def update_collection_rule(coll_gid, old_cond, new_cond):
    """Swap a single VENDOR EQUALS rule's condition old->new, preserving all others."""
    cur = gql(COLL_RULE_Q % coll_gid)["data"]["collection"]
    rs = cur["ruleSet"]
    new_rules = []
    changed = False
    for r in rs["rules"]:
        if r["column"] == "VENDOR" and r["relation"] == "EQUALS" and r["condition"] == old_cond:
            new_rules.append({"column": "VENDOR", "relation": "EQUALS", "condition": new_cond})
            changed = True
        else:
            new_rules.append({"column": r["column"], "relation": r["relation"], "condition": r["condition"]})
    if not changed:
        return {"status": "RULE_NOT_FOUND", "coll": cur["title"]}
    inp = {"id": coll_gid, "ruleSet": {
        "appliedDisjunctively": rs["appliedDisjunctively"], "rules": new_rules}}
    res = gql(COLL_UPDATE, {"input": inp})["data"]["collectionUpdate"]
    return {"status": "RULE_UPDATED" if not res["userErrors"] else "RULE_ERROR",
            "coll": cur["title"], "errors": res["userErrors"]}

def rename_vendor(old_vendor, new_vendor, coupled_collections, apply=False, gap=0.55):
    """coupled_collections = list of {id(gid), title}. Update rules FIRST, then rename products."""
    rec = {"old": old_vendor, "new": new_vendor, "coupled": [], "renamed": 0,
           "errors": [], "status": "PLANNED"}
    prods = fetch_vendor_products(old_vendor)
    rec["product_count"] = len(prods)
    if apply:
        # 1) update coupled collection rules to new condition first
        for c in coupled_collections:
            r = update_collection_rule(c["id"], old_vendor, new_vendor)
            rec["coupled"].append(r)
            if r["status"] not in ("RULE_UPDATED",):
                rec["errors"].append(r)
            time.sleep(0.4)
        # 2) PG mirror
        mirror_pg(old_vendor, new_vendor)
        # 3) rename each product's vendor
        for i, p in enumerate(prods):
            res = gql(PROD_UPDATE, {"input": {"id": p["id"], "vendor": new_vendor}})["data"]["productUpdate"]
            if res["userErrors"]:
                rec["errors"].append({"id": p["id"], "errs": res["userErrors"]})
            else:
                rec["renamed"] += 1
            time.sleep(gap)
            if (i + 1) % 50 == 0:
                sys.stderr.write(f"    {old_vendor}: {i+1}/{len(prods)}\n")
        rec["status"] = "APPLIED" if not rec["errors"] else "PARTIAL"
    return rec

if __name__ == "__main__":
    coupled_map = json.load(open("/tmp/job2_coupled.json"))
    # dedup coupled collections per vendor by gid
    def coup(v):
        seen = {}
        for c in coupled_map.get(v, []):
            seen[c["id"]] = {"id": c["id"], "title": c["title"]}
        return list(seen.values())

    mode = sys.argv[1] if len(sys.argv) > 1 else "plan"
    if mode == "plan":
        for v, nv in RENAME.items():
            r = rename_vendor(v, nv, coup(v), apply=False)
            print(f"{r['product_count']:>6}  {v!r} -> {nv!r}  coupled={len(coup(v))}")
        print("HELD:", HELD)
    elif mode == "one":
        v = sys.argv[2]; nv = RENAME[v]
        r = rename_vendor(v, nv, coup(v), apply=True)
        json.dump(r, open(f"/tmp/job2_{v.replace(' ','_')}.json", "w"), indent=2)
        print("STATUS:", r["status"], "renamed:", r["renamed"], "coupled:", r["coupled"], "errors:", len(r["errors"]))