`exec/lib/shopify.py`

85 lines
#!/usr/bin/env python3
"""Shared Shopify Admin API (2024-10) client for the Elitis reprice job.
Reads token from the DW repo .env. All calls are $0 (Shopify Admin API is free).
"""
import os, json, time, urllib.request, urllib.error, sys

API_VERSION = "2024-10"

def _load_env():
    dom = os.environ.get("SHOPIFY_STORE_DOMAIN")
    tok = os.environ.get("SHOPIFY_ADMIN_ACCESS_TOKEN")
    if dom and tok:
        return dom, tok
    envp = os.path.expanduser("~/Projects/Designer-Wallcoverings/.env")
    with open(envp) as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            k, v = line.split("=", 1)
            v = v.strip().strip('"').strip("'")
            if k == "SHOPIFY_STORE_DOMAIN":
                dom = v
            elif k == "SHOPIFY_ADMIN_ACCESS_TOKEN":
                tok = v
    if not dom or not tok:
        sys.exit("Missing SHOPIFY_STORE_DOMAIN / SHOPIFY_ADMIN_ACCESS_TOKEN")
    return dom, tok

DOMAIN, TOKEN = _load_env()
GQL_URL = f"https://{DOMAIN}/admin/api/{API_VERSION}/graphql.json"

def gql(query, variables=None, retries=5):
    body = json.dumps({"query": query, "variables": variables or {}}).encode()
    req = urllib.request.Request(GQL_URL, data=body, method="POST", headers={
        "X-Shopify-Access-Token": TOKEN,
        "Content-Type": "application/json",
    })
    last = None
    for attempt in range(retries):
        try:
            with urllib.request.urlopen(req, timeout=60) as r:
                data = json.loads(r.read())
        except urllib.error.HTTPError as e:
            last = f"HTTP {e.code}: {e.read().decode()[:300]}"
            if e.code in (429, 500, 502, 503):
                time.sleep(2 * (attempt + 1)); continue
            raise RuntimeError(last)
        except urllib.error.URLError as e:
            last = str(e); time.sleep(2 * (attempt + 1)); continue
        if "errors" in data and data["errors"]:
            raise RuntimeError("GQL errors: " + json.dumps(data["errors"])[:500])
        # Handle throttle inside userErrors-free responses
        ext = data.get("extensions", {}).get("cost", {}).get("throttleStatus", {})
        avail = ext.get("currentlyAvailable")
        if avail is not None and avail < 200:
            time.sleep(1.5)
        return data["data"]
    raise RuntimeError(f"gql failed after {retries}: {last}")

def find_product_by_title(title):
    """Exact title match (title is unique per Elitis product). Returns node or None."""
    q = """query($q:String!){ products(first:10, query:$q){ edges{ node{
      id legacyResourceId title status handle vendor productType tags
      options{name values}
      variants(first:30){ edges{ node{
        id title sku price inventoryPolicy inventoryItem{ id tracked }
        selectedOptions{name value} } } }
      metafields(first:50){ edges{ node{ namespace key value type } } }
    } } } }"""
    # Escape quotes/backslashes for the Shopify search-query string literal
    safe = title.replace("\\", "\\\\").replace('"', '\\"')
    data = gql(q, {"q": f'title:"{safe}"'})
    edges = data["products"]["edges"]
    for e in edges:
        if e["node"]["title"] == title:
            return e["node"]
    # fall back: single result even if not exact
    return edges[0]["node"] if len(edges) == 1 else None

if __name__ == "__main__":
    # smoke test
    n = find_product_by_title(sys.argv[1] if len(sys.argv) > 1 else "Lin Expérience Minimaliste | Elitis")
    print(json.dumps(n, indent=2, ensure_ascii=False) if n else "NOT FOUND")