""" Product domain orchestration: use Shopify and Digistore services and persist locally. """ import logging from typing import Any from django.db import transaction from django.db.models import Q from products.models import Product from products.services.shopify_service import ShopifyService from products.services.digistore_service import DigistoreService logger = logging.getLogger(__name__) _SYNC_ERROR_MAX_LEN = 8000 def _truncate_product_sync_error(message: str) -> str: s = (message or "").strip() if len(s) <= _SYNC_ERROR_MAX_LEN: return s return s[: _SYNC_ERROR_MAX_LEN - 3] + "..." class ProductService: """Business logic for products (Shopify → Digistore).""" @staticmethod def fetch_from_shopify(product_id: str) -> dict[str, Any] | None: """Fetch product data from Shopify by ID.""" return ShopifyService.fetch_product(product_id) @staticmethod def _set_product_sync_error(shopify_product_id: str, message: str) -> None: sid = str(shopify_product_id or "").strip() if not sid: return err = _truncate_product_sync_error(message) updated = Product.objects.filter(shopify_product_id=sid).update(sync_error=err) if updated: return # Ensure we never drop sync errors even when product row # does not exist yet (e.g. early webhook failure path). product, created = Product.objects.get_or_create( shopify_product_id=sid, defaults={"sync_error": err}, ) if not created and product.sync_error != err: product.sync_error = err product.save(update_fields=["sync_error", "updated_at"]) @staticmethod def push_to_digistore(product_data: dict[str, Any]) -> str | None: """ Send product data to Digistore24 via DigistoreService and mirror it locally. """ shopify_id = product_data.get("id") or product_data.get("product_id") if not shopify_id: raise RuntimeError("Shopify product payload missing 'id'") title = product_data.get("title", "") default_variant_id = product_data.get("default_variant_id") # Prevent duplicate Digistore creates on concurrent webhook deliveries. # We lock the product row (or create and then lock it) so only one worker # can decide whether to create vs update for the same Shopify product. with transaction.atomic(): product, _ = Product.objects.select_for_update().get_or_create( shopify_product_id=str(shopify_id), defaults={ "title": title, "shopify_variant_id": str(default_variant_id) if default_variant_id else "", "raw_payload": product_data, }, ) existing_digistore_id = (product.digistore_product_id or "").strip() if existing_digistore_id: # Update existing Digistore product instead of creating a new one. DigistoreService.update_product(existing_digistore_id, product_data) digistore_id = existing_digistore_id payment_plan_id = "" else: # No mapping yet -> create a new Digistore product exactly once. digistore_id, payment_plan_id = DigistoreService.push_product(product_data) # Mirror latest state to our local DB, including default Shopify variant id product.title = title old_ds = (product.digistore_product_id or "").strip() new_ds = str(digistore_id) if digistore_id is not None else "" if old_ds != new_ds: product.digistore_paymentplan_id = "" product.digistore_product_id = new_ds if payment_plan_id and new_ds: product.digistore_paymentplan_id = str(payment_plan_id) product.shopify_variant_id = ( str(default_variant_id) if default_variant_id else "" ) product.raw_payload = product_data product.sync_error = "" product.save( update_fields=[ "title", "digistore_product_id", "digistore_paymentplan_id", "shopify_variant_id", "raw_payload", "sync_error", "updated_at", ] ) return product.digistore_product_id or digistore_id @staticmethod def sync_product(shopify_product_id: str) -> dict[str, Any]: """Full sync: fetch Shopify → push Digistore.""" sid = str(shopify_product_id).strip() data = ProductService.fetch_from_shopify(sid) if not data: ProductService._set_product_sync_error(sid, "product_not_found") return {"ok": False, "error": "product_not_found"} try: digistore_id = ProductService.push_to_digistore(data) except Exception as exc: logger.exception("push_to_digistore failed shopify_product_id=%s", sid) ProductService._set_product_sync_error(sid, str(exc)) return {"ok": False, "error": str(exc)} # Shopify metafields are pushed from `products.signals` on Product.post_save # after `push_to_digistore` persists `digistore_product_id`. return {"ok": True, "digistore_product_id": digistore_id} @staticmethod def list_ids_from_shopify(limit: int = 250) -> list[str]: """ List Shopify product IDs using the Admin API. Returns a list of string product IDs. """ return ShopifyService.list_product_ids(limit=limit) @staticmethod def lookup_digistore_product_id(shopify_product_or_variant_id: str) -> str | None: """ Resolve Digistore product id from a Shopify product id or variant id. """ sid = str(shopify_product_or_variant_id or "").strip() if not sid: return None product = Product.objects.filter( Q(shopify_product_id=sid) | Q(shopify_variant_id=sid) ).first() if not product: return None ds = (product.digistore_product_id or "").strip() return ds or None @staticmethod def list_from_db(limit: int = 100, offset: int = 0) -> list[dict[str, Any]]: """List products from our DB (synced records).""" qs = Product.objects.order_by("-created_at")[offset : offset + limit] return [ { "shopify_product_id": p.shopify_product_id, "digistore_product_id": p.digistore_product_id or None, "title": p.title, "created_at": p.created_at.isoformat(), } for p in qs ] @staticmethod def sync_all(limit: int = 250) -> dict[str, Any]: """ Sync all products: get IDs from Shopify, then sync each. """ ids = ProductService.list_ids_from_shopify(limit=limit) synced = 0 failed = 0 results = [] for pid in ids: r = ProductService.sync_product(pid) if r.get("ok"): synced += 1 results.append({"shopify_product_id": pid, "ok": True, **r}) else: failed += 1 results.append({"shopify_product_id": pid, "ok": False, "error": r.get("error", "unknown")}) return {"ok": True, "synced": synced, "failed": failed, "total": len(ids), "results": results} @staticmethod def _volume_metafield_payload(product: Product) -> dict[str, Any]: rules = product.pricing_rules.filter(is_active=True).order_by("quantity") currency = "EUR" if isinstance(product.raw_payload, dict): cur = product.raw_payload.get("currency") if cur and isinstance(cur, str): currency = cur.strip().upper()[:3] or currency return { "currency": currency, "is_discount_enabled": bool(product.is_volume_pricing_enabled), "tiers": [ {"qty": r.quantity, "price": float(r.unit_price), "is_default": r.is_default} for r in rules ], } @staticmethod def sync_shopify_metafields_for_product(product: Product) -> None: """ Push Shopify product metafields: Digistore product id + volume pricing JSON. Requires Admin API scopes for metafields (e.g. write_products / metafields). """ sid = (product.shopify_product_id or "").strip() if not sid: logger.warning("sync_shopify_metafields skipped: no shopify_product_id product_pk=%s", product.pk) return ds = (product.digistore_product_id or "").strip() if ds: try: ShopifyService.set_digistore_metafield(sid, ds) logger.info( "Shopify metafield digistore24_product_id set shopify_product_id=%s", sid, ) except Exception as exc: logger.exception( "Shopify set_digistore_metafield failed shopify_product_id=%s: %s", sid, exc, ) try: payload = ProductService._volume_metafield_payload(product) ShopifyService.set_volume_pricing_metafield(sid, payload) logger.info( "Shopify metafield volume_pricing set shopify_product_id=%s tiers=%s enabled=%s", sid, len(payload.get("tiers") or []), payload.get("is_discount_enabled"), ) except Exception as exc: logger.exception( "Shopify set_volume_pricing_metafield failed shopify_product_id=%s: %s", sid, exc, )