`job1_tags.py`

129 lines
#!/usr/bin/env python3
"""JOB 1 — fix banned 'wallpaper' word in TAGS of 238 active DW products.
Whole-word, case-preserving: wallpaper->wallcovering, wallpapers->wallcoverings.
PG-first mirror to dw_unified.shopify_products, then Shopify tagsAdd/tagsRemove.
ACTIVE products only. Tag/vendor edits create NO variants.
"""
import json, re, sys, time, urllib.request, subprocess

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"

# whole-word, case-preserving replacement
def fix_tag(t):
    def f(m):
        s = m.group(0)
        rep = 'wallcoverings' if s.lower().endswith('s') else 'wallcovering'
        if s[0].isupper():
            rep = rep.capitalize()
        return rep
    return re.sub(r'[Ww]allpapers?', f, t)

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))
            # throttle backoff
            cost = d.get("extensions", {}).get("cost", {}).get("throttleStatus", {})
            if cost.get("currentlyAvailable", 1000) < 200:
                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")

TAGS_ADD = """mutation($id:ID!,$tags:[String!]!){ tagsAdd(id:$id, tags:$tags){ userErrors{field message} } }"""
TAGS_REMOVE = """mutation($id:ID!,$tags:[String!]!){ tagsRemove(id:$id, tags:$tags){ userErrors{field message} } }"""

def fetch_product(handle):
    q = '{ products(first:1, query:"handle:%s") { edges { node { id handle status tags } } } }' % handle
    edges = gql(q)["data"]["products"]["edges"]
    return edges[0]["node"] if edges else None

def gid_to_numeric(gid):
    return gid.rsplit("/", 1)[-1]

def mirror_pg(shopify_numeric_id, new_tags_list):
    new_tags_str = ", ".join(new_tags_list)
    # match either bare numeric id or gid form stored in shopify_id
    sql = """UPDATE shopify_products SET tags=%s
             WHERE shopify_id=%s OR shopify_id=%s;"""
    gid = f"gid://shopify/Product/{shopify_numeric_id}"
    subprocess.run(
        ["psql", "-d", "dw_unified", "-v", "ON_ERROR_STOP=1", "-c",
         "UPDATE shopify_products SET tags = $$%s$$ WHERE shopify_id=$$%s$$ OR shopify_id=$$%s$$;"
         % (new_tags_str.replace("$$",""), str(shopify_numeric_id), gid)],
        check=True, capture_output=True, text=True)

def process(handles, apply=False, gap=0.6):
    results = []
    for i, h in enumerate(handles):
        node = fetch_product(h)
        if not node:
            results.append({"handle": h, "status": "NOT_FOUND"}); continue
        if node["status"] != "ACTIVE":
            results.append({"handle": h, "status": "SKIP_NOT_ACTIVE"}); continue
        old_tags = node["tags"]
        banned = [t for t in old_tags if re.search(r'[Ww]allpapers?', t)]
        if not banned:
            results.append({"handle": h, "status": "NO_BANNED"}); continue
        new_for_banned = [fix_tag(t) for t in banned]
        # final tag set: remove banned, add fixed (dedup, preserve order)
        final = [t for t in old_tags if t not in banned]
        for nt in new_for_banned:
            if nt not in final:
                final.append(nt)
        rec = {"handle": h, "id": node["id"], "removed": banned,
               "added": new_for_banned, "final": final, "status": "PLANNED"}
        if apply:
            num = gid_to_numeric(node["id"])
            # PG FIRST
            mirror_pg(num, final)
            # add fixed tags
            ra = gql(TAGS_ADD, {"id": node["id"], "tags": new_for_banned})
            ea = ra["data"]["tagsAdd"]["userErrors"]
            # remove banned tags
            rr = gql(TAGS_REMOVE, {"id": node["id"], "tags": banned})
            er = rr["data"]["tagsRemove"]["userErrors"]
            rec["status"] = "APPLIED" if not ea and not er else "ERROR"
            if ea or er:
                rec["errors"] = {"add": ea, "remove": er}
            time.sleep(gap)
        results.append(rec)
        if (i + 1) % 25 == 0:
            sys.stderr.write(f"  ...{i+1}/{len(handles)}\n")
    return results

if __name__ == "__main__":
    mode = sys.argv[1] if len(sys.argv) > 1 else "plan"
    src = json.load(open("/tmp/wp_tag_hits.json"))
    handles = [r["handle"] for r in src]
    if mode == "canary":
        n = int(sys.argv[2]) if len(sys.argv) > 2 else 3
        res = process(handles[:n], apply=True)
    elif mode == "apply":
        res = process(handles, apply=True)
    elif mode == "apply-rest":
        skip = int(sys.argv[2])
        res = process(handles[skip:], apply=True)
    else:
        res = process(handles, apply=False)
    out = f"/tmp/job1_{mode}.json"
    json.dump(res, open(out, "w"), indent=2)
    from collections import Counter
    print("STATUS:", dict(Counter(r["status"] for r in res)))
    print("AUDIT:", out)