from __future__ import annotations from collections import defaultdict from dataclasses import dataclass from typing import Any, Callable, Dict, Iterable, List, Optional from sqlalchemy import select from sqlalchemy.orm import Session from channel.url_canonical import normalize_plp_path from db.channel_listing_path import upsert_listing_path from db.master_taxonomy_intent import MATCH_MATCHED, _NodeIndex, _ProductLike, resolve_path_taxonomy from db.models import ( BrochureTaxonomyGroup, BrochureTaxonomyGroupSku, MasterCollectionRegistry, MasterProduct, MasterTaxonomyNode, ) ACTIVE_STATUS = "active" INACTIVE_STATUS = "inactive" @dataclass class ExtractedBrochureGroup: category_l1: str brochure_l1_label: Optional[str] brochure_l2_label: Optional[str] brochure_leaf_label: Optional[str] brochure_path_text: str canonical_taxonomy_path_slug: Optional[str] canonical_taxonomy_path_label: Optional[str] mapping_source: Optional[str] raw_payload: Dict[str, Any] sku_items: List[Dict[str, Any]] def extract_brochure_taxonomy_groups(payload: Dict[str, Any]) -> List[ExtractedBrochureGroup]: collection = payload.get("collection") if isinstance(payload.get("collection"), dict) else {} category_l1 = _clean(collection.get("category_l1")) or "Kitchen Cabinets" raw_groups = payload.get("taxonomy_groups") groups: List[ExtractedBrochureGroup] = [] if isinstance(raw_groups, list): for item in raw_groups: if not isinstance(item, dict): continue group = _group_from_dict(item, category_l1=category_l1) if group is not None: groups.append(group) return groups if isinstance(raw_groups, dict): for l1_label, level2 in raw_groups.items(): groups.extend(_groups_from_nested(level2, category_l1=category_l1, l1_label=_clean(l1_label))) if groups: return groups sku_catalog = payload.get("sku_catalog") if isinstance(sku_catalog, dict): groups.extend(_groups_from_sku_catalog(sku_catalog, category_l1=category_l1)) return [group for group in groups if not _is_probably_non_taxonomy_group(group)] def audit_brochure_taxonomy_payload( payload: Dict[str, Any], *, session: Optional[Session] = None, ) -> Dict[str, Any]: extracted = extract_brochure_taxonomy_groups(payload) resolved: List[Dict[str, Any]] = [] counts = defaultdict(int) index = _NodeIndex.load(session) if session is not None else None for group in extracted: resolution = _resolve_canonical_group(session, index=index, group=group) counts[resolution["mapping_status"]] += 1 resolved.append( { "brochure_path_text": group.brochure_path_text, "brochure_l1_label": group.brochure_l1_label, "brochure_l2_label": group.brochure_l2_label, "brochure_leaf_label": group.brochure_leaf_label, "canonical_taxonomy_path_slug": resolution.get("canonical_taxonomy_path_slug"), "canonical_taxonomy_path_label": resolution.get("canonical_taxonomy_path_label"), "mapping_status": resolution["mapping_status"], "mapping_source": resolution.get("mapping_source"), "sku_count": len(group.sku_items), "sku_sample": [item.get("brochure_sku") for item in group.sku_items[:10]], } ) return { "group_count": len(resolved), "mapping_status_counts": dict(counts), "groups": resolved, } def seed_brochure_taxonomy( session: Session, payload: Dict[str, Any], *, collection_registry_id: Optional[int] = None, ) -> Dict[str, Any]: source_document = payload.get("source_document") if isinstance(payload.get("source_document"), dict) else {} brochure_file_name = _clean(source_document.get("file_name")) brochure_collection_code = _clean(source_document.get("collection_code")) extracted = extract_brochure_taxonomy_groups(payload) index = _NodeIndex.load(session) created_groups = 0 updated_groups = 0 created_skus = 0 updated_skus = 0 deactivated_groups = 0 deactivated_skus = 0 active_group_ids: set[int] = set() for item in extracted: resolution = _resolve_canonical_group(session, index=index, group=item) row = session.scalar( select(BrochureTaxonomyGroup).where( BrochureTaxonomyGroup.collection_registry_id == collection_registry_id, BrochureTaxonomyGroup.category_l1 == item.category_l1, BrochureTaxonomyGroup.brochure_l1_label == item.brochure_l1_label, BrochureTaxonomyGroup.brochure_l2_label == item.brochure_l2_label, BrochureTaxonomyGroup.brochure_leaf_label == item.brochure_leaf_label, ) ) if row is None: row = BrochureTaxonomyGroup( collection_registry_id=collection_registry_id, category_l1=item.category_l1, brochure_l1_label=item.brochure_l1_label, brochure_l2_label=item.brochure_l2_label, brochure_leaf_label=item.brochure_leaf_label, brochure_path_text=item.brochure_path_text, is_active=True, ) session.add(row) created_groups += 1 else: updated_groups += 1 row.brochure_collection_code = brochure_collection_code row.brochure_file_name = brochure_file_name row.brochure_path_text = item.brochure_path_text row.canonical_taxonomy_path_slug = resolution.get("canonical_taxonomy_path_slug") row.canonical_taxonomy_path_label = resolution.get("canonical_taxonomy_path_label") row.mapping_status = resolution["mapping_status"] row.mapping_source = resolution.get("mapping_source") row.raw_payload = item.raw_payload row.notes = _compose_group_notes(item, resolution) row.is_active = True session.flush() active_group_ids.add(int(row.id)) seen_skus: set[str] = set() for sku_item in item.sku_items: brochure_sku = _clean(sku_item.get("brochure_sku")) if not brochure_sku or brochure_sku in seen_skus: continue seen_skus.add(brochure_sku) expected_master_sku = _expected_master_sku( brochure_collection_code=brochure_collection_code, brochure_sku=brochure_sku, explicit_master_sku=_clean(sku_item.get("expected_master_sku")), ) sku_row = session.scalar( select(BrochureTaxonomyGroupSku).where( BrochureTaxonomyGroupSku.group_id == row.id, BrochureTaxonomyGroupSku.brochure_sku == brochure_sku, ) ) if sku_row is None: sku_row = BrochureTaxonomyGroupSku( group_id=row.id, brochure_sku=brochure_sku, ) session.add(sku_row) created_skus += 1 else: updated_skus += 1 sku_row.expected_master_sku = expected_master_sku sku_row.assignment_status = ACTIVE_STATUS sku_row.raw_payload = sku_item sku_row.notes = None # Drop SKUs that are no longer part of this group in the current payload. for sku_row in session.scalars( select(BrochureTaxonomyGroupSku).where(BrochureTaxonomyGroupSku.group_id == row.id) ).all(): if sku_row.brochure_sku in seen_skus: continue if sku_row.assignment_status != INACTIVE_STATUS: sku_row.assignment_status = INACTIVE_STATUS deactivated_skus += 1 # Deactivate stale groups for this collection that are not in the current payload. if collection_registry_id is not None: stale_groups = session.scalars( select(BrochureTaxonomyGroup).where( BrochureTaxonomyGroup.collection_registry_id == collection_registry_id, BrochureTaxonomyGroup.is_active.is_(True), ) ).all() for group in stale_groups: if int(group.id) in active_group_ids: continue group.is_active = False deactivated_groups += 1 for sku_row in session.scalars( select(BrochureTaxonomyGroupSku).where( BrochureTaxonomyGroupSku.group_id == group.id, BrochureTaxonomyGroupSku.assignment_status == ACTIVE_STATUS, ) ).all(): sku_row.assignment_status = INACTIVE_STATUS deactivated_skus += 1 session.flush() return { "group_count": len(extracted), "created_groups": created_groups, "updated_groups": updated_groups, "created_skus": created_skus, "updated_skus": updated_skus, "deactivated_groups": deactivated_groups, "deactivated_skus": deactivated_skus, } def reconcile_brochure_taxonomy_assignments( session: Session, *, brochure_collection_code: Optional[str] = None, replace_existing: bool = True, dry_run: bool = True, progress: Optional[Callable[[str], None]] = None, ) -> Dict[str, Any]: """Assign brochure-mapped leaves, then expand to full Kitchen Cabinets path set. Each Kitchen Cabinets SKU should end with: - all-products - kitchen-cabinets (L1 hub) - kitchen-cabinets/{facet} (L2) - kitchen-cabinets/{facet}/{leaf} (L3) - kitchen-cabinets/{collection} (collection) """ from db.channel_listing_path import ( bulk_upsert_listing_assignments, deactivate_listing_assignments_for_skus, ) from db.collection_landing_pages import canonical_collection, collection_path_slug from db.master_taxonomy_product_paths import ( SHOPPING_FACET_SEGMENT_SLUGS, all_products_path_slug, expand_shopping_facet_ancestor_paths, ) def _progress(message: str) -> None: if progress is None: return try: progress(message) except Exception: pass stmt = ( select(BrochureTaxonomyGroup) .where(BrochureTaxonomyGroup.is_active.is_(True)) .where(BrochureTaxonomyGroup.mapping_status == MATCH_MATCHED) ) if brochure_collection_code: stmt = stmt.where(BrochureTaxonomyGroup.brochure_collection_code == brochure_collection_code) groups = list(session.scalars(stmt).all()) _progress(f"reconcile: loaded {len(groups)} matched brochure groups") leaf_by_sku: Dict[str, set[str]] = defaultdict(set) products_by_sku: Dict[str, MasterProduct] = {} skipped_groups = 0 index = _NodeIndex.load(session) root_slug = all_products_path_slug() ensured_paths: set[str] = set() def _ensure_path(slug: str, *, title: str, path_kind: str, notes: str) -> None: slug = normalize_plp_path(slug or "") if not slug or slug in ensured_paths: return upsert_listing_path( session, path_slug=slug, title=title, path_kind=path_kind, notes=notes, ) ensured_paths.add(slug) group_ids = [int(g.id) for g in groups] sku_rows_by_group: Dict[int, List[BrochureTaxonomyGroupSku]] = defaultdict(list) if group_ids: for sku_row in session.scalars( select(BrochureTaxonomyGroupSku).where( BrochureTaxonomyGroupSku.group_id.in_(group_ids), BrochureTaxonomyGroupSku.assignment_status == ACTIVE_STATUS, ) ).all(): sku_rows_by_group[int(sku_row.group_id)].append(sku_row) candidate_skus: List[str] = [] for group in groups: slug = normalize_plp_path(group.canonical_taxonomy_path_slug or "") if not slug: skipped_groups += 1 continue node = index.by_slug.get(slug) if node is not None and node.is_active: _ensure_path( slug, title=node.name, path_kind=node.node_kind, notes="linked by brochure taxonomy reconcile", ) for sku_row in sku_rows_by_group.get(int(group.id), []): master_sku = _clean(sku_row.expected_master_sku) if not master_sku: continue leaf_by_sku[master_sku].add(slug) candidate_skus.append(master_sku) # Batch-load master products once. missing = sorted({sku for sku in candidate_skus if sku not in products_by_sku}) chunk = 500 for i in range(0, len(missing), chunk): sku_chunk = missing[i : i + chunk] for product in session.scalars(select(MasterProduct).where(MasterProduct.sku.in_(sku_chunk))).all(): products_by_sku[product.sku] = product _progress(f"reconcile: resolved {len(products_by_sku)} master products for {len(leaf_by_sku)} brochure SKUs") # Drop SKUs with no master product row. for master_sku in list(leaf_by_sku.keys()): if master_sku not in products_by_sku: del leaf_by_sku[master_sku] # Promote L2-only shopping-facet leaves (e.g. panels-and-fillers) to L3 when possible. from db.master_taxonomy_product_paths import canonical_master_taxonomy_path_slugs l2_promote_candidates = 0 for master_sku, leaves in list(leaf_by_sku.items()): product = products_by_sku.get(master_sku) if product is None: continue l2_only = { slug for slug in leaves if len([p for p in slug.split("/") if p]) == 2 and slug.split("/")[-1] in SHOPPING_FACET_SEGMENT_SLUGS } if not l2_only: continue l2_promote_candidates += 1 resolved = canonical_master_taxonomy_path_slugs(session, product, index=index) for slug in list(l2_only): facet = slug.split("/")[-1] l3 = next( ( cand for cand in resolved if len([p for p in cand.split("/") if p]) >= 3 and cand.split("/")[1] == facet ), None, ) if l3: leaves.discard(slug) leaves.add(normalize_plp_path(l3)) if l2_promote_candidates: _progress(f"reconcile: L2-only promotion attempted for {l2_promote_candidates} SKUs") assignments_by_slug: Dict[str, set[str]] = defaultdict(set) per_sku_paths: Dict[str, List[str]] = {} sku_items = sorted(leaf_by_sku.items()) for idx, (master_sku, leaves) in enumerate(sku_items, start=1): product = products_by_sku[master_sku] hub = normalize_plp_path(product.category_l1 or "Kitchen Cabinets") expanded = list( expand_shopping_facet_ancestor_paths( sorted(leaves), index=index, include_hub=(hub == "kitchen-cabinets"), ) ) coll = canonical_collection(product.collection or "") if coll: coll_slug = normalize_plp_path( collection_path_slug(product.category_l1 or "Kitchen Cabinets", coll) or "" ) if coll_slug and coll_slug not in expanded: expanded.append(coll_slug) coll_node = index.by_slug.get(coll_slug) _ensure_path( coll_slug, title=(coll_node.name if coll_node is not None else (product.collection or coll_slug)), path_kind=(coll_node.node_kind if coll_node is not None else "collection"), notes="linked by brochure taxonomy reconcile (collection)", ) paths: List[str] = [] seen: set[str] = set() for slug in ([root_slug] if root_slug else []) + expanded: slug = normalize_plp_path(slug or "") if not slug or slug in seen: continue seen.add(slug) paths.append(slug) if slug == root_slug: _ensure_path( slug, title="All Products", path_kind="category", notes="global Magento Default/Root + Shopify All membership", ) else: node = index.by_slug.get(slug) if node is not None and node.is_active: _ensure_path( slug, title=node.name, path_kind=node.node_kind, notes="linked by brochure taxonomy reconcile", ) else: _ensure_path( slug, title=slug.replace("-", " ").title(), path_kind="category", notes="linked by brochure taxonomy reconcile (expanded)", ) assignments_by_slug[slug].add(master_sku) per_sku_paths[master_sku] = paths if idx % 1000 == 0: _progress(f"reconcile: expanded paths for {idx}/{len(sku_items)} SKUs") all_skus = sorted(per_sku_paths) deactivated = 0 if replace_existing and not dry_run: _progress(f"reconcile: deactivating prior listing assignments for {len(all_skus)} SKUs") deactivated = deactivate_listing_assignments_for_skus(session, skus=all_skus) upserted = 0 samples: List[Dict[str, Any]] = [] if dry_run: for slug, skus in sorted(assignments_by_slug.items()): upserted += len(skus) samples.append( { "path_slug": slug, "sku_count": len(skus), "sku_sample": sorted(skus)[:10], } ) else: from db.models import ChannelListingPath path_keys = list(assignments_by_slug.keys()) path_rows: Dict[str, ChannelListingPath] = {} if path_keys: for row in session.scalars( select(ChannelListingPath).where(ChannelListingPath.path_slug.in_(path_keys)) ).all(): path_rows[row.path_slug] = row pair_list: List[tuple[str, int]] = [] for slug, skus in sorted(assignments_by_slug.items()): path = path_rows.get(slug) if path is None: continue for sku in skus: pair_list.append((sku, int(path.id))) samples.append( { "path_slug": slug, "sku_count": len(skus), "sku_sample": sorted(skus)[:10], } ) _progress(f"reconcile: bulk upserting {len(pair_list)} listing assignments across {len(assignments_by_slug)} paths") upserted = bulk_upsert_listing_assignments( session, assignments=pair_list, match_source="brochure_taxonomy_reconcile", ) _progress( f"reconcile: done sku_count={len(all_skus)} path_count={len(assignments_by_slug)} " f"upserted={upserted} deactivated={deactivated} dry_run={dry_run}" ) return { "group_count": len(groups), "path_count": len(assignments_by_slug), "sku_count": len(all_skus), "replace_existing": replace_existing, "dry_run": dry_run, "deactivated": deactivated if not dry_run else 0, "upserted": upserted if not dry_run else 0, "would_upsert": upserted if dry_run else upserted, "skipped_groups": skipped_groups, "samples": samples[:25], "path_set_sample": [ {"master_sku": sku, "path_slugs": per_sku_paths[sku]} for sku in all_skus[:10] ], } def _group_from_dict(item: Dict[str, Any], *, category_l1: str) -> Optional[ExtractedBrochureGroup]: brochure_l1_label = _clean( item.get("brochure_l1_label") or item.get("l1") or item.get("section") or item.get("category") ) brochure_l2_label = _clean( item.get("brochure_l2_label") or item.get("l2") or item.get("bucket") or item.get("group") ) brochure_leaf_label = _clean( item.get("brochure_leaf_label") or item.get("leaf") or item.get("subcategory") or item.get("label") ) canonical_path_slug = normalize_plp_path(item.get("canonical_taxonomy_path_slug") or "") canonical_path_label = _clean(item.get("canonical_taxonomy_path_label") or item.get("canonical_path")) sku_items = _sku_items_from_value(item.get("skus") or item.get("sku_items")) if not (brochure_l1_label or brochure_l2_label or brochure_leaf_label) or not sku_items: return None return ExtractedBrochureGroup( category_l1=category_l1, brochure_l1_label=brochure_l1_label, brochure_l2_label=brochure_l2_label, brochure_leaf_label=brochure_leaf_label, brochure_path_text=_brochure_path_text(category_l1, brochure_l1_label, brochure_l2_label, brochure_leaf_label), canonical_taxonomy_path_slug=canonical_path_slug or None, canonical_taxonomy_path_label=canonical_path_label, mapping_source="payload_explicit" if canonical_path_slug or canonical_path_label else None, raw_payload=dict(item), sku_items=sku_items, ) def _groups_from_nested( value: Any, *, category_l1: str, l1_label: Optional[str], l2_label: Optional[str] = None, ) -> List[ExtractedBrochureGroup]: out: List[ExtractedBrochureGroup] = [] if isinstance(value, dict): for key, child in value.items(): label = _clean(key) if isinstance(child, (list, tuple)): sku_items = _sku_items_from_value(child) if sku_items: out.append( ExtractedBrochureGroup( category_l1=category_l1, brochure_l1_label=l1_label, brochure_l2_label=l2_label, brochure_leaf_label=label, brochure_path_text=_brochure_path_text(category_l1, l1_label, l2_label, label), canonical_taxonomy_path_slug=None, canonical_taxonomy_path_label=None, mapping_source=None, raw_payload={"leaf": label, "skus": child}, sku_items=sku_items, ) ) continue next_l2 = l2_label if l2_label is not None else label next_leaf = None if l2_label is None else label if next_leaf is not None and isinstance(child, dict) and any(isinstance(v, (list, tuple)) for v in child.values()): out.extend(_groups_from_nested(child, category_l1=category_l1, l1_label=l1_label, l2_label=next_leaf)) else: out.extend(_groups_from_nested(child, category_l1=category_l1, l1_label=l1_label, l2_label=next_l2)) elif isinstance(value, (list, tuple)) and l2_label: sku_items = _sku_items_from_value(value) if sku_items: out.append( ExtractedBrochureGroup( category_l1=category_l1, brochure_l1_label=l1_label, brochure_l2_label=l2_label, brochure_leaf_label=None, brochure_path_text=_brochure_path_text(category_l1, l1_label, l2_label, None), canonical_taxonomy_path_slug=None, canonical_taxonomy_path_label=None, mapping_source=None, raw_payload={"skus": list(value)}, sku_items=sku_items, ) ) return out def _groups_from_sku_catalog( sku_catalog: Dict[str, Any], *, category_l1: str, ) -> List[ExtractedBrochureGroup]: out: List[ExtractedBrochureGroup] = [] for raw_key, value in sku_catalog.items(): key = _clean(raw_key) if not key: continue brochure_l1_label, brochure_l2_label = _sku_catalog_bucket_labels(key) structured_groups = _groups_from_structured_sku_catalog_items( value, category_l1=category_l1, brochure_l1_label=brochure_l1_label, brochure_l2_label=brochure_l2_label, raw_key=key, ) if structured_groups: out.extend(structured_groups) continue if isinstance(value, dict): for leaf_key, leaf_value in value.items(): leaf_label = _humanize_catalog_key(leaf_key) sku_items = _sku_items_from_catalog_value(leaf_value) if sku_items: out.append( ExtractedBrochureGroup( category_l1=category_l1, brochure_l1_label=brochure_l1_label, brochure_l2_label=brochure_l2_label, brochure_leaf_label=leaf_label, brochure_path_text=_brochure_path_text( category_l1, brochure_l1_label, brochure_l2_label, leaf_label, ), canonical_taxonomy_path_slug=None, canonical_taxonomy_path_label=None, mapping_source="sku_catalog", raw_payload={"sku_catalog_key": key, "leaf_key": leaf_key, "value": leaf_value}, sku_items=sku_items, ) ) continue if isinstance(leaf_value, dict): for sub_leaf_key, sub_leaf_value in leaf_value.items(): sub_leaf_label = _humanize_catalog_key(sub_leaf_key) sub_sku_items = _sku_items_from_catalog_value(sub_leaf_value) if not sub_sku_items: continue out.append( ExtractedBrochureGroup( category_l1=category_l1, brochure_l1_label=brochure_l1_label, brochure_l2_label=brochure_l2_label, brochure_leaf_label=f"{leaf_label} / {sub_leaf_label}", brochure_path_text=_brochure_path_text( category_l1, brochure_l1_label, brochure_l2_label, f"{leaf_label} / {sub_leaf_label}", ), canonical_taxonomy_path_slug=None, canonical_taxonomy_path_label=None, mapping_source="sku_catalog", raw_payload={ "sku_catalog_key": key, "leaf_key": leaf_key, "sub_leaf_key": sub_leaf_key, "value": sub_leaf_value, }, sku_items=sub_sku_items, ) ) else: sku_items = _sku_items_from_catalog_value(value) if sku_items: out.append( ExtractedBrochureGroup( category_l1=category_l1, brochure_l1_label=brochure_l1_label, brochure_l2_label=brochure_l2_label, brochure_leaf_label=None, brochure_path_text=_brochure_path_text(category_l1, brochure_l1_label, brochure_l2_label, None), canonical_taxonomy_path_slug=None, canonical_taxonomy_path_label=None, mapping_source="sku_catalog", raw_payload={"sku_catalog_key": key, "value": value}, sku_items=sku_items, ) ) return out def _groups_from_structured_sku_catalog_items( value: Any, *, category_l1: str, brochure_l1_label: str, brochure_l2_label: Optional[str], raw_key: str, ) -> List[ExtractedBrochureGroup]: if not isinstance(value, list) or not value or not all(isinstance(item, dict) for item in value): return [] grouped: Dict[tuple[Optional[str], Optional[str], Optional[str]], List[Dict[str, Any]]] = {} for item in value: classification = _classify_structured_sku_catalog_item( item, brochure_l1_label=brochure_l1_label, brochure_l2_label=brochure_l2_label, ) if classification is None: continue brochure_sku = _clean(item.get("sku") or item.get("brochure_sku")) if not brochure_sku: continue item_payload = dict(item) item_payload["brochure_sku"] = brochure_sku key = ( classification["brochure_l1_label"], classification["brochure_l2_label"], classification["brochure_leaf_label"], ) grouped.setdefault(key, []).append(item_payload) out: List[ExtractedBrochureGroup] = [] for (l1_label, l2_label, leaf_label), items in grouped.items(): out.append( ExtractedBrochureGroup( category_l1=category_l1, brochure_l1_label=l1_label, brochure_l2_label=l2_label, brochure_leaf_label=leaf_label, brochure_path_text=_brochure_path_text(category_l1, l1_label, l2_label, leaf_label), canonical_taxonomy_path_slug=None, canonical_taxonomy_path_label=None, mapping_source="sku_catalog", raw_payload={"sku_catalog_key": raw_key, "value": items}, sku_items=[ { "brochure_sku": item["brochure_sku"], "expected_master_sku": _clean(item.get("expected_master_sku")), } for item in items ], ) ) return out def _classify_structured_sku_catalog_item( item: Dict[str, Any], *, brochure_l1_label: str, brochure_l2_label: Optional[str], ) -> Optional[Dict[str, Optional[str]]]: sku_type = _clean(item.get("sku_type")) door_count = _int_or_none(item.get("door_count")) glass = _boolish(item.get("glass")) if brochure_l1_label == "Wall Cabinets" and brochure_l2_label == "Basic Wall": if glass: # glass=true basic_wall rows must never stay under Basic Wall. # Missing door_count is intentionally skipped for manual review. if door_count is None: return None leaf = "Double Door Glass Wall Cabinet" if door_count >= 2 else "Single Door Glass Wall Cabinet" return { "brochure_l1_label": brochure_l1_label, "brochure_l2_label": leaf, "brochure_leaf_label": None, } return { "brochure_l1_label": brochure_l1_label, "brochure_l2_label": "Basic Wall", "brochure_leaf_label": None, } return None def _resolve_canonical_group( session: Optional[Session], *, index: Optional[_NodeIndex], group: ExtractedBrochureGroup, ) -> Dict[str, Any]: slug = normalize_plp_path(group.canonical_taxonomy_path_slug or "") if slug and session is not None and index is not None: node = index.by_slug.get(slug) if node is not None and node.is_active: return { "mapping_status": MATCH_MATCHED, "canonical_taxonomy_path_slug": slug, "canonical_taxonomy_path_label": node.name, "mapping_source": group.mapping_source or "payload_explicit", } return { "mapping_status": "invalid", "canonical_taxonomy_path_slug": slug, "canonical_taxonomy_path_label": group.canonical_taxonomy_path_label, "mapping_source": group.mapping_source or "payload_explicit", } if group.canonical_taxonomy_path_label and session is not None and index is not None: explicit = _resolve_path_label(session, group=group, path=group.canonical_taxonomy_path_label, index=index) if explicit is not None: return explicit alias_path = _canonical_alias_path(group) if alias_path and session is not None and index is not None: aliased = _resolve_path_label(session, group=group, path=alias_path, index=index) if aliased is not None: aliased["mapping_source"] = "brochure_alias" return aliased if session is None or index is None: return { "mapping_status": "unmapped", "canonical_taxonomy_path_slug": slug or None, "canonical_taxonomy_path_label": group.canonical_taxonomy_path_label, "mapping_source": group.mapping_source, } candidates = [ _brochure_candidate_path(group.category_l1, group.brochure_l1_label, group.brochure_leaf_label), _brochure_candidate_path(group.category_l1, group.brochure_l1_label, group.brochure_l2_label), ] for candidate in candidates: if not candidate: continue resolved = _resolve_path_label(session, group=group, path=candidate, index=index) if resolved is not None: return resolved return { "mapping_status": "unmapped", "canonical_taxonomy_path_slug": None, "canonical_taxonomy_path_label": None, "mapping_source": None, } def _resolve_path_label( session: Session, *, group: ExtractedBrochureGroup, path: str, index: _NodeIndex, ) -> Optional[Dict[str, Any]]: product = _ProductLike( sku=_clean(group.sku_items[0].get("expected_master_sku")) or _clean(group.sku_items[0].get("brochure_sku")) or "BROCHURE", category_l1=group.category_l1, category_l2=None, collection=None, brand=None, product_family="Cabinets", ) resolution = resolve_path_taxonomy(session, path, product, index=index) if resolution.get("status") != MATCH_MATCHED: return None return { "mapping_status": MATCH_MATCHED, "canonical_taxonomy_path_slug": normalize_plp_path(resolution.get("path_slug") or ""), "canonical_taxonomy_path_label": resolution.get("name"), "mapping_source": "path_match", } def _brochure_candidate_path( category_l1: str, brochure_l1_label: Optional[str], leaf_label: Optional[str], ) -> Optional[str]: if not brochure_l1_label or not leaf_label: return None return f"{category_l1}/{brochure_l1_label}/{leaf_label}" def _sku_items_from_value(value: Any) -> List[Dict[str, Any]]: out: List[Dict[str, Any]] = [] if isinstance(value, (list, tuple)): for item in value: if isinstance(item, dict): brochure_sku = _clean(item.get("brochure_sku") or item.get("sku")) if brochure_sku: out.append( { "brochure_sku": brochure_sku, "expected_master_sku": _clean(item.get("expected_master_sku")), } ) else: brochure_sku = _clean(item) if brochure_sku: out.append({"brochure_sku": brochure_sku}) return out def _sku_items_from_catalog_value(value: Any) -> List[Dict[str, Any]]: if isinstance(value, dict): if "skus" in value: return _sku_items_from_value(value.get("skus")) if "items" in value: return _sku_items_from_value(value.get("items")) if "sku_list" in value: return _sku_items_from_value(value.get("sku_list")) if "brochure_skus" in value: return _sku_items_from_value(value.get("brochure_skus")) return [] return _sku_items_from_value(value) def _expected_master_sku( *, brochure_collection_code: Optional[str], brochure_sku: str, explicit_master_sku: Optional[str], ) -> Optional[str]: if explicit_master_sku: return explicit_master_sku code = _clean(brochure_collection_code) sku = _clean(brochure_sku) if not sku: return None # Brochure SKUs often contain hyphens (GD-W1530, OCM8-1). Only skip prefixing # when the value already starts with the collection code. if code: prefix = f"{code}-" if sku.upper().startswith(prefix.upper()): return sku return f"{prefix}{sku}" return sku def _brochure_path_text( category_l1: str, brochure_l1_label: Optional[str], brochure_l2_label: Optional[str], brochure_leaf_label: Optional[str], ) -> str: parts = [category_l1, brochure_l1_label, brochure_l2_label, brochure_leaf_label] return " > ".join(part for part in parts if _clean(part)) def _compose_group_notes(group: ExtractedBrochureGroup, resolution: Dict[str, Any]) -> Optional[str]: notes = [] if resolution.get("mapping_status") == "unmapped": notes.append("Awaiting canonical master taxonomy mapping.") elif resolution.get("mapping_status") == "invalid": notes.append("Explicit canonical taxonomy path did not match an active master taxonomy node.") return " ".join(notes) or None def _clean(value: Any) -> Optional[str]: text = str(value or "").strip() if not text: return None return text def _int_or_none(value: Any) -> Optional[int]: try: text = str(value).strip() if not text: return None return int(float(text)) except (TypeError, ValueError): return None def _boolish(value: Any) -> bool: if isinstance(value, bool): return value text = str(value or "").strip().lower() return text in {"1", "true", "yes", "y"} def _humanize_catalog_key(value: Any) -> Optional[str]: text = _clean(value) if not text: return None pieces = str(text).replace("-", " ").replace("_", " ").split() return " ".join(piece.capitalize() for piece in pieces) def _sku_catalog_bucket_labels(key: str) -> tuple[str, Optional[str]]: norm = str(key).strip().lower() mapped = { "basic_wall": ("Wall Cabinets", "Basic Wall"), "stacked_wall": ("Wall Cabinets", "Stacked Wall"), "bridge_wall": ("Wall Cabinets", "Bridge Wall"), "corner_wall": ("Wall Cabinets", "Corner Wall"), "special_wall": ("Wall Cabinets", "Special Wall"), "basic_bases": ("Base Cabinets", "Basic Bases"), "sink_bases": ("Base Cabinets", "Sink Bases"), "drawer_bases": ("Base Cabinets", "Drawer Bases"), "special_bases": ("Base Cabinets", "Special Bases"), "tall_cabinets": ("Base Cabinets", "Tall Cabinets"), "base_cabinets": ("Base Cabinets", "Base Cabinets"), "panels_and_fillers": ("Panels and Fillers", None), "moldings": ("Mouldings", None), "mouldings": ("Mouldings", None), "accessories": ("Accessories", None), } if norm in mapped: return mapped[norm] human = _humanize_catalog_key(norm) or norm return human, None def _canonical_alias_path(group: ExtractedBrochureGroup) -> Optional[str]: l1 = _clean(group.brochure_l1_label) l2 = _clean(group.brochure_l2_label) leaf = _clean(group.brochure_leaf_label) category_l1 = _clean(group.category_l1) or "Kitchen Cabinets" aliases = { ("Wall Cabinets", "Bridge Wall"): f"{category_l1}/Wall Cabinets/Bridge Wall", ("Base Cabinets", "Basic Bases"): f"{category_l1}/Base Cabinets/Basic Base", ("Base Cabinets", "Sink Bases"): f"{category_l1}/Base Cabinets/Sink Base", ("Base Cabinets", "Drawer Bases"): f"{category_l1}/Base Cabinets/Drawer Base", ("Base Cabinets", "Special Bases"): f"{category_l1}/Base Cabinets/Special Base", ("Panels and Fillers", None): f"{category_l1}/Panels and Fillers", ("Mouldings", None): f"{category_l1}/Mouldings", ("Accessories", None): f"{category_l1}/Accessories", } direct = aliases.get((l1, l2)) if direct: return direct if l1 == "Base Cabinets" and l2 == "Base Cabinets" and _looks_like_corner_base_group(group): return f"{category_l1}/Base Cabinets/Corner Base" if l1 == "Base Cabinets" and l2 == "Tall Cabinets": return f"{category_l1}/Tall Cabinets" if l1 == "Accessories" and leaf: return f"{category_l1}/Accessories" return None def _looks_like_corner_base_group(group: ExtractedBrochureGroup) -> bool: samples = [str(item.get("brochure_sku") or "").strip().upper() for item in group.sku_items] if not samples: return False prefixes = ("BCOR", "BBC", "EZR", "EZ", "PBWLS") return all(sample.startswith(prefixes) for sample in samples if sample) def _is_probably_non_taxonomy_group(group: ExtractedBrochureGroup) -> bool: label_parts = " ".join( str(part or "") for part in (group.brochure_l1_label, group.brochure_l2_label, group.brochure_leaf_label) ).lower() if "note" in label_parts: return True if "touch up" in label_parts: return True samples = [str(item.get("brochure_sku") or "").strip() for item in group.sku_items] if not samples: return True non_sku_phrases = 0 for sample in samples: if not sample: continue if len(sample.split()) >= 4 or "." in sample: non_sku_phrases += 1 continue upper = sample.upper() if not any(ch.isdigit() for ch in upper) and "-" not in upper and "/" not in upper: if upper not in {"SD", "SC10", "V54"}: non_sku_phrases += 1 return non_sku_phrases >= max(2, len(samples) // 2)