"""Report where each curated shopping_* value comes from and whether a taxonomy node backs it. The outbound lock trusts manual taxonomy assignments and Start Shopping configs, so a duplicate variant curated there (moldings vs mouldings, tall vs tall-cabinets, single-door-glass vs single-door-glass-wall-cabinet) is treated as legitimate and keeps reaching Magento. This report attributes every curated value to its source rows and flags the ones with no active master_taxonomy_node behind them. python -m app.jobs.report_shopping_vocabulary_variants python -m app.jobs.report_shopping_vocabulary_variants --code shopping_l2 """ from __future__ import annotations import argparse import json from collections import defaultdict from typing import Any, Dict, List, Optional, Sequence, Set, Tuple from sqlalchemy import func, select from sqlalchemy.orm import Session from db.models import ( CollectionLandingPage, ManualTaxonomyAssignment, ManualTaxonomyAssignmentCollection, MasterTaxonomyNode, ) from db.session import get_session from db.shopping_taxonomy_vocabulary import ( SHOPPING_TAXONOMY_CODES, normalize_shopping_value, ) def _node_backed_values(session: Session) -> Dict[str, Set[str]]: """Values that an active master_taxonomy_node lineage can actually produce.""" from db.collection_landing_pages import ( _relative_shopping_collection_value, _relative_shopping_l1_value, _relative_shopping_l2_value, ) paths = [ str(path or "").strip() for path in session.scalars( select(MasterTaxonomyNode.path_slug).where(MasterTaxonomyNode.is_active.is_(True)) ).all() if str(path or "").strip() ] out: Dict[str, Set[str]] = {code: set() for code in SHOPPING_TAXONOMY_CODES} for path in paths: parts = [part for part in path.split("/") if part] if len(parts) == 2: out["shopping_collection"].add(normalize_shopping_value(_relative_shopping_collection_value(path))) out["shopping_l1"].add(normalize_shopping_value(_relative_shopping_l1_value(path))) if len(parts) >= 3: out["shopping_l2"].add(normalize_shopping_value(_relative_shopping_l2_value(path))) return {code: {value for value in values if value} for code, values in out.items()} def _origins(session: Session) -> Dict[str, Dict[str, List[Dict[str, Any]]]]: """origins[code][value] = [{source, reference, sku_count}]""" origins: Dict[str, Dict[str, List[Dict[str, Any]]]] = { code: defaultdict(list) for code in SHOPPING_TAXONOMY_CODES } sku_counts = dict( session.execute( select( ManualTaxonomyAssignmentCollection.assignment_id, func.count(ManualTaxonomyAssignmentCollection.id), ) .where(ManualTaxonomyAssignmentCollection.is_active.is_(True)) .group_by(ManualTaxonomyAssignmentCollection.assignment_id) ).all() ) rows = session.execute( select( ManualTaxonomyAssignment.id, ManualTaxonomyAssignment.source_sku, ManualTaxonomyAssignment.canonical_taxonomy_path_slug, ManualTaxonomyAssignment.shopping_l1, ManualTaxonomyAssignment.shopping_l2, ).where(ManualTaxonomyAssignment.assignment_status == "active") ).all() for assignment_id, source_sku, canonical_path, shopping_l1, shopping_l2 in rows: reference = f"{source_sku} @ {canonical_path}" sku_count = int(sku_counts.get(assignment_id) or 0) for code, raw in (("shopping_l1", shopping_l1), ("shopping_l2", shopping_l2)): value = normalize_shopping_value(raw) if value: origins[code][value].append( {"source": "manual_taxonomy", "reference": reference, "sku_count": sku_count} ) collection_rows = session.execute( select( ManualTaxonomyAssignmentCollection.shopping_collection, ManualTaxonomyAssignmentCollection.collection_code, ).where(ManualTaxonomyAssignmentCollection.is_active.is_(True)) ).all() seen_collections: Set[Tuple[str, str]] = set() for shopping_collection, collection_code in collection_rows: value = normalize_shopping_value(shopping_collection) key = (value, str(collection_code or "")) if not value or key in seen_collections: continue seen_collections.add(key) origins["shopping_collection"][value].append( {"source": "manual_taxonomy", "reference": str(collection_code or ""), "sku_count": None} ) from db.collection_landing_pages import ( _normalize_start_shopping_config, _relative_shopping_collection_value, _relative_shopping_l2_value, _start_shopping_l1_value, ) landing_rows = session.execute( select(CollectionLandingPage.path_slug, CollectionLandingPage.start_shopping_config) .where(CollectionLandingPage.is_active.is_(True)) ).all() for path_slug, raw_config in landing_rows: collection_value = normalize_shopping_value(_relative_shopping_collection_value(path_slug)) if collection_value: origins["shopping_collection"][collection_value].append( {"source": "collection_landing_page", "reference": str(path_slug), "sku_count": None} ) config = _normalize_start_shopping_config(raw_config) for item in config.get("items") or []: l1_value = normalize_shopping_value(_start_shopping_l1_value(item)) if l1_value: origins["shopping_l1"][l1_value].append( {"source": "start_shopping_config", "reference": str(path_slug), "sku_count": None} ) for l2_path in item.get("l2_path_slugs") or []: l2_value = normalize_shopping_value(_relative_shopping_l2_value(l2_path)) if l2_value: origins["shopping_l2"][l2_value].append( {"source": "start_shopping_config", "reference": str(path_slug), "sku_count": None} ) for candidate in item.get("l2_values") or []: slug = str((candidate or {}).get("slug") or "").strip() if slug and l1_value: origins["shopping_l2"][normalize_shopping_value(f"{l1_value}/{slug}")].append( {"source": "start_shopping_config", "reference": str(path_slug), "sku_count": None} ) return origins def _variant_key(code: str, value: str) -> str: """Collapse known spelling/suffix variants so duplicates cluster together.""" parts = [part for part in value.split("/") if part] normalized: List[str] = [] for part in parts: segment = part.replace("moulding", "molding") for suffix in ("-wall-cabinet", "-base-cabinet", "-cabinet", "-molding", "-cabinets"): if segment.endswith(suffix) and len(segment) > len(suffix) + 2: segment = segment[: -len(suffix)] break normalized.append(segment.rstrip("s")) return "/".join(normalized) def run(*, codes: Optional[Sequence[str]] = None) -> Dict[str, Any]: wanted = [str(code).strip() for code in (codes or SHOPPING_TAXONOMY_CODES) if str(code).strip()] with get_session() as session: origins = _origins(session) node_backed = _node_backed_values(session) out: Dict[str, Any] = {"codes": []} for code in wanted: values = origins.get(code) or {} entries = [] for value in sorted(values): entries.append( { "value": value, "node_backed": value in (node_backed.get(code) or set()), "sources": sorted({item["source"] for item in values[value]}), "curated_by": values[value][:8], "manual_sku_rows": sum(int(item["sku_count"] or 0) for item in values[value]), } ) clusters = defaultdict(list) for entry in entries: clusters[_variant_key(code, entry["value"])].append(entry["value"]) out["codes"].append( { "attribute_code": code, "curated_count": len(entries), "not_backed_by_active_node": [e["value"] for e in entries if not e["node_backed"]], "variant_clusters": { key: sorted(members) for key, members in sorted(clusters.items()) if len(members) > 1 }, "values": entries, } ) return out def main() -> int: parser = argparse.ArgumentParser(description="Report curated shopping_* value provenance and variants") parser.add_argument( "--code", action="append", dest="codes", help=f"Limit to specific codes (default: {', '.join(SHOPPING_TAXONOMY_CODES)})", ) args = parser.parse_args() print(json.dumps(run(codes=args.codes), indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())