#!/usr/bin/env python3 """ Pak Jabit Store — Bulk Product Import via Spree Admin API v3 Creates products, uploads images, publishes to Online Store channel, and sets stock — all through the official Admin API. Usage: python3 bulk_import.py [--dry-run] [--start N] [--limit N] """ import json import os import sys import time import requests import argparse from pathlib import Path import os as _os def _spree_admin_key() -> str: """The Spree admin API key. Required — never defaulted.""" key = _os.environ.get("SPREE_ADMIN_KEY", "").strip() if not key: raise SystemExit( "SPREE_ADMIN_KEY is not set.\n" " export SPREE_ADMIN_KEY=\n" "It is deliberately not stored in this file — see README.md." ) return key # --- Config --- API_BASE = "https://api.jux-lab.com/api/v3/admin" API_KEY = _spree_admin_key() HEADERS = {"X-Spree-Api-Key": API_KEY, "Content-Type": "application/json"} CHANNEL_ID = "ch_UkLWZg9DAJ" # Online Store MYR_TO_USD = 4.8 PUBLISHED_AT = "2026-09-15T00:00:00Z" CATALOGUE_DIR = Path("/root/Opus/Projects/PeakyBuildr/scraped-catalogue") PRODUCTS_DIR = CATALOGUE_DIR / "products" BATCH_FILES = ["products-batch2.jsonl", "products-batch3.jsonl"] LOG_FILE = CATALOGUE_DIR / "import_log.jsonl" def api(method, path, **kwargs): url = f"{API_BASE}{path}" headers = dict(HEADERS) if "files" in kwargs: headers.pop("Content-Type", None) r = getattr(requests, method)(url, headers=headers, timeout=60, **kwargs) try: return r.json() except Exception: return {"error": {"code": r.status_code, "message": r.text[:200]}} def find_product_folder(code, name): """Match JSONL entry to local image folder.""" for folder in sorted(PRODUCTS_DIR.iterdir()): info_file = folder / "info.json" if info_file.exists(): info = json.loads(info_file.read_text()) if info.get("source_code") == code: return folder return None def create_product(name, brand, price_myr, slug=None): price = round(float(price_myr) * MYR_TO_USD, 2) desc = f"{brand} - {name}" # A scalar "price" is silently dropped; Spree wants a prices array, and a # product with no price row is excluded from the storefront's # available(currency) scope no matter how it is published. payload = { "name": name, "description": desc, "prices": [{"amount": price, "currency": "USD"}], "status": "active", "available_on": "2026-09-15T00:00:00Z", } if slug: payload["slug"] = slug return api("post", "/products", json=payload) def upload_image(product_id, variant_id, image_path): with open(image_path, "rb") as f: return api("post", f"/products/{product_id}/variants/{variant_id}/media", files={"attachment": (image_path.name, f, "image/jpeg")}) def publish_to_channel(product_id): # Without published_at the publication row exists but is not live. return api("post", f"/channels/{CHANNEL_ID}/add_products", json={"product_ids": [product_id], "published_at": PUBLISHED_AT}) def make_purchasable(product_id, variant_id): return api("patch", f"/products/{product_id}/variants/{variant_id}", json={"track_inventory": False}) def log_result(entry): with open(LOG_FILE, "a") as f: f.write(json.dumps(entry) + "\n") def get_existing_slugs(): slugs = set() page = 1 while True: data = api("get", f"/products?per_page=100&page={page}") products = data.get("data", []) for p in products: slugs.add(p.get("slug", "")) meta = data.get("meta", {}) if page >= meta.get("pages", 1): break page += 1 return slugs def main(): parser = argparse.ArgumentParser() parser.add_argument("--dry-run", action="store_true") parser.add_argument("--start", type=int, default=0) parser.add_argument("--limit", type=int, default=999) args = parser.parse_args() # Load all products from batch files products = [] for batch_file in BATCH_FILES: path = CATALOGUE_DIR / batch_file if path.exists(): with open(path) as f: for line in f: products.append(json.loads(line.strip())) print(f"Loaded {len(products)} products from {len(BATCH_FILES)} batch files") # Get existing to skip duplicates print("Fetching existing products...") existing = get_existing_slugs() print(f" {len(existing)} products already in store") # Process imported = 0 skipped = 0 failed = 0 products_slice = products[args.start:args.start + args.limit] for i, prod in enumerate(products_slice): idx = args.start + i name = prod["name"] brand = prod["brand"] code = prod["code"] price = prod["price"] folder = find_product_folder(code, name) if not folder: print(f" [{idx+1}] SKIP (no folder): {brand} - {name}") skipped += 1 continue # Check slug collision info = json.loads((folder / "info.json").read_text()) slug = folder.name if slug in existing: print(f" [{idx+1}] SKIP (exists): {slug}") skipped += 1 continue if args.dry_run: images = list(folder.glob("image-*.jpg")) print(f" [{idx+1}] DRY: {brand} - {name} ({len(images)} imgs, ${float(price)*MYR_TO_USD:.2f})") continue # 1. Create product result = create_product(name, brand, price, slug) if "error" in result: print(f" [{idx+1}] FAIL create: {result['error']}") log_result({"code": code, "action": "create", "error": result["error"]}) failed += 1 continue product_id = result["id"] variant_id = result.get("default_variant_id", "") print(f" [{idx+1}] Created: {product_id} — {brand} - {name}") # 2. Upload images images = sorted(folder.glob("image-*.jpg")) img_ok = 0 for img in images: r = upload_image(product_id, variant_id, img) if "error" not in r: img_ok += 1 else: print(f" WARN image {img.name}: {r['error']}") time.sleep(0.3) print(f" Images: {img_ok}/{len(images)}") # 3. Publish to Online Store pub = publish_to_channel(product_id) if "error" in pub: print(f" WARN publish: {pub['error']}") # 4. Make purchasable stk = make_purchasable(product_id, variant_id) if "error" in stk: print(f" WARN stock: {stk['error']}") log_result({ "code": code, "product_id": product_id, "slug": slug, "brand": brand, "name": name, "images": img_ok, "price_usd": round(float(price) * MYR_TO_USD, 2) }) imported += 1 existing.add(slug) time.sleep(0.5) print(f"\nDone! Imported: {imported}, Skipped: {skipped}, Failed: {failed}") if not args.dry_run and imported > 0: print(f"Log: {LOG_FILE}") if __name__ == "__main__": main()