"""DB-owned master taxonomy registries — brands, families, types, collections.""" from __future__ import annotations import re from typing import Any, Dict, List, Optional, Tuple, Type, Union from sqlalchemy import func, or_, select from sqlalchemy.orm import Session from channel.url_canonical import normalize_plp_path, slugify_segment from db.collection_landing_pages import ( canonical_category, canonical_collection, collection_path_slug, DEFAULT_CATEGORY_L1, ) from db.models import ( CollectionLandingPage, MasterBrand, MasterCollectionRegistry, MasterProduct, MasterProductFamily, MasterProductType, ) RegistryModel = Union[MasterBrand, MasterProductFamily, MasterProductType, MasterCollectionRegistry] MATCH_MATCHED = "matched" MATCH_PROPOSED = "proposed" MATCH_AMBIGUOUS = "ambiguous" MATCH_NONE = "none" def registry_code_from_name(name: Optional[str]) -> str: text = str(name or "").strip() if not text: return "unknown" code = slugify_segment(text) return code or re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-") or "unknown" def aliases_to_json(items: Optional[List[str]]) -> Optional[Dict[str, Any]]: if isinstance(items, dict): payload = dict(items) raw_items = payload.get("items") if isinstance(raw_items, list): payload["items"] = [str(item).strip() for item in raw_items if str(item).strip()] return payload or None values = [str(item).strip() for item in (items or []) if str(item).strip()] return {"items": values} if values else None def aliases_from_json(value: Any) -> List[str]: values: List[str] = [] if isinstance(value, dict): items = value.get("items") if isinstance(items, list): values.extend(str(item).strip() for item in items if str(item).strip()) for key in ("source_collection_code", "source_code", "collection_code", "canonical_code"): item = value.get(key) if str(item or "").strip(): values.append(str(item).strip()) for key in ("code_aliases", "alternate_codes", "matching_styles"): items = value.get(key) if isinstance(items, list): values.extend(str(item).strip() for item in items if str(item).strip()) line_name = str(value.get("line_name") or "").strip() color_label = str(value.get("color_label") or "").strip() if line_name and color_label: values.append(f"{line_name} {color_label}") elif line_name: values.append(line_name) if isinstance(value, list): values.extend(str(item).strip() for item in value if str(item).strip()) seen = set() deduped: List[str] = [] for item in values: key = _normalize_match_text(item) if not key or key in seen: continue seen.add(key) deduped.append(item) return deduped def alias_codes_from_json(value: Any) -> List[str]: values: List[str] = [] if isinstance(value, dict): for key in ("source_collection_code", "source_code", "collection_code"): item = value.get(key) if str(item or "").strip(): values.append(str(item).strip().upper()) for key in ("code_aliases", "alternate_codes"): items = value.get(key) if isinstance(items, list): values.extend(str(item).strip().upper() for item in items if str(item).strip()) if isinstance(value, list): values.extend(str(item).strip().upper() for item in value if str(item).strip()) seen = set() deduped: List[str] = [] for item in values: if not item or item in seen: continue seen.add(item) deduped.append(item) return deduped def _norm_key(value: Optional[str]) -> str: return re.sub(r"[^a-z0-9]+", "", str(value or "").lower()) def _normalize_match_text(value: Optional[str]) -> str: text = str(value or "").strip().lower() return text.replace("grey", "gray") def _entity_dict(row: RegistryModel) -> Dict[str, Any]: aliases = aliases_from_json(row.aliases) base = { "id": row.id, "code": row.code, "name": row.name, "aliases": aliases, "is_active": row.is_active, "notes": row.notes, } if isinstance(row, MasterCollectionRegistry): alias_codes = alias_codes_from_json(row.aliases) base.update( { "path_slug": row.path_slug, "brand_id": row.brand_id, "family_id": row.family_id, "source_collection_code": str((row.aliases or {}).get("source_collection_code") or "").strip() if isinstance(row.aliases, dict) else "", "alias_codes": alias_codes, "sku_prefix": alias_codes[0] if alias_codes else row.code, "aliases_raw": row.aliases, } ) if isinstance(row, MasterProductType): base["family_id"] = row.family_id return base def list_brands(session: Session, *, search: Optional[str] = None, limit: int = 500) -> List[Dict[str, Any]]: return _list_entities(session, MasterBrand, search=search, limit=limit) def list_families(session: Session, *, search: Optional[str] = None, limit: int = 500) -> List[Dict[str, Any]]: return _list_entities(session, MasterProductFamily, search=search, limit=limit) def list_product_types( session: Session, *, family_id: Optional[int] = None, search: Optional[str] = None, limit: int = 500, ) -> List[Dict[str, Any]]: stmt = select(MasterProductType).where(MasterProductType.is_active.is_(True)).order_by(MasterProductType.name) if family_id is not None: stmt = stmt.where(MasterProductType.family_id == family_id) if search: pattern = f"%{search.strip()}%" stmt = stmt.where(or_(MasterProductType.name.ilike(pattern), MasterProductType.code.ilike(pattern))) if limit: stmt = stmt.limit(limit) return [_entity_dict(row) for row in session.scalars(stmt).all()] def list_collections( session: Session, *, family_id: Optional[int] = None, brand_id: Optional[int] = None, search: Optional[str] = None, limit: int = 500, ) -> List[Dict[str, Any]]: stmt = select(MasterCollectionRegistry).where(MasterCollectionRegistry.is_active.is_(True)).order_by( MasterCollectionRegistry.name ) if family_id is not None: stmt = stmt.where(MasterCollectionRegistry.family_id == family_id) if brand_id is not None: stmt = stmt.where(MasterCollectionRegistry.brand_id == brand_id) if search: pattern = f"%{search.strip()}%" stmt = stmt.where( or_( MasterCollectionRegistry.name.ilike(pattern), MasterCollectionRegistry.code.ilike(pattern), MasterCollectionRegistry.path_slug.ilike(pattern), ) ) if limit: stmt = stmt.limit(limit) return [_entity_dict(row) for row in session.scalars(stmt).all()] def upsert_brand(session: Session, payload: Dict[str, Any]) -> Dict[str, Any]: return _upsert_entity(session, MasterBrand, payload) def upsert_family(session: Session, payload: Dict[str, Any]) -> Dict[str, Any]: return _upsert_entity(session, MasterProductFamily, payload) def upsert_product_type(session: Session, payload: Dict[str, Any]) -> Dict[str, Any]: return _upsert_entity(session, MasterProductType, payload, extra_fields={"family_id": payload.get("family_id")}) def upsert_collection_registry( session: Session, payload: Dict[str, Any], *, allow_invent: Optional[bool] = True, ) -> Dict[str, Any]: """Upsert a collection registry row. Intentional dashboard/API callers default allow_invent=True. Auto invent paths (hierarchy/backfill) must pass allow_invent=False when frozen, or rely on their own invent gates before calling. """ from db.taxonomy_invent_policy import can_auto_invent name = str(payload.get("name") or "").strip() category_l1 = canonical_category(payload.get("category_l1")) or DEFAULT_CATEGORY_L1 path_slug = payload.get("path_slug") if not path_slug and name: path_slug = collection_path_slug(category_l1, name) extra = { "path_slug": normalize_plp_path(path_slug) if path_slug else None, "brand_id": payload.get("brand_id"), "family_id": payload.get("family_id"), } if not can_auto_invent(allow_invent=allow_invent): code = str(payload.get("code") or registry_code_from_name(name)).strip() existing = None if payload.get("id") is not None: existing = session.get(MasterCollectionRegistry, int(payload["id"])) if existing is None and code: existing = session.scalar( select(MasterCollectionRegistry).where(MasterCollectionRegistry.code == code).limit(1) ) if existing is None: raise ValueError( f"Taxonomy invent is frozen; collection registry does not exist for code={code!r}. " "Unlock invent or create the collection via the dashboard API." ) # Under freeze without unlock, return existing row unchanged (no auto mutations). return _entity_dict(existing) return _upsert_entity(session, MasterCollectionRegistry, payload, extra_fields=extra) def deactivate_brand(session: Session, brand_id: int) -> bool: return _deactivate_entity(session, MasterBrand, brand_id) def deactivate_family(session: Session, family_id: int) -> bool: return _deactivate_entity(session, MasterProductFamily, family_id) def deactivate_product_type(session: Session, product_type_id: int) -> bool: return _deactivate_entity(session, MasterProductType, product_type_id) def deactivate_collection_registry(session: Session, collection_id: int) -> bool: return _deactivate_entity(session, MasterCollectionRegistry, collection_id) def get_collection_registry_links(session: Session, *, code: Optional[str] = None, registry_id: Optional[int] = None) -> Dict[str, Any]: row = _get_collection_registry(session, code=code, registry_id=registry_id) if row is None: raise ValueError("Collection registry not found") landing = session.scalar( select(CollectionLandingPage).where(CollectionLandingPage.collection_registry_id == row.id).limit(1) ) if landing is None and row.path_slug: landing = session.scalar( select(CollectionLandingPage).where(CollectionLandingPage.path_slug == row.path_slug).limit(1) ) sku_count = session.scalar( select(func.count()) .select_from(MasterProduct) .where(MasterProduct.collection_registry_id == row.id) .where(MasterProduct.is_active.is_(True)) ) return { "registry": _entity_dict(row), "landing_page": {"path_slug": landing.path_slug, "title": landing.title} if landing else None, "sku_count": int(sku_count or 0), } def resolve_or_propose_registry_entities( session: Session, record: Dict[str, Any], ) -> Dict[str, Any]: """Resolve brand/family/collection registry entities for one master row payload.""" brand_name = str(record.get("brand") or "").strip() family_name = str(record.get("product_family") or record.get("category_l1") or "").strip() collection_name = canonical_collection(record.get("collection")) category_l1 = canonical_category(record.get("category_l1")) or DEFAULT_CATEGORY_L1 return { "brand": _resolve_entity(session, MasterBrand, brand_name), "family": _resolve_entity(session, MasterProductFamily, family_name or category_l1), "collection": _resolve_collection(session, collection_name, category_l1=category_l1), "product_type": _resolve_product_type(session, record), } def apply_registry_fks_to_product( session: Session, product: MasterProduct, *, resolution: Optional[Dict[str, Any]] = None, create_missing: bool = False, allow_invent: Optional[bool] = None, ) -> Dict[str, Any]: """Write registry FKs on a master_product from text fields or precomputed resolution.""" from db.taxonomy_invent_policy import can_auto_invent record = { "brand": product.brand, "product_family": product.product_family, "category_l1": product.category_l1, "collection": product.collection, } resolved = resolution or resolve_or_propose_registry_entities(session, record) applied: Dict[str, Any] = {} invent_ok = can_auto_invent(allow_invent=allow_invent) for key, attr, upsert_fn in ( ("brand", "brand_id", upsert_brand), ("family", "family_id", upsert_family), ("collection", "collection_registry_id", upsert_collection_registry), ): item = resolved.get(key) or {} entity_id = item.get("id") if entity_id is None and create_missing and invent_ok and item.get("status") == MATCH_PROPOSED: proposed = item.get("proposed") or {} if key == "collection": created = upsert_fn(session, proposed, allow_invent=True) else: created = upsert_fn(session, proposed) entity_id = created.get("id") if entity_id is not None and item.get("status") != MATCH_AMBIGUOUS: setattr(product, attr, int(entity_id)) applied[attr] = int(entity_id) type_item = resolved.get("product_type") or {} if type_item.get("id") and type_item.get("status") != MATCH_AMBIGUOUS: product.product_type_id = int(type_item["id"]) applied["product_type_id"] = int(type_item["id"]) session.flush() return applied def summarize_import_registry_intent(session: Session, records: List[Dict[str, Any]]) -> Dict[str, Any]: """Aggregate registry resolution for a batch of master import rows.""" proposed_brands: List[Dict[str, Any]] = [] proposed_collections: List[Dict[str, Any]] = [] proposed_families: List[Dict[str, Any]] = [] ambiguous: List[Dict[str, Any]] = [] product_type_inferences: List[Dict[str, Any]] = [] matched_counts = {"brand": 0, "family": 0, "collection": 0, "product_type": 0} seen_brands: set[str] = set() seen_collections: set[str] = set() seen_families: set[str] = set() seen_types: set[str] = set() for record in records: resolution = resolve_or_propose_registry_entities(session, record) for key, bucket, seen in ( ("brand", proposed_brands, seen_brands), ("family", proposed_families, seen_families), ("collection", proposed_collections, seen_collections), ): item = resolution.get(key) or {} name = str(item.get("name") or "").strip().lower() if item.get("status") == MATCH_MATCHED: matched_counts[key] += 1 elif item.get("status") == MATCH_PROPOSED and name and name not in seen: seen.add(name) bucket.append(item.get("proposed") or {"name": item.get("name")}) elif item.get("status") == MATCH_AMBIGUOUS and name: ambiguous.append({"dimension": key, "name": item.get("name"), "candidates": item.get("candidates")}) type_item = resolution.get("product_type") or {} type_name = str(type_item.get("name") or "").strip().lower() if type_item.get("status") == MATCH_MATCHED: matched_counts["product_type"] += 1 elif type_item.get("status") == MATCH_PROPOSED and type_name and type_name not in seen_types: seen_types.add(type_name) product_type_inferences.append(type_item.get("proposed") or {"name": type_item.get("name")}) elif type_item.get("status") == MATCH_AMBIGUOUS and type_name: ambiguous.append( {"dimension": "product_type", "name": type_item.get("name"), "candidates": type_item.get("candidates")} ) return { "row_count": len(records), "matched_counts": matched_counts, "proposed_brands": proposed_brands, "proposed_families": proposed_families, "proposed_collections": proposed_collections, "product_type_inferences": product_type_inferences, "ambiguous": ambiguous, } def backfill_master_taxonomy_from_products( session: Session, *, dry_run: bool = False, allow_invent: Optional[bool] = None, ) -> Dict[str, Any]: """Seed registry tables from distinct master_product text fields and link FKs.""" from db.collection_path_guard import is_polluted_collection_name, known_collection_names_for_hub from db.taxonomy_invent_policy import can_auto_invent stats = { "brands_created": 0, "families_created": 0, "collections_created": 0, "products_linked": 0, "landing_pages_linked": 0, "skipped_ambiguous": 0, "skipped_polluted": 0, "invent_frozen": False, } if not can_auto_invent(allow_invent=allow_invent): stats["invent_frozen"] = True if not dry_run: # Still link FKs for products that already match existing registry rows. return {**_backfill_link_existing_only(session), **stats, "dry_run": dry_run} return {**stats, "dry_run": True, "message": "invent frozen; no registry creates"} for brand_name in _distinct_values(session, MasterProduct.brand): if not brand_name: continue resolved = _resolve_entity(session, MasterBrand, brand_name) if resolved["status"] == MATCH_AMBIGUOUS: stats["skipped_ambiguous"] += 1 continue if resolved["status"] != MATCH_PROPOSED: continue if dry_run: stats["brands_created"] += 1 continue upsert_brand( session, resolved.get("proposed") or {"name": brand_name, "code": registry_code_from_name(brand_name)}, ) stats["brands_created"] += 1 for family_name in _distinct_values(session, MasterProduct.product_family): if not family_name: continue resolved = _resolve_entity(session, MasterProductFamily, family_name) if resolved["status"] == MATCH_AMBIGUOUS: stats["skipped_ambiguous"] += 1 continue if resolved["status"] != MATCH_PROPOSED: continue if dry_run: stats["families_created"] += 1 continue upsert_family( session, resolved.get("proposed") or {"name": family_name, "code": registry_code_from_name(family_name)}, ) stats["families_created"] += 1 rows = session.execute( select(MasterProduct.category_l1, MasterProduct.collection, MasterProduct.brand) .where(MasterProduct.is_active.is_(True)) .where(MasterProduct.collection.is_not(None)) .distinct() ).all() known_by_hub: Dict[str, set[str]] = {} for category_l1, collection, brand in rows: coll = canonical_collection(collection) if not coll: continue hub = canonical_category(category_l1) or DEFAULT_CATEGORY_L1 known = known_by_hub.get(hub) if known is None: known = known_collection_names_for_hub(session, category_l1=hub) known_by_hub[hub] = known if is_polluted_collection_name(session, coll, category_l1=hub, known_names=known): stats["skipped_polluted"] += 1 continue resolved = _resolve_collection(session, coll, category_l1=hub) if resolved["status"] == MATCH_AMBIGUOUS: stats["skipped_ambiguous"] += 1 continue if resolved["status"] != MATCH_PROPOSED: continue if dry_run: stats["collections_created"] += 1 continue brand_resolved = _resolve_entity(session, MasterBrand, brand) if brand else {"id": None} family_resolved = _resolve_entity( session, MasterProductFamily, canonical_category(category_l1) or "", ) proposed = resolved.get("proposed") or { "name": coll, "code": registry_code_from_name(coll), "category_l1": category_l1, } proposed.setdefault("brand_id", brand_resolved.get("id")) proposed.setdefault("family_id", family_resolved.get("id")) upsert_collection_registry(session, proposed, allow_invent=True) stats["collections_created"] += 1 if dry_run: return {**stats, "dry_run": True} products = session.scalars(select(MasterProduct).where(MasterProduct.is_active.is_(True))).all() for product in products: before = (product.brand_id, product.family_id, product.collection_registry_id) apply_registry_fks_to_product(session, product, create_missing=False) after = (product.brand_id, product.family_id, product.collection_registry_id) if before != after: stats["products_linked"] += 1 for landing in session.scalars(select(CollectionLandingPage)).all(): if landing.collection_registry_id: continue coll = canonical_collection(landing.collection) if not coll: continue resolved = _resolve_collection(session, coll, category_l1=landing.category_l1) if resolved.get("id"): landing.collection_registry_id = int(resolved["id"]) stats["landing_pages_linked"] += 1 session.flush() return stats def _backfill_link_existing_only(session: Session) -> Dict[str, Any]: """Link product/landing FKs to existing registry rows without creating any.""" stats = { "brands_created": 0, "families_created": 0, "collections_created": 0, "products_linked": 0, "landing_pages_linked": 0, "skipped_ambiguous": 0, "skipped_polluted": 0, } products = session.scalars(select(MasterProduct).where(MasterProduct.is_active.is_(True))).all() for product in products: before = (product.brand_id, product.family_id, product.collection_registry_id) apply_registry_fks_to_product(session, product, create_missing=False) after = (product.brand_id, product.family_id, product.collection_registry_id) if before != after: stats["products_linked"] += 1 for landing in session.scalars(select(CollectionLandingPage)).all(): if landing.collection_registry_id: continue coll = canonical_collection(landing.collection) if not coll: continue resolved = _resolve_collection(session, coll, category_l1=landing.category_l1) if resolved.get("id"): landing.collection_registry_id = int(resolved["id"]) stats["landing_pages_linked"] += 1 session.flush() return stats def _list_entities( session: Session, model: Type[RegistryModel], *, search: Optional[str], limit: int, ) -> List[Dict[str, Any]]: stmt = select(model).where(model.is_active.is_(True)).order_by(model.name) # type: ignore[attr-defined] if search: pattern = f"%{search.strip()}%" stmt = stmt.where(or_(model.name.ilike(pattern), model.code.ilike(pattern))) # type: ignore[attr-defined] if limit: stmt = stmt.limit(limit) return [_entity_dict(row) for row in session.scalars(stmt).all()] def _upsert_entity( session: Session, model: Type[RegistryModel], payload: Dict[str, Any], *, extra_fields: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: name = str(payload.get("name") or "").strip() if not name: raise ValueError("name is required") code = str(payload.get("code") or registry_code_from_name(name)).strip() entity_id = payload.get("id") row = None if entity_id is not None: row = session.get(model, int(entity_id)) if row is None: row = session.scalar(select(model).where(model.code == code).limit(1)) # type: ignore[attr-defined] aliases = payload.get("aliases") if "aliases" in payload: alias_json = aliases_to_json(aliases) if aliases is not None else None elif row is not None: alias_json = row.aliases else: alias_json = None data = { "code": code, "name": name, "is_active": bool(payload.get("is_active", True)), } if "notes" in payload: data["notes"] = payload.get("notes") elif row is None: data["notes"] = None if alias_json is not None or row is None or "aliases" in payload: data["aliases"] = alias_json if extra_fields: data.update({k: v for k, v in extra_fields.items() if k not in {"code", "name"}}) if row is None: row = model(**data) session.add(row) else: for key, value in data.items(): setattr(row, key, value) session.flush() return _entity_dict(row) def _deactivate_entity(session: Session, model: Type[RegistryModel], entity_id: int) -> bool: row = session.get(model, entity_id) if row is None: return False row.is_active = False session.flush() return True def _get_collection_registry( session: Session, *, code: Optional[str] = None, registry_id: Optional[int] = None, ) -> Optional[MasterCollectionRegistry]: if registry_id is not None: return session.get(MasterCollectionRegistry, int(registry_id)) if code: return session.scalar( select(MasterCollectionRegistry).where(MasterCollectionRegistry.code == str(code).strip()).limit(1) ) return None def _resolve_entity(session: Session, model: Type[RegistryModel], name: Optional[str]) -> Dict[str, Any]: text = str(name or "").strip() if not text: return {"status": MATCH_NONE, "id": None, "name": None, "candidates": []} code = registry_code_from_name(text) norm = _normalize_match_text(text) by_code = session.scalar( select(model).where(model.is_active.is_(True)).where(model.code == code).limit(1) # type: ignore[attr-defined] ) if by_code is not None: names = {_normalize_match_text(by_code.name), _normalize_match_text(by_code.code)} names.update(_normalize_match_text(alias) for alias in aliases_from_json(by_code.aliases)) if norm in names: return {**_entity_dict(by_code), "status": MATCH_MATCHED} candidates: List[Dict[str, Any]] = [] for row in session.scalars(select(model).where(model.is_active.is_(True))).all(): # type: ignore[attr-defined] names = {_normalize_match_text(row.name), _normalize_match_text(row.code)} names.update(_normalize_match_text(alias) for alias in aliases_from_json(row.aliases)) if norm in names: candidates.append(_entity_dict(row)) if len(candidates) == 1: return {**candidates[0], "status": MATCH_MATCHED} if len(candidates) > 1: return {"status": MATCH_AMBIGUOUS, "id": None, "name": text, "candidates": candidates} return { "status": MATCH_PROPOSED, "id": None, "name": text, "proposed": {"name": text, "code": code}, "candidates": [], } def _resolve_collection( session: Session, collection_name: Optional[str], *, category_l1: Optional[str] = None, ) -> Dict[str, Any]: coll = canonical_collection(collection_name) if not coll: return {"status": MATCH_NONE, "id": None, "name": None, "candidates": []} resolved = _resolve_entity(session, MasterCollectionRegistry, coll) if resolved["status"] == MATCH_PROPOSED: resolved["proposed"] = { "name": coll, "code": registry_code_from_name(coll), "category_l1": category_l1, "path_slug": collection_path_slug(category_l1 or DEFAULT_CATEGORY_L1, coll), } if resolved["status"] == MATCH_MATCHED and not resolved.get("path_slug"): row = session.get(MasterCollectionRegistry, int(resolved["id"])) if row and not row.path_slug: row.path_slug = collection_path_slug(category_l1 or DEFAULT_CATEGORY_L1, coll) session.flush() resolved["path_slug"] = row.path_slug return resolved def _resolve_product_type(session: Session, record: Dict[str, Any]) -> Dict[str, Any]: attrs = record.get("attributes") if isinstance(record.get("attributes"), dict) else {} type_name = ( str(record.get("product_type") or "").strip() or str(attrs.get("cabinet_type_sub_category_1") or attrs.get("product_type") or "").strip() ) if not type_name: return {"status": MATCH_NONE, "id": None, "name": None, "candidates": []} family_resolved = _resolve_entity(session, MasterProductFamily, record.get("product_family") or record.get("category_l1")) family_id = family_resolved.get("id") resolved = _resolve_entity(session, MasterProductType, type_name) if resolved["status"] == MATCH_PROPOSED: resolved["proposed"] = {"name": type_name, "code": registry_code_from_name(type_name), "family_id": family_id} return resolved def _distinct_values(session: Session, column) -> List[str]: rows = session.scalars( select(column).where(column.is_not(None)).where(column != "").distinct().order_by(column) ).all() return [str(row).strip() for row in rows if str(row).strip()]