"""RBS REGAL — E-commerce live sync worker. Fetches products + orders from a configured Shopify or WooCommerce store and stores them in MongoDB. Designed to be called both interactively (POST /sync) and from a future scheduled background task. Uses `httpx` for async HTTP. No external SDK to keep footprint small. """ from datetime import datetime, timezone import asyncio import httpx import re REQUEST_TIMEOUT = 30.0 PAGE_SIZE = 250 MAX_PAGES = 50 # safety cap def _now(): return datetime.now(timezone.utc).isoformat() def _ser_log_id(d): if d and "_id" in d: d = dict(d) d.pop("_id", None) return d # ---------------- Shopify ---------------- async def _shopify_paginated(client: httpx.AsyncClient, base_url: str, endpoint: str, token: str, params: dict = None): """Iterate Shopify pages using the Link: ...rel=next header.""" url = f"{base_url}/admin/api/2024-10/{endpoint}.json" headers = {"X-Shopify-Access-Token": token, "Accept": "application/json"} out = [] page = 0 while url and page < MAX_PAGES: page += 1 for attempt in range(3): try: resp = await client.get(url, headers=headers, params=params, timeout=REQUEST_TIMEOUT) if resp.status_code == 429: await asyncio.sleep(int(resp.headers.get("Retry-After", "2"))) continue resp.raise_for_status() break except (httpx.HTTPError, httpx.ReadTimeout): if attempt == 2: raise await asyncio.sleep(1.5 * (attempt + 1)) data = resp.json() # Endpoint key is the same as the singular name plural ("products", "orders") items = data.get(endpoint, []) or [] out.extend(items) # Pagination via Link header link = resp.headers.get("link") or resp.headers.get("Link") or "" m = re.search(r'<([^>]+)>;\s*rel="next"', link) url = m.group(1) if m else None params = None # next URL already includes them return out async def _sync_shopify(conn: dict, db) -> dict: base_raw = (conn.get("store_url") or "").rstrip("/") if not base_raw.startswith("http"): base_raw = f"https://{base_raw}" token = conn.get("access_token") if not token: raise ValueError("Shopify access_token missing") company_id = conn["company_id"] cid = str(conn.get("_id")) results = {"products": 0, "orders": 0, "errors": []} async with httpx.AsyncClient() as client: if conn.get("sync_products", True): try: products = await _shopify_paginated(client, base_raw, "products", token, {"limit": PAGE_SIZE}) for p in products: await db.ecommerce_synced_products.update_one( {"connection_id": cid, "remote_id": str(p.get("id"))}, {"$set": { "connection_id": cid, "company_id": company_id, "provider": "shopify", "remote_id": str(p.get("id")), "title": p.get("title"), "vendor": p.get("vendor"), "product_type": p.get("product_type"), "status": p.get("status"), "variants_count": len(p.get("variants") or []), "raw": p, "synced_at": _now(), }}, upsert=True, ) results["products"] = len(products) except Exception as e: results["errors"].append(f"products: {type(e).__name__}: {str(e)[:200]}") if conn.get("sync_orders", True): try: orders = await _shopify_paginated(client, base_raw, "orders", token, {"limit": PAGE_SIZE, "status": "any"}) for o in orders: await db.ecommerce_synced_orders.update_one( {"connection_id": cid, "remote_id": str(o.get("id"))}, {"$set": { "connection_id": cid, "company_id": company_id, "provider": "shopify", "remote_id": str(o.get("id")), "order_no": o.get("name") or o.get("order_number"), "customer_email": (o.get("customer") or {}).get("email") or o.get("email"), "total": float(o.get("total_price") or 0), "currency": o.get("currency"), "status": o.get("financial_status"), "created_at_remote": o.get("created_at"), "line_items_count": len(o.get("line_items") or []), "raw": o, "synced_at": _now(), }}, upsert=True, ) results["orders"] = len(orders) except Exception as e: results["errors"].append(f"orders: {type(e).__name__}: {str(e)[:200]}") return results # ---------------- WooCommerce ---------------- async def _woo_paginated(client: httpx.AsyncClient, base_url: str, endpoint: str, ck: str, cs: str): """WooCommerce REST v3 — page query param.""" url = f"{base_url.rstrip('/')}/wp-json/wc/v3/{endpoint}" out = [] page = 1 while page <= MAX_PAGES: params = {"per_page": 100, "page": page} for attempt in range(3): try: resp = await client.get(url, auth=(ck, cs), params=params, timeout=REQUEST_TIMEOUT) if resp.status_code == 429: await asyncio.sleep(int(resp.headers.get("Retry-After", "2"))) continue resp.raise_for_status() break except (httpx.HTTPError, httpx.ReadTimeout): if attempt == 2: raise await asyncio.sleep(1.5 * (attempt + 1)) items = resp.json() if not isinstance(items, list) or not items: break out.extend(items) if len(items) < 100: break page += 1 return out async def _sync_woocommerce(conn: dict, db) -> dict: base = (conn.get("store_url") or "").rstrip("/") if not base.startswith("http"): base = f"https://{base}" ck = conn.get("consumer_key"); cs = conn.get("consumer_secret") if not (ck and cs): raise ValueError("WooCommerce consumer_key/secret missing") company_id = conn["company_id"] cid = str(conn.get("_id")) results = {"products": 0, "orders": 0, "errors": []} async with httpx.AsyncClient() as client: if conn.get("sync_products", True): try: products = await _woo_paginated(client, base, "products", ck, cs) for p in products: await db.ecommerce_synced_products.update_one( {"connection_id": cid, "remote_id": str(p.get("id"))}, {"$set": { "connection_id": cid, "company_id": company_id, "provider": "woocommerce", "remote_id": str(p.get("id")), "title": p.get("name"), "vendor": "", "product_type": p.get("type"), "status": p.get("status"), "raw": p, "synced_at": _now(), }}, upsert=True, ) results["products"] = len(products) except Exception as e: results["errors"].append(f"products: {type(e).__name__}: {str(e)[:200]}") if conn.get("sync_orders", True): try: orders = await _woo_paginated(client, base, "orders", ck, cs) for o in orders: await db.ecommerce_synced_orders.update_one( {"connection_id": cid, "remote_id": str(o.get("id"))}, {"$set": { "connection_id": cid, "company_id": company_id, "provider": "woocommerce", "remote_id": str(o.get("id")), "order_no": str(o.get("number") or o.get("id")), "customer_email": (o.get("billing") or {}).get("email"), "total": float(o.get("total") or 0), "currency": o.get("currency"), "status": o.get("status"), "created_at_remote": o.get("date_created"), "line_items_count": len(o.get("line_items") or []), "raw": o, "synced_at": _now(), }}, upsert=True, ) results["orders"] = len(orders) except Exception as e: results["errors"].append(f"orders: {type(e).__name__}: {str(e)[:200]}") return results # ---------------- Orchestrator ---------------- async def sync_connection(conn: dict, db) -> dict: """Dispatch to the right provider and return summary {products, orders, errors}.""" provider = (conn.get("provider") or "").lower() if provider == "shopify": return await _sync_shopify(conn, db) if provider == "woocommerce": return await _sync_woocommerce(conn, db) if provider == "wordpress": # WordPress (without WooCommerce) doesn't have product/order APIs return {"products": 0, "orders": 0, "errors": ["WordPress provider does not expose product/order APIs — install WooCommerce on the same site"]} raise ValueError(f"Unknown provider: {provider}")