r"""Remap references away from inactive master collection registry rows, then delete them. Default mode is dry-run and prints a summary only. Example: python -m app.jobs.cleanup_inactive_collection_registry python -m app.jobs.cleanup_inactive_collection_registry --dump-path E:\2026\plytixmage_dbs_data\master_collection_registry.csv python -m app.jobs.cleanup_inactive_collection_registry --dump-path E:\2026\plytixmage_dbs_data\master_collection_registry.csv --apply """ from __future__ import annotations import argparse import csv import json from collections import defaultdict from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, Iterable, Optional from sqlalchemy import select from sqlalchemy.orm import Session from db.models import ( BrochureTaxonomyGroup, BrochureTaxonomyGroupSku, ChannelShellProduct, CollectionCommerceProfile, CollectionLandingPage, ManualAttributeLockRule, ManualTaxonomyAssignment, ManualTaxonomyAssignmentCollection, MasterCollectionRegistry, MasterProduct, MasterProductCollectionMembership, MasterTaxonomyNode, ) from db.master_taxonomy_registry import alias_codes_from_json from db.session import get_session @dataclass(frozen=True) class RegistryDumpRow: id: int code: str name: str path_slug: str brand_id: Optional[int] family_id: Optional[int] aliases: dict[str, Any] is_active: bool notes: str def cleanup_inactive_collection_registry( session: Session, *, dump_path: Optional[str | Path] = None, dry_run: bool = True, note: Optional[str] = None, ) -> dict[str, Any]: rows = _load_dump_rows(dump_path) if dump_path else _load_db_rows(session) active_rows = [row for row in rows if row.is_active] inactive_rows = [row for row in rows if not row.is_active] plan = _build_remap_plan(active_rows, inactive_rows) cleanup_note = note or f"inactive collection registry cleanup from dump at {datetime.now(timezone.utc).isoformat()}" summary: dict[str, Any] = { "status": "ok", "dry_run": dry_run, "source": "dump" if dump_path else "database", "dump_path": str(dump_path) if dump_path else None, "active_rows": len(active_rows), "inactive_rows": len(inactive_rows), "mapped_inactive_rows": sum(1 for item in plan.values() if item["target_id"] is not None), "unmapped_inactive_rows": sum(1 for item in plan.values() if item["target_id"] is None), "missing_registry_rows": [], "steps": {}, } inactive_ids = sorted(plan.keys()) db_rows: list[MasterCollectionRegistry] = [] if inactive_ids: db_rows = list( session.scalars(select(MasterCollectionRegistry).where(MasterCollectionRegistry.id.in_(inactive_ids))).all() ) db_ids = {int(row.id) for row in db_rows} summary["missing_registry_rows"] = [item for item in inactive_ids if item not in db_ids] summary["steps"]["master_product"] = _rewrite_fk_rows( session, MasterProduct, plan, dry_run=dry_run, note=cleanup_note, ) summary["steps"]["collection_landing_page"] = _rewrite_fk_rows( session, CollectionLandingPage, plan, dry_run=dry_run, note=cleanup_note, after_update=_sync_landing_page_fields, ) summary["steps"]["collection_commerce_profile"] = _rewrite_fk_rows( session, CollectionCommerceProfile, plan, dry_run=dry_run, note=cleanup_note, dedupe_key=_commerce_profile_key, ) summary["steps"]["brochure_taxonomy_group"] = _rewrite_fk_rows( session, BrochureTaxonomyGroup, plan, dry_run=dry_run, note=cleanup_note, after_update=_sync_brochure_group_fields, dedupe_key=_brochure_group_key, merge_duplicate=_merge_brochure_group_duplicate, ) summary["steps"]["master_taxonomy_node"] = _rewrite_fk_rows( session, MasterTaxonomyNode, plan, dry_run=dry_run, note=cleanup_note, after_update=_sync_taxonomy_node_fields, ) summary["steps"]["channel_shell_product"] = _rewrite_fk_rows( session, ChannelShellProduct, plan, dry_run=dry_run, note=cleanup_note, ) summary["steps"]["master_product_collection_membership"] = _rewrite_fk_rows( session, MasterProductCollectionMembership, plan, dry_run=dry_run, note=cleanup_note, after_update=_sync_membership_fields, ) summary["steps"]["manual_attribute_lock_rule"] = _rewrite_fk_rows( session, ManualAttributeLockRule, plan, dry_run=dry_run, note=cleanup_note, after_update=_sync_lock_rule_fields, dedupe_key=_manual_attribute_lock_key, ) summary["steps"]["manual_taxonomy_assignment_collection"] = _rewrite_assignment_collections( session, plan, dry_run=dry_run, note=cleanup_note, ) summary["steps"]["manual_taxonomy_assignment_normalization"] = _normalize_manual_taxonomy_assignments( session, dry_run=dry_run, note=cleanup_note, ) summary["steps"]["master_collection_registry"] = _delete_inactive_registry_rows( session, db_rows, dry_run=dry_run, ) if not dry_run: session.flush() return summary def _rewrite_fk_rows( session: Session, model: Any, plan: dict[int, dict[str, Any]], *, dry_run: bool, note: str, after_update: Optional[Any] = None, dedupe_key: Optional[Any] = None, merge_duplicate: Optional[Any] = None, ) -> dict[str, Any]: source_ids = sorted(plan.keys()) result = { "candidate_rows": 0, "would_update": 0, "updated": 0, "would_clear": 0, "cleared": 0, "would_delete_duplicates": 0, "deleted_duplicates": 0, } if not source_ids: return result target_ids = sorted({int(item["target_id"]) for item in plan.values() if item.get("target_id") is not None}) rows = list(session.scalars(select(model).where(model.collection_registry_id.in_(source_ids))).all()) result["candidate_rows"] = len(rows) survivors: dict[Any, Any] = {} pending_updates: list[tuple[Any, Optional[dict[str, Any]], bool]] = [] if dedupe_key is not None and target_ids: existing_rows = list( session.scalars( select(model) .where(model.collection_registry_id.in_(target_ids)) .where(~model.collection_registry_id.in_(source_ids)) ).all() ) for row in existing_rows: key = dedupe_key( row, { "target_id": row.collection_registry_id, "target_code": getattr(row, "collection_code", None), }, ) if key is not None and key not in survivors: survivors[key] = row for row in rows: source_id = int(row.collection_registry_id) target = plan.get(source_id) target_id = target.get("target_id") if target else None will_clear = target_id is None if dedupe_key is not None: key = dedupe_key(row, target) if key is not None: existing = survivors.get(key) if existing is None: survivors[key] = row else: if dry_run: result["would_delete_duplicates"] += 1 else: if merge_duplicate is not None: merge_duplicate(session, existing, row, note) _merge_notes(existing, row, note) session.delete(row) result["deleted_duplicates"] += 1 continue if dry_run: if will_clear: result["would_clear"] += 1 else: result["would_update"] += 1 continue pending_updates.append((row, target, will_clear)) if not dry_run and result["deleted_duplicates"] > 0: session.flush() for row, target, will_clear in pending_updates: row.collection_registry_id = target.get("target_id") if target else None if after_update is not None: after_update(row, target) if hasattr(row, "notes"): row.notes = _append_note(getattr(row, "notes"), note) if will_clear: result["cleared"] += 1 else: result["updated"] += 1 return result def _rewrite_assignment_collections( session: Session, plan: dict[int, dict[str, Any]], *, dry_run: bool, note: str, ) -> dict[str, Any]: source_ids = sorted(plan.keys()) result = { "candidate_rows": 0, "would_update": 0, "updated": 0, "would_delete_orphans": 0, "deleted_orphans": 0, "would_delete_duplicates": 0, "deleted_duplicates": 0, } if not source_ids: return result rows = list( session.scalars( select(ManualTaxonomyAssignmentCollection).where( ManualTaxonomyAssignmentCollection.collection_registry_id.in_(source_ids) ) ).all() ) result["candidate_rows"] = len(rows) survivors: dict[tuple[int, str], ManualTaxonomyAssignmentCollection] = {} target_codes = sorted({str(item["target_code"]) for item in plan.values() if item.get("target_code")}) pending_updates: list[tuple[ManualTaxonomyAssignmentCollection, dict[str, Any]]] = [] if target_codes: existing_rows = list( session.scalars( select(ManualTaxonomyAssignmentCollection).where( ManualTaxonomyAssignmentCollection.collection_code.in_(target_codes) ) ).all() ) for row in existing_rows: if row.collection_registry_id in source_ids: continue survivors[(int(row.assignment_id), str(row.collection_code))] = row for row in rows: source_id = int(row.collection_registry_id) target = plan.get(source_id) target_id = target.get("target_id") if target else None if target_id is None: if dry_run: result["would_delete_orphans"] += 1 else: session.delete(row) result["deleted_orphans"] += 1 continue key = (int(row.assignment_id), str(target["target_code"])) existing = survivors.get(key) if existing is None: survivors[key] = row else: if dry_run: result["would_delete_duplicates"] += 1 else: _merge_notes(existing, row, note) session.delete(row) result["deleted_duplicates"] += 1 continue if dry_run: result["would_update"] += 1 continue pending_updates.append((row, target)) if not dry_run and (result["deleted_orphans"] > 0 or result["deleted_duplicates"] > 0): session.flush() for row, target in pending_updates: row.collection_registry_id = target["target_id"] row.collection_code = target["target_code"] row.source_collection_code = target["target_source_code"] or target["target_code"] row.collection_name = target["target_name"] row.collection_path_slug = target["target_path_slug"] row.shopping_collection = target["target_name"] if not str(row.sku_prefix or "").strip() or str(row.sku_prefix).strip() == target["source_code"]: row.sku_prefix = target["target_code"] row.notes = _append_note(row.notes, note) result["updated"] += 1 return result def _delete_inactive_registry_rows( session: Session, rows: Iterable[MasterCollectionRegistry], *, dry_run: bool, ) -> dict[str, Any]: all_rows = list(rows) result = { "candidate_rows": len(all_rows), "would_delete": len(all_rows) if dry_run else 0, "deleted": 0, } if dry_run: return result for row in all_rows: session.delete(row) result["deleted"] += 1 return result def _normalize_manual_taxonomy_assignments( session: Session, *, dry_run: bool, note: str, ) -> dict[str, Any]: lookup = _active_collection_lookup(session) assignments = list( session.scalars( select(ManualTaxonomyAssignment) .where(ManualTaxonomyAssignment.assignment_status == "active") .order_by(ManualTaxonomyAssignment.id) ).all() ) if not assignments: return { "assignment_count": 0, "candidate_child_rows": 0, "would_update_child_rows": 0, "updated_child_rows": 0, "would_delete_duplicate_child_rows": 0, "deleted_duplicate_child_rows": 0, "would_update_payloads": 0, "updated_payloads": 0, } assignment_ids = [int(row.id) for row in assignments] children = list( session.scalars( select(ManualTaxonomyAssignmentCollection) .where(ManualTaxonomyAssignmentCollection.assignment_id.in_(assignment_ids)) .where(ManualTaxonomyAssignmentCollection.is_active.is_(True)) .order_by( ManualTaxonomyAssignmentCollection.assignment_id, ManualTaxonomyAssignmentCollection.sort_order, ManualTaxonomyAssignmentCollection.id, ) ).all() ) by_assignment: dict[int, list[ManualTaxonomyAssignmentCollection]] = defaultdict(list) for child in children: by_assignment[int(child.assignment_id)].append(child) result = { "assignment_count": len(assignments), "candidate_child_rows": len(children), "would_update_child_rows": 0, "updated_child_rows": 0, "would_delete_duplicate_child_rows": 0, "deleted_duplicate_child_rows": 0, "would_update_payloads": 0, "updated_payloads": 0, } for assignment in assignments: assignment_children = by_assignment.get(int(assignment.id), []) normalized_children: list[ManualTaxonomyAssignmentCollection] = [] seen_codes: set[str] = set() for child in assignment_children: target = _resolve_active_collection_target( lookup, collection_registry_id=child.collection_registry_id, collection_name=child.collection_name, collection_code=child.collection_code, source_collection_code=child.source_collection_code, sku_prefix=child.sku_prefix, ) if target is None: normalized_children.append(child) continue target_code = str(target["code"] or "").strip().upper() existing_suffix = _manual_assignment_suffix(str(child.master_sku or "").strip(), str(assignment.source_sku or "").strip()) normalized_master_sku = f"{target['sku_prefix']}-{existing_suffix}".strip("-").upper() should_update = any( [ child.collection_registry_id != target["id"], str(child.collection_code or "").strip().upper() != target_code, str(child.source_collection_code or "").strip().upper() != str(target["source_collection_code"] or target_code).strip().upper(), str(child.sku_prefix or "").strip().upper() != str(target["sku_prefix"] or target_code).strip().upper(), str(child.collection_name or "").strip() != str(target["name"] or "").strip(), str(child.collection_path_slug or "").strip() != str(target["path_slug"] or "").strip(), str(child.master_sku or "").strip().upper() != normalized_master_sku, (child.alias_codes if isinstance(child.alias_codes, list) else []) != target["alias_codes"], ] ) duplicate_key = target_code if duplicate_key in seen_codes: if dry_run: result["would_delete_duplicate_child_rows"] += 1 else: session.delete(child) result["deleted_duplicate_child_rows"] += 1 continue seen_codes.add(duplicate_key) if should_update: if dry_run: result["would_update_child_rows"] += 1 else: child.collection_registry_id = target["id"] child.collection_code = target_code child.source_collection_code = target["source_collection_code"] or target_code child.sku_prefix = target["sku_prefix"] or target_code child.collection_name = target["name"] or None child.collection_path_slug = target["path_slug"] or None child.alias_codes = target["alias_codes"] or None child.master_sku = normalized_master_sku child.notes = _append_note(child.notes, note) result["updated_child_rows"] += 1 normalized_children.append(child) if not dry_run and result["deleted_duplicate_child_rows"] > 0: session.flush() rebuilt_payload = _rebuild_manual_assignment_payload(assignment, normalized_children) current_payload = assignment.raw_payload if isinstance(assignment.raw_payload, dict) else {} if rebuilt_payload != current_payload: if dry_run: result["would_update_payloads"] += 1 else: assignment.raw_payload = rebuilt_payload result["updated_payloads"] += 1 return result def _rebuild_manual_assignment_payload( assignment: ManualTaxonomyAssignment, children: list[ManualTaxonomyAssignmentCollection], ) -> dict[str, Any]: payload = dict(assignment.raw_payload) if isinstance(assignment.raw_payload, dict) else {} payload["source_sku"] = str(assignment.source_sku or "").strip().upper() payload["canonical_taxonomy_path_slug"] = str(assignment.canonical_taxonomy_path_slug or "").strip() payload["canonical_taxonomy_path_label"] = str(assignment.canonical_taxonomy_path_label or "").strip() or None payload["shopping_l1"] = str(assignment.shopping_l1 or "").strip() or None payload["shopping_l2"] = str(assignment.shopping_l2 or "").strip() or None payload["l2_category"] = str(assignment.l2_category or "").strip() or None payload["l3_category"] = str(assignment.l3_category or "").strip() or None payload["collections"] = [ { "master_sku": str(child.master_sku or "").strip().upper(), "sku_prefix": str(child.sku_prefix or "").strip().upper(), "alias_codes": child.alias_codes if isinstance(child.alias_codes, list) else [], "collection_code": str(child.collection_code or "").strip().upper(), "source_collection_code": str(child.source_collection_code or "").strip().upper() or None, "collection_name": str(child.collection_name or "").strip(), "collection_path_slug": str(child.collection_path_slug or "").strip() or None, "shopping_collection": str(child.shopping_collection or "").strip() or None, } for child in sorted(children, key=lambda item: (int(item.sort_order or 0), int(item.id or 0))) if child.is_active ] return payload def _manual_assignment_suffix(master_sku: str, source_sku: str) -> str: normalized_master = str(master_sku or "").strip().upper() normalized_source = str(source_sku or "").strip().upper() if "-" in normalized_master: return normalized_master.split("-", 1)[1].strip().upper() return normalized_source def _active_collection_lookup(session: Session) -> dict[str, Any]: rows = list( session.scalars( select(MasterCollectionRegistry) .where(MasterCollectionRegistry.is_active.is_(True)) .order_by(MasterCollectionRegistry.id) ).all() ) by_id: dict[int, dict[str, Any]] = {} by_name: dict[str, dict[str, Any]] = {} by_code: dict[str, dict[str, Any]] = {} for row in rows: source_code = ( str((row.aliases or {}).get("source_collection_code") or "").strip().upper() if isinstance(row.aliases, dict) else "" ) alias_codes = [str(code or "").strip().upper() for code in alias_codes_from_json(row.aliases) if str(code or "").strip()] payload = { "id": int(row.id), "code": str(row.code or "").strip().upper(), "name": str(row.name or "").strip(), "path_slug": str(row.path_slug or "").strip(), "source_collection_code": source_code, "sku_prefix": source_code or str(row.code or "").strip().upper(), "alias_codes": alias_codes, } by_id[int(row.id)] = payload if payload["name"]: by_name.setdefault(payload["name"].strip().lower(), payload) for key in {payload["code"], payload["source_collection_code"], payload["sku_prefix"], *alias_codes}: key_text = str(key or "").strip().upper() if key_text: by_code.setdefault(key_text, payload) return {"by_id": by_id, "by_name": by_name, "by_code": by_code} def _resolve_active_collection_target( lookup: dict[str, Any], *, collection_registry_id: Optional[int] = None, collection_name: Optional[str] = None, collection_code: Optional[str] = None, source_collection_code: Optional[str] = None, sku_prefix: Optional[str] = None, ) -> Optional[dict[str, Any]]: if collection_registry_id is not None: target = lookup["by_id"].get(int(collection_registry_id)) if target is not None: return target name_key = str(collection_name or "").strip().lower() if name_key: target = lookup["by_name"].get(name_key) if target is not None: return target for key in (collection_code, source_collection_code, sku_prefix): code_key = str(key or "").strip().upper() if code_key: target = lookup["by_code"].get(code_key) if target is not None: return target return None def _sync_landing_page_fields(row: CollectionLandingPage, target: dict[str, Any]) -> None: if not target or target.get("target_id") is None: return row.collection = target["target_name"] def _sync_brochure_group_fields(row: BrochureTaxonomyGroup, target: dict[str, Any]) -> None: if not target: return row.brochure_collection_code = target["target_source_code"] or target["target_code"] if target.get("target_id") else None def _merge_brochure_group_duplicate( session: Session, keep_row: BrochureTaxonomyGroup, drop_row: BrochureTaxonomyGroup, note: str, ) -> None: keep_children = { str(row.brochure_sku or "").strip().upper(): row for row in session.scalars( select(BrochureTaxonomyGroupSku).where(BrochureTaxonomyGroupSku.group_id == int(keep_row.id)) ).all() } drop_children = list( session.scalars( select(BrochureTaxonomyGroupSku).where(BrochureTaxonomyGroupSku.group_id == int(drop_row.id)) ).all() ) for child in drop_children: key = str(child.brochure_sku or "").strip().upper() existing = keep_children.get(key) if existing is not None: _merge_notes(existing, child, note) session.delete(child) continue child.group_id = int(keep_row.id) child.notes = _append_note(child.notes, note) keep_children[key] = child session.flush() def _sync_taxonomy_node_fields(row: MasterTaxonomyNode, target: dict[str, Any]) -> None: if not target or target.get("target_id") is None: return if row.node_kind == "collection": row.code = target["target_code"] row.name = target["target_name"] def _sync_membership_fields(row: MasterProductCollectionMembership, target: dict[str, Any]) -> None: if not target or target.get("target_id") is None: row.collection_name = None return row.collection_name = target["target_name"] def _sync_lock_rule_fields(row: ManualAttributeLockRule, target: dict[str, Any]) -> None: if not target or target.get("target_id") is None: row.collection_code = None row.collection_name = None return row.collection_code = target["target_code"] row.collection_name = target["target_name"] def _commerce_profile_key(row: CollectionCommerceProfile, target: dict[str, Any]) -> Optional[tuple[Any, ...]]: return ( target.get("target_id"), str(row.channel_code or ""), int(row.connection_id) if row.connection_id is not None else None, ) def _brochure_group_key(row: BrochureTaxonomyGroup, target: dict[str, Any]) -> Optional[tuple[Any, ...]]: return ( target.get("target_id"), str(row.category_l1 or ""), str(row.brochure_l1_label or ""), str(row.brochure_l2_label or ""), str(row.brochure_leaf_label or ""), ) def _manual_attribute_lock_key(row: ManualAttributeLockRule, target: dict[str, Any]) -> Optional[tuple[Any, ...]]: return ( str(row.canonical_taxonomy_path_slug or ""), str(row.path_match_kind or ""), str((target or {}).get("target_code") or ""), ) def _merge_notes(target_row: Any, source_row: Any, note: str) -> None: extra = f"merged duplicate row id={getattr(source_row, 'id', '?')}" target_row.notes = _append_note(getattr(target_row, "notes", None), f"{note}; {extra}") def _append_note(existing: Optional[str], addition: str) -> str: base = str(existing or "").strip() extra = str(addition or "").strip() if not base: return extra if not extra: return base return f"{base}\n{extra}" def _build_remap_plan( active_rows: Iterable[RegistryDumpRow], inactive_rows: Iterable[RegistryDumpRow], ) -> dict[int, dict[str, Any]]: by_slug: dict[str, list[RegistryDumpRow]] = defaultdict(list) by_name: dict[str, list[RegistryDumpRow]] = defaultdict(list) by_code: dict[str, list[RegistryDumpRow]] = defaultdict(list) by_source_code: dict[str, list[RegistryDumpRow]] = defaultdict(list) active_rows = list(active_rows) for row in active_rows: if row.path_slug: by_slug[_norm(row.path_slug)].append(row) if row.name: by_name[_norm(row.name)].append(row) if row.code: by_code[_norm(row.code)].append(row) source_code = _norm(row.aliases.get("source_collection_code")) if source_code: by_source_code[source_code].append(row) plan: dict[int, dict[str, Any]] = {} for row in inactive_rows: target: Optional[RegistryDumpRow] = None reason: Optional[str] = None if _norm(row.path_slug) in by_slug: target = by_slug[_norm(row.path_slug)][0] reason = "path_slug" elif _norm(row.name) in by_name: target = by_name[_norm(row.name)][0] reason = "name" elif _norm(row.code) in by_code: target = by_code[_norm(row.code)][0] reason = "code" elif _norm(row.code) in by_source_code: target = by_source_code[_norm(row.code)][0] reason = "source_collection_code" plan[row.id] = { "source_id": row.id, "source_code": row.code, "source_name": row.name, "source_path_slug": row.path_slug, "target_id": target.id if target else None, "target_code": target.code if target else None, "target_name": target.name if target else None, "target_path_slug": target.path_slug if target else None, "target_source_code": target.aliases.get("source_collection_code") if target else None, "match_reason": reason, } return plan def _load_dump_rows(path: str | Path) -> list[RegistryDumpRow]: dump_path = Path(path) with dump_path.open("r", encoding="utf-8-sig", newline="") as handle: rows = [] for raw in csv.DictReader(handle): rows.append( RegistryDumpRow( id=int(raw["id"]), code=str(raw.get("code") or "").strip(), name=str(raw.get("name") or "").strip(), path_slug=str(raw.get("path_slug") or "").strip(), brand_id=_parse_int(raw.get("brand_id")), family_id=_parse_int(raw.get("family_id")), aliases=_parse_json(raw.get("aliases")), is_active=_parse_bool(raw.get("is_active")), notes=str(raw.get("notes") or "").strip(), ) ) return rows def _load_db_rows(session: Session) -> list[RegistryDumpRow]: rows = list(session.scalars(select(MasterCollectionRegistry).order_by(MasterCollectionRegistry.id)).all()) return [ RegistryDumpRow( id=int(row.id), code=str(row.code or "").strip(), name=str(row.name or "").strip(), path_slug=str(row.path_slug or "").strip(), brand_id=int(row.brand_id) if row.brand_id is not None else None, family_id=int(row.family_id) if row.family_id is not None else None, aliases=row.aliases if isinstance(row.aliases, dict) else {}, is_active=bool(row.is_active), notes=str(row.notes or "").strip(), ) for row in rows ] def _parse_bool(value: Any) -> bool: return str(value or "").strip().lower() in {"1", "true", "t", "yes", "y"} def _parse_int(value: Any) -> Optional[int]: text = str(value or "").strip() if not text or text == r"\N": return None return int(text) if text.isdigit() else None def _parse_json(value: Any) -> dict[str, Any]: text = str(value or "").strip() if not text or text.lower() == "null": return {} try: data = json.loads(text) except Exception: return {} return data if isinstance(data, dict) else {} def _norm(value: Any) -> str: return str(value or "").strip().lower() def main() -> int: parser = argparse.ArgumentParser(description="Delete inactive collection registry rows after remapping references.") parser.add_argument("--dump-path", help="Optional path to master_collection_registry.csv dump.") parser.add_argument("--apply", action="store_true", help="Apply changes. Default is dry-run.") args = parser.parse_args() with get_session() as session: summary = cleanup_inactive_collection_registry( session, dump_path=args.dump_path, dry_run=not args.apply, ) print(json.dumps(summary, indent=2, sort_keys=True, default=str)) return 0 if __name__ == "__main__": raise SystemExit(main())