"""Audit canonical storefront left-pane filter values before channel push. Examples: python -m app.jobs.audit_storefront_filter_readiness python -m app.jobs.audit_storefront_filter_readiness --limit 250 python -m app.jobs.audit_storefront_filter_readiness --attribute item_availability --attribute color_finish """ from __future__ import annotations import argparse import json import re from collections import Counter from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence from sqlalchemy import select from sqlalchemy.orm import Session from db.canonical_attributes import ( _normalize_measurement_value, canonicalize_attribute_fields, derive_composite_attribute_value, ) from db.channel_exports import ( _attach_location_availability_fields, _core_fields, _location_availability, compose_canonical_fields, ) from db.master_taxonomy_hierarchy import _attrs_by_sku from db.models import MasterProduct, MasterProductRelation from db.session import get_session from db.shopping_facet_catalog import COLLECTION_AVAILABILITY_OPTIONS, KITCHEN_CABINET_SHOPPING_OPTIONS from db.source_imports import normalize_column_key Validator = Callable[[str], Optional[str]] def _allowed_value_validator(values: Iterable[str]) -> Validator: allowed = {str(value) for value in values if str(value).strip()} def _validate(value: str) -> Optional[str]: return None if value in allowed else f"unexpected value '{value}'" return _validate def _numeric_string_validator(value: str) -> Optional[str]: return None if re.fullmatch(r"\d+(?:\.\d+)?", str(value or "").strip()) else f"expected numeric label, got '{value}'" LEFT_PANE_FILTERS: List[Dict[str, Any]] = [ { "code": "item_availability", "label": "Item Availability", "required": True, "validator": _allowed_value_validator(row["label"] for row in COLLECTION_AVAILABILITY_OPTIONS), }, {"code": "manufacturer", "label": "Manufacturer", "required": True}, {"code": "family", "label": "Family", "required": True}, {"code": "collection_style", "label": "Collection Style", "required": True}, {"code": "color_finish", "label": "Color Finish", "required": True}, { "code": "door_style", "label": "Door Style", "required": True, "validator": _allowed_value_validator( ["Shaker", "Slim Shaker", "Recessed Panel", "Wide Rail", "Raised Panel", "Flat Panel", "Slab"] ), }, { "code": "overlay", "label": "Overlay", "required": True, "validator": _allowed_value_validator(["Full Overlay - Framed", "Partial Overlay - Framed", "Inset"]), }, { "code": "drawer_slides", "label": "Drawer Slides", "validator": _allowed_value_validator( ["Blum Soft-Close Slides", "Ball-Bearing Slides", "Standard Side-Mount", "Undermount Slides"] ), }, { "code": "hinges", "label": "Hinges", "validator": _allowed_value_validator( ["Full Overlay Hinges", "Concealed Soft-Close Hinges", "Standard Hinges"] ), }, {"code": "drawer_box", "label": "Drawer Box"}, { "code": "box_construction", "label": "Box Construction", "validator": _allowed_value_validator(["Framed", "Frameless", "Inset Cabinets"]), }, { "code": "shelf_clips", "label": "Shelf Clips", "validator": _allowed_value_validator(["Stainless Steel (locking)", "Plastic", "Metal"]), }, {"code": "materials_face_frame", "label": "Materials Face Frame"}, {"code": "materials_box", "label": "Materials Box"}, {"code": "width", "label": "Width (in)", "required": True, "validator": _numeric_string_validator}, {"code": "height", "label": "Height (in)", "required": True, "validator": _numeric_string_validator}, {"code": "angle", "label": "Angle (direction)"}, {"code": "depth", "label": "Depth (in)", "required": True, "validator": _numeric_string_validator}, { "code": "soft_close", "label": "Soft Close", "required": True, "validator": _allowed_value_validator(["Yes", "No"]), }, { "code": "shelves_included", "label": "Shelves Included", "validator": _allowed_value_validator(["Yes", "No"]), }, { "code": "cabinet_type", "label": "Start Shopping Group", "required": True, "validator": _allowed_value_validator(item["magento_label"] for item in KITCHEN_CABINET_SHOPPING_OPTIONS), }, { "code": "base_cabinets", "label": "Base Cabinets Detail", "group": "base cabinets", "validator": _allowed_value_validator(["basic base", "drawer", "sink", "corner", "special", "tall"]), }, { "code": "wall_cabinets", "label": "Wall Cabinets Detail", "group": "wall cabinets", "validator": _allowed_value_validator(["basic wall", "stacked wall", "bridge wall", "corner wall", "special wall"]), }, { "code": "panels_and_fillers", "label": "Panels and Fillers Detail", "group": "panels and fillers", "validator": _allowed_value_validator(["panels", "fillers"]), }, {"code": "mouldings", "label": "Mouldings Detail", "group": "mouldings"}, { "code": "accessories", "label": "Accessories Detail", "group": "accessories", "validator": _allowed_value_validator( ["floating shelf", "rollout tray", "glass door", "touch up", "valance arch", "corbel", "knee drawer", "post"] ), }, ] def audit_storefront_filter_readiness( session: Session, *, attributes: Optional[Sequence[str]] = None, limit: Optional[int] = None, sku_sample_limit: int = 25, value_sample_limit: int = 20, exclude_parents: bool = True, ) -> Dict[str, Any]: products = _active_products(session, limit=limit, exclude_parents=exclude_parents) attrs_by_sku = _attrs_by_sku(session, [product.id for product in products]) availability_by_product = _location_availability(session, [product.id for product in products]) selected_filters = _selected_filters(attributes) filter_reports: List[Dict[str, Any]] = [] products_missing_required: List[Dict[str, Any]] = [] for definition in selected_filters: report = _audit_single_filter( products, attrs_by_sku, availability_by_product, definition, sku_sample_limit=sku_sample_limit, value_sample_limit=value_sample_limit, ) filter_reports.append(report) required_codes = {row["code"] for row in selected_filters if row.get("required")} if required_codes: for product in products: fields = _normalized_product_fields( product, attrs_by_sku.get(product.sku, {}), availability_by_product.get(product.id) or {}, ) missing = [] for code in sorted(required_codes): definition = next(item for item in selected_filters if item["code"] == code) if not _is_applicable(definition, fields): continue if not _derive_filter_value(code, fields): missing.append(code) if missing: products_missing_required.append({"sku": product.sku, "missing": missing}) return { "status": "ok", "dry_run": True, "active_sellable_product_count": len(products), "limited": limit is not None, "exclude_parents": exclude_parents, "filters_checked": [row["code"] for row in selected_filters], "required_filter_codes": sorted(required_codes), "products_missing_required_count": len(products_missing_required), "sample_products_missing_required": products_missing_required[:sku_sample_limit], "filters": filter_reports, "ready": not any( report["required"] and (report["missing_count"] > 0 or report["invalid_value_count"] > 0) for report in filter_reports ), "notes": [ "Values are audited after canonical storefront normalization, not raw master CSV values.", "Parent/shell products are excluded by default because left-pane filters should reflect sellable rows.", "Group detail filters (base_cabinets, wall_cabinets, panels_and_fillers, mouldings, accessories) are only audited when cabinet_type matches that group.", ], } def _selected_filters(attributes: Optional[Sequence[str]]) -> List[Dict[str, Any]]: if not attributes: return list(LEFT_PANE_FILTERS) wanted = {normalize_column_key(value) for value in attributes if str(value or "").strip()} return [row for row in LEFT_PANE_FILTERS if row["code"] in wanted] def _active_products(session: Session, *, limit: Optional[int], exclude_parents: bool) -> List[MasterProduct]: stmt = select(MasterProduct).where(MasterProduct.is_active.is_(True)).order_by(MasterProduct.sku) rows = list(session.scalars(stmt).all()) if exclude_parents: parent_skus = { str(value) for value in session.scalars(select(MasterProductRelation.parent_sku).distinct()).all() if str(value or "").strip() } rows = [ row for row in rows if row.sku not in parent_skus and "-PARENT" not in str(row.sku or "").upper() ] if limit: rows = rows[: int(limit)] return rows def _normalized_product_fields( product: MasterProduct, attrs: Dict[str, Any], availability: Dict[str, Any], ) -> Dict[str, Any]: fields, _ = compose_canonical_fields(_core_fields(product), attrs, {}) _attach_location_availability_fields(fields, availability) return canonicalize_attribute_fields(fields) def _audit_single_filter( products: Sequence[MasterProduct], attrs_by_sku: Dict[str, Dict[str, str]], availability_by_product: Dict[int, Dict[str, Any]], definition: Dict[str, Any], *, sku_sample_limit: int, value_sample_limit: int, ) -> Dict[str, Any]: counter: Counter[str] = Counter() missing_skus: List[str] = [] invalid_rows: List[Dict[str, Any]] = [] applicable_count = 0 for product in products: fields = _normalized_product_fields( product, attrs_by_sku.get(product.sku, {}), availability_by_product.get(product.id) or {}, ) if not _is_applicable(definition, fields): continue applicable_count += 1 value = _derive_filter_value(definition["code"], fields) if not value: missing_skus.append(product.sku) continue counter[value] += 1 validator = definition.get("validator") if validator: issue = validator(value) if issue: invalid_rows.append({"sku": product.sku, "value": value, "issue": issue}) return { "code": definition["code"], "label": definition["label"], "required": bool(definition.get("required")), "applicable_count": applicable_count, "populated_count": sum(counter.values()), "missing_count": len(missing_skus), "invalid_value_count": len(invalid_rows), "distinct_value_count": len(counter), "top_values": counter.most_common(value_sample_limit), "sample_missing_skus": missing_skus[:sku_sample_limit], "sample_invalid_values": invalid_rows[:sku_sample_limit], } def _is_applicable(definition: Dict[str, Any], fields: Dict[str, Any]) -> bool: group = str(definition.get("group") or "").strip().lower() if not group: return True return str(fields.get("cabinet_type") or "").strip().lower() == group def _derive_filter_value(code: str, fields: Dict[str, Any]) -> Optional[str]: direct = { "item_availability": fields.get("item_availability"), "manufacturer": fields.get("manufacturer") or fields.get("brand"), "family": fields.get("family") or fields.get("product_family"), "collection_style": derive_composite_attribute_value("collection_style", fields) or fields.get("collection"), "color_finish": derive_composite_attribute_value("color_finish", fields) or fields.get("color_finish"), "door_style": fields.get("door_style") or fields.get("cabinet_door_style"), "overlay": derive_composite_attribute_value("overlay", fields) or fields.get("overlay"), "drawer_slides": derive_composite_attribute_value("drawer_slides", fields), "hinges": derive_composite_attribute_value("hinges", fields), "drawer_box": derive_composite_attribute_value("drawer_box", fields), "box_construction": derive_composite_attribute_value("box_construction", fields), "shelf_clips": derive_composite_attribute_value("shelf_clips", fields), "materials_face_frame": derive_composite_attribute_value("materials_face_frame", fields), "materials_box": derive_composite_attribute_value("materials_box", fields), "width": _dimension_value(fields, "width"), "height": _dimension_value(fields, "height"), "angle": derive_composite_attribute_value("angle", fields), "depth": fields.get("depth"), "soft_close": derive_composite_attribute_value("soft_close", fields), "shelves_included": derive_composite_attribute_value("shelves_included", fields), "cabinet_type": fields.get("cabinet_type"), "base_cabinets": derive_composite_attribute_value("base_cabinets", fields), "wall_cabinets": derive_composite_attribute_value("wall_cabinets", fields), "panels_and_fillers": derive_composite_attribute_value("panels_and_fillers", fields), "mouldings": derive_composite_attribute_value("mouldings", fields), "accessories": derive_composite_attribute_value("accessories", fields), } value = direct.get(code) text = str(value or "").strip() return text or None def _dimension_value(fields: Dict[str, Any], axis: str) -> Optional[str]: assembly = str(fields.get("assembly_type") or "").strip().lower() if axis == "width": codes = ["variation_width_in", "width_in", "width_inches", "variation_width_inches"] if assembly == "assembled": codes = ["assembled_width", "assembled_width_in", *codes] elif axis == "height": codes = ["variation_height_in", "height_in", "height_inches", "variation_height_inches"] if assembly == "assembled": codes = ["assembled_height", "assembled_height_in", *codes] else: codes = ["variation_depth_in", "depth_in", "depth_inches", "variation_depth_inches"] if assembly == "assembled": codes = ["assembled_depth", "assembled_depth_in", *codes] for code in codes: value = _normalize_measurement_value(fields.get(code)) if value: return value return None def main(argv: Optional[Sequence[str]] = None) -> int: parser = argparse.ArgumentParser(description="Audit storefront left-pane filter readiness") parser.add_argument("--attribute", action="append", default=None, help="Canonical left-pane filter code to audit") parser.add_argument("--limit", type=int, default=None, help="Limit products for a quick sample") parser.add_argument("--sku-sample-limit", type=int, default=25, help="Max sample SKUs per issue bucket") parser.add_argument("--value-sample-limit", type=int, default=20, help="Max top values shown per filter") parser.add_argument("--include-parents", action="store_true", help="Include variation/shell parent SKUs in the audit") args = parser.parse_args(list(argv) if argv is not None else None) with get_session() as session: result = audit_storefront_filter_readiness( session, attributes=args.attribute, limit=args.limit, sku_sample_limit=args.sku_sample_limit, value_sample_limit=args.value_sample_limit, exclude_parents=not args.include_parents, ) print(json.dumps(result, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())