`scripts/schumacher-265-live-check.py`

68 lines
#!/usr/bin/env python3
"""READ-ONLY live-Shopify completeness check for the Schumacher-265 unpublished-ACTIVE cohort.
GET only. No writes. Paces requests to respect the 2 req/s REST bucket.
Usage: python3 schumacher-265-live-check.py [limit]
Outputs CSV to stdout."""
import os, sys, time, json, urllib.request, subprocess

LIMIT = int(sys.argv[1]) if len(sys.argv) > 1 else 0
STORE = "designer-laboratory-sandbox.myshopify.com"
VER = "2024-10"

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

TOK = token()

def get_ids():
    q = ("SELECT regexp_replace(shopify_id,'\\D','','g') FROM shopify_products "
         "WHERE vendor='Schumacher' AND status='ACTIVE' AND online_store_published IS NOT TRUE "
         "ORDER BY shopify_id")
    out = subprocess.check_output(["psql","-d","dw_unified","-At","-c",q]).decode()
    ids = [x for x in out.splitlines() if x.strip()]
    return ids[:LIMIT] if LIMIT else ids

def fetch(pid, tries=4):
    url = f"https://{STORE}/admin/api/{VER}/products/{pid}.json"
    req = urllib.request.Request(url, headers={"X-Shopify-Access-Token": TOK})
    for t in range(tries):
        try:
            with urllib.request.urlopen(req, timeout=30) as r:
                return json.loads(r.read().decode())
        except urllib.error.HTTPError as e:
            if e.code == 429:
                time.sleep(2*(t+1)); continue
            if e.code == 404:
                return {"_404": True}
            time.sleep(1)
        except Exception:
            time.sleep(1)
    return None

print("id,sku,title,published_at,images,body_len,nvariants,prod_price,sample_price,has_sample,inv_tracked")
ids = get_ids()
for i, pid in enumerate(ids):
    d = fetch(pid)
    time.sleep(0.55)  # ~1.8 req/s
    if not d or "_404" in d:
        print(f"{pid},,,FETCH_FAIL,,,,,,,"); continue
    p = d.get("product", {})
    vs = p.get("variants") or []
    prod = [v for v in vs if not (v.get("sku") or "").endswith("-Sample")]
    samp = [v for v in vs if (v.get("sku") or "").endswith("-Sample")]
    pv = prod[0] if prod else {}
    sv = samp[0] if samp else {}
    title = (p.get("title") or "").replace(",", " ")
    print(",".join(str(x) for x in [
        pid, pv.get("sku") or "", title, p.get("published_at"),
        len(p.get("images") or []), len(p.get("body_html") or ""),
        len(vs), pv.get("price"), sv.get("price"),
        "Y" if samp else "N", pv.get("inventory_management") or "none",
    ]))
    sys.stderr.write(f"\r{i+1}/{len(ids)}"); sys.stderr.flush()
sys.stderr.write("\n")