`publish_batch.py`

111 lines
#!/usr/bin/env python3
"""
Reid Witlin / Architectural Fabrics — publish the 1,005 newly-ACTIVE products
to the same 5 sales channels the existing 190 live products already carry
(Online Store, Point of Sale, Buy Button, Facebook & Instagram, Houzz) —
verified against a live sample product before writing this script.

status:ACTIVE alone does NOT put a product on the storefront; it must also be
published to the "Online Store" publication. This step is what actually makes
them visible at designerwallcoverings.com/products/<handle>.

SAFETY: DRY_RUN=1 default. Reads the live product set fresh via GraphQL.
"""
import os, json, time, urllib.request

DRY_RUN = os.environ.get("DRY_RUN", "1") != "0"
LIMIT = int(os.environ.get("LIMIT", "0")) or None

def _tok():
    p = os.path.expanduser("~/Projects/secrets-manager/.env")
    for line in open(p):
        if line.startswith("SHOPIFY_ADMIN_TOKEN="):
            return line.split("=", 1)[1].strip().strip('"')
    raise SystemExit("SHOPIFY_ADMIN_TOKEN not found")

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):
    body = json.dumps({"query": q, "variables": v or {}}).encode()
    req = urllib.request.Request(URL, body,
        {"X-Shopify-Access-Token": TOKEN, "Content-Type": "application/json"})
    for a in range(8):
        try:
            d = json.load(urllib.request.urlopen(req, timeout=90))
            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")

# Matches the existing 190 already-live Reid Witlin products' channel set exactly.
CHANNELS = {
    "Online Store": "gid://shopify/Publication/22208643184",
    "Point of Sale": "gid://shopify/Publication/37904089153",
    "Buy Button": "gid://shopify/Publication/22497296496",
    "Facebook & Instagram": "gid://shopify/Publication/29739483201",
    "Houzz": "gid://shopify/Publication/29776969793",
}

LIST_Q = """
query($cursor: String) {
  products(first: 250, after: $cursor, query: "vendor:'Architectural Fabrics' status:active") {
    pageInfo { hasNextPage endCursor }
    nodes { id handle publishedAt }
  }
}"""

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

def fetch_unpublished():
    out, cursor = [], None
    while True:
        d = gql(LIST_Q, {"cursor": cursor})
        data = d.get("data", {}).get("products", {})
        out.extend([p for p in data.get("nodes", []) if not p.get("publishedAt")])
        pi = data.get("pageInfo", {})
        if not pi.get("hasNextPage"):
            return out
        cursor = pi.get("endCursor")

def main():
    products = fetch_unpublished()
    if LIMIT:
        products = products[:LIMIT]
    print(f"{'DRY-RUN' if DRY_RUN else 'LIVE'}: {len(products)} unpublished ACTIVE products "
          f"({'no writes' if DRY_RUN else 'publishing to ' + ', '.join(CHANNELS)})")
    published, errs = 0, []
    pub_input = [{"publicationId": pid} for pid in CHANNELS.values()]
    for i, p in enumerate(products):
        if DRY_RUN:
            if i < 3:
                print(f"  would publish {p['handle']} to {len(CHANNELS)} channels")
            published += 1
            continue
        d = gql(PUBLISH_M, {"id": p["id"], "input": pub_input})
        r = (d.get("data") or {}).get("publishablePublish") or {}
        ue = r.get("userErrors") or []
        if ue:
            errs.append({"id": p["id"], "handle": p["handle"], "errors": ue[:2]})
        else:
            published += 1
        if i % 50 == 0:
            print(f"  ...{i}/{len(products)} published={published} errs={len(errs)}")
        time.sleep(0.25)
    out = {"published": published, "errors": errs[:20], "total": len(products)}
    json.dump(out, open("publish-results.json", "w"), indent=2)
    print(f"\nDONE. published={published} userErrors={len(errs)} "
          f"({'DRY-RUN — nothing written' if DRY_RUN else 'live on storefront'})")
    if errs:
        print("sample errors:", json.dumps(errs[:3]))

if __name__ == "__main__":
    main()