"""Kitchen package domain: resolve brochure packages into fixed-price bundle DTOs.""" from __future__ import annotations import hashlib from dataclasses import asdict, dataclass, field from typing import Any, Dict, Iterable, List, Optional, Sequence from sqlalchemy import select from sqlalchemy.orm import Session from db.brochure_sku_resolver import resolve_brochure_sku_from_description from db.brochure_taxonomy import _expected_master_sku from db.models import CollectionCommerceProfile, MasterCollectionRegistry, MasterProduct, MasterProductPrice PACKAGE_SKU_SUFFIX: Dict[str, str] = { "l_shape_8x10": "PKG-L810", "u_shape_10x10x10": "PKG-U101010", } KNOWN_PACKAGE_CODES = frozenset(PACKAGE_SKU_SUFFIX) @dataclass(frozen=True) class KitchenPackageComponent: brochure_sku: str master_sku: str qty: int description: str = "" mapping_confidence: str = "" master_exists: bool = False @dataclass(frozen=True) class KitchenPackage: package_code: str label: str price_usd: float collection_code: str collection_name: str category_l1: str package_master_sku: str components: List[KitchenPackageComponent] = field(default_factory=list) source_profile_id: Optional[int] = None brand: Optional[str] = None def missing_component_skus(self) -> List[str]: return [c.master_sku for c in self.components if not c.master_exists] def to_dict(self) -> Dict[str, Any]: return asdict(self) def package_master_sku(collection_code: str, package_code: str) -> str: code = str(collection_code or "").strip().upper() pkg = str(package_code or "").strip().lower() suffix = PACKAGE_SKU_SUFFIX.get(pkg) if not code: raise ValueError("collection_code is required") if not suffix: # Fallback: sanitize package_code into SKU-safe suffix safe = re.sub(r"[^A-Z0-9]+", "-", pkg.upper()).strip("-") suffix = f"PKG-{safe}" if safe else "PKG" return f"{code}-{suffix}" def normalize_kitchen_package_dicts(raw: Any) -> List[Dict[str, Any]]: if not isinstance(raw, list): return [] out: List[Dict[str, Any]] = [] for item in raw: if not isinstance(item, dict): continue package_code = str(item.get("package_code") or item.get("code") or "").strip().lower() if not package_code: continue label = str(item.get("label") or item.get("name") or package_code).strip() price = item.get("price_usd") if price is None: price = item.get("price") try: price_usd = float(price) except (TypeError, ValueError): continue includes = item.get("includes") if isinstance(item.get("includes"), list) else [] out.append( { "package_code": package_code, "label": label, "price_usd": price_usd, "includes": includes, } ) return out def extract_packages_from_landing_metadata(metadata: Any) -> List[Dict[str, Any]]: if not isinstance(metadata, dict): return [] packages = metadata.get("kitchen_packages") if packages is None and isinstance(metadata.get("landing_metadata"), dict): packages = metadata["landing_metadata"].get("kitchen_packages") return normalize_kitchen_package_dicts(packages) def resolve_kitchen_packages_from_payload( payload: Dict[str, Any], *, session: Optional[Session] = None, ) -> List[KitchenPackage]: """Resolve packages from a brochure JSON payload (no DB profile required).""" collection = payload.get("collection") if isinstance(payload.get("collection"), dict) else {} source_document = payload.get("source_document") if isinstance(payload.get("source_document"), dict) else {} collection_code = ( str(source_document.get("collection_code") or "").strip() or str(collection.get("collection_code") or "").strip() or str(collection.get("canonical_code") or "").strip() ) collection_name = str(collection.get("display_name") or collection.get("name") or "").strip() category_l1 = str(collection.get("category_l1") or "Kitchen Cabinets").strip() or "Kitchen Cabinets" brand = str(collection.get("brand_label") or "").strip() or None raw_packages = payload.get("kitchen_packages") if raw_packages is None: landing = payload.get("landing_metadata") if isinstance(payload.get("landing_metadata"), dict) else {} raw_packages = landing.get("kitchen_packages") packages = normalize_kitchen_package_dicts(raw_packages) return _build_packages( session=session, packages=packages, collection_code=collection_code, collection_name=collection_name, category_l1=category_l1, brand=brand, source_profile_id=None, ) def list_kitchen_packages_from_profiles( session: Session, *, collection_codes: Optional[Sequence[str]] = None, collection_names: Optional[Sequence[str]] = None, only_known_codes: bool = True, ) -> List[KitchenPackage]: """Load kitchen packages from active CollectionCommerceProfile rows.""" stmt = ( select(CollectionCommerceProfile, MasterCollectionRegistry) .outerjoin( MasterCollectionRegistry, MasterCollectionRegistry.id == CollectionCommerceProfile.collection_registry_id, ) .where(CollectionCommerceProfile.is_active.is_(True)) .order_by(CollectionCommerceProfile.id) ) wanted_codes = {str(c).strip().upper() for c in (collection_codes or []) if str(c).strip()} wanted_names = {str(n).strip().lower() for n in (collection_names or []) if str(n).strip()} out: List[KitchenPackage] = [] seen: set[str] = set() for profile, registry in session.execute(stmt).all(): metadata = profile.landing_metadata_json if isinstance(profile.landing_metadata_json, dict) else {} packages = extract_packages_from_landing_metadata(metadata) if not packages: continue collection_code = "" collection_name = str(profile.collection_name or "").strip() brand = None if registry is not None: collection_code = str(registry.code or "").strip() collection_name = collection_name or str(registry.name or "").strip() aliases = registry.aliases if isinstance(registry.aliases, dict) else {} brand = str(aliases.get("brand_label") or "").strip() or None if not collection_code: collection_meta = metadata.get("collection") if isinstance(metadata.get("collection"), dict) else {} source_document = metadata.get("source_document") if isinstance(metadata.get("source_document"), dict) else {} collection_code = ( str(source_document.get("collection_code") or "").strip() or str(collection_meta.get("collection_code") or "").strip() or str(collection_meta.get("canonical_code") or "").strip() ) if wanted_codes and collection_code.upper() not in wanted_codes: continue if wanted_names and collection_name.lower() not in wanted_names: continue category_l1 = str(profile.category_l1 or "Kitchen Cabinets").strip() or "Kitchen Cabinets" built = _build_packages( session=session, packages=packages, collection_code=collection_code, collection_name=collection_name, category_l1=category_l1, brand=brand, source_profile_id=int(profile.id), only_known_codes=only_known_codes, ) for pkg in built: if pkg.package_master_sku in seen: continue seen.add(pkg.package_master_sku) out.append(pkg) return out def upsert_package_master_products(session: Session, packages: Iterable[KitchenPackage]) -> Dict[str, Any]: """Persist package shell SKUs + base prices into master catalog.""" created = 0 updated = 0 for pkg in packages: sku = pkg.package_master_sku row = session.scalar(select(MasterProduct).where(MasterProduct.sku == sku)) name = f"{pkg.collection_name} — {pkg.label}".strip(" —") payload = { "product_role": "kitchen_package_bundle", "package_code": pkg.package_code, "collection_code": pkg.collection_code, "components": [ { "master_sku": c.master_sku, "brochure_sku": c.brochure_sku, "qty": c.qty, "description": c.description, } for c in pkg.components ], } row_hash = hashlib.sha256( f"{sku}|{pkg.price_usd}|{pkg.package_code}|{len(pkg.components)}".encode("utf-8") ).hexdigest() if row is None: row = MasterProduct( sku=sku, name=name, product_family="Kitchen Package", category_l1=pkg.category_l1, category_l2="Kitchen Packages", category_l3=pkg.label, brand=pkg.brand, collection=pkg.collection_name, assembly_type="bundle", status="enabled", normalization_status="ready", raw_payload=payload, row_hash=row_hash, is_active=True, ) session.add(row) session.flush() created += 1 else: row.name = name row.product_family = "Kitchen Package" row.category_l1 = pkg.category_l1 row.category_l2 = "Kitchen Packages" row.category_l3 = pkg.label row.brand = pkg.brand or row.brand row.collection = pkg.collection_name row.assembly_type = "bundle" row.status = "enabled" row.normalization_status = "ready" row.raw_payload = payload row.row_hash = row_hash row.is_active = True updated += 1 session.flush() price = session.scalar( select(MasterProductPrice).where( MasterProductPrice.product_id == row.id, MasterProductPrice.price_type == "base", ) ) if price is None: session.add( MasterProductPrice( product_id=row.id, sku=sku, price_type="base", amount=float(pkg.price_usd), currency="USD", source_label="kitchen_package", ) ) else: price.amount = float(pkg.price_usd) price.currency = "USD" price.source_label = "kitchen_package" price.sku = sku return {"created": created, "updated": updated} def _build_packages( *, session: Optional[Session], packages: List[Dict[str, Any]], collection_code: str, collection_name: str, category_l1: str, brand: Optional[str], source_profile_id: Optional[int], only_known_codes: bool = False, ) -> List[KitchenPackage]: if not collection_code: return [] existing: Optional[set[str]] = None if session is not None: prefix = f"{collection_code}-%" rows = session.scalars( select(MasterProduct.sku).where( MasterProduct.is_active.is_(True), MasterProduct.sku.ilike(prefix), ) ).all() existing = {str(s).strip() for s in rows if str(s).strip()} out: List[KitchenPackage] = [] for item in packages: package_code = str(item.get("package_code") or "").strip().lower() if only_known_codes and package_code not in KNOWN_PACKAGE_CODES: continue components: List[KitchenPackageComponent] = [] for inc in item.get("includes") or []: if not isinstance(inc, dict): continue brochure_sku = _resolve_package_component_brochure_sku(inc) if not brochure_sku: continue try: qty = int(inc.get("qty") or 1) except (TypeError, ValueError): qty = 1 qty = max(1, qty) master_sku = _expected_master_sku( brochure_collection_code=collection_code, brochure_sku=brochure_sku, explicit_master_sku=str(inc.get("master_sku") or "").strip() or None, ) if not master_sku: continue components.append( KitchenPackageComponent( brochure_sku=brochure_sku, master_sku=master_sku, qty=qty, description=str(inc.get("description") or "").strip(), mapping_confidence=str(inc.get("mapping_confidence") or "").strip(), master_exists=bool(existing is not None and master_sku in existing), ) ) if not components: continue out.append( KitchenPackage( package_code=package_code, label=str(item.get("label") or package_code), price_usd=float(item["price_usd"]), collection_code=collection_code, collection_name=collection_name or collection_code, category_l1=category_l1, package_master_sku=package_master_sku(collection_code, package_code), components=components, source_profile_id=source_profile_id, brand=brand, ) ) return out def _resolve_package_component_brochure_sku(item: Dict[str, Any]) -> str: description = str(item.get("description") or "").strip() inferred = resolve_brochure_sku_from_description(description) if inferred: return inferred return str(item.get("mapped_brochure_sku") or item.get("brochure_sku") or "").strip()