"""Shopify Admin API client for Core-side store data collection and webhook registration.""" from __future__ import annotations import asyncio from typing import Any, Optional import httpx import structlog logger = structlog.get_logger() _SHOPIFY_API_VERSION = "2024-01" _PAGE_LIMIT = 250 class ShopifyAdminClient: def __init__(self, shop_domain: str, access_token: str) -> None: self._shop = shop_domain.strip().rstrip("/") self._token = access_token self._base = f"https://{self._shop}/admin/api/{_SHOPIFY_API_VERSION}" self._headers = { "X-Shopify-Access-Token": access_token, "Content-Type": "application/json", } async def collect_store_understanding(self) -> dict[str, Any]: """Collect all artifacts needed for a Store Understanding run.""" async with httpx.AsyncClient(timeout=60.0) as client: store, products, custom_cols, smart_cols, pages, blogs, themes = await asyncio.gather( self._get(client, "/shop.json"), self._get_all(client, "/products.json", "products"), self._get_all(client, "/custom_collections.json", "custom_collections"), self._get_all(client, "/smart_collections.json", "smart_collections"), self._get_all(client, "/pages.json", "pages"), self._get_all(client, "/blogs.json", "blogs"), self._get(client, "/themes.json"), ) active_theme = next( (t for t in (themes.get("themes") or []) if t.get("role") == "main"), None, ) theme_assets: dict[str, str] = {} if active_theme: async with httpx.AsyncClient(timeout=60.0) as client: theme_assets = await self._collect_theme_assets(client, active_theme["id"]) articles: list[dict] = [] async with httpx.AsyncClient(timeout=30.0) as client: for blog in blogs: blog_id = blog.get("id") if blog_id: items = await self._get_all( client, f"/blogs/{blog_id}/articles.json", "articles" ) for article in items: article["blog_id"] = blog_id articles.extend(items) navigation = await self._collect_navigation() collections = [*custom_cols, *smart_cols] for b in blogs: b["articles"] = [a for a in articles if a.get("blog_id") == b.get("id")] return { "schema_version": 1, "store": _redact_store(store.get("shop") or {}), "products": [_redact_product(p) for p in products], "collections": collections, "pages": pages, "blogs": blogs, "navigation": navigation, "theme": { **(active_theme or {}), "assets": theme_assets, }, "changes": [], } async def register_webhooks(self, bff_url: str, topics: list[str]) -> list[dict]: """Register webhook subscriptions pointing at the BFF. Idempotent: skips existing.""" bff_url = bff_url.rstrip("/") results: list[dict] = [] async with httpx.AsyncClient(timeout=30.0) as client: existing_resp = await self._get(client, "/webhooks.json") existing = {w["topic"]: w for w in (existing_resp.get("webhooks") or [])} for topic in topics: # customers/data_request -> customers-data-request: the BFF routes on hyphens. slug = topic.replace("/", "-").replace("_", "-") address = f"{bff_url}/webhooks/shopify/{slug}" if topic in existing and existing[topic].get("address") == address: results.append({"topic": topic, "action": "exists", "id": existing[topic]["id"]}) continue resp = await client.post( f"{self._base}/webhooks.json", headers=self._headers, json={"webhook": {"topic": topic, "address": address, "format": "json"}}, ) if resp.status_code in (200, 201): results.append({"topic": topic, "action": "created", **resp.json().get("webhook", {})}) else: logger.warning( "shopify_webhook_registration_failed", topic=topic, status=resp.status_code, body=resp.text[:200], ) results.append({"topic": topic, "action": "failed", "status": resp.status_code}) return results async def _get(self, client: httpx.AsyncClient, path: str) -> dict: resp = await client.get(f"{self._base}{path}", headers=self._headers) resp.raise_for_status() return resp.json() async def _get_all( self, client: httpx.AsyncClient, path: str, key: str ) -> list[dict]: url = f"{self._base}{path}" params: dict[str, Any] = {"limit": _PAGE_LIMIT} results: list[dict] = [] while url: resp = await client.get(url, headers=self._headers, params=params) resp.raise_for_status() data = resp.json() results.extend(data.get(key) or []) url = _next_page_url(resp) params = {} return results async def _collect_theme_assets( self, client: httpx.AsyncClient, theme_id: int ) -> dict[str, str]: resp = await client.get( f"{self._base}/themes/{theme_id}/assets.json", headers=self._headers ) resp.raise_for_status() asset_list = resp.json().get("assets") or [] text_keys = [ a["key"] for a in asset_list if isinstance(a.get("key"), str) and a["key"].endswith((".liquid", ".json", ".css", ".js", ".scss")) ] # Deduplicate and cap text_keys = list(dict.fromkeys(text_keys))[:200] assets: dict[str, str] = {} for key in text_keys: try: r = await client.get( f"{self._base}/themes/{theme_id}/assets.json", headers=self._headers, params={"asset[key]": key}, ) if r.status_code == 200: asset = r.json().get("asset") or {} content = asset.get("value") or asset.get("attachment") or "" if isinstance(content, str): assets[key] = content except Exception: pass return assets async def _collect_navigation(self) -> list[dict]: query = """ { menus(first: 20) { edges { node { title handle items { title url type items { title url type } } } } } } """ try: async with httpx.AsyncClient(timeout=20.0) as client: resp = await client.post( f"https://{self._shop}/admin/api/{_SHOPIFY_API_VERSION}/graphql.json", headers=self._headers, json={"query": query}, ) if resp.status_code == 200: data = resp.json() edges = ( (data.get("data") or {}).get("menus", {}).get("edges") or [] ) return [e["node"] for e in edges if "node" in e] except Exception: pass return [] def _next_page_url(response: httpx.Response) -> Optional[str]: link = response.headers.get("link", "") for part in link.split(","): if 'rel="next"' in part: url_part = part.split(";")[0].strip().strip("<>") return url_part return None def _redact_store(store: dict) -> dict: sensitive = {"email", "customer_email", "phone", "myshopify_domain"} return {k: v for k, v in store.items() if k not in sensitive} def _redact_product(product: dict) -> dict: keep = { "id", "title", "handle", "body_html", "vendor", "product_type", "status", "tags", "variants", "options", "images", "metafields", "published_at", "created_at", "updated_at", } return {k: v for k, v in product.items() if k in keep}