"""Import brochure-derived collection JSON into normalization registries and commerce profile.""" from __future__ import annotations import json import re from pathlib import Path from typing import Any, Dict, Optional from sqlalchemy import select from sqlalchemy.orm import Session from channel.url_canonical import normalize_plp_path from db.brochure_taxonomy import reconcile_brochure_taxonomy_assignments from db.brochure_taxonomy import seed_brochure_taxonomy from db.location_registry import active_location_directory, ensure_default_home_surplus_locations from db.models import CollectionCommerceProfile, MasterBrand, MasterCollectionRegistry, MasterLocationRegistry, MasterProductFamily def load_collection_brochure_seed(path: str | Path) -> Dict[str, Any]: return json.loads(Path(path).read_text(encoding="utf-8")) def seed_collection_brochure( session: Session, payload: Dict[str, Any], *, channel_code: Optional[str] = None, connection_id: Optional[int] = None, sample_category_path_slug: str = "product-category/sample_doors", sample_cta_label: str = "Order Sample", appointment_cta_label: str = "Book Appointment", appointment_url: Optional[str] = None, is_active: bool = True, ) -> Dict[str, Any]: collection = payload.get("collection") or {} if not isinstance(collection, dict): raise ValueError("payload.collection must be an object") display_name = _clean_str(collection.get("display_name")) if not display_name: raise ValueError("collection.display_name is required") category_l1 = _clean_str(collection.get("category_l1")) or "Kitchen Cabinets" brand_label = _clean_str(collection.get("brand_label")) family_name = _clean_str(collection.get("product_family")) or "Cabinets" line_name = _clean_str(collection.get("line_name")) color_label = _clean_str(collection.get("color_label")) source_document = payload.get("source_document") if isinstance(payload.get("source_document"), dict) else {} source_collection_code = _clean_str(source_document.get("collection_code")) or _clean_str(collection.get("collection_code")) collection_code = source_collection_code or _clean_str(collection.get("canonical_code")) or _slug_code(display_name) brand = _upsert_brand(session, brand_label) family = _upsert_family(session, family_name) ensure_default_home_surplus_locations(session, brand_label=brand_label) registry = _upsert_collection_registry( session, code=collection_code, name=display_name, path_slug=normalize_plp_path(f"{category_l1}/{display_name}") or None, brand_id=brand.id if brand else None, family_id=family.id if family else None, aliases={ "canonical_code": _clean_str(collection.get("canonical_code")), "line_name": line_name, "color_label": color_label, "source_collection_code": source_collection_code, "matching_styles": payload.get("matching_styles") or [], }, notes=_compose_registry_notes(payload), ) sample_master_sku = _sample_master_sku(payload, collection_code) availability_modes = _availability_modes(payload) badges_payload = _availability_badges_payload(payload) landing_metadata = _landing_metadata(payload) normalized_availability = _normalized_availability(session, payload) if normalized_availability: landing_metadata["availability"] = normalized_availability notes = _compose_profile_notes(payload) row = session.scalar( select(CollectionCommerceProfile).where( CollectionCommerceProfile.collection_registry_id == registry.id, CollectionCommerceProfile.channel_code == channel_code, CollectionCommerceProfile.connection_id == connection_id, ) ) if row is None: row = CollectionCommerceProfile( collection_registry_id=registry.id, channel_code=channel_code, connection_id=connection_id, ) session.add(row) row.category_l1 = category_l1 row.collection_name = display_name row.sample_master_sku = sample_master_sku row.sample_category_path_slug = sample_category_path_slug row.availability_modes_json = {"modes": availability_modes} if availability_modes else None row.appointment_url = _clean_str(appointment_url) row.sample_cta_label = sample_cta_label row.appointment_cta_label = appointment_cta_label row.badges_json = badges_payload row.landing_metadata_json = landing_metadata row.is_active = bool(is_active) row.notes = notes session.flush() brochure_taxonomy = seed_brochure_taxonomy( session, payload, collection_registry_id=registry.id, ) sample_ensure = None if sample_master_sku: from db.sample_category import ensure_sample_master_products sample_ensure = ensure_sample_master_products( session, skus=[sample_master_sku], dry_run=False, ) return { "brand": {"id": brand.id, "code": brand.code, "name": brand.name} if brand else None, "family": {"id": family.id, "code": family.code, "name": family.name} if family else None, "collection_registry": {"id": registry.id, "code": registry.code, "name": registry.name}, "collection_commerce_profile": { "id": row.id, "collection_name": row.collection_name, "sample_master_sku": row.sample_master_sku, "availability_modes": availability_modes, "channel_code": row.channel_code, "connection_id": row.connection_id, }, "brochure_taxonomy": brochure_taxonomy, "sample_master_product": sample_ensure, } def seed_and_reconcile_collection_brochure_payloads( session: Session, payloads: list[Dict[str, Any]], *, channel_code: Optional[str] = None, connection_id: Optional[int] = None, sample_category_path_slug: str = "product-category/sample_doors", appointment_url: Optional[str] = None, replace_existing_master_assignments: bool = True, dry_run: bool = True, ) -> Dict[str, Any]: """Seed brochure JSON payloads, then fix master taxonomy assignments for those collections.""" seeded_results: list[Dict[str, Any]] = [] collection_codes: list[str] = [] for payload in payloads: seed_result = seed_collection_brochure( session, payload, channel_code=channel_code, connection_id=connection_id, sample_category_path_slug=sample_category_path_slug, appointment_url=appointment_url, ) seeded_results.append(seed_result) code = str((seed_result.get("collection_registry") or {}).get("code") or "").strip() if code: collection_codes.append(code) reconcile_results: list[Dict[str, Any]] = [] for code in sorted(set(collection_codes)): reconcile_results.append( { "collection_code": code, "result": reconcile_brochure_taxonomy_assignments( session, brochure_collection_code=code, replace_existing=replace_existing_master_assignments, dry_run=dry_run, ), } ) return { "payload_count": len(payloads), "seeded_count": len(seeded_results), "collection_codes": sorted(set(collection_codes)), "replace_existing_master_assignments": replace_existing_master_assignments, "dry_run": dry_run, "seed_results": seeded_results, "reconcile_results": reconcile_results, } def _upsert_brand(session: Session, brand_label: Optional[str]) -> Optional[MasterBrand]: if not brand_label: return None code = _slug_code(brand_label) row = session.scalar(select(MasterBrand).where(MasterBrand.code == code)) if row is None: row = MasterBrand(code=code, name=brand_label, aliases={"labels": [brand_label]}, is_active=True) session.add(row) else: row.name = brand_label row.aliases = _merge_label_aliases(row.aliases, brand_label) row.is_active = True session.flush() return row def _upsert_family(session: Session, family_name: Optional[str]) -> Optional[MasterProductFamily]: if not family_name: return None code = _slug_code(family_name) row = session.scalar(select(MasterProductFamily).where(MasterProductFamily.code == code)) if row is None: row = MasterProductFamily(code=code, name=family_name, aliases={"labels": [family_name]}, is_active=True) session.add(row) else: row.name = family_name row.aliases = _merge_label_aliases(row.aliases, family_name) row.is_active = True session.flush() return row def _upsert_collection_registry( session: Session, *, code: str, name: str, path_slug: Optional[str], brand_id: Optional[int], family_id: Optional[int], aliases: Dict[str, Any], notes: Optional[str], ) -> MasterCollectionRegistry: row = session.scalar(select(MasterCollectionRegistry).where(MasterCollectionRegistry.code == code)) if row is None: row = MasterCollectionRegistry(code=code, name=name, is_active=True) session.add(row) row.name = name row.path_slug = path_slug row.brand_id = brand_id row.family_id = family_id row.aliases = aliases row.notes = notes row.is_active = True session.flush() return row def _landing_metadata(payload: Dict[str, Any]) -> Dict[str, Any]: metadata = payload.get("landing_metadata") result = dict(metadata) if isinstance(metadata, dict) else {} if "kitchen_packages" not in result and isinstance(payload.get("kitchen_packages"), list): result["kitchen_packages"] = payload.get("kitchen_packages") for key in ( "source_document", "collection", "normalized_attributes", "specifications", "availability", "matching_styles", "sample_products", "sku_groups", "ambiguities", "normalization_seed_hints", ): if key in payload: result[key] = payload.get(key) return result def _normalized_availability(session: Session, payload: Dict[str, Any]) -> Dict[str, Any]: availability = payload.get("availability") if not isinstance(availability, dict): return {} directory = active_location_directory(session) by_label = _location_lookup(directory) modes = [] for item in availability.get("modes") or []: if not isinstance(item, dict): continue raw_scope = item.get("location_scope") normalized_item = dict(item) resolved = _resolve_location_scope(raw_scope, directory, by_label) if resolved: normalized_item.update(resolved) modes.append(normalized_item) out = dict(availability) out["modes"] = modes out["active_location_directory"] = directory return out def _availability_modes(payload: Dict[str, Any]) -> list[str]: availability = payload.get("availability") if not isinstance(availability, dict): return [] out: list[str] = [] for item in availability.get("modes") or []: if isinstance(item, dict): code = _clean_str(item.get("code")) if code: out.append(code) return out def _availability_badges(payload: Dict[str, Any]) -> list[str]: availability = payload.get("availability") if not isinstance(availability, dict): return [] out: list[str] = [] for item in availability.get("modes") or []: if isinstance(item, dict): label = _clean_str(item.get("label")) if label: out.append(label) return out def _availability_badges_payload(payload: Dict[str, Any]) -> Optional[dict]: """Build structured badges JSON (badges + availability_locations) from brochure availability.""" from db.collection_landing_schema import normalize_collection_badges availability = payload.get("availability") if not isinstance(availability, dict): return None badges = _availability_badges(payload) locations: list[dict] = [] for item in availability.get("modes") or []: if not isinstance(item, dict): continue label = _clean_str(item.get("label")) meaning = _clean_str(item.get("meaning") or item.get("description")) raw_locations = item.get("locations") if raw_locations is None: raw_locations = item.get("stores") or item.get("store_names") entry: dict = {"badge": label or "", "locations": raw_locations if raw_locations is not None else []} if meaning: entry["meaning"] = meaning if entry["badge"] or entry["locations"] or meaning: locations.append(entry) return normalize_collection_badges( { "badges": badges, "availability_locations": locations, } ) def _sample_master_sku(payload: Dict[str, Any], collection_code: str) -> Optional[str]: if "sample_products" in payload and payload.get("sample_products") == []: return None sample_products = payload.get("sample_products") or [] for item in sample_products: if not isinstance(item, dict): continue exact = _clean_str(item.get("expected_master_sku")) if exact: return exact suffix = _clean_str(item.get("sample_sku_suffix")) if suffix: return f"{collection_code}-{suffix}" brochure_sku = _clean_str(item.get("brochure_sku")) if brochure_sku: return f"{collection_code}-{brochure_sku}" return f"{collection_code}-SD" def _compose_registry_notes(payload: Dict[str, Any]) -> Optional[str]: source_document = payload.get("source_document") if isinstance(payload.get("source_document"), dict) else {} parts = [] if source_document.get("source_url"): parts.append(f"brochure={source_document['source_url']}") if source_document.get("file_name"): parts.append(f"file={source_document['file_name']}") if source_document.get("document_version_code"): parts.append(f"version={source_document['document_version_code']}") return "; ".join(parts) or None def _compose_profile_notes(payload: Dict[str, Any]) -> Optional[str]: source_document = payload.get("source_document") if isinstance(payload.get("source_document"), dict) else {} note_parts = [] if source_document.get("file_name"): note_parts.append(f"seeded from {source_document['file_name']}") notes = source_document.get("notes") if isinstance(notes, list) and notes: note_parts.extend(str(item).strip() for item in notes if str(item).strip()) return " | ".join(note_parts) or None def _location_lookup(directory: list[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: lookup: Dict[str, Dict[str, Any]] = {} for item in directory: labels = [item.get("name"), *item.get("labels", [])] for label in labels: text = _norm(label) if text: lookup[text] = item return lookup def _resolve_location_scope( raw_scope: Any, directory: list[Dict[str, Any]], by_label: Dict[str, Dict[str, Any]], ) -> Optional[Dict[str, Any]]: if raw_scope is None: return None if isinstance(raw_scope, str): text = _norm(raw_scope) if text == "all locations": return { "all_active_locations": True, "location_codes": [item["code"] for item in directory], "location_labels": [item["name"] for item in directory], } if text in {"not location-specific in brochure", "not location specific in brochure"}: return {"all_active_locations": False, "location_codes": [], "location_labels": [], "location_scope_raw": raw_scope} raw_scope = [raw_scope] if isinstance(raw_scope, list): labels: list[str] = [] codes: list[str] = [] unresolved: list[str] = [] for value in raw_scope: label = _clean_str(value) if not label: continue match = by_label.get(_norm(label)) if match is None: unresolved.append(label) labels.append(label) continue codes.append(match["code"]) labels.append(match["name"]) out = {"all_active_locations": False, "location_codes": codes, "location_labels": labels} if unresolved: out["unresolved_location_labels"] = unresolved return out return {"location_scope_raw": raw_scope} def _merge_label_aliases(existing: Any, value: str) -> Dict[str, Any]: labels = [] if isinstance(existing, dict) and isinstance(existing.get("labels"), list): labels = [str(item).strip() for item in existing["labels"] if str(item).strip()] if value not in labels: labels.append(value) return {"labels": labels} def _slug_code(value: str) -> str: text = re.sub(r"[^a-z0-9]+", "-", str(value).strip().lower()).strip("-") return text or "unknown" def _clean_str(value: Any) -> Optional[str]: if value is None: return None text = str(value).strip() return text or None def _norm(value: Any) -> str: return re.sub(r"[^a-z0-9]+", " ", str(value or "").strip().lower()).strip()