"""Process Shopify Store Understanding exports into internal JSON artifacts.""" from __future__ import annotations import json import re from datetime import UTC, datetime from uuid import UUID import structlog from src.core.database import async_session_factory from src.core.security.encryption import decrypt_value from src.domain.external_admin.models import ( SITE_UNDERSTANDING_STATUS_COMPLETED, SITE_UNDERSTANDING_STATUS_FAILED, SITE_UNDERSTANDING_STATUS_QUEUED, SITE_UNDERSTANDING_STATUS_RUNNING, ) from src.domain.external_admin.repository import ( ExternalAdminInstallationRepository, ExternalCredentialRepository, SiteUnderstandingRunRepository, ) from src.domain.projects.repository import ProjectRepoRepository from src.integrations.cloud_control_client import get_cloud_client_for_workspace from src.integrations.github_client import GitHubAppClient from src.integrations.github_installation import resolve_installation_id_for_workspace from src.integrations.shopify_admin_client import ShopifyAdminClient logger = structlog.get_logger() SITE_UNDERSTANDING_BRANCH = "p2hs-site-understanding" async def process_shopify_site_understanding_run(ctx: dict, run_id: str) -> dict: """Commit sanitized Shopify store-understanding JSON to the internal repo branch.""" db_factory = ctx.get("db_factory", async_session_factory) try: su_run_id = UUID(run_id) except ValueError: logger.warning("shopify_site_understanding_invalid_run_id", run_id=run_id) return {"status": "invalid_run_id"} async with db_factory() as db: run_repo = SiteUnderstandingRunRepository(db) run = await run_repo.get(su_run_id) if not run: logger.warning("shopify_site_understanding_run_not_found", run_id=run_id) return {"status": "not_found"} if run.status == SITE_UNDERSTANDING_STATUS_COMPLETED: return {"status": "already_completed"} if run.status not in (SITE_UNDERSTANDING_STATUS_QUEUED, SITE_UNDERSTANDING_STATUS_FAILED): return {"status": run.status} run.status = SITE_UNDERSTANDING_STATUS_RUNNING run.started_at = datetime.now(UTC) await run_repo.save(run) await db.commit() try: async with db_factory() as db: run_repo = SiteUnderstandingRunRepository(db) project_repo = ProjectRepoRepository(db) run = await run_repo.get(su_run_id) if not run: return {"status": "not_found"} project_repo_row = await project_repo.get_by_project_id(run.project_id) if not project_repo_row: raise RuntimeError("project_repo_not_found") installation = await ExternalAdminInstallationRepository(db).get_by_id( run.installation_id ) if not installation: raise RuntimeError("external_admin_installation_not_found") # If no payload was provided, collect from Shopify Admin API directly. payload = run.payload_json or {} if not payload.get("store") and installation.shop_domain: cred_repo = ExternalCredentialRepository(db) cred = await cred_repo.get_active(installation.id, "shopify_access_token") if cred and cred.value_enc: try: access_token = decrypt_value(cred.value_enc) client = ShopifyAdminClient(installation.shop_domain, access_token) payload = await client.collect_store_understanding() run.payload_json = payload await run_repo.save(run) await db.commit() logger.info( "shopify_site_understanding_collected", run_id=run_id, shop=installation.shop_domain, ) except Exception as exc: logger.warning( "shopify_site_understanding_collection_failed", run_id=run_id, error=str(exc), ) files = _build_artifact_files(payload, run_id=str(run.id)) installation_id = await resolve_installation_id_for_workspace( db_factory, installation.workspace_id, ) github = GitHubAppClient(installation_id=installation_id) commit_sha = await github.create_commit( project_repo_row.repo_full_name, SITE_UNDERSTANDING_BRANCH, files, f"chore: sync Shopify store understanding {str(su_run_id)[:8]}", ) async with db_factory() as db: run_repo = SiteUnderstandingRunRepository(db) installation_repo = ExternalAdminInstallationRepository(db) run = await run_repo.get(su_run_id) if run: installation = await installation_repo.get_by_id(run.installation_id) summary = dict(run.summary_json or {}) summary.update( { "message": "Store Understanding JSON artifacts committed.", "artifact_file_count": len(files), "site_understanding_branch": SITE_UNDERSTANDING_BRANCH, "commit_sha": commit_sha, } ) run.status = SITE_UNDERSTANDING_STATUS_COMPLETED run.completed_at = datetime.now(UTC) run.summary_json = summary run.artifacts_json = { "kind": "managed_repo_branch", "branch": SITE_UNDERSTANDING_BRANCH, "commit_sha": commit_sha, "file_count": len(files), "manifest_path": "p2hs/store-understanding/manifest.json", } if installation: cloud = await get_cloud_client_for_workspace(installation.workspace_id, db) await cloud.upsert_wp_site_shape_summary( project_id=str(run.project_id), payload={ "run_id": str(run.id), "workspace_id": str(installation.workspace_id), "project_id": str(run.project_id), "installation_id": str(run.installation_id), "shop_domain": installation.shop_domain or "", "platform": "shopify", "status": SITE_UNDERSTANDING_STATUS_COMPLETED, "reason": run.reason, "summary": summary, "artifacts": run.artifacts_json, }, ) await run_repo.save(run) await db.commit() logger.info("shopify_site_understanding_completed", run_id=run_id, commit_sha=commit_sha) return {"status": SITE_UNDERSTANDING_STATUS_COMPLETED, "commit_sha": commit_sha} except Exception as exc: logger.exception("shopify_site_understanding_failed", run_id=run_id, error=str(exc)) async with db_factory() as db: run_repo = SiteUnderstandingRunRepository(db) run = await run_repo.get(su_run_id) if run: run.status = SITE_UNDERSTANDING_STATUS_FAILED run.error = "shopify_site_understanding_failed" summary = dict(run.summary_json or {}) summary["message"] = "Store Understanding processing failed." summary["error"] = str(exc)[:500] run.summary_json = summary await run_repo.save(run) await db.commit() return {"status": SITE_UNDERSTANDING_STATUS_FAILED} def _stable_json(value: object) -> str: return json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n" def _safe_slug(value: object, fallback: str = "item") -> str: text = str(value or "").strip().lower() text = re.sub(r"[^a-z0-9._-]+", "-", text).strip("-._") return text[:80] or fallback def _build_artifact_files(payload: dict, *, run_id: str) -> dict[str, str]: generated_at = str(payload.get("generated_at") or datetime.now(UTC).isoformat()) store = payload.get("store") if isinstance(payload.get("store"), dict) else {} products = payload.get("products") if isinstance(payload.get("products"), list) else [] collections = payload.get("collections") if isinstance(payload.get("collections"), list) else [] pages = payload.get("pages") if isinstance(payload.get("pages"), list) else [] blogs = payload.get("blogs") if isinstance(payload.get("blogs"), list) else [] navigation = payload.get("navigation") if isinstance(payload.get("navigation"), list) else [] theme = payload.get("theme") if isinstance(payload.get("theme"), dict) else {} theme_assets = theme.get("assets") if isinstance(theme.get("assets"), dict) else {} changes = payload.get("changes") if isinstance(payload.get("changes"), list) else [] files: dict[str, str] = { "p2hs/store-understanding/manifest.json": _stable_json( { "schema_version": payload.get("schema_version") or 1, "run_id": run_id, "generated_at": generated_at, "branch_role": "site_understanding", "platform": "shopify", "counts": { "products": len(products), "collections": len(collections), "pages": len(pages), "blogs": len(blogs), "navigation": len(navigation), "theme_assets": len(theme_assets), "unsynced_changes": len(changes), }, } ), "docs/STORE_UNDERSTANDING.md": _store_understanding_markdown(payload), "store/store.json": _stable_json(store), "content/products/index.json": _stable_json(products), "content/collections/index.json": _stable_json(collections), "content/pages/index.json": _stable_json(pages), "content/blogs/index.json": _stable_json(blogs), "content/navigation/index.json": _stable_json(navigation), "changes/latest.json": _stable_json(changes), } for product in products: if isinstance(product, dict): identifier = _safe_slug(product.get("handle") or product.get("id"), "product") files[f"content/products/{identifier}.json"] = _stable_json(product) for page in pages: if isinstance(page, dict): identifier = _safe_slug(page.get("handle") or page.get("id"), "page") files[f"content/pages/{identifier}.json"] = _stable_json(page) files["theme/theme.json"] = _stable_json({k: v for k, v in theme.items() if k != "assets"}) for asset_path, asset_content in theme_assets.items(): if not isinstance(asset_path, str) or not isinstance(asset_content, str): continue safe_path = asset_path.replace("\\", "/").lstrip("/") files[f"theme/assets/{safe_path}"] = asset_content return files def _store_understanding_markdown(payload: dict) -> str: store = payload.get("store") if isinstance(payload.get("store"), dict) else {} products = payload.get("products") if isinstance(payload.get("products"), list) else [] collections = payload.get("collections") if isinstance(payload.get("collections"), list) else [] pages = payload.get("pages") if isinstance(payload.get("pages"), list) else [] theme = payload.get("theme") if isinstance(payload.get("theme"), dict) else {} theme_assets = theme.get("assets") if isinstance(theme.get("assets"), dict) else {} changes = payload.get("changes") if isinstance(payload.get("changes"), list) else [] return "\n".join( [ "# Store Understanding", "", "This document is generated from sanitized Shopify export JSON.", "", "## Store", "", f"- Domain: {store.get('domain') or store.get('myshopify_domain') or 'unknown'}", f"- Currency: {store.get('currency') or 'unknown'}", f"- Timezone: {store.get('timezone') or 'unknown'}", "", "## Catalog", "", f"- Products: {len(products)}", f"- Collections: {len(collections)}", f"- Pages: {len(pages)}", "", "## Theme", "", f"- Active theme: {theme.get('name') or 'unknown'}", f"- Theme assets: {len(theme_assets)}", "", "## Changes", "", f"- Unsynced change events: {len(changes)}", "", ] )