"""Audit SKU taxonomy, filter attributes, and configurable parent readiness. Examples: python -m app.jobs.audit_product_taxonomy_readiness python -m app.jobs.audit_product_taxonomy_readiness --limit 200 python -m app.jobs.audit_product_taxonomy_readiness --attribute cabinet_type --attribute color """ from __future__ import annotations import argparse import json from collections import Counter, defaultdict from typing import Any, Dict, Iterable, List, Optional, Sequence from sqlalchemy import func, select from sqlalchemy.orm import Session from channel.url_canonical import normalize_plp_path from db.channel_listing_path import resolve_listing_paths_for_sku from db.master_taxonomy_hierarchy import _attrs_by_sku from db.master_taxonomy_product_paths import canonical_master_taxonomy_path_slugs from db.models import MasterProduct, MasterProductAttributeValue, MasterProductRelation from db.session import get_session from magento.payload_mapper import get_axis_attribute_codes DEFAULT_FILTER_ATTRIBUTES = [ "brand", "collection", "assembly_type", "stocked", "cabinet_type", "cabinet_type_sub_category_1", "cabinet_type_sub_category_2", "door_style", "color", "finish", "cabinet_construction", "filter_type", "item_size", "variation_width_in", "variation_height_in", "variation_depth_in", ] def audit_product_taxonomy_readiness( session: Session, *, attributes: Optional[Sequence[str]] = None, limit: Optional[int] = None, sku_sample_limit: int = 25, ) -> Dict[str, Any]: products = _active_products(session, limit=limit) attrs_by_sku = _attrs_by_sku(session, [p.id for p in products]) product_by_sku = {p.sku: p for p in products} wanted_attributes = _normalize_attributes(attributes or DEFAULT_FILTER_ATTRIBUTES) taxonomy = _audit_taxonomy_assignments( session, products, attrs_by_sku, sku_sample_limit=sku_sample_limit, ) filters = _audit_filter_attributes( products, attrs_by_sku, wanted_attributes, sku_sample_limit=sku_sample_limit, ) configurables = _audit_configurable_parents( session, product_by_sku, attrs_by_sku, sku_sample_limit=sku_sample_limit, ) return { "status": "ok", "dry_run": True, "active_product_count": len(products), "limited": limit is not None, "taxonomy": taxonomy, "filter_attributes": filters, "configurable_parents": configurables, "ready": ( taxonomy["missing_expected_count"] == 0 and taxonomy["stale_assigned_count"] == 0 and filters["products_missing_any_count"] == 0 and configurables["parent_without_children_count"] == 0 and configurables["duplicate_axis_signature_parent_count"] == 0 and configurables["missing_option_mapping_parent_count"] == 0 ), } def _active_products(session: Session, *, limit: Optional[int]) -> List[MasterProduct]: stmt = select(MasterProduct).where(MasterProduct.is_active.is_(True)).order_by(MasterProduct.sku) if limit: stmt = stmt.limit(int(limit)) return list(session.scalars(stmt).all()) def _normalize_attributes(attributes: Iterable[str]) -> List[str]: out: List[str] = [] seen: set[str] = set() for attr in attributes: code = str(attr or "").strip().lower() if not code or code in seen: continue seen.add(code) out.append(code) return out def _product_value(product: MasterProduct, attrs: Dict[str, str], code: str) -> str: if code == "stocked": if product.stocked is None: return "" return "true" if product.stocked else "false" value = getattr(product, code, None) if hasattr(product, code) else None if value is None or str(value).strip() == "": value = attrs.get(code) return str(value or "").strip() def _audit_filter_attributes( products: Sequence[MasterProduct], attrs_by_sku: Dict[str, Dict[str, str]], attributes: Sequence[str], *, sku_sample_limit: int, ) -> Dict[str, Any]: missing_by_attr: Dict[str, List[str]] = {attr: [] for attr in attributes} products_missing_any: List[Dict[str, Any]] = [] value_counts: Dict[str, Counter[str]] = {attr: Counter() for attr in attributes} for product in products: attrs = attrs_by_sku.get(product.sku, {}) missing: List[str] = [] for attr in attributes: value = _product_value(product, attrs, attr) if value: value_counts[attr][value] += 1 else: missing_by_attr[attr].append(product.sku) missing.append(attr) if missing: products_missing_any.append({"sku": product.sku, "missing": missing}) return { "attributes_checked": list(attributes), "products_missing_any_count": len(products_missing_any), "missing_by_attribute": { attr: {"count": len(skus), "sample_skus": skus[:sku_sample_limit]} for attr, skus in missing_by_attr.items() if skus }, "top_values": { attr: value_counts[attr].most_common(15) for attr in attributes if value_counts[attr] }, "sample_products_missing_any": products_missing_any[:sku_sample_limit], } def _audit_taxonomy_assignments( session: Session, products: Sequence[MasterProduct], attrs_by_sku: Dict[str, Dict[str, str]], *, sku_sample_limit: int, ) -> Dict[str, Any]: missing_expected: List[Dict[str, Any]] = [] stale_assigned: List[Dict[str, Any]] = [] no_expected: List[str] = [] for product in products: expected = set( canonical_master_taxonomy_path_slugs( session, product, attrs_by_sku.get(product.sku, {}), ) ) assigned = { normalize_plp_path(path.get("path_slug") or "") for path in resolve_listing_paths_for_sku(session, product.sku) } assigned.discard("") if not expected: no_expected.append(product.sku) missing = sorted(expected - assigned) stale = sorted(assigned - expected) if missing: missing_expected.append({"sku": product.sku, "missing_paths": missing, "assigned_paths": sorted(assigned)}) if stale: stale_assigned.append({"sku": product.sku, "stale_paths": stale, "expected_paths": sorted(expected)}) return { "missing_expected_count": len(missing_expected), "stale_assigned_count": len(stale_assigned), "no_expected_path_count": len(no_expected), "sample_missing_expected": missing_expected[:sku_sample_limit], "sample_stale_assigned": stale_assigned[:sku_sample_limit], "sample_no_expected_path_skus": no_expected[:sku_sample_limit], } def _audit_configurable_parents( session: Session, product_by_sku: Dict[str, MasterProduct], attrs_by_sku: Dict[str, Dict[str, str]], *, sku_sample_limit: int, ) -> Dict[str, Any]: relation_rows = list( session.scalars(select(MasterProductRelation).order_by(MasterProductRelation.parent_sku, MasterProductRelation.sort_order)).all() ) by_parent: Dict[str, List[MasterProductRelation]] = defaultdict(list) child_skus: set[str] = set() for relation in relation_rows: by_parent[relation.parent_sku].append(relation) child_skus.add(relation.child_sku) parent_skus = set(by_parent) missing_parent_rows = sorted(sku for sku in parent_skus if sku not in product_by_sku) missing_child_rows = sorted(sku for sku in child_skus if sku not in product_by_sku) parent_without_children: List[str] = [] duplicate_axis_parents: List[Dict[str, Any]] = [] missing_mapping_parents: List[Dict[str, Any]] = [] axis_counts: Counter[str] = Counter() for parent_sku, relations in by_parent.items(): if len(relations) < 1: parent_without_children.append(parent_sku) continue parent = product_by_sku.get(parent_sku) row = _parent_relation_row(parent_sku, parent, relations) axes = get_axis_attribute_codes(row) axis_counts[",".join(axes) or "(none)"] += 1 signatures: Dict[tuple[str, ...], List[str]] = defaultdict(list) missing_mappings: List[str] = [] for relation in relations: mapping = relation.option_mapping if isinstance(relation.option_mapping, dict) else {} if not mapping: missing_mappings.append(relation.child_sku) continue signature = tuple(str(mapping.get(axis) or "").strip().lower() for axis in axes) if not axes or any(not value for value in signature): missing_mappings.append(relation.child_sku) continue signatures[signature].append(relation.child_sku) duplicates = [ {"axis_values": dict(zip(axes, signature)), "children": skus} for signature, skus in signatures.items() if len(skus) > 1 ] if duplicates: duplicate_axis_parents.append({"parent_sku": parent_sku, "duplicates": duplicates}) if missing_mappings: missing_mapping_parents.append({"parent_sku": parent_sku, "children": missing_mappings[:sku_sample_limit], "axes": axes}) duplicate_title_rows = _duplicate_parent_titles(session, parent_skus, sku_sample_limit=sku_sample_limit) return { "parent_count": len(parent_skus), "relation_count": len(relation_rows), "parent_without_children_count": len(parent_without_children), "missing_parent_product_count": len(missing_parent_rows), "missing_child_product_count": len(missing_child_rows), "duplicate_parent_title_group_count": len(duplicate_title_rows), "duplicate_axis_signature_parent_count": len(duplicate_axis_parents), "missing_option_mapping_parent_count": len(missing_mapping_parents), "axis_sets": axis_counts.most_common(20), "sample_parent_without_children": parent_without_children[:sku_sample_limit], "sample_missing_parent_products": missing_parent_rows[:sku_sample_limit], "sample_missing_child_products": missing_child_rows[:sku_sample_limit], "sample_duplicate_parent_titles": duplicate_title_rows[:sku_sample_limit], "sample_duplicate_axis_signatures": duplicate_axis_parents[:sku_sample_limit], "sample_missing_option_mappings": missing_mapping_parents[:sku_sample_limit], } def _parent_relation_row( parent_sku: str, parent: Optional[MasterProduct], relations: Sequence[MasterProductRelation], ) -> Dict[str, Any]: variations = [] for relation in relations: mapping = relation.option_mapping if isinstance(relation.option_mapping, dict) else {} variations.append({"sku": relation.child_sku, **mapping}) axes = sorted({code for item in variations for code in item if code != "sku"}) return { "sku": parent_sku, "name": (parent.name if parent else None) or parent_sku, "type": "configurable", "product_type": "configurable", "variant_list": ",".join(relation.child_sku for relation in relations), "configurable_attributes": ",".join(axes), "configurable_variations": json.dumps(variations), } def _duplicate_parent_titles( session: Session, parent_skus: set[str], *, sku_sample_limit: int, ) -> List[Dict[str, Any]]: if not parent_skus: return [] rows = session.execute( select(func.lower(func.trim(MasterProduct.name)), func.count(MasterProduct.sku)) .where(MasterProduct.sku.in_(parent_skus)) .group_by(func.lower(func.trim(MasterProduct.name))) .having(func.count(MasterProduct.sku) > 1) ).all() out: List[Dict[str, Any]] = [] for title, count in rows: skus = [ sku for sku in session.scalars( select(MasterProduct.sku) .where(MasterProduct.sku.in_(parent_skus)) .where(func.lower(func.trim(MasterProduct.name)) == title) .order_by(MasterProduct.sku) .limit(sku_sample_limit) ).all() ] out.append({"title": title, "count": int(count), "sample_skus": skus}) return sorted(out, key=lambda row: (-int(row["count"]), str(row["title"]))) def main() -> int: parser = argparse.ArgumentParser(description="Audit product taxonomy/channel readiness before Magento/Shopify push") parser.add_argument("--attribute", action="append", dest="attributes", help="Filter attribute code to check; repeatable") parser.add_argument("--limit", type=int, default=None, help="Limit active products for a quick sample") parser.add_argument("--sample-limit", type=int, default=25) args = parser.parse_args() with get_session() as session: result = audit_product_taxonomy_readiness( session, attributes=args.attributes, limit=args.limit, sku_sample_limit=args.sample_limit, ) print(json.dumps(result, indent=2, default=str)) return 0 if result.get("status") == "ok" else 1 if __name__ == "__main__": raise SystemExit(main())