""" Examples: python -m app.jobs.audit_manual_taxonomy_suffix_mismatches python -m app.jobs.audit_manual_taxonomy_suffix_mismatches --source-sku BF3 python -m app.jobs.audit_manual_taxonomy_suffix_mismatches --source-sku BF3 --batch-size 200 --sample-limit 50 """ from __future__ import annotations import argparse import json from typing import Any, Dict, List, Optional, Sequence from sqlalchemy import func, select from db.channel_listing_path import resolve_listing_paths_for_sku from db.manual_taxonomy_assignments import resolve_manual_taxonomy_import_override from db.master_taxonomy_product_paths import canonical_master_taxonomy_path_slugs, leaf_taxonomy_path_slugs from db.models import MasterProduct from db.session import get_session def _clean(value: Any) -> str: return str(value or "").strip() def _mismatch_record(product: MasterProduct, manual: Dict[str, Any]) -> Dict[str, Any]: current = { "category_l1": _clean(product.category_l1), "category_l2": _clean(product.category_l2), "category_l3": _clean(product.category_l3), } expected = { "category_l1": _clean(manual.get("category_l1")), "category_l2": _clean(manual.get("category_l2")), "category_l3": _clean(manual.get("category_l3")), } wrong_fields = [ field for field in ("category_l1", "category_l2", "category_l3") if current[field] != expected[field] ] return { "master_sku": product.sku, "source_item": _clean(product.source_item), "collection": _clean(product.collection), "membership_mode": _clean(product.membership_mode), "match_scope": _clean(manual.get("match_scope")), "manual_source_sku": _clean(manual.get("source_sku")), "manual_master_sku": _clean(manual.get("master_sku")), "canonical_taxonomy_path_slug": _clean(manual.get("canonical_taxonomy_path_slug")), "current": current, "expected": expected, "wrong_fields": wrong_fields, } def _taxonomy_path_mismatch_record( session: Any, product: MasterProduct, manual: Dict[str, Any], ) -> Dict[str, Any]: expected_paths = { _clean(path) for path in leaf_taxonomy_path_slugs(canonical_master_taxonomy_path_slugs(session, product, attrs={})) if _clean(path) } assigned_paths = { _clean(path.get("path_slug")) for path in resolve_listing_paths_for_sku(session, product.sku) if _clean(path.get("path_slug")) } missing_expected = sorted(expected_paths - assigned_paths) stale_assigned = sorted(assigned_paths - expected_paths) return { "master_sku": product.sku, "source_item": _clean(product.source_item), "collection": _clean(product.collection), "membership_mode": _clean(product.membership_mode), "match_scope": _clean(manual.get("match_scope")), "manual_source_sku": _clean(manual.get("source_sku")), "manual_master_sku": _clean(manual.get("master_sku")), "canonical_taxonomy_path_slug": _clean(manual.get("canonical_taxonomy_path_slug")), "expected_leaf_paths": sorted(expected_paths), "assigned_leaf_paths": sorted(assigned_paths), "missing_expected_paths": missing_expected, "stale_assigned_paths": stale_assigned, } def audit_manual_taxonomy_suffix_mismatches( *, source_sku: Optional[str] = None, batch_size: int = 500, start_offset: int = 0, max_batches: Optional[int] = None, sample_limit: int = 25, ) -> Dict[str, Any]: batch_size = max(int(batch_size or 500), 1) start_offset = max(int(start_offset or 0), 0) max_batches = None if max_batches in (None, 0) else max(int(max_batches), 1) normalized_source = _clean(source_sku).upper() with get_session() as session: stmt = select(MasterProduct).where(MasterProduct.is_active.is_(True)).order_by(MasterProduct.sku) if normalized_source: stmt = stmt.where(func.upper(func.coalesce(MasterProduct.sku, "")).like(f"%-{normalized_source}")) products = list(session.scalars(stmt).all()) products = products[start_offset:] if not products: return { "status": "ok", "source_sku": normalized_source or None, "batch_size": batch_size, "start_offset": start_offset, "max_batches": max_batches, "product_count": 0, "checked_count": 0, "manual_match_count": 0, "field_mismatch_count": 0, "path_mismatch_count": 0, "mismatch_count": 0, "sample_field_mismatches": [], "sample_path_mismatches": [], } ranges = list(range(0, len(products), batch_size)) if max_batches is not None: ranges = ranges[:max_batches] checked_count = 0 manual_match_count = 0 field_mismatch_count = 0 path_mismatch_count = 0 sample_field_mismatches: List[Dict[str, Any]] = [] sample_path_mismatches: List[Dict[str, Any]] = [] by_scope = {"exact": 0, "source_sku": 0} for start in ranges: batch = products[start : start + batch_size] for product in batch: checked_count += 1 manual = resolve_manual_taxonomy_import_override( session, master_sku=product.sku, source_sku=product.source_item, ) if not manual: continue manual_match_count += 1 scope = _clean(manual.get("match_scope")) or "unknown" if scope in by_scope: by_scope[scope] += 1 current_l1 = _clean(product.category_l1) current_l2 = _clean(product.category_l2) current_l3 = _clean(product.category_l3) expected_l1 = _clean(manual.get("category_l1")) expected_l2 = _clean(manual.get("category_l2")) expected_l3 = _clean(manual.get("category_l3")) if (current_l1, current_l2, current_l3) != (expected_l1, expected_l2, expected_l3): field_mismatch_count += 1 if len(sample_field_mismatches) < sample_limit: sample_field_mismatches.append(_mismatch_record(product, manual)) path_record = _taxonomy_path_mismatch_record(session, product, manual) if path_record["missing_expected_paths"] or path_record["stale_assigned_paths"]: path_mismatch_count += 1 if len(sample_path_mismatches) < sample_limit: sample_path_mismatches.append(path_record) return { "status": "ok", "source_sku": normalized_source or None, "batch_size": batch_size, "start_offset": start_offset, "max_batches": max_batches, "product_count": len(products), "checked_count": checked_count, "manual_match_count": manual_match_count, "field_mismatch_count": field_mismatch_count, "path_mismatch_count": path_mismatch_count, "mismatch_count": field_mismatch_count + path_mismatch_count, "manual_match_scope_counts": by_scope, "sample_field_mismatches": sample_field_mismatches, "sample_path_mismatches": sample_path_mismatches, } def main(argv: Optional[Sequence[str]] = None) -> int: parser = argparse.ArgumentParser(description="Audit master_product rows against effective manual taxonomy overrides") parser.add_argument("--source-sku", default=None, help="Optional base/source SKU suffix like BF3") parser.add_argument("--batch-size", type=int, default=500, help="Rows to inspect per batch") parser.add_argument("--start-offset", type=int, default=0, help="Start from this zero-based product offset") parser.add_argument("--max-batches", type=int, default=0, help="Optional limit on how many batches to inspect") parser.add_argument("--sample-limit", type=int, default=25, help="Number of mismatch rows to include in the sample") args = parser.parse_args(list(argv) if argv is not None else None) result = audit_manual_taxonomy_suffix_mismatches( source_sku=args.source_sku, batch_size=args.batch_size, start_offset=args.start_offset, max_batches=args.max_batches, sample_limit=args.sample_limit, ) print(json.dumps(result, indent=2, default=str)) return 0 if __name__ == "__main__": raise SystemExit(main())