`_shopify_channel_fill.py`

114 lines
#!/usr/bin/env python3
"""Publish active products to all reachable sales-channel publications.
Excludes peel-and-stick lines (Surface Stick, PS Removable Wallpaper) per Steve.
Modes: --test N  (dry-ish: publish only N products, verbose)
       --run      (full run, cursor pagination, log progress)
"""
import os, sys, json, time, urllib.request

STORE = "designer-laboratory-sandbox"
API = f"https://{STORE}.myshopify.com/admin/api/2024-10/graphql.json"
EXCLUDE_VENDORS = {"Surface Stick", "PS Removable Wallpaper"}

def token():
    with open(os.path.expanduser("~/Projects/secrets-manager/.env")) as f:
        for line in f:
            if line.startswith("SHOPIFY_ADMIN_TOKEN="):
                return line.split("=",1)[1].strip().strip('"').strip("'")
    raise SystemExit("no token")

TOKEN = token()

def gql(query, variables=None):
    body = json.dumps({"query": query, "variables": variables or {}}).encode()
    req = urllib.request.Request(API, data=body, headers={
        "X-Shopify-Access-Token": TOKEN, "Content-Type": "application/json"})
    for attempt in range(6):
        try:
            with urllib.request.urlopen(req, timeout=60) as r:
                d = json.loads(r.read())
            if "errors" in d and any("throttled" in json.dumps(e).lower() for e in d.get("errors",[])):
                time.sleep(2 + attempt*2); continue
            return d
        except Exception as e:
            time.sleep(2 + attempt*2)
    return {"errors":"exhausted"}

def get_publications():
    d = gql("{ publications(first:50){ edges{ node{ id name } } } }")
    return [(e["node"]["id"], e["node"]["name"]) for e in d["data"]["publications"]["edges"]]

PUBS = get_publications()
PUB_IDS = [p[0] for p in PUBS]

PAGE_Q = """
query($cursor:String){
  products(first:60, after:$cursor, query:"status:active"){
    pageInfo{ hasNextPage endCursor }
    edges{ node{ id vendor
      resourcePublications(first:30){ edges{ node{ publication{ id } } } }
    } }
  }
}"""

PUBLISH_M = """
mutation($id:ID!, $pubs:[PublicationInput!]!){
  publishablePublish(id:$id, input:$pubs){ userErrors{ field message } }
}"""

def missing_pubs(node):
    on = {e["node"]["publication"]["id"] for e in node["resourcePublications"]["edges"]}
    return [pid for pid in PUB_IDS if pid not in on]

def run(limit=None, test=False):
    print(f"Publications reachable: {len(PUBS)}")
    for pid,name in PUBS: print(f"  - {name}")
    print(f"Excluding vendors: {EXCLUDE_VENDORS}\n")
    cursor=None; seen=0; skipped_ps=0; already=0; published=0; errs=0; pages=0
    t0=time.time()
    while True:
        d = gql(PAGE_Q, {"cursor": cursor})
        try:
            conn = d["data"]["products"]
        except Exception:
            print("PAGE ERROR:", json.dumps(d)[:300]); break
        pages+=1
        for e in conn["edges"]:
            n=e["node"]; seen+=1
            if n["vendor"] in EXCLUDE_VENDORS:
                skipped_ps+=1; continue
            miss = missing_pubs(n)
            if not miss:
                already+=1; continue
            res = gql(PUBLISH_M, {"id": n["id"], "pubs":[{"publicationId":p} for p in miss]})
            ue = res.get("data",{}).get("publishablePublish",{}).get("userErrors") if res.get("data") else None
            if ue:
                errs+=1
                if errs<=10: print(f"  userErr {n['id']}: {ue[:1]}")
            else:
                published+=1
                if test: print(f"  published {n['id']} (+{len(miss)} chans)")
            if limit and (published+already+skipped_ps)>=limit and test:
                print("\nTEST LIMIT reached"); _summary(seen,skipped_ps,already,published,errs,pages,t0); return
            time.sleep(0.08)
        if seen % 600 == 0 or test:
            print(f"[{seen} seen | pub {published} | already {already} | ps-skip {skipped_ps} | err {errs} | {int(time.time()-t0)}s]")
        if not conn["pageInfo"]["hasNextPage"]:
            break
        cursor = conn["pageInfo"]["endCursor"]
        if limit and test and seen>=limit: break
    _summary(seen,skipped_ps,already,published,errs,pages,t0)

def _summary(seen,ps,already,pub,errs,pages,t0):
    print(f"\nSUMMARY seen={seen} ps_skipped={ps} already_full={already} published={pub} errors={errs} pages={pages} secs={int(time.time()-t0)}")
    print("DONE")

if __name__=="__main__":
    if "--test" in sys.argv:
        i=sys.argv.index("--test"); N=int(sys.argv[i+1]) if len(sys.argv)>i+1 else 5
        run(limit=N, test=True)
    elif "--run" in sys.argv:
        run()
    else:
        print("usage: --test N | --run")