← back to Dw Wallpaper Cleanup
job2_finish.py
188 lines
#!/usr/bin/env python3
"""JOB 2 COMPLETION — idempotent finisher for the 3 coupled vendor renames.
Synchronous. PG-first mirror to dw_unified.shopify_products, then Shopify.
ACTIVE products only. Vendor edits create NO variants.
Per-vendor logic (idempotent):
1. Capture LIVE before: old vendor count, new vendor count, each coupled
collection's productsCount + ruleSet.
2. For each coupled collection: ensure a VENDOR EQUALS <new> rule exists.
- If a VENDOR EQUALS <old> rule exists, REPLACE it with <new>.
- If neither old nor new exists but disjunctive (Malibu case where new
already present), leave as-is.
This guarantees membership is preserved at all times.
3. PG mirror UPDATE old->new for ACTIVE rows.
4. Rename each remaining ACTIVE product old->new via productUpdate, rate-limited.
5. Verify after: old vendor -> 0, collection count did not drop below the
captured floor.
"""
import json, sys, time, subprocess, urllib.request, urllib.error
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"
URL = f"https://{DOMAIN}/admin/api/2024-10/graphql.json"
PLAN = [
{"old": "Malibu Wallpaper", "new": "Malibu Wallcoverings",
"colls": ["gid://shopify/Collection/298507403315"]},
{"old": "Apartment Wallpaper", "new": "Apartment Wallcoverings",
"colls": ["gid://shopify/Collection/298970972211",
"gid://shopify/Collection/299533828147"]},
{"old": "Wallpaper NYC", "new": "Wallcovering NYC",
"colls": ["gid://shopify/Collection/298971430963"]},
]
def gql(q, v=None):
body={"query":q}
if v is not None: body["variables"]=v
req=urllib.request.Request(URL,data=json.dumps(body).encode(),
headers={"X-Shopify-Access-Token":TOKEN,"Content-Type":"application/json"})
for a in range(8):
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*(a+1)); continue
time.sleep(2*(a+1))
except Exception: time.sleep(2*(a+1))
raise RuntimeError("gql failed")
def vcount(v):
return gql('{ productsCount(query:%s){count} }'%json.dumps('status:active AND vendor:"%s"'%v))["data"]["productsCount"]["count"]
def vcount_all(v):
return gql('{ productsCount(query:%s){count} }'%json.dumps('vendor:"%s"'%v))["data"]["productsCount"]["count"]
def coll_state(gid):
q='{ collection(id:"%s"){ id title productsCount{count} ruleSet{ appliedDisjunctively rules{column relation condition} } } }'%gid
return gql(q)["data"]["collection"]
PROD_UPDATE="""mutation($input:ProductInput!){ productUpdate(input:$input){ product{id vendor} userErrors{field message} } }"""
COLL_UPDATE="""mutation($input:CollectionInput!){ collectionUpdate(input:$input){ collection{id title} userErrors{field message} } }"""
def ensure_vendor_rule(gid, old, new):
"""Make sure collection has VENDOR EQUALS <new>; replace VENDOR EQUALS <old> if present.
Returns action taken."""
c=coll_state(gid)
rs=c.get("ruleSet")
if not rs:
return {"coll":gid,"action":"NO_RULESET_SKIP"}
rules=rs["rules"]
has_new=any(r["column"]=="VENDOR" and r["relation"]=="EQUALS" and r["condition"]==new for r in rules)
has_old=any(r["column"]=="VENDOR" and r["relation"]=="EQUALS" and r["condition"]==old for r in rules)
if has_new and not has_old:
return {"coll":c["title"],"action":"ALREADY_NEW"}
new_rules=[]
for r in rules:
if r["column"]=="VENDOR" and r["relation"]=="EQUALS" and r["condition"]==old:
if not has_new:
new_rules.append({"column":"VENDOR","relation":"EQUALS","condition":new})
# if has_new already, just drop the old one (dedup)
else:
new_rules.append({"column":r["column"],"relation":r["relation"],"condition":r["condition"]})
inp={"id":gid,"ruleSet":{"appliedDisjunctively":rs["appliedDisjunctively"],"rules":new_rules}}
res=gql(COLL_UPDATE,{"input":inp})["data"]["collectionUpdate"]
return {"coll":c["title"],"action":"RULE_SWAPPED" if not res["userErrors"] else "RULE_ERROR",
"errors":res["userErrors"]}
def fetch_all_status(vendor):
"""ALL statuses (active+draft+archived) — DTD verdict A: banned word is
catalog-wide. Vendor edit keeps status unchanged + creates no variants."""
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('vendor:"%s"'%vendor), after)
d=gql(q)["data"]["products"]
for e in d["edges"]:
n=e["node"]
if n["vendor"]==vendor:
out.append(n["id"])
if not d["pageInfo"]["hasNextPage"]: break
cursor=d["pageInfo"]["endCursor"]
return out
def mirror_pg(old,new):
"""All statuses — match the all-status Shopify rename."""
try:
subprocess.run(["psql","-d","dw_unified","-v","ON_ERROR_STOP=1","-c",
"UPDATE shopify_products SET vendor=$$%s$$ WHERE vendor=$$%s$$;"%(new,old)],
check=True, capture_output=True, text=True)
return "OK"
except subprocess.CalledProcessError as e:
return "PG_ERR: "+(e.stderr or "")[:200]
except Exception as e:
return "PG_ERR: "+str(e)[:200]
def run_vendor(p, gap=0.5):
old,new=p["old"],p["new"]
rec={"old":old,"new":new}
rec["before"]={"old_count":vcount(old),"old_count_all":vcount_all(old),
"new_count":vcount(new),"new_count_all":vcount_all(new),
"colls":[{"id":g, **{k:coll_state(g).get(k) for k in ("title",)},
"count":coll_state(g)["productsCount"]["count"]} for g in p["colls"]]}
# capture collection floors
floors={g:coll_state(g)["productsCount"]["count"] for g in p["colls"]}
sys.stderr.write(f"[{old}] before: old={rec['before']['old_count']} new={rec['before']['new_count']} floors={floors}\n")
# 1) ensure rules
rec["rule_actions"]=[]
for g in p["colls"]:
a=ensure_vendor_rule(g, old, new)
rec["rule_actions"].append(a)
sys.stderr.write(f" rule {g}: {a}\n")
time.sleep(0.4)
# 2) PG mirror
rec["pg"]=mirror_pg(old,new)
sys.stderr.write(f" pg mirror: {rec['pg']}\n")
# 3) rename products (all statuses — DTD verdict A)
ids=fetch_all_status(old)
rec["to_rename"]=len(ids)
renamed=0; errors=[]
for i,pid in enumerate(ids):
res=gql(PROD_UPDATE,{"input":{"id":pid,"vendor":new}})["data"]["productUpdate"]
if res["userErrors"]:
errors.append({"id":pid,"errs":res["userErrors"]})
else:
renamed+=1
time.sleep(gap)
if (i+1)%50==0:
sys.stderr.write(f" {old}: {i+1}/{len(ids)} renamed\n")
rec["renamed"]=renamed; rec["rename_errors"]=errors
# 4) verify
time.sleep(3)
after={"old_count":vcount(old),"old_count_all":vcount_all(old),
"new_count":vcount(new),"new_count_all":vcount_all(new),
"colls":[{"id":g,"count":coll_state(g)["productsCount"]["count"]} for g in p["colls"]]}
rec["after"]=after
rec["floors"]=floors
coll_ok=all(after["colls"][i]["count"]>=floors[p["colls"][i]] for i in range(len(p["colls"])))
rec["verify"]={"old_zero_all":after["old_count_all"]==0,"coll_not_dropped":coll_ok,
"new_increased":after["new_count_all"]>=rec["before"]["new_count_all"]}
rec["status"]="OK" if (after["old_count_all"]==0 and coll_ok and not errors) else "CHECK"
sys.stderr.write(f"[{old}] after: old_all={after['old_count_all']} new_all={after['new_count_all']} colls={after['colls']} status={rec['status']}\n")
return rec
if __name__=="__main__":
only=sys.argv[1] if len(sys.argv)>1 else None
results=[]
for p in PLAN:
if only and p["old"]!=only: continue
r=run_vendor(p)
results.append(r)
json.dump(r, open(f"job2_done_{p['old'].replace(' ','_')}.json","w"), indent=2)
json.dump(results, open("job2_finish_results.json","w"), indent=2)
print("\n=== SUMMARY ===")
for r in results:
print(f"{r['old']} -> {r['new']}: status={r['status']} renamed={r['renamed']} "
f"old_after={r['after']['old_count']} new_after={r['after']['new_count']} "
f"colls={[c['count'] for c in r['after']['colls']]}")