from __future__ import annotations import json from collections import defaultdict from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Tuple from channel.url_canonical import normalize_plp_path from db.brochure_taxonomy import extract_brochure_taxonomy_groups from db.collection_brochure_seed import load_collection_brochure_seed def transform_brochure_vocabulary_to_kitchen_taxonomy( *, vocabulary_path: str | Path, brochure_folder: str | Path, ) -> Dict[str, Any]: vocabulary = json.loads(Path(vocabulary_path).read_text(encoding="utf-8")) types = vocabulary.get("types") or [] sku_index, ambiguous_skus = _build_vocabulary_sku_index(types) collections: Dict[str, Dict[str, Any]] = {} confirmed_hierarchy: Dict[str, List[str]] = defaultdict(list) skipped_skus: List[Dict[str, Any]] = [] brochure_dir = Path(brochure_folder) for brochure_path in sorted(brochure_dir.glob("*.json")): payload = load_collection_brochure_seed(brochure_path) 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 {} canonical_code = str(collection.get("canonical_code") or "").strip() display_name = str(collection.get("display_name") or "").strip() collection_code = str(source_document.get("collection_code") or collection.get("collection_code") or "").strip() if not canonical_code or not display_name or not collection_code: continue assignments: List[Dict[str, Any]] = [] seen: set[Tuple[str, str]] = set() for group in extract_brochure_taxonomy_groups(payload): for sku_item in group.sku_items: brochure_sku = str(sku_item.get("brochure_sku") or "").strip().upper() if not brochure_sku: continue mapped = sku_index.get(brochure_sku) if not mapped: skipped_skus.append( { "collection_code": collection_code, "canonical_code": canonical_code, "brochure_sku": brochure_sku, "reason": "missing_in_vocabulary", } ) continue canonical_slug = normalize_plp_path(str(mapped.get("canonical_taxonomy_path_slug") or "")) if not canonical_slug: skipped_skus.append( { "collection_code": collection_code, "canonical_code": canonical_code, "brochure_sku": brochure_sku, "reason": "missing_canonical_taxonomy_path_slug", } ) continue dedupe_key = (brochure_sku, canonical_slug) if dedupe_key in seen: continue seen.add(dedupe_key) l2_label, l3_label = _labels_from_canonical_slug(canonical_slug) if l2_label and l3_label and l3_label not in confirmed_hierarchy[l2_label]: confirmed_hierarchy[l2_label].append(l3_label) assignments.append( { "sku": brochure_sku, "l2_category": l2_label, "l3_category": l3_label, "canonical_taxonomy_path_slug": canonical_slug, "canonical_taxonomy_path_label": _label_from_slug(canonical_slug), "type_key": mapped.get("type_key"), "shopping_l1": mapped.get("shopping_l1"), "shopping_l2": mapped.get("shopping_l2"), "brochure_l1_label": mapped.get("brochure_l1_label"), "brochure_l2_label": mapped.get("brochure_l2_label"), "type_label": mapped.get("type_label"), "source_family": mapped.get("type_key"), } ) if not assignments: continue collections[canonical_code] = { "display_name": display_name, "source_collection_code": collection_code, "sku_count": len(assignments), "assignments": sorted(assignments, key=lambda item: (str(item.get("canonical_taxonomy_path_slug") or ""), str(item.get("sku") or ""))), } return { "master_taxonomy_reference": { "confirmed_hierarchy": { key: sorted(values) for key, values in sorted(confirmed_hierarchy.items()) }, "proposed_new_l3_categories": {}, "source": "brochure_sku_vocabulary_transform", }, "collections": collections, "transform_summary": { "vocabulary_type_count": len(types), "collection_count": len(collections), "ambiguous_vocabulary_skus": sorted(ambiguous_skus), "skipped_sku_count": len(skipped_skus), "skipped_sku_sample": skipped_skus[:100], }, } def _build_vocabulary_sku_index(types: Iterable[Dict[str, Any]]) -> Tuple[Dict[str, Dict[str, Any]], set[str]]: sku_to_rows: Dict[str, List[Dict[str, Any]]] = defaultdict(list) for row in types: if not isinstance(row, dict): continue merged = { "type_key": row.get("type_key"), "canonical_taxonomy_path_slug": row.get("canonical_taxonomy_path_slug"), "shopping_l1": row.get("shopping_l1"), "shopping_l2": row.get("shopping_l2"), "brochure_l1_label": row.get("brochure_l1_label"), "brochure_l2_label": row.get("brochure_l2_label"), "type_label": row.get("type_label"), } for variant in row.get("variants") or []: if not isinstance(variant, dict): continue sku = str(variant.get("sku") or "").strip().upper() if not sku: continue sku_to_rows[sku].append(merged) index: Dict[str, Dict[str, Any]] = {} ambiguous: set[str] = set() for sku, rows in sku_to_rows.items(): unique = { ( str(row.get("type_key") or ""), str(row.get("canonical_taxonomy_path_slug") or ""), ): row for row in rows } if len(unique) > 1: ambiguous.add(sku) continue index[sku] = next(iter(unique.values())) return index, ambiguous def _labels_from_canonical_slug(path_slug: str) -> Tuple[str, str]: parts = [part for part in normalize_plp_path(path_slug).split("/") if part] if len(parts) < 3: return "", "" return _title_from_slug(parts[1]), _title_from_slug(parts[-1]) def _title_from_slug(value: str) -> str: return str(value or "").replace("-", " ").strip().title() def _label_from_slug(path_slug: str) -> str: parts = [part for part in normalize_plp_path(path_slug).split("/") if part] return " / ".join(_title_from_slug(part) for part in parts)