"""Audit Magento PLP filter readiness: aliases, registry attrs, and Digisilk store codes. Compares PlytixMage canonical filter attributes against: 1) channel_attribute_alias / static field mappings 2) magento_attribute_registry (what exists on each connection) 3) Digisilk storefront filterable attribute codes (HSMageDSSGit setup patches) Examples: python -m app.jobs.audit_magento_filter_readiness python -m app.jobs.audit_magento_filter_readiness --connection-id 1 python -m app.jobs.audit_magento_filter_readiness --connection-id 2 """ from __future__ import annotations import argparse import json import sys from collections import defaultdict from typing import Any, Dict, List, Optional, Sequence, Set from sqlalchemy import func, select from sqlalchemy.orm import Session from app.jobs.audit_product_taxonomy_readiness import DEFAULT_FILTER_ATTRIBUTES from db.channel_aliases import export_channel_attribute_aliases from db.channel_exports import _build_alias_map, compose_canonical_fields from db.models import ( MagentoAttributeOptionRegistry, MagentoAttributeRegistry, MagentoConnection, MasterProduct, MasterProductAttributeValue, MasterStaticFieldMapping, ) from db.session import get_session # Digisilk Magento filterable selects from CreateCustomAttributesDropdownV2 + cabinet_type patches. DIGISILK_STORE_FILTERABLE: Set[str] = { "brand", "family", "collection_style", "color_finish", "door_style", "overlay", "finish_type", "drawer_slides", "hinges", "drawer_box", "box_construction", "shelf_clips", "materials_face_frame", "materials_box", "cabinet_type", "base_cabinets", "wall_cabinets", "panels_and_fillers", "mouldings", "accessories", "width", } # Canonical master codes that drive storefront filters → preferred Digisilk Magento codes. # When using owned/custom attributes instead, preferred may equal canonical. FILTER_ATTR_EXPECTATIONS: List[Dict[str, Any]] = [ { "canonical": "cabinet_type", "preferred_magento": "cabinet_type", "role": "Start Shopping L1 + horizontal nav", "required": True, }, { "canonical": "door_style", "preferred_magento": "door_style", "role": "Sidebar layered nav", "required": True, }, { "canonical": "color", "preferred_magento": "color_finish", "also_ok": ["color"], "role": "Sidebar Color/Finish filter", "required": True, }, { "canonical": "cabinet_construction", "preferred_magento": "box_construction", "also_ok": ["cabinet_construction"], "role": "Framed/Frameless filter (theme JS also looks for cabinet_construction)", "required": True, }, { "canonical": "finish", "preferred_magento": "finish_type", "also_ok": ["finish"], "role": "Sidebar finish filter", "required": False, }, { "canonical": "collection", "preferred_magento": "collection_style", "also_ok": ["family", "collection"], "role": "Collection/style filter", "required": False, }, { "canonical": "brand", "preferred_magento": "brand", "also_ok": ["family"], "role": "Brand filter (default static alias maps brand→family)", "required": False, }, { "canonical": "variation_width_in", "preferred_magento": "width", "also_ok": ["width_in", "variation_width_in"], "role": "Width layered nav / configurable axis", "required": False, }, { "canonical": "assembly_type", "preferred_magento": "assembly_type", "also_ok": ["assembly"], "role": "Assembled vs RTA filter", "required": False, }, ] def audit_magento_filter_readiness( session: Session, *, connection_id: Optional[int] = None, sku_sample_limit: int = 10, ) -> Dict[str, Any]: aliases = [ row for row in export_channel_attribute_aliases(session) if row["channel_code"] == "magento" and row.get("is_active", True) ] alias_by_canonical: Dict[str, List[str]] = defaultdict(list) for row in aliases: alias_by_canonical[row["canonical_code"]].append(row["channel_attribute_code"]) runtime_alias_map = _build_alias_map(session, "magento") static_rows = list(session.scalars(select(MasterStaticFieldMapping).where(MasterStaticFieldMapping.is_active.is_(True))).all()) static_by_master = { row.master_attribute_code: row.magento_attribute_code for row in static_rows if row.magento_attribute_code } connections = _connections(session, connection_id=connection_id) connection_reports = [ _audit_connection(session, conn, alias_by_canonical, runtime_alias_map) for conn in connections ] attr_checks = [ _check_filter_attr( expectation, alias_by_canonical=alias_by_canonical, runtime_alias_map=runtime_alias_map, static_by_master=static_by_master, connection_reports=connection_reports, ) for expectation in FILTER_ATTR_EXPECTATIONS ] master_coverage = _master_filter_coverage(session, sku_sample_limit=sku_sample_limit) not_ready = [item for item in attr_checks if item["status"] == "NOT_READY"] review = [item for item in attr_checks if item["status"] == "REVIEW"] ready = [item for item in attr_checks if item["status"] == "READY"] return { "status": "ok", "dry_run": True, "summary": { "ready_count": len(ready), "review_count": len(review), "not_ready_count": len(not_ready), "required_not_ready": [ item["canonical"] for item in not_ready if item.get("required") ], "filter_push_blocked": any(item.get("required") for item in not_ready), }, "magento_alias_count": len(aliases), "magento_aliases": aliases, "runtime_alias_map": {k: v for k, v in sorted(runtime_alias_map.items()) if k in _watched_canonicals()}, "static_magento_mappings": { k: v for k, v in sorted(static_by_master.items()) if k in _watched_canonicals() }, "filter_attributes": attr_checks, "connections": connection_reports, "master_coverage": master_coverage, "digisilk_store_filterable": sorted(DIGISILK_STORE_FILTERABLE), "notes": [ "Digisilk HSMageDSSGit uses Mageplaza layered nav on native EAV attrs (not Amasty Shopby).", "If you provisioned owned/custom Magento attrs with canonical codes, also_ok codes are acceptable.", "Theme JS still references amshopby[cabinet_construction]; prefer cabinet_construction on owned stores.", "cabinet_type option labels must match shopping facet catalog (e.g. 'base cabinets').", ], } def _watched_canonicals() -> Set[str]: watched = {item["canonical"] for item in FILTER_ATTR_EXPECTATIONS} watched.update(DEFAULT_FILTER_ATTRIBUTES) watched.update({"color_finish", "box_construction", "collection_style", "finish_type", "family", "width"}) return watched def _connections(session: Session, *, connection_id: Optional[int]) -> List[MagentoConnection]: stmt = select(MagentoConnection).order_by(MagentoConnection.id) if connection_id is not None: stmt = stmt.where(MagentoConnection.id == int(connection_id)) return list(session.scalars(stmt).all()) def _audit_connection( session: Session, conn: MagentoConnection, alias_by_canonical: Dict[str, List[str]], runtime_alias_map: Dict[str, List[str]], ) -> Dict[str, Any]: regs = list( session.scalars( select(MagentoAttributeRegistry) .where(MagentoAttributeRegistry.connection_id == conn.id) .order_by(MagentoAttributeRegistry.attribute_code) ).all() ) present = {row.attribute_code for row in regs} interesting = _watched_canonicals() | DIGISILK_STORE_FILTERABLE attributes: List[Dict[str, Any]] = [] for row in regs: if row.attribute_code not in interesting: continue opt_count = session.scalar( select(func.count()) .select_from(MagentoAttributeOptionRegistry) .where(MagentoAttributeOptionRegistry.connection_id == conn.id) .where(MagentoAttributeOptionRegistry.attribute_code == row.attribute_code) ) sample = list( session.scalars( select(MagentoAttributeOptionRegistry.option_label) .where(MagentoAttributeOptionRegistry.connection_id == conn.id) .where(MagentoAttributeOptionRegistry.attribute_code == row.attribute_code) .order_by(MagentoAttributeOptionRegistry.option_label) .limit(8) ).all() ) attributes.append( { "attribute_code": row.attribute_code, "frontend_input": row.frontend_input, "frontend_label": row.frontend_label, "option_count": int(opt_count or 0), "sample_options": sample, "in_digisilk_store_set": row.attribute_code in DIGISILK_STORE_FILTERABLE, } ) return { "connection_id": conn.id, "store_base_url": conn.store_base_url, "environment": conn.environment, "status": conn.status, "registry_attribute_count": len(regs), "store_filterable_present": sorted(DIGISILK_STORE_FILTERABLE & present), "store_filterable_missing": sorted(DIGISILK_STORE_FILTERABLE - present), "canonical_codes_present": sorted(_watched_canonicals() & present), "attributes": attributes, "resolved_targets_present": { canonical: [ code for code in runtime_alias_map.get(canonical, alias_by_canonical.get(canonical, [canonical])) if code in present ] for canonical in (item["canonical"] for item in FILTER_ATTR_EXPECTATIONS) }, } def _check_filter_attr( expectation: Dict[str, Any], *, alias_by_canonical: Dict[str, List[str]], runtime_alias_map: Dict[str, List[str]], static_by_master: Dict[str, Optional[str]], connection_reports: Sequence[Dict[str, Any]], ) -> Dict[str, Any]: canonical = expectation["canonical"] preferred = expectation["preferred_magento"] also_ok = set(expectation.get("also_ok") or []) acceptable = {preferred, *also_ok, canonical} mapped = list(runtime_alias_map.get(canonical) or alias_by_canonical.get(canonical) or []) static_target = static_by_master.get(canonical) effective_targets = mapped or ([static_target] if static_target else [canonical]) effective_targets = [str(code).strip() for code in effective_targets if str(code or "").strip()] targets_ok = any(code in acceptable for code in effective_targets) preferred_hit = preferred in effective_targets custom_hit = any(code in also_ok for code in effective_targets) present_on: Dict[str, List[str]] = {} missing_on: List[int] = [] for conn in connection_reports: present_targets = [ code for code in effective_targets if code in {a["attribute_code"] for a in conn["attributes"]} or code in set(conn.get("canonical_codes_present") or []) or code in set(conn.get("store_filterable_present") or []) ] # broader presence check against registry lists registry_codes = {a["attribute_code"] for a in conn["attributes"]} registry_codes.update(conn.get("canonical_codes_present") or []) registry_codes.update(conn.get("store_filterable_present") or []) present_targets = [code for code in effective_targets if code in registry_codes] present_on[str(conn["connection_id"])] = present_targets if not present_targets: missing_on.append(int(conn["connection_id"])) if not connection_reports: status = "REVIEW" reason = "No Magento connections found in DB" elif not targets_ok: status = "NOT_READY" reason = f"Maps to {effective_targets}; expected one of {sorted(acceptable)}" elif missing_on and expectation.get("required"): status = "NOT_READY" reason = f"Target attribute missing from Magento registry on connection(s) {missing_on}" elif missing_on: status = "REVIEW" reason = f"Target attribute missing from Magento registry on connection(s) {missing_on}" elif preferred_hit: status = "READY" reason = f"Aliased/pass-through to Digisilk-preferred code '{preferred}'" elif custom_hit: status = "READY" reason = f"Using owned/custom Magento code(s) {effective_targets} (acceptable)" else: status = "REVIEW" reason = f"Targets {effective_targets} are acceptable but verify is_filterable + options on Magento" return { "canonical": canonical, "preferred_magento": preferred, "also_ok": sorted(also_ok), "role": expectation["role"], "required": bool(expectation.get("required")), "effective_targets": effective_targets, "static_mapping": static_target, "alias_rows": alias_by_canonical.get(canonical, []), "present_on_connections": present_on, "status": status, "reason": reason, } def _master_filter_coverage(session: Session, *, sku_sample_limit: int) -> Dict[str, Any]: products = list(session.scalars(select(MasterProduct).where(MasterProduct.is_active.is_(True))).all()) active_count = len(products) product_ids = [product.id for product in products] attrs_by_product: Dict[int, Dict[str, Any]] = defaultdict(dict) if product_ids: for row in session.scalars( select(MasterProductAttributeValue).where(MasterProductAttributeValue.product_id.in_(product_ids)) ).all(): if row.value is not None and str(row.value).strip(): attrs_by_product[int(row.product_id)][row.attribute_code] = row.value derived_by_attr: Dict[str, Dict[str, Any]] = { code: {"count": 0, "sample_missing_skus": []} for code in DEFAULT_FILTER_ATTRIBUTES } for product in products: fields, _ = compose_canonical_fields( _core_fields(product), attrs_by_product.get(product.id, {}), {}, ) for code in DEFAULT_FILTER_ATTRIBUTES: if str(fields.get(code) or "").strip(): derived_by_attr[code]["count"] += 1 elif len(derived_by_attr[code]["sample_missing_skus"]) < sku_sample_limit: derived_by_attr[code]["sample_missing_skus"].append(product.sku) by_attr: Dict[str, Any] = {} for code in DEFAULT_FILTER_ATTRIBUTES: filled = int( session.scalar( select(func.count(func.distinct(MasterProductAttributeValue.product_id))) .join(MasterProduct, MasterProduct.id == MasterProductAttributeValue.product_id) .where(MasterProduct.is_active.is_(True)) .where(MasterProductAttributeValue.attribute_code == code) .where(MasterProductAttributeValue.value.isnot(None)) .where(MasterProductAttributeValue.value != "") ) or 0 ) # Also count core columns when present on MasterProduct. # Boolean/int columns must not be compared to '' (Postgres rejects that). core_filled = 0 if hasattr(MasterProduct, code): col = getattr(MasterProduct, code) stmt = ( select(func.count()) .select_from(MasterProduct) .where(MasterProduct.is_active.is_(True)) .where(col.isnot(None)) ) if _is_stringish_column(col): stmt = stmt.where(col != "") core_filled = int(session.scalar(stmt) or 0) effective = max(filled, core_filled) pipeline_filled = int(derived_by_attr.get(code, {}).get("count") or 0) by_attr[code] = { "filled_count": max(effective, pipeline_filled), "attr_table_count": filled, "core_column_count": core_filled, "pipeline_derived_count": pipeline_filled, "coverage_pct": round((100.0 * max(effective, pipeline_filled) / active_count), 1) if active_count else 0.0, "sample_missing_skus": derived_by_attr.get(code, {}).get("sample_missing_skus", []), } return {"active_product_count": active_count, "by_attribute": by_attr, "sample_limit": sku_sample_limit} def _core_fields(product: MasterProduct) -> Dict[str, Any]: return { "title": product.name, "brand": product.brand, "manufacturer": product.manufacturer, "product_family": product.product_family, "category_l1": product.category_l1, "category_l2": product.category_l2, "category_l3": product.category_l3, "collection": product.collection, "assembly_type": product.assembly_type, "item_size": product.item_size, "status": product.status, "stocked": product.stocked, } def _is_stringish_column(col: Any) -> bool: try: python_type = col.type.python_type except (NotImplementedError, AttributeError): return False return python_type is str def main() -> int: parser = argparse.ArgumentParser(description="Audit Magento category/collection filter readiness") parser.add_argument("--connection-id", type=int, default=None) parser.add_argument("--sample-limit", type=int, default=10) args = parser.parse_args() with get_session() as session: result = audit_magento_filter_readiness( session, connection_id=args.connection_id, sku_sample_limit=args.sample_limit, ) print(json.dumps(result, indent=2, default=str)) return 1 if result["summary"]["filter_push_blocked"] else 0 if __name__ == "__main__": sys.exit(main())