from __future__ import annotations import json from pathlib import Path from django.conf import settings from django.core.management.base import BaseCommand, CommandError from django.db.models import Q from oscar.core.loading import get_model from mandelstudio.content_hygiene import BLOCKED_DEMO_PAGE_SLUGS, DEMO_MARKERS from mandelstudio.idea_marketplace import ( FULL_DESCRIPTION_ATTRIBUTE_CODE, IDEA_PRODUCT_CLASS_NAME, IDEA_PRODUCTS, SHORT_DESCRIPTION_ATTRIBUTE_CODE, ) from mandelstudio.launch_validation import ( get_checkout_apps, get_declared_payment_apps, idea_marketplace_payments_enabled, validate_payment_provider_config, ) class Command(BaseCommand): help = ( "Fail-fast launch validation for idea marketplace: payment provider, " "catalog integrity, digital/non-shipping behavior, and EUR pricing." ) def handle(self, *args, **options): Product = get_model("catalogue", "Product") ProductClass = get_model("catalogue", "ProductClass") ProductAttribute = get_model("catalogue", "ProductAttribute") StockRecord = get_model("partner", "StockRecord") Page = get_model("wagtailcore", "Page") validate_payment_provider_config() installed_apps = list(settings.INSTALLED_APPS) payments_enabled = idea_marketplace_payments_enabled() payment_apps = get_declared_payment_apps(installed_apps) checkout_apps = get_checkout_apps() config_path = Path(__file__).resolve().parents[2] / "ocyan.json" if config_path.exists(): with config_path.open("r", encoding="utf-8") as handle: config_payload = json.load(handle) config_plugins = [ str(plugin) for plugin in config_payload.get("ocyan_plugins", []) ] if any( "demodata" in "".join(ch for ch in str(plugin).lower() if ch.isalnum()) for plugin in config_plugins ): raise CommandError( "Demo data plugin detected in ocyan.json. Remove it before launch." ) settings_payload = config_payload.get("settings", {}) domain = str(settings_payload.get("django", {}).get("domain", "")).strip() shop_base_url = str( settings_payload.get("oscar", {}).get("shop_base_url", "") ).strip("/") if not domain or domain.upper() == "CHANGE_ME": raise CommandError( "settings.django.domain is missing/placeholder in ocyan.json." ) if not shop_base_url: raise CommandError( "settings.oscar.shop_base_url is missing in ocyan.json." ) currency = getattr(settings, "OSCAR_DEFAULT_CURRENCY", "EUR") if currency != "EUR": raise CommandError(f"OSCAR_DEFAULT_CURRENCY must be EUR, got '{currency}'.") product_class = ProductClass.objects.filter( name=IDEA_PRODUCT_CLASS_NAME ).first() if product_class is None: raise CommandError(f"Missing ProductClass '{IDEA_PRODUCT_CLASS_NAME}'.") if product_class.requires_shipping: raise CommandError("Idea Product class requires_shipping must be False.") short_attr_exists = ProductAttribute.objects.filter( product_class=product_class, code=SHORT_DESCRIPTION_ATTRIBUTE_CODE ).exists() full_attr_exists = ProductAttribute.objects.filter( product_class=product_class, code=FULL_DESCRIPTION_ATTRIBUTE_CODE ).exists() if not short_attr_exists or not full_attr_exists: raise CommandError( "Missing required idea product attributes: short_description and/or full_description." ) expected_titles = {item.title for item in IDEA_PRODUCTS} expected_prices = {item.title: item.price_eur for item in IDEA_PRODUCTS} found_products = Product.objects.filter(product_class=product_class) found_titles = set(found_products.values_list("title", flat=True)) missing_titles = sorted(expected_titles - found_titles) if missing_titles: raise CommandError( f"Missing seeded idea products: {', '.join(missing_titles)}." ) non_public_idea_titles = list( found_products.filter( title__in=expected_titles, is_public=False ).values_list("title", flat=True) ) if non_public_idea_titles: raise CommandError( "Seeded idea products must be public to appear in the storefront. " f"Examples: {', '.join(sorted(non_public_idea_titles))}" ) invalid_shipping_products = [ product.title for product in found_products if getattr(product, "is_shipping_required", False) ] if invalid_shipping_products: raise CommandError( "Some idea products still require shipping; expected digital-only products: " + ", ".join(invalid_shipping_products) ) # Validate each seeded idea has EUR stockrecord pricing in the expected range. invalid_stockrecords: list[str] = [] missing_stockrecords: list[str] = [] for product in found_products.filter(title__in=expected_titles): stockrecord = ( StockRecord.objects.filter(product=product).order_by("id").first() ) if stockrecord is None: missing_stockrecords.append(product.title) continue if stockrecord.price_currency != "EUR": invalid_stockrecords.append( f"{product.title} (currency={stockrecord.price_currency})" ) continue expected = expected_prices[product.title] actual = stockrecord.price_excl_tax if actual is None or actual != expected: invalid_stockrecords.append( f"{product.title} (price_excl_tax={actual}, expected={expected})" ) continue if actual < 29 or actual > 149: invalid_stockrecords.append( f"{product.title} (out-of-range price_excl_tax={actual})" ) if missing_stockrecords: raise CommandError( "Missing stockrecords for seeded idea products: " + ", ".join(sorted(missing_stockrecords)) ) if invalid_stockrecords: raise CommandError( "Invalid stockrecord pricing for seeded idea products: " + "; ".join(sorted(invalid_stockrecords)) ) non_idea_public_titles = list( Product.objects.exclude(title__in=expected_titles) .filter(is_public=True) .values_list("title", flat=True)[:10] ) if non_idea_public_titles: raise CommandError( "Non-idea products are still public. Retire them before launch. " f"Examples: {', '.join(non_idea_public_titles)}" ) demo_page_filter = Q() for marker in DEMO_MARKERS: demo_page_filter |= ( Q(title__icontains=marker) | Q(slug__icontains=marker) | Q(search_description__icontains=marker) ) live_demo_pages = ( Page.objects.live() .public() .exclude(depth__lte=2) .filter(demo_page_filter | Q(slug__in=BLOCKED_DEMO_PAGE_SLUGS)) .values_list("title", "slug")[:10] ) if live_demo_pages: formatted = ", ".join( f"{title} ({slug})" for title, slug in live_demo_pages ) raise CommandError( "Demo-like pages are still live/public. Purge them before launch. " f"Examples: {formatted}" ) self.stdout.write( self.style.SUCCESS( "Idea marketplace launch validation passed: " f"{len(found_titles)} products, EUR currency, checkout apps={checkout_apps}, " f"payment apps={payment_apps}, payments_enabled={payments_enabled}." ) )