"""Audit master SKU data and channel parity before Magento/Shopify pushes. Examples: python -m app.jobs.audit_product_channel_parity python -m app.jobs.audit_product_channel_parity --limit 200 python -m app.jobs.audit_product_channel_parity --sku ABC --sku XYZ """ from __future__ import annotations import argparse import json from collections import defaultdict from typing import Any, Dict, Iterable, List, Optional, Sequence from sqlalchemy import or_, select from sqlalchemy.orm import Session from channel.url_canonical import build_pdp_slug, normalize_plp_path, slugify_segment from db.channel_exports import build_channel_product_payloads from db.channel_listing_path import resolve_listing_paths_for_sku from db.master_relations import children_by_parent, parent_skus_with_children from db.models import ( MagentoAttributeOptionRegistry, MagentoCatalogState, MagentoCategoryRegistry, MagentoProductSourceSnapshot, ChannelListingPath, ChannelListingPathTarget, MasterProduct, MasterProductRelation, MasterTaxonomyNode, ShopifyCollectionRegistry, ShopifyProductSourceSnapshot, ) from db.session import get_session from magento.master_catalog_rows import build_magento_snapshot_rows_from_master from magento.payload_mapper import get_axis_attribute_codes from magento.payload_mapper import snapshot_to_magento_product from magento.configurable_requirements import resolve_magento_axis_code from magento.sync_hashes import _parse_children_from_row from channel.attribute_value_resolver import norm_option_label PRODUCT_DESCRIPTOR_TOKENS = { "accessory", "base", "cabinet", "drawer", "door", "filler", "moulding", "molding", "panel", "sample", "sink", "tall", "trim", "vanity", "wall", } def audit_product_channel_parity( session: Session, *, skus: Optional[Sequence[str]] = None, limit: Optional[int] = None, magento_connection_id: Optional[int] = None, shopify_connection_id: Optional[int] = None, sku_sample_limit: int = 25, ) -> Dict[str, Any]: wanted = _wanted_skus(session, skus=skus, limit=limit) master = _master_rows(session, wanted) master_by_sku = {row.sku: row for row in master} listing_paths = _listing_paths_by_sku(session, wanted) parent_children = children_by_parent(session, wanted) parent_skus = parent_skus_with_children(session, wanted) master_issues = _audit_master_products( master, listing_paths, parent_skus=parent_skus, parent_children=parent_children, sku_sample_limit=sku_sample_limit, ) magento = _audit_magento_payloads( session, wanted, master_by_sku, listing_paths, connection_id=magento_connection_id, sku_sample_limit=sku_sample_limit, ) shopify = _audit_shopify_payloads( session, wanted, master_by_sku, listing_paths, connection_id=shopify_connection_id, sku_sample_limit=sku_sample_limit, ) taxonomy_channels = _audit_taxonomy_channel_parity( session, magento_connection_id=magento_connection_id, shopify_connection_id=shopify_connection_id, sku_sample_limit=sku_sample_limit, ) configurable_options = _audit_configurable_option_mapping( session, wanted, magento_connection_id=magento_connection_id, sku_sample_limit=sku_sample_limit, ) issue_count = ( master_issues["issue_count"] + magento["issue_count"] + shopify["issue_count"] + taxonomy_channels["issue_count"] + configurable_options["issue_count"] ) return { "status": "ok", "dry_run": True, "sku_count": len(wanted), "limited": limit is not None, "connections": { "magento_connection_id": magento_connection_id, "shopify_connection_id": shopify_connection_id, }, "master_data": master_issues, "magento": magento, "shopify": shopify, "taxonomy_channels": taxonomy_channels, "configurable_options": configurable_options, "ready": issue_count == 0, } def _wanted_skus(session: Session, *, skus: Optional[Sequence[str]], limit: Optional[int]) -> List[str]: if skus: return sorted({str(sku or "").strip() for sku in skus if str(sku or "").strip()}) stmt = select(MasterProduct.sku).where(MasterProduct.is_active.is_(True)).order_by(MasterProduct.sku) if limit: stmt = stmt.limit(int(limit)) return [str(sku) for sku in session.scalars(stmt).all()] def _master_rows(session: Session, skus: Sequence[str]) -> List[MasterProduct]: if not skus: return [] return list( session.scalars( select(MasterProduct) .where(MasterProduct.sku.in_(list(skus))) .order_by(MasterProduct.sku) ).all() ) def _listing_paths_by_sku(session: Session, skus: Sequence[str]) -> Dict[str, List[Dict[str, Any]]]: return {sku: resolve_listing_paths_for_sku(session, sku) for sku in skus} def _audit_master_products( products: Sequence[MasterProduct], listing_paths: Dict[str, List[Dict[str, Any]]], *, parent_skus: set[str], parent_children: Dict[str, List[Dict[str, Any]]], sku_sample_limit: int, ) -> Dict[str, Any]: missing_title: List[str] = [] missing_url: List[Dict[str, Any]] = [] no_listing_paths: List[str] = [] parent_title_suspects: List[Dict[str, Any]] = [] duplicate_titles: Dict[str, List[str]] = defaultdict(list) for product in products: title = _text(product.name) if not title: missing_title.append(product.sku) else: duplicate_titles[_norm(title)].append(product.sku) raw_slug = _raw_payload_value(product.raw_payload, "product_url_slug", "url_key", "handle") expected_slug = _expected_product_slug(product, explicit_slug=raw_slug) if raw_slug and _norm_slug(raw_slug) != _norm_slug(expected_slug): missing_url.append( { "sku": product.sku, "master_title": title, "stored_slug": raw_slug, "expected_slug": expected_slug, } ) paths = listing_paths.get(product.sku) or [] if not paths: no_listing_paths.append(product.sku) if product.sku in parent_skus and _looks_like_collection_only_title(product, paths): parent_title_suspects.append( { "sku": product.sku, "title": title, "collection": _text(product.collection), "category_l2": _text(product.category_l2), "child_count": len(parent_children.get(product.sku) or []), "paths": [p.get("path_slug") for p in paths[:5]], "reason": "parent_title_matches_collection_or_category_without_product_descriptor", } ) duplicate_title_groups = [ {"title": title, "count": len(skus), "sample_skus": skus[:sku_sample_limit]} for title, skus in sorted(duplicate_titles.items()) if len(skus) > 1 ] issue_count = ( len(missing_title) + len(missing_url) + len(no_listing_paths) + len(parent_title_suspects) + len(duplicate_title_groups) ) return { "issue_count": issue_count, "missing_title_count": len(missing_title), "stored_url_mismatch_count": len(missing_url), "no_listing_path_count": len(no_listing_paths), "parent_collection_title_suspect_count": len(parent_title_suspects), "duplicate_title_group_count": len(duplicate_title_groups), "sample_missing_title_skus": missing_title[:sku_sample_limit], "sample_stored_url_mismatches": missing_url[:sku_sample_limit], "sample_no_listing_path_skus": no_listing_paths[:sku_sample_limit], "sample_parent_collection_title_suspects": parent_title_suspects[:sku_sample_limit], "sample_duplicate_titles": duplicate_title_groups[:sku_sample_limit], } def _audit_magento_payloads( session: Session, skus: Sequence[str], master_by_sku: Dict[str, MasterProduct], listing_paths: Dict[str, List[Dict[str, Any]]], *, connection_id: Optional[int], sku_sample_limit: int, ) -> Dict[str, Any]: rows = build_magento_snapshot_rows_from_master( session, skus=list(skus), only_assigned=False, connection_id=connection_id, expand_relations=False, ) rows_by_master = {str(row.get("master_sku") or row.get("sku")): row for row in rows} rows_by_sku = {str(row.get("sku") or ""): row for row in rows} issues: List[Dict[str, Any]] = [] category_issues: List[Dict[str, Any]] = [] for master_sku, row in rows_by_master.items(): product = master_by_sku.get(master_sku) if product is None: continue payload = snapshot_to_magento_product(row, rows_by_sku=rows_by_sku) expected_title = _expected_title(product) pushed_title = _text(payload.get("name")) if expected_title and pushed_title and _norm(expected_title) != _norm(pushed_title): if "configurable" not in str(row.get("product_type") or "").lower(): issues.append( { "sku": master_sku, "field": "name", "master": expected_title, "would_push": pushed_title, } ) expected_url = _expected_product_slug_from_row(row, product, fallback_sku=master_sku) pushed_url = _custom_attribute_value(payload, "url_key") if pushed_url and _norm_slug(expected_url) != _norm_slug(pushed_url): issues.append( { "sku": master_sku, "field": "url_key", "expected": expected_url, "would_push": pushed_url, } ) expected_paths = { normalize_plp_path(path.get("path_slug") or "") for path in listing_paths.get(master_sku, []) if path.get("path_slug") } pushed_ids = [int(cid) for cid in (row.get("category_ids") or []) if str(cid).isdigit()] registry_by_id = _magento_registry_paths(session, connection_id, pushed_ids) pushed_paths = { _normalize_channel_category_path(path) for path in _split_paths(row.get("categories_raw") or row.get("categories")) } pushed_paths.discard("") # Prefer Magento registry path_names for the IDs that would actually be pushed. registry_paths = { _normalize_channel_category_path(registry_by_id[cid]) for cid in pushed_ids if registry_by_id.get(cid) } registry_paths.discard("") compare_paths = registry_paths or pushed_paths path_mismatch = bool(expected_paths and compare_paths) and not all( _magento_path_covers_expected(expected, compare_paths) for expected in expected_paths ) if path_mismatch: category_issues.append( { "sku": master_sku, "expected_paths": sorted(expected_paths), "would_push_paths": sorted(compare_paths), "category_ids": pushed_ids or (row.get("category_ids") or []), "registry_paths": sorted(registry_paths), } ) category_groups = _group_category_mismatches(category_issues, sample_limit=sku_sample_limit) live = _audit_live_magento_state( session, rows_by_master, connection_id=connection_id, sku_sample_limit=sku_sample_limit, ) issue_count = len(issues) + len(category_issues) + live["issue_count"] return { "issue_count": issue_count, "payload_count": len(rows), "payload_field_mismatch_count": len(issues), "payload_category_mismatch_count": len(category_issues), "payload_category_mismatch_group_count": len(category_groups), "sample_payload_field_mismatches": issues[:sku_sample_limit], "sample_payload_category_mismatches": category_issues[:sku_sample_limit], "payload_category_mismatch_groups": category_groups[:sku_sample_limit], "live_state": live, } def _audit_live_magento_state( session: Session, rows_by_master: Dict[str, Dict[str, Any]], *, connection_id: Optional[int], sku_sample_limit: int, ) -> Dict[str, Any]: if not rows_by_master: return {"issue_count": 0, "checked_count": 0, "sample_mismatches": []} stmt = select(MagentoProductSourceSnapshot).where( MagentoProductSourceSnapshot.valid_to.is_(None), MagentoProductSourceSnapshot.sku.in_([str(row.get("sku") or sku) for sku, row in rows_by_master.items()]), ) if connection_id is not None: stmt = stmt.where(MagentoProductSourceSnapshot.connection_id == connection_id) live_by_sku = {row.sku: row for row in session.scalars(stmt).all()} mismatches: List[Dict[str, Any]] = [] for master_sku, row in rows_by_master.items(): channel_sku = str(row.get("sku") or master_sku) live = live_by_sku.get(channel_sku) if live is None: continue expected_payload = snapshot_to_magento_product(row) live_payload = live.payload if isinstance(live.payload, dict) else {} expected_name = _text(expected_payload.get("name")) live_name = _text(live_payload.get("name") or live_payload.get("title")) if expected_name and live_name and _norm(expected_name) != _norm(live_name): mismatches.append( { "sku": master_sku, "channel_sku": channel_sku, "field": "name", "expected": expected_name, "live": live_name, } ) expected_url = _custom_attribute_value(expected_payload, "url_key") live_url = _custom_attribute_value(live_payload, "url_key") or _text(live_payload.get("url_key")) if expected_url and live_url and _norm_slug(expected_url) != _norm_slug(live_url): mismatches.append( { "sku": master_sku, "channel_sku": channel_sku, "field": "url_key", "expected": expected_url, "live": live_url, } ) return { "issue_count": len(mismatches), "checked_count": len(live_by_sku), "sample_mismatches": mismatches[:sku_sample_limit], } def _audit_shopify_payloads( session: Session, skus: Sequence[str], master_by_sku: Dict[str, MasterProduct], listing_paths: Dict[str, List[Dict[str, Any]]], *, connection_id: Optional[int], sku_sample_limit: int, ) -> Dict[str, Any]: # Shopify is flat: exclude configurable parent shells the same way a real push does. payloads = build_channel_product_payloads( session, "shopify", skus=list(skus), only_assigned=False, connection_id=connection_id, product_structure="flat", ) issues: List[Dict[str, Any]] = [] collection_issues: List[Dict[str, Any]] = [] parent_leaks: List[Dict[str, Any]] = [] for payload in payloads: master_sku = payload["sku"] product = master_by_sku.get(master_sku) if product is None: continue fields = payload.get("fields") or {} expected_title = _expected_title(product) pushed_title = _text(fields.get("title") or fields.get("name")) if expected_title and pushed_title and _norm(expected_title) != _norm(pushed_title): issues.append( { "sku": master_sku, "field": "title", "master": expected_title, "would_push": pushed_title, } ) expected_handle = _expected_product_slug_for_payload(payload, product, fallback_sku=master_sku) pushed_handle = _text(fields.get("handle") or fields.get("product_url_slug")) if pushed_handle and _norm_slug(expected_handle) != _norm_slug(pushed_handle): issues.append( { "sku": master_sku, "field": "handle", "expected": expected_handle, "would_push": pushed_handle, } ) if payload.get("product_role") == "parent": parent_leaks.append( { "sku": master_sku, "title": pushed_title, "reason": "configurable_parent_shell_in_shopify_flat_payload", } ) targets = payload.get("taxonomy_targets") or [] expected_paths = { normalize_plp_path(path.get("path_slug") or "") for path in listing_paths.get(master_sku, []) if path.get("path_slug") } target_paths = { normalize_plp_path(t.get("remote_path") or t.get("path_slug") or "") for t in targets if isinstance(t, dict) } if expected_paths and target_paths and not expected_paths.intersection(target_paths): collection_issues.append( { "sku": master_sku, "expected_paths": sorted(expected_paths), "taxonomy_targets": targets[:10], } ) live = _audit_live_shopify_state( session, payloads, connection_id=connection_id, sku_sample_limit=sku_sample_limit, ) issue_count = len(issues) + len(collection_issues) + len(parent_leaks) + live["issue_count"] return { "issue_count": issue_count, "payload_count": len(payloads), "payload_field_mismatch_count": len(issues), "collection_target_mismatch_count": len(collection_issues), "parent_shell_payload_count": len(parent_leaks), "sample_payload_field_mismatches": issues[:sku_sample_limit], "sample_collection_target_mismatches": collection_issues[:sku_sample_limit], "sample_parent_shell_payloads": parent_leaks[:sku_sample_limit], "live_state": live, } def _audit_live_shopify_state( session: Session, payloads: Sequence[Dict[str, Any]], *, connection_id: Optional[int], sku_sample_limit: int, ) -> Dict[str, Any]: if not payloads: return {"issue_count": 0, "checked_count": 0, "sample_mismatches": []} by_channel_sku = {str(p.get("channel_sku") or p.get("sku")): p for p in payloads} stmt = select(ShopifyProductSourceSnapshot).where( ShopifyProductSourceSnapshot.valid_to.is_(None), ShopifyProductSourceSnapshot.sku.in_(list(by_channel_sku)), ) if connection_id is not None: stmt = stmt.where(ShopifyProductSourceSnapshot.connection_id == connection_id) live_by_sku = {row.sku: row for row in session.scalars(stmt).all()} mismatches: List[Dict[str, Any]] = [] for channel_sku, payload in by_channel_sku.items(): live = live_by_sku.get(channel_sku) if live is None: continue fields = payload.get("fields") or {} live_payload = live.payload if isinstance(live.payload, dict) else {} expected_title = _text(fields.get("title") or fields.get("name")) live_title = _text(live_payload.get("title")) if expected_title and live_title and _norm(expected_title) != _norm(live_title): mismatches.append( { "sku": payload.get("sku"), "channel_sku": channel_sku, "field": "title", "expected": expected_title, "live": live_title, } ) expected_handle = _text(fields.get("handle") or fields.get("product_url_slug")) live_handle = _text(live_payload.get("handle")) if expected_handle and live_handle and _norm_slug(expected_handle) != _norm_slug(live_handle): mismatches.append( { "sku": payload.get("sku"), "channel_sku": channel_sku, "field": "handle", "expected": expected_handle, "live": live_handle, } ) return { "issue_count": len(mismatches), "checked_count": len(live_by_sku), "sample_mismatches": mismatches[:sku_sample_limit], } def _audit_taxonomy_channel_parity( session: Session, *, magento_connection_id: Optional[int], shopify_connection_id: Optional[int], sku_sample_limit: int, ) -> Dict[str, Any]: paths = _active_listing_paths_by_slug(session) nodes = _active_taxonomy_nodes_by_slug(session) target_rows = list( session.scalars( select(ChannelListingPathTarget) .where(ChannelListingPathTarget.is_active.is_(True)) .order_by(ChannelListingPathTarget.channel_code, ChannelListingPathTarget.remote_id) ).all() ) magento_registry = _magento_registry_by_id(session, magento_connection_id) shopify_registry = _shopify_registry_by_id(session, shopify_connection_id) missing_master_path: List[Dict[str, Any]] = [] title_mismatches: List[Dict[str, Any]] = [] url_mismatches: List[Dict[str, Any]] = [] missing_registry: List[Dict[str, Any]] = [] for target in target_rows: if str(target.taxonomy_kind or "") in {"category_filter", "collection_filter", "intersection"}: continue if target.channel_code == "magento" and magento_connection_id is not None and target.connection_id not in {None, magento_connection_id}: continue if target.channel_code == "shopify" and shopify_connection_id is not None and target.connection_id not in {None, shopify_connection_id}: continue path = paths.get(int(target.listing_path_id)) if path is None: missing_master_path.append( { "target_id": target.id, "channel": target.channel_code, "remote_id": target.remote_id, "reason": "target_without_active_listing_path", } ) continue node = nodes.get(path.path_slug) master_title = _text(node.name if node else path.title) master_slug = normalize_plp_path(node.path_slug if node else path.path_slug) if target.channel_code == "magento": registry = magento_registry.get(str(target.remote_id)) if registry is None: missing_registry.append(_target_summary(path, target, "missing_magento_registry_row")) continue live_title = _text(registry.name) expected_leaf_title = _leaf_title(master_title or master_slug) if expected_leaf_title and live_title and _norm(expected_leaf_title) != _norm(live_title): title_mismatches.append( { **_target_summary(path, target, "magento_category_title_mismatch"), "master_title": expected_leaf_title, "master_full_title": master_title, "live_title": live_title, } ) live_leaf_slug = slugify_segment(live_title) master_leaf_slug = master_slug.split("/")[-1] if master_leaf_slug and live_leaf_slug and master_leaf_slug != live_leaf_slug: url_mismatches.append( { **_target_summary(path, target, "magento_category_leaf_url_mismatch"), "master_leaf_slug": master_leaf_slug, "live_leaf_slug_from_title": live_leaf_slug, } ) elif target.channel_code == "shopify": registry = shopify_registry.get(str(target.remote_id)) if registry is None: missing_registry.append(_target_summary(path, target, "missing_shopify_registry_row")) continue live_title = _text(registry.title) expected_shopify_titles = { _norm(master_title), _norm_title_for_channel(master_title), _norm(master_slug.replace("/", " ")), _norm_title_for_channel(master_slug.replace("/", " ")), _norm(target.remote_path or ""), _norm_title_for_channel(str(target.remote_path or "")), } if live_title and _norm(live_title) not in {value for value in expected_shopify_titles if value}: title_mismatches.append( { **_target_summary(path, target, "shopify_collection_title_mismatch"), "master_title": master_title, "master_path_title": master_slug.replace("/", " ").title(), "live_title": live_title, } ) live_handle = _text(registry.handle) expected_handles = { master_slug.split("/")[-1], slugify_segment(master_title), slugify_segment(master_slug.replace("/", " ")), master_slug.replace("/", "-"), slugify_segment(str(target.remote_path or "")), } if live_handle and live_handle not in {value for value in expected_handles if value}: url_mismatches.append( { **_target_summary(path, target, "shopify_collection_handle_mismatch"), "expected_handles": sorted(value for value in expected_handles if value), "live_handle": live_handle, } ) issue_count = len(missing_master_path) + len(title_mismatches) + len(url_mismatches) + len(missing_registry) return { "issue_count": issue_count, "target_count": len(target_rows), "missing_master_path_count": len(missing_master_path), "missing_registry_count": len(missing_registry), "title_mismatch_count": len(title_mismatches), "url_mismatch_count": len(url_mismatches), "sample_missing_master_paths": missing_master_path[:sku_sample_limit], "sample_missing_registry": missing_registry[:sku_sample_limit], "sample_title_mismatches": title_mismatches[:sku_sample_limit], "sample_url_mismatches": url_mismatches[:sku_sample_limit], } def _active_listing_paths_by_slug(session: Session) -> Dict[int, ChannelListingPath]: rows = session.scalars(select(ChannelListingPath).where(ChannelListingPath.is_active.is_(True))).all() return {int(row.id): row for row in rows} def _active_taxonomy_nodes_by_slug(session: Session) -> Dict[str, MasterTaxonomyNode]: rows = session.scalars(select(MasterTaxonomyNode).where(MasterTaxonomyNode.is_active.is_(True))).all() return {normalize_plp_path(row.path_slug): row for row in rows} def _magento_registry_by_id(session: Session, connection_id: Optional[int]) -> Dict[str, MagentoCategoryRegistry]: stmt = select(MagentoCategoryRegistry) if connection_id is not None: stmt = stmt.where(MagentoCategoryRegistry.connection_id == connection_id) return {str(row.category_id): row for row in session.scalars(stmt).all()} def _shopify_registry_by_id(session: Session, connection_id: Optional[int]) -> Dict[str, ShopifyCollectionRegistry]: stmt = select(ShopifyCollectionRegistry) if connection_id is not None: stmt = stmt.where(ShopifyCollectionRegistry.connection_id == connection_id) return {str(row.collection_id): row for row in session.scalars(stmt).all()} def _target_summary(path: ChannelListingPath, target: ChannelListingPathTarget, reason: str) -> Dict[str, Any]: return { "reason": reason, "channel": target.channel_code, "remote_id": target.remote_id, "taxonomy_kind": target.taxonomy_kind, "path_slug": path.path_slug, "path_title": path.title, "remote_path": target.remote_path, } def _audit_configurable_option_mapping( session: Session, skus: Sequence[str], *, magento_connection_id: Optional[int], sku_sample_limit: int, ) -> Dict[str, Any]: relation_rows = list( session.scalars( select(MasterProductRelation) .where( or_( MasterProductRelation.parent_sku.in_(list(skus)), MasterProductRelation.child_sku.in_(list(skus)), ) ) .order_by(MasterProductRelation.parent_sku, MasterProductRelation.sort_order) ).all() ) option_index = _magento_option_index(session, magento_connection_id) if relation_rows: by_parent: Dict[str, List[MasterProductRelation]] = defaultdict(list) for row in relation_rows: by_parent[row.parent_sku].append(row) report = _audit_master_product_relations( by_parent, option_index, sku_sample_limit=sku_sample_limit, ) report["missing_persisted_relation_parent_count"] = 0 report["sample_missing_persisted_relations"] = [] return report missing_persisted = _variation_builder_parents_missing_relations(session, skus) source = "magento_snapshot_rows" parent_rows = build_magento_snapshot_rows_from_master( session, skus=list(skus), only_assigned=False, connection_id=magento_connection_id, expand_relations=True, ) report = _audit_configurable_rows( parent_rows, option_index, source=source, sku_sample_limit=sku_sample_limit, ) report["missing_persisted_relation_parent_count"] = len(missing_persisted) report["sample_missing_persisted_relations"] = missing_persisted[:sku_sample_limit] report["issue_count"] = int(report.get("issue_count") or 0) + len(missing_persisted) if missing_persisted and not report.get("parent_count"): report["source"] = "variation_builder_payload_without_relations" return report def _audit_master_product_relations( by_parent: Dict[str, List[MasterProductRelation]], option_index: Dict[str, Dict[str, int]], *, sku_sample_limit: int, ) -> Dict[str, Any]: """Audit option mappings directly from persisted relations (no string round-trip).""" missing_options: List[Dict[str, Any]] = [] missing_mappings: List[Dict[str, Any]] = [] relation_count = 0 for parent_sku, relations in sorted(by_parent.items()): relation_count += len(relations) if not relations: missing_mappings.append( { "parent_sku": parent_sku, "axes": [], "reason": "no_configurable_children", } ) continue axes = sorted( { str(code).strip() for relation in relations for code in dict(relation.option_mapping or {}).keys() if str(code).strip() and str(dict(relation.option_mapping or {}).get(code) or "").strip() } ) for relation in relations: child_sku = str(relation.child_sku or "").strip() mapping = { str(k).strip(): str(v).strip() for k, v in dict(relation.option_mapping or {}).items() if str(k).strip() and str(v or "").strip() } if not mapping: missing_mappings.append( { "parent_sku": parent_sku, "child_sku": child_sku, "axes": axes, "reason": "empty_option_mapping", } ) continue for axis in axes or sorted(mapping.keys()): value = _axis_mapping_value(mapping, axis) if not value: missing_mappings.append( { "parent_sku": parent_sku, "child_sku": child_sku, "axis": axis, "mapping": mapping, "reason": "missing_axis_value", } ) continue magento_axis = resolve_magento_axis_code(axis) or axis options = option_index.get(str(magento_axis).lower(), {}) or option_index.get(axis.lower(), {}) value_index = _resolve_value_index_for_audit(str(value), options) if value_index is None and options: missing_options.append( { "parent_sku": parent_sku, "child_sku": child_sku, "axis": axis, "magento_axis": magento_axis, "value": value, "normalized_value": norm_option_label(str(value)), "known_option_count": len(options), "sample_known_options": list(options)[:15], "reason": "magento_option_value_index_missing", } ) return { "issue_count": len(missing_options) + len(missing_mappings), "source": "master_product_relations", "parent_count": len(by_parent), "relation_count": relation_count, "missing_option_count": len(missing_options), "missing_mapping_count": len(missing_mappings), "sample_missing_options": missing_options[:sku_sample_limit], "sample_missing_mappings": missing_mappings[:sku_sample_limit], } def _variation_builder_parents_missing_relations( session: Session, skus: Sequence[str], ) -> List[Dict[str, Any]]: """Parents that still only have children in raw_payload (relations never persisted).""" from db.variation_builder import _variation_builder_parent_products wanted = {str(sku or "").strip() for sku in skus if str(sku or "").strip()} if not wanted: return [] out: List[Dict[str, Any]] = [] for parent_sku, product in _variation_builder_parent_products(session).items(): payload = product.raw_payload if isinstance(product.raw_payload, dict) else {} children = [ str(child or "").strip() for child in (payload.get("children") or []) if str(child or "").strip() ] if not children: continue touches = parent_sku in wanted or any(child in wanted for child in children) if not touches: continue out.append( { "parent_sku": parent_sku, "child_count": len(children), "sample_children": children[:5], "reason": "variation_builder_payload_without_master_product_relation", "repair_hint": "python -m app.jobs.repair_variation_relations", } ) return out def _audit_configurable_rows( rows: Sequence[Dict[str, Any]], option_index: Dict[str, Dict[str, int]], *, source: str, sku_sample_limit: int, ) -> Dict[str, Any]: configurable_rows: List[Dict[str, Any]] = [] for row in rows: product_type = str(row.get("product_type") or "").strip().lower() has_relation_fields = bool(str(row.get("variant_list") or "").strip() or str(row.get("configurable_variations") or "").strip()) if "configurable" in product_type or has_relation_fields: configurable_rows.append(row) missing_options: List[Dict[str, Any]] = [] missing_mappings: List[Dict[str, Any]] = [] relation_count = 0 for parent_row in configurable_rows: parent_sku = str(parent_row.get("master_sku") or parent_row.get("sku") or "").strip() children, mappings, child_to_mapping = _parse_children_from_row(parent_row) relation_count += len(children) axes = sorted(get_axis_attribute_codes(parent_row)) if not axes: missing_mappings.append( { "parent_sku": parent_sku, "reason": "no_configurable_axes", "child_count": len(children), } ) continue if not children: missing_mappings.append( { "parent_sku": parent_sku, "axes": axes, "reason": "no_configurable_children", } ) continue for index, child_sku in enumerate(children): mapping = child_to_mapping.get(child_sku) or (mappings[index] if index < len(mappings) else {}) if not mapping: missing_mappings.append( { "parent_sku": parent_sku, "child_sku": child_sku, "axes": axes, "reason": "empty_option_mapping", } ) continue for axis in axes: value = _axis_mapping_value(mapping, axis) if not value: missing_mappings.append( { "parent_sku": parent_sku, "child_sku": child_sku, "axis": axis, "mapping": mapping, "reason": "missing_axis_value", } ) continue norm = norm_option_label(str(value)) options = option_index.get(axis, {}) value_index = _resolve_value_index_for_audit(str(value), options) if value_index is None: missing_options.append( { "parent_sku": parent_sku, "child_sku": child_sku, "axis": axis, "value": value, "normalized_value": norm, "known_option_count": len(options), "sample_known_options": list(options)[:15], "reason": "magento_option_value_index_missing", } ) return { "issue_count": len(missing_options) + len(missing_mappings), "source": source, "parent_count": len(configurable_rows), "relation_count": relation_count, "missing_option_count": len(missing_options), "missing_mapping_count": len(missing_mappings), "sample_missing_options": missing_options[:sku_sample_limit], "sample_missing_mappings": missing_mappings[:sku_sample_limit], } def _magento_option_index(session: Session, connection_id: Optional[int]) -> Dict[str, Dict[str, int]]: stmt = select(MagentoAttributeOptionRegistry) if connection_id is not None: stmt = stmt.where(MagentoAttributeOptionRegistry.connection_id == connection_id) out: Dict[str, Dict[str, int]] = defaultdict(dict) for row in session.scalars(stmt).all(): out[str(row.attribute_code or "").strip().lower()][str(row.option_label_norm or "").strip()] = int(row.value_index) return out def _parent_relation_row_for_audit(parent_sku: str, relations: Sequence[MasterProductRelation]) -> Dict[str, Any]: """Build a Magento-style row for parsers that expect sku=attr pipe segments.""" segments: List[str] = [] axes: set[str] = set() for relation in relations: mapping = relation.option_mapping if isinstance(relation.option_mapping, dict) else {} parts = [f"sku={relation.child_sku}"] for key, value in sorted(mapping.items()): code = str(key or "").strip() text = str(value or "").strip() if not code or not text: continue channel_key = resolve_magento_axis_code(code) or code axes.add(channel_key) parts.append(f"{channel_key}={text}") segments.append(",".join(parts)) return { "sku": parent_sku, "product_type": "configurable", "variant_list": ",".join(relation.child_sku for relation in relations), "configurable_attributes": ",".join(sorted(axes)), "configurable_variations": "|".join(segments), } def _group_category_mismatches( category_issues: Sequence[Dict[str, Any]], *, sample_limit: int, ) -> List[Dict[str, Any]]: """Collapse per-SKU category mismatches into expected→actual path groups.""" groups: Dict[tuple, Dict[str, Any]] = {} for issue in category_issues: expected = tuple(issue.get("expected_paths") or []) actual = tuple(issue.get("would_push_paths") or []) key = (expected, actual) group = groups.get(key) if group is None: group = { "expected_paths": list(expected), "would_push_paths": list(actual), "count": 0, "sample_skus": [], "sample_category_ids": [], } groups[key] = group group["count"] += 1 sku = str(issue.get("sku") or "").strip() if sku and len(group["sample_skus"]) < 5: group["sample_skus"].append(sku) ids = issue.get("category_ids") or [] if ids and not group["sample_category_ids"]: group["sample_category_ids"] = list(ids) ordered = sorted(groups.values(), key=lambda row: (-int(row["count"]), row["expected_paths"], row["would_push_paths"])) return ordered[:sample_limit] def _axis_mapping_value(mapping: Dict[str, Any], axis: str) -> str: for key in (axis, axis.replace("_", ""), f"{axis}_in", f"{axis}_inches"): value = mapping.get(key) if value is not None and str(value).strip(): return str(value).strip() return "" def _resolve_value_index_for_audit(value: str, options: Dict[str, int]) -> Optional[int]: if not value: return None norm = norm_option_label(value) direct = options.get(norm) if direct is not None: return int(direct) flat = norm.replace(" ", "") for option_label, value_index in options.items(): if option_label.replace(" ", "") == flat: return int(value_index) if value.isdigit(): numeric = int(value) if numeric in set(options.values()): return numeric return None def _expected_title(product: MasterProduct) -> str: return _text(product.name) def _expected_product_slug( product: MasterProduct, *, explicit_slug: Optional[str] = None, fallback_sku: Optional[str] = None, ) -> str: slug = _text(explicit_slug) or _raw_payload_value(product.raw_payload, "product_url_slug", "url_key", "handle") if slug: return slug sku = fallback_sku or product.sku return build_pdp_slug(_expected_title(product) or sku, sku) def _expected_product_slug_from_row( row: Dict[str, Any], product: MasterProduct, *, fallback_sku: str, ) -> str: explicit = _text(row.get("product_url_slug") or row.get("url_key") or row.get("handle")) if explicit: return explicit sku = str(row.get("sku") or fallback_sku) title = _text(row.get("seo_title") or row.get("meta_title") or _expected_title(product) or fallback_sku) return build_pdp_slug(title, sku) def _expected_product_slug_for_payload( payload: Dict[str, Any], product: MasterProduct, *, fallback_sku: str, ) -> str: fields = payload.get("fields") if isinstance(payload.get("fields"), dict) else {} explicit = _text(fields.get("handle") or fields.get("product_url_slug") or fields.get("url_key")) if explicit: return explicit return _expected_product_slug(product, fallback_sku=str(payload.get("channel_sku") or fallback_sku)) def _looks_like_collection_only_title(product: MasterProduct, paths: Sequence[Dict[str, Any]]) -> bool: title = _norm(_text(product.name)) if not title: return False comparables = { _norm(_text(product.collection)), _norm(_text(product.category_l2)), _norm(_text(product.category_l3)), } for path in paths: slug = normalize_plp_path(path.get("path_slug") or "") if slug: comparables.add(_norm(slug.split("/")[-1].replace("-", " "))) tokens = set(title.split()) has_descriptor = bool(tokens & PRODUCT_DESCRIPTOR_TOKENS) return title in {value for value in comparables if value} and not has_descriptor def _custom_attribute_value(payload: Dict[str, Any], code: str) -> Optional[str]: for item in payload.get("custom_attributes") or []: if isinstance(item, dict) and item.get("attribute_code") == code: return _text(item.get("value")) attrs = payload.get("attrs") if isinstance(payload.get("attrs"), dict) else {} if code in attrs: return _text(attrs.get(code)) return None def _raw_payload_value(payload: Optional[dict], *codes: str) -> str: if not isinstance(payload, dict): return "" for code in codes: value = payload.get(code) if value is not None and str(value).strip(): return str(value).strip() return "" def _split_paths(value: Any) -> List[str]: if value is None: return [] if isinstance(value, list): return [str(v).strip() for v in value if str(v).strip()] return [part.strip() for part in str(value).replace(",", "|").split("|") if part.strip()] def _normalize_channel_category_path(value: Any) -> str: path = normalize_plp_path(str(value or "")) for prefix in ("default-category/", "root-catalog/"): if path.startswith(prefix): return path[len(prefix):] return path def _magento_registry_paths( session: Session, connection_id: Optional[int], category_ids: Sequence[int], ) -> Dict[int, str]: ids = sorted({int(cid) for cid in category_ids if cid is not None}) if not ids: return {} stmt = select(MagentoCategoryRegistry).where(MagentoCategoryRegistry.category_id.in_(ids)) if connection_id is not None: stmt = stmt.where(MagentoCategoryRegistry.connection_id == connection_id) out: Dict[int, str] = {} for row in session.scalars(stmt).all(): out[int(row.category_id)] = str(row.path_names or row.name or "") return out def _magento_path_covers_expected(expected: str, pushed_paths: Iterable[str]) -> bool: """True when an expected PLP path is represented by Magento registry/push paths. Magento ``path_names`` and master PLP slugs are different namespaces. Require the leaf segment, and for depth>=3 paths also the parent facet (base-cabinets vs wall-cabinets) so leaf collisions like ``single-door`` cannot silently pass. """ expected_path = normalize_plp_path(expected) if not expected_path: return True expected_parts = [part for part in expected_path.split("/") if part] if not expected_parts: return True leaf = expected_parts[-1] parent = expected_parts[-2] if len(expected_parts) >= 3 else None for raw in pushed_paths: pushed = normalize_plp_path(str(raw or "")) if not pushed: continue if pushed == expected_path: return True pushed_parts = [part for part in pushed.split("/") if part] if not pushed_parts: continue if leaf not in pushed_parts: continue if parent is None: return True if parent in pushed_parts: return True return False def _leaf_title(value: str) -> str: text = _text(value).replace("/", " / ") parts = [part.strip() for part in text.split("/") if part.strip()] return parts[-1] if parts else text def _text(value: Any) -> str: return str(value or "").strip() def _norm(value: str) -> str: return " ".join(_text(value).lower().split()) def _norm_title_for_channel(value: str) -> str: text = _text(value).replace("/", " ").replace("-", " ") return _norm(text) def _norm_slug(value: str) -> str: return normalize_plp_path(_text(value)) def main() -> int: parser = argparse.ArgumentParser(description="Audit master SKU data and Magento/Shopify parity") parser.add_argument("--sku", action="append", dest="skus", help="Specific master SKU to audit; repeatable") parser.add_argument("--limit", type=int, default=None, help="Limit active SKUs for a quick sample") parser.add_argument("--sample-limit", type=int, default=25) parser.add_argument("--magento-connection-id", type=int, default=None) parser.add_argument("--shopify-connection-id", type=int, default=None) args = parser.parse_args() with get_session() as session: result = audit_product_channel_parity( session, skus=args.skus, limit=args.limit, magento_connection_id=args.magento_connection_id, shopify_connection_id=args.shopify_connection_id, 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())