`pull_catalog.py`

46 lines
#!/usr/bin/env python3
"""Pull the full Novasuede catalog from Shopify (read-only) to data/novasuede_raw.ndjson."""
import os, json, time, urllib.request, re, pathlib

ROOT = pathlib.Path(__file__).resolve().parent
DATA = ROOT / "data"; DATA.mkdir(exist_ok=True)

# read token from secrets-manager .env without printing it
env = pathlib.Path.home() / "Projects/secrets-manager/.env"
TOKEN = None
for ln in env.read_text().splitlines():
    if ln.startswith("SHOPIFY_ADMIN_TOKEN="):
        TOKEN = ln.split("=", 1)[1].strip().strip('"'); break
assert TOKEN, "no SHOPIFY_ADMIN_TOKEN"

SHOP = "designer-laboratory-sandbox.myshopify.com"
API = f"https://{SHOP}/admin/api/2024-01"
FIELDS = "id,title,handle,status,tags,product_type,body_html,images,image,variants,vendor"

def get(url):
    req = urllib.request.Request(url, headers={"X-Shopify-Access-Token": TOKEN})
    for attempt in range(5):
        try:
            with urllib.request.urlopen(req, timeout=30) as r:
                return json.load(r)
        except urllib.error.HTTPError as e:
            if e.code == 429:
                time.sleep(2); continue
            raise
    raise RuntimeError("too many retries")

out = (DATA / "novasuede_raw.ndjson").open("w")
since = 0; total = 0
while True:
    url = f"{API}/products.json?vendor=Novasuede&limit=250&since_id={since}&fields={FIELDS}"
    ps = get(url).get("products", [])
    if not ps: break
    for p in ps:
        out.write(json.dumps(p) + "\n")
    total += len(ps)
    since = ps[-1]["id"]
    if len(ps) < 250: break
    time.sleep(0.4)
out.close()
print(f"pulled {total} products -> {DATA/'novasuede_raw.ndjson'}")