Harden demo export cleanup safety

backend/scripts/cleanup_demo_artifacts.py

@@ -40,6 +40,13 @@ def select_cleanup_candidates(exports: list[Any], keep_latest: int) -> tuple[lis
     return ordered[:keep_latest], ordered[keep_latest:]
 
 
+def filter_exports_by_type(exports: list[Any], export_types: list[str] | None) -> list[Any]:
+    if not export_types:
+        return exports
+    allowed = set(export_types)
+    return [export for export in exports if getattr(export, "export_type", None) in allowed]
+
+
 def export_path(export: Any) -> Path:
     return Path(str(getattr(export, "storage_path")))
 
@@ -58,15 +65,27 @@ def prune_empty_parents(start_path: Path, storage_root: Path) -> list[str]:
     return pruned
 
 
-def cleanup_demo_exports(project_name: str, keep_latest: int, apply: bool) -> dict[str, Any]:
+def cleanup_demo_exports(
+    project_name: str,
+    keep_latest: int,
+    apply: bool,
+    max_delete: int,
+    export_types: list[str] | None = None,
+) -> dict[str, Any]:
+    if max_delete < 0:
+        raise ValueError("max_delete must be greater than or equal to zero")
     settings = get_settings()
     storage_root = Path(settings.storage_root).resolve()
     summary: dict[str, Any] = {
         "dry_run": not apply,
         "project_name": project_name,
+        "keep_latest": keep_latest,
+        "max_delete": max_delete,
+        "export_types": export_types or [],
         "storage_root": str(storage_root),
         "projects": [],
         "matched_export_count": 0,
+        "type_filtered_export_count": 0,
         "selected_export_count": 0,
         "deleted_export_count": 0,
         "candidate_files": [],
@@ -92,12 +111,21 @@ def cleanup_demo_exports(project_name: str, keep_latest: int, apply: bool) -> di
                 .order_by(Export.created_at.desc())
                 .all()
             )
-            kept, candidates = select_cleanup_candidates(exports, keep_latest)
+            filtered_exports = filter_exports_by_type(exports, export_types)
+            kept, candidates = select_cleanup_candidates(filtered_exports, keep_latest)
             summary["projects"].append(str(project.id))
             summary["matched_export_count"] += len(exports)
+            summary["type_filtered_export_count"] += len(filtered_exports)
             summary["selected_export_count"] += len(candidates)
             summary["kept_export_ids"].extend(str(export.id) for export in kept)
 
+            if apply and len(candidates) > max_delete:
+                summary["blocked_reason"] = (
+                    f"selected_export_count {len(candidates)} exceeds --max-delete {max_delete}; "
+                    "raise --max-delete after reviewing a dry run"
+                )
+                continue
+
             for export in candidates:
                 path = export_path(export)
                 if not is_within_storage_root(path, storage_root):
@@ -140,6 +168,18 @@ def build_parser() -> argparse.ArgumentParser:
         default=3,
         help="Number of newest export records/files to keep per matching project.",
     )
+    parser.add_argument(
+        "--max-delete",
+        type=int,
+        default=25,
+        help="Maximum export rows/files allowed to be deleted per matching project when --apply is set.",
+    )
+    parser.add_argument(
+        "--export-type",
+        action="append",
+        default=None,
+        help="Restrict cleanup to an export_type. Repeat for multiple types.",
+    )
     parser.add_argument("--apply", action="store_true", help="Delete selected export rows and files.")
     return parser
 
@@ -149,11 +189,15 @@ def main() -> int:
     args = parser.parse_args()
     if args.keep_latest < 0:
         parser.error("--keep-latest must be greater than or equal to zero")
+    if args.max_delete < 0:
+        parser.error("--max-delete must be greater than or equal to zero")
 
     summary = cleanup_demo_exports(
         project_name=args.project_name,
         keep_latest=args.keep_latest,
         apply=args.apply,
+        max_delete=args.max_delete,
+        export_types=args.export_type,
     )
     print(json.dumps(summary, indent=2, sort_keys=True))
     return 0