`apply_tags.py`

95 lines
#!/usr/bin/env python3
"""
Wave-1 WRITE: apply a `material-group-<bucket>` tag to each FREE-resolved
New-Arrival Wallcovering (tier1 Material: line + tier1.5 body statement).

Boost SD filters on tags natively, so this tag becomes the Material facet source.
Adding a tag is INVISIBLE on the storefront until the Boost filter is configured to
use it, and is fully reversible (remove the tag).

  python3 apply_tags.py                 # DRY RUN: print the write set + tag counts
  python3 apply_tags.py --canary 3      # WRITE to first 3 products, then verify
  python3 apply_tags.py --apply         # WRITE to all resolved products

Idempotent: skips a product that already carries its material-group tag.
Reads SHOPIFY_ADMIN_TOKEN from ~/Projects/secrets-manager/.env. No cost (Admin API
writes are free).
"""
import os, re, sys, json, time, subprocess, urllib.request, urllib.error
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import classify as C

DOM = "designer-laboratory-sandbox.myshopify.com"
ENV = os.path.expanduser("~/Projects/secrets-manager/.env")

def token():
    for line in open(ENV):
        if line.startswith("SHOPIFY_ADMIN_TOKEN="):
            return line.split("=",1)[1].strip().strip('"').strip("'")
    sys.exit("no SHOPIFY_ADMIN_TOKEN")

def slug(bucket):
    return "material-group-" + re.sub(r"[^a-z0-9]+","-",bucket.lower()).strip("-")

def resolved_set():
    """[(numeric_id, title, bucket, tag)] for FREE-resolved rows only."""
    out=[]
    for r in C.load_rows(scope_all=("--all" in sys.argv)):
        tier,bucket,ev = C.classify(r)
        if bucket and tier in ("tier1_material_line","tier15_body"):
            gid=r.get("shopify_id") or ""
            nid=gid.rsplit("/",1)[-1]
            if nid.isdigit():
                out.append((nid, r["title"], bucket, slug(bucket)))
    return out

def api(method, path, tok, body=None):
    req=urllib.request.Request(f"https://{DOM}/admin/api/2024-10/{path}",
        method=method, headers={"X-Shopify-Access-Token":tok,"Content-Type":"application/json"},
        data=json.dumps(body).encode() if body else None)
    return json.load(urllib.request.urlopen(req))

def apply_one(nid, tag, tok):
    p=api("GET", f"products/{nid}.json?fields=id,tags", tok)["product"]
    tags=[t.strip() for t in (p.get("tags") or "").split(",") if t.strip()]
    if tag in tags:
        return "skip(present)"
    tags.append(tag)
    api("PUT", f"products/{nid}.json", tok, {"product":{"id":int(nid),"tags":", ".join(tags)}})
    # verify
    v=api("GET", f"products/{nid}.json?fields=id,tags", tok)["product"]
    return "OK" if tag in (v.get("tags") or "") else "FAIL(verify)"

def main():
    rows=resolved_set()
    from collections import Counter
    by_tag=Counter(t for _,_,_,t in rows)
    print(f"Wave-1 tag write set: {len(rows)} products")
    for t,n in by_tag.most_common(): print(f"  {t:28s} {n}")

    canary = None
    if "--canary" in sys.argv:
        canary = int(sys.argv[sys.argv.index("--canary")+1])
    do_write = ("--apply" in sys.argv) or (canary is not None)
    if not do_write:
        print("\nDRY RUN — no writes. Re-run with --canary N or --apply."); return

    tok=token()
    target = rows[:canary] if canary is not None else rows
    print(f"\nWRITING to {len(target)} products (canary={canary})...")
    res=Counter()
    for nid,title,bucket,tag in target:
        try:
            r=apply_one(nid,tag,tok)
        except urllib.error.HTTPError as e:
            r=f"HTTP{e.code}"
        res[r]+=1
        if canary is not None:
            print(f"  {r:16s} {tag:26s} {title[:40]}")
        time.sleep(0.3)  # gentle on the 2 req/s REST limit
    print("result:", dict(res))
    print("Cost: $0 (Admin API writes are free).")

if __name__=="__main__":
    main()