`archive_disco.py`

76 lines
#!/usr/bin/env python3
"""
Archive vendor-DISCONTINUED EUR- products — GATED (Steve runs).

Sets status ACTIVE -> ARCHIVED for the 188 held EUR- products whose mfr pattern
appears on the vendor DISCONTINUED lists (DiscontinuedList.pdf + the DG
discontinued list). These are already parked (no current trade price) AND
officially discontinued by the vendor. ARCHIVED hides them from the storefront
but preserves them (reversible; not deleted).

Scope note: matched at pattern (base-code) level — a product is archived if it's
parked (unpriceable) AND its base pattern is on a discontinued list. Not the
LIMITED-stock list (those stay orderable). Reads archive_list.json.

SAFETY: DRY_RUN=1 default. Smoke: DRY_RUN=0 LIMIT=1 python3 archive_disco.py
"""
import os, json, time, urllib.request

HERE = os.path.dirname(os.path.abspath(__file__))
DRY_RUN = os.environ.get("DRY_RUN", "1") != "0"
LIMIT = int(os.environ.get("LIMIT", "0")) or None

def _tok():
    for l in open(os.path.expanduser("~/Projects/secrets-manager/.env")):
        if l.startswith("SHOPIFY_ADMIN_TOKEN="): return l.split("=", 1)[1].strip().strip('"')
    raise SystemExit("token")
TOKEN = os.environ.get("AT") or _tok()
URL = "https://designer-laboratory-sandbox.myshopify.com/admin/api/2024-10/graphql.json"

def gql(q, v=None):
    b = json.dumps({"query": q, "variables": v or {}}).encode()
    req = urllib.request.Request(URL, b, {"X-Shopify-Access-Token": TOKEN, "Content-Type": "application/json"})
    for a in range(6):
        try:
            d = json.load(urllib.request.urlopen(req, timeout=60))
            if "errors" in d and any("THROTTLED" in str(e) for e in d["errors"]):
                time.sleep(2 * (a + 1)); continue
            return d
        except Exception:
            time.sleep(2 * (a + 1))
    raise RuntimeError("gql failed")

FIND = 'query($q:String!){products(first:1,query:$q){edges{node{id status title}}}}'
UPD = 'mutation($id:ID!){productUpdate(input:{id:$id,status:ARCHIVED}){product{status} userErrors{message}}}'

def main():
    items = json.load(open(os.path.join(HERE, "archive_list.json")))
    if LIMIT: items = items[:LIMIT]
    done = skip = 0; errs = []
    print(f"{'DRY-RUN' if DRY_RUN else 'LIVE'}: archiving {len(items)} vendor-discontinued products")
    for i, it in enumerate(items):
        sku = it["sku"]
        d = gql(FIND, {"q": "sku:" + sku})
        e = (d.get("data") or {}).get("products", {}).get("edges", [])
        if not e:
            errs.append({"sku": sku, "err": "not found"}); continue
        node = e[0]["node"]
        if node["status"] == "ARCHIVED":
            skip += 1; continue
        if DRY_RUN:
            if i < 6: print(f"  would archive {sku} [{it['vendor']}] {node['title'][:34]}")
            done += 1; continue
        r = gql(UPD, {"id": node["id"]})
        ue = ((r.get("data") or {}).get("productUpdate") or {}).get("userErrors") or []
        if ue: errs.append({"sku": sku, "err": ue[:2]})
        else: done += 1
        if i % 25 == 0: print(f"  ...{i}/{len(items)} archived={done} err={len(errs)}")
        time.sleep(0.3)
    json.dump({"archived": done, "skipped": skip, "errors": errs[:25]},
              open(os.path.join(HERE, "archive-results.json"), "w"), indent=2)
    print(f"\nDONE {'(DRY-RUN)' if DRY_RUN else ''}: archived={done} skip={skip} err={len(errs)}")
    if errs: print("errs:", json.dumps(errs[:3]))

if __name__ == "__main__":
    main()