"""Apply taxonomy pollution cleanup from generated audit CSVs. Default mode is dry-run. The job only touches rows present in the audit report files, which makes the cleanup traceable back to the inspected dump. Example: python -m app.jobs.cleanup_taxonomy_pollution_from_audit --audit-dir E:\\2026\\plytixmage_dbs_data\\audit_reports python -m app.jobs.cleanup_taxonomy_pollution_from_audit --audit-dir E:\\2026\\plytixmage_dbs_data\\audit_reports --apply """ from __future__ import annotations import argparse import csv import json import os import sys from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, Iterable, Optional, Sequence, Set from sqlalchemy import select from sqlalchemy.orm import Session from db.models import ( ChannelListingPath, ChannelListingPathTarget, CollectionLandingPage, MagentoCategoryRegistry, MasterCollectionRegistry, MasterProduct, MasterTaxonomyChannelLink, MasterTaxonomyNode, MasterTaxonomyPathAlias, ProductChannelTaxonomyAssignment, ProductListingPathAssignment, ) from db.session import get_session ACTIVE = "active" INACTIVE = "inactive" DEFAULT_PROTECTED_SLUGS = {"doors"} DEFAULT_STRUCTURAL_REMOTE_LABELS = {"accessories", "doors"} def cleanup_taxonomy_pollution_from_audit( session: Session, *, audit_dir: str | os.PathLike[str], dry_run: bool = True, deactivate_listing_paths: bool = True, deactivate_listing_assignments: bool = True, deactivate_listing_targets: bool = True, deactivate_taxonomy_assignments: bool = True, deactivate_inactive_node_links: bool = True, delete_inactive_taxonomy_nodes: bool = True, db_inactive_taxonomy_nodes: bool = False, inactive_node_path_prefixes: Optional[Iterable[str]] = None, mark_magento_registry_inactive: bool = False, merge_duplicate_collections: bool = False, include_structural_remote_labels: bool = False, protected_slugs: Optional[Iterable[str]] = None, note: Optional[str] = None, ) -> Dict[str, Any]: audit_path = Path(audit_dir) cleanup_note = note or f"taxonomy pollution cleanup from audit at {datetime.now(timezone.utc).isoformat()}" protected = {_norm_slug(item) for item in (protected_slugs or DEFAULT_PROTECTED_SLUGS) if _norm_slug(item)} structural = set() if include_structural_remote_labels else set(DEFAULT_STRUCTURAL_REMOTE_LABELS) normalized_prefixes = [_norm_slug(item) for item in (inactive_node_path_prefixes or []) if _norm_slug(item)] summary: Dict[str, Any] = { "status": "ok", "dry_run": dry_run, "audit_dir": str(audit_path), "protected_slugs": sorted(protected), "structural_remote_labels_skipped": sorted(structural), "steps": {}, "warnings": [], } if deactivate_taxonomy_assignments: rows = _read_csv(audit_path / "bad_product_channel_taxonomy_assignments.csv") rows = [row for row in rows if not _is_structural_remote_row(row, structural)] summary["steps"]["product_channel_taxonomy_assignments"] = _deactivate_by_ids( session, ProductChannelTaxonomyAssignment, _ids(rows), dry_run=dry_run, note=cleanup_note, status_field="assignment_status", active_value=ACTIVE, inactive_value=INACTIVE, ) if deactivate_listing_assignments: rows = _filter_protected_paths( _read_csv(audit_path / "bad_product_listing_path_assignments.csv"), protected, ) summary["steps"]["product_listing_path_assignments"] = _deactivate_by_ids( session, ProductListingPathAssignment, _ids(rows), dry_run=dry_run, note=cleanup_note, status_field="assignment_status", active_value=ACTIVE, inactive_value=INACTIVE, ) if deactivate_listing_targets: rows = _filter_protected_paths( _read_csv(audit_path / "bad_channel_listing_path_targets.csv"), protected, ) summary["steps"]["channel_listing_path_targets"] = _deactivate_by_ids( session, ChannelListingPathTarget, _ids(rows), dry_run=dry_run, note=cleanup_note, status_field="is_active", active_value=True, inactive_value=False, note_field=None, ) if deactivate_listing_paths: rows = _filter_protected_paths(_read_csv(audit_path / "bad_listing_paths.csv"), protected) listing_path_result = _deactivate_by_ids( session, ChannelListingPath, _ids(rows), dry_run=dry_run, note=cleanup_note, status_field="is_active", active_value=True, inactive_value=False, ) landing_result = _deactivate_collection_landing_pages_for_paths( session, [str(row.get("path_slug") or "").strip() for row in rows], dry_run=dry_run, note=cleanup_note, ) summary["steps"]["channel_listing_paths"] = listing_path_result summary["steps"]["collection_landing_pages"] = landing_result inactive_node_link_rows: list[Dict[str, str]] = [] if deactivate_inactive_node_links: inactive_node_link_rows = ( _db_inactive_taxonomy_link_rows( session, protected_slugs=protected, path_prefixes=normalized_prefixes, ) if db_inactive_taxonomy_nodes else _filter_protected_paths( _read_csv(audit_path / "links_to_inactive_taxonomy_nodes.csv"), protected, ) ) summary["steps"]["master_taxonomy_channel_links"] = _deactivate_by_ids( session, MasterTaxonomyChannelLink, _ids(inactive_node_link_rows), dry_run=dry_run, note=cleanup_note, status_field="is_active", active_value=True, inactive_value=False, note_field=None, ) if delete_inactive_taxonomy_nodes: inactive_node_rows = ( _db_inactive_taxonomy_node_rows( session, protected_slugs=protected, path_prefixes=normalized_prefixes, ) if db_inactive_taxonomy_nodes else _filter_protected_paths( _read_csv(audit_path / "inactive_master_taxonomy_nodes.csv"), protected, ) ) if not inactive_node_rows: inactive_node_rows = inactive_node_link_rows summary["steps"]["master_taxonomy_node"] = _delete_inactive_taxonomy_nodes( session, inactive_node_rows, dry_run=dry_run, planned_link_ids=_ids(inactive_node_link_rows), ) if mark_magento_registry_inactive: rows = _read_csv(audit_path / "bad_magento_categories.csv") rows = [row for row in rows if not _is_structural_remote_row(row, structural)] summary["steps"]["magento_category_registry"] = _deactivate_by_ids( session, MagentoCategoryRegistry, _ids(rows), dry_run=dry_run, note=cleanup_note, status_field="is_active", active_value=True, inactive_value=False, note_field=None, ) if merge_duplicate_collections: rows = _read_csv(audit_path / "duplicate_collection_registry_rows.csv") summary["steps"]["duplicate_collection_registry"] = _merge_duplicate_collections( session, rows, dry_run=dry_run, note=cleanup_note, ) if not dry_run: session.flush() return summary def _deactivate_by_ids( session: Session, model: Any, ids: Sequence[int], *, dry_run: bool, note: str, status_field: str, active_value: Any, inactive_value: Any, note_field: Optional[str] = "notes", ) -> Dict[str, Any]: wanted = [int(item) for item in ids] result = { "candidate_count": len(wanted), "found_count": 0, "would_deactivate": 0, "deactivated": 0, "already_inactive": 0, "missing_ids": [], } if not wanted: return result rows = list(session.scalars(select(model).where(model.id.in_(wanted))).all()) found_ids = {int(row.id) for row in rows} result["found_count"] = len(rows) result["missing_ids"] = [item for item in wanted if item not in found_ids] for row in rows: current = getattr(row, status_field) if not _status_matches(current, active_value): result["already_inactive"] += 1 continue if dry_run: result["would_deactivate"] += 1 continue setattr(row, status_field, inactive_value) if note_field and hasattr(row, note_field): setattr(row, note_field, _append_note(getattr(row, note_field), note)) result["deactivated"] += 1 return result def _deactivate_collection_landing_pages_for_paths( session: Session, path_slugs: Sequence[str], *, dry_run: bool, note: str, ) -> Dict[str, Any]: slugs = sorted({slug for slug in path_slugs if slug}) result = {"candidate_count": len(slugs), "found_count": 0, "would_deactivate": 0, "deactivated": 0} if not slugs: return result rows = list( session.scalars( select(CollectionLandingPage) .where(CollectionLandingPage.path_slug.in_(slugs)) .where(CollectionLandingPage.is_active.is_(True)) ).all() ) result["found_count"] = len(rows) for row in rows: if dry_run: result["would_deactivate"] += 1 continue row.is_active = False row.notes = _append_note(row.notes, note) result["deactivated"] += 1 return result def _db_inactive_taxonomy_node_rows( session: Session, *, protected_slugs: Set[str], path_prefixes: Sequence[str], ) -> list[Dict[str, str]]: stmt = select(MasterTaxonomyNode).where(MasterTaxonomyNode.is_active.is_(False)).order_by(MasterTaxonomyNode.path_slug) rows = list(session.scalars(stmt).all()) out: list[Dict[str, str]] = [] for row in rows: slug = _norm_slug(row.path_slug) if slug in protected_slugs: continue if path_prefixes and not any(slug == prefix or slug.startswith(f"{prefix}/") for prefix in path_prefixes): continue out.append({"id": str(row.id), "node_id": str(row.id), "path_slug": row.path_slug or ""}) return out def _db_inactive_taxonomy_link_rows( session: Session, *, protected_slugs: Set[str], path_prefixes: Sequence[str], ) -> list[Dict[str, str]]: node_rows = _db_inactive_taxonomy_node_rows( session, protected_slugs=protected_slugs, path_prefixes=path_prefixes, ) node_ids = _taxonomy_node_ids(node_rows) if not node_ids: return [] rows = list( session.scalars( select(MasterTaxonomyChannelLink).where(MasterTaxonomyChannelLink.taxonomy_node_id.in_(node_ids)) ).all() ) return [ { "id": str(row.id), "taxonomy_node_id": str(row.taxonomy_node_id), "path_slug": next( ( node_row.get("path_slug", "") for node_row in node_rows if str(node_row.get("node_id") or node_row.get("id") or "") == str(row.taxonomy_node_id) ), "", ), } for row in rows ] def _delete_inactive_taxonomy_nodes( session: Session, rows: Sequence[Dict[str, str]], *, dry_run: bool, planned_link_ids: Sequence[int] = (), ) -> Dict[str, Any]: node_ids = _taxonomy_node_ids(rows) result: Dict[str, Any] = { "candidate_count": len(node_ids), "found_count": 0, "missing_ids": [], "skipped_active": 0, "blocked_by_children": 0, "blocked_by_active_links": 0, "link_rows_found": 0, "alias_rows_found": 0, "would_delete_links": 0, "deleted_links": 0, "would_delete_aliases": 0, "deleted_aliases": 0, "would_delete_nodes": 0, "deleted_nodes": 0, "blocked": [], } if not node_ids: return result node_rows = list(session.scalars(select(MasterTaxonomyNode).where(MasterTaxonomyNode.id.in_(node_ids))).all()) found_ids = {int(row.id) for row in node_rows} result["found_count"] = len(node_rows) result["missing_ids"] = [item for item in node_ids if item not in found_ids] child_rows = list( session.scalars(select(MasterTaxonomyNode).where(MasterTaxonomyNode.parent_id.in_(node_ids))).all() ) children_by_parent: Dict[int, list[MasterTaxonomyNode]] = {} for child in child_rows: if child.parent_id is None: continue children_by_parent.setdefault(int(child.parent_id), []).append(child) link_rows = list( session.scalars( select(MasterTaxonomyChannelLink).where(MasterTaxonomyChannelLink.taxonomy_node_id.in_(node_ids)) ).all() ) links_by_node: Dict[int, list[MasterTaxonomyChannelLink]] = {} for row in link_rows: links_by_node.setdefault(int(row.taxonomy_node_id), []).append(row) result["link_rows_found"] = len(link_rows) alias_rows = list( session.scalars(select(MasterTaxonomyPathAlias).where(MasterTaxonomyPathAlias.canonical_node_id.in_(node_ids))).all() ) aliases_by_node: Dict[int, list[MasterTaxonomyPathAlias]] = {} for row in alias_rows: aliases_by_node.setdefault(int(row.canonical_node_id), []).append(row) result["alias_rows_found"] = len(alias_rows) delete_set = set(node_ids) planned_link_id_set = {int(item) for item in planned_link_ids} deletable_nodes: list[MasterTaxonomyNode] = [] for node in sorted(node_rows, key=lambda item: (str(item.path_slug or "").count("/"), int(item.id)), reverse=True): if node.is_active: result["skipped_active"] += 1 result["blocked"].append({"node_id": int(node.id), "path_slug": node.path_slug, "reason": "active"}) continue blocking_children = sorted( int(child.id) for child in children_by_parent.get(int(node.id), []) if int(child.id) not in delete_set ) if blocking_children: result["blocked_by_children"] += 1 result["blocked"].append( { "node_id": int(node.id), "path_slug": node.path_slug, "reason": "child_rows", "child_ids": blocking_children, } ) continue blocking_links = sorted( int(link.id) for link in links_by_node.get(int(node.id), []) if link.is_active and int(link.id) not in planned_link_id_set ) if blocking_links: result["blocked_by_active_links"] += 1 result["blocked"].append( { "node_id": int(node.id), "path_slug": node.path_slug, "reason": "active_links", "link_ids": blocking_links, } ) continue deletable_nodes.append(node) if dry_run: result["would_delete_nodes"] = len(deletable_nodes) result["would_delete_links"] = sum(len(links_by_node.get(int(node.id), [])) for node in deletable_nodes) result["would_delete_aliases"] = sum(len(aliases_by_node.get(int(node.id), [])) for node in deletable_nodes) return result for node in deletable_nodes: for link in links_by_node.get(int(node.id), []): session.delete(link) result["deleted_links"] += 1 for alias in aliases_by_node.get(int(node.id), []): session.delete(alias) result["deleted_aliases"] += 1 if deletable_nodes: session.flush() # Flush each node delete leaf-first so Postgres never sees a parent row # removed before its descendant rows are gone. for node in sorted(deletable_nodes, key=lambda item: (str(item.path_slug or "").count("/"), int(item.id)), reverse=True): session.delete(node) result["deleted_nodes"] += 1 session.flush() return result def _merge_duplicate_collections( session: Session, rows: Sequence[Dict[str, str]], *, dry_run: bool, note: str, ) -> Dict[str, Any]: groups: Dict[str, list[Dict[str, str]]] = {} for row in rows: key = str(row.get("path_slug") or "").strip().lower() if key: groups.setdefault(key, []).append(row) result = { "group_count": 0, "duplicate_rows": 0, "would_update_products": 0, "updated_products": 0, "would_update_taxonomy_nodes": 0, "updated_taxonomy_nodes": 0, "would_update_landing_pages": 0, "updated_landing_pages": 0, "would_deactivate_registry_rows": 0, "deactivated_registry_rows": 0, "groups": [], } for path_slug, group_rows in sorted(groups.items()): ids = [int(row["id"]) for row in group_rows if str(row.get("id") or "").isdigit()] if len(ids) <= 1: continue keep_id = _choose_collection_keeper(group_rows) duplicate_ids = [item for item in ids if item != keep_id] result["group_count"] += 1 result["duplicate_rows"] += len(duplicate_ids) result["groups"].append({"path_slug": path_slug, "keep_id": keep_id, "deactivate_ids": duplicate_ids}) product_rows = list( session.scalars(select(MasterProduct).where(MasterProduct.collection_registry_id.in_(duplicate_ids))).all() ) taxonomy_rows = list( session.scalars( select(MasterTaxonomyNode).where(MasterTaxonomyNode.collection_registry_id.in_(duplicate_ids)) ).all() ) landing_rows = list( session.scalars( select(CollectionLandingPage).where(CollectionLandingPage.collection_registry_id.in_(duplicate_ids)) ).all() ) duplicate_registry_rows = list( session.scalars( select(MasterCollectionRegistry) .where(MasterCollectionRegistry.id.in_(duplicate_ids)) .where(MasterCollectionRegistry.is_active.is_(True)) ).all() ) result["would_update_products"] += len(product_rows) result["would_update_taxonomy_nodes"] += len(taxonomy_rows) result["would_update_landing_pages"] += len(landing_rows) result["would_deactivate_registry_rows"] += len(duplicate_registry_rows) if dry_run: continue for row in product_rows: row.collection_registry_id = keep_id result["updated_products"] += 1 for row in taxonomy_rows: row.collection_registry_id = keep_id result["updated_taxonomy_nodes"] += 1 for row in landing_rows: row.collection_registry_id = keep_id result["updated_landing_pages"] += 1 for row in duplicate_registry_rows: row.is_active = False row.notes = _append_note(row.notes, f"{note}; merged into master_collection_registry.id={keep_id}") result["deactivated_registry_rows"] += 1 return result def _choose_collection_keeper(rows: Sequence[Dict[str, str]]) -> int: def score(row: Dict[str, str]) -> tuple[int, int, int]: row_id = int(row["id"]) has_brand = 1 if str(row.get("brand_id") or "").strip().isdigit() else 0 simple_code = 1 if "/" not in str(row.get("code") or "") and not str(row.get("code") or "").startswith("kitchen-cabinets-") and not str(row.get("code") or "").startswith("bathroom-vanities-") else 0 return (has_brand, simple_code, -row_id) return int(max(rows, key=score)["id"]) def _read_csv(path: Path) -> list[Dict[str, str]]: if not path.exists(): return [] with path.open("r", encoding="utf-8-sig", newline="") as handle: return [dict(row) for row in csv.DictReader(handle)] def _ids(rows: Sequence[Dict[str, str]]) -> list[int]: out: list[int] = [] for row in rows: value = str(row.get("id") or "").strip() if value.isdigit(): out.append(int(value)) return out def _taxonomy_node_ids(rows: Sequence[Dict[str, str]]) -> list[int]: out: list[int] = [] seen: set[int] = set() for row in rows: for key in ("taxonomy_node_id", "node_id", "id"): value = str(row.get(key) or "").strip() if value.isdigit(): node_id = int(value) if node_id not in seen: seen.add(node_id) out.append(node_id) break return out def _filter_protected_paths(rows: Sequence[Dict[str, str]], protected_slugs: Set[str]) -> list[Dict[str, str]]: out: list[Dict[str, str]] = [] for row in rows: slug = _norm_slug(row.get("path_slug") or row.get("path_slug_path")) if slug in protected_slugs: continue out.append(row) return out def _is_structural_remote_row(row: Dict[str, str], structural_labels: Set[str]) -> bool: if not structural_labels: return False labels = { str(row.get("remote_path") or "").strip().lower(), str(row.get("name") or "").strip().lower(), str(row.get("title") or "").strip().lower(), } return bool(labels & structural_labels) def _status_matches(current: Any, expected: Any) -> bool: if isinstance(expected, bool): return bool(current) is expected return str(current or "").strip().lower() == str(expected).strip().lower() def _append_note(existing: Optional[str], note: str) -> str: current = str(existing or "").strip() return f"{current}\n{note}".strip() if current else note def _norm_slug(value: Any) -> str: return str(value or "").strip().lower().strip("/") def main() -> int: parser = argparse.ArgumentParser(description="Clean taxonomy pollution rows identified by audit CSVs") parser.add_argument("--audit-dir", default=r"E:\2026\plytixmage_dbs_data\audit_reports") parser.add_argument("--apply", action="store_true", help="Apply changes. Default is dry-run.") parser.add_argument("--skip-listing-paths", action="store_true") parser.add_argument("--skip-listing-assignments", action="store_true") parser.add_argument("--skip-listing-targets", action="store_true") parser.add_argument("--skip-taxonomy-assignments", action="store_true") parser.add_argument("--skip-inactive-node-links", action="store_true") parser.add_argument("--skip-inactive-taxonomy-nodes", action="store_true") parser.add_argument( "--db-inactive-taxonomy-nodes", action="store_true", help="Use inactive master_taxonomy_node rows from the database directly instead of audit CSV files.", ) parser.add_argument( "--inactive-node-path-prefix", action="append", default=None, help="Limit DB inactive taxonomy node cleanup to this path prefix. Repeatable.", ) parser.add_argument( "--mark-magento-registry-inactive", action="store_true", help="Also mark bad Magento registry rows inactive locally. Does not delete remote Magento categories.", ) parser.add_argument( "--merge-duplicate-collections", action="store_true", help="Merge duplicate master_collection_registry rows by path_slug and deactivate duplicate registry rows.", ) parser.add_argument( "--include-structural-remote-labels", action="store_true", help="Also process rows with remote/name/title Accessories or Doors. Default skips them for review.", ) parser.add_argument( "--protect-slug", action="append", default=None, help="Listing path slug to protect from cleanup. Repeatable. Default protects doors.", ) args = parser.parse_args() with get_session() as session: try: summary = cleanup_taxonomy_pollution_from_audit( session, audit_dir=args.audit_dir, dry_run=not args.apply, deactivate_listing_paths=not args.skip_listing_paths, deactivate_listing_assignments=not args.skip_listing_assignments, deactivate_listing_targets=not args.skip_listing_targets, deactivate_taxonomy_assignments=not args.skip_taxonomy_assignments, deactivate_inactive_node_links=not args.skip_inactive_node_links, delete_inactive_taxonomy_nodes=not args.skip_inactive_taxonomy_nodes, db_inactive_taxonomy_nodes=args.db_inactive_taxonomy_nodes, inactive_node_path_prefixes=args.inactive_node_path_prefix, mark_magento_registry_inactive=args.mark_magento_registry_inactive, merge_duplicate_collections=args.merge_duplicate_collections, include_structural_remote_labels=args.include_structural_remote_labels, protected_slugs=args.protect_slug or DEFAULT_PROTECTED_SLUGS, ) if not args.apply: session.rollback() except Exception: session.rollback() raise print(json.dumps(summary, indent=2, default=str)) return 0 if summary.get("status") == "ok" else 1 if __name__ == "__main__": sys.exit(main())