from __future__ import annotations import argparse import json import sys from datetime import datetime from pathlib import Path from typing import Any BACKEND_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(BACKEND_ROOT)) REPOSITORY_ROOT = BACKEND_ROOT.parent SCRIPTS_ROOT = REPOSITORY_ROOT / "scripts" if SCRIPTS_ROOT.is_dir(): sys.path.insert(0, str(SCRIPTS_ROOT)) from app.core.config import get_settings # noqa: E402 - imported after backend path bootstrap from app.db.session import SessionLocal # noqa: E402 - imported after backend path bootstrap from app.models import Export, Project # noqa: E402 - imported after backend path bootstrap from release_backup_guard import ( # noqa: E402 - imported after scripts path bootstrap require_confirmation, verify_current_backup, ) DEMO_PROJECT_NAME = "GeoIntel Demo - Building QA" DELETE_CONFIRMATION = "DELETE_DEMO_EXPORTS" def is_within_storage_root(path: Path, storage_root: Path) -> bool: try: path.resolve().relative_to(storage_root.resolve()) except ValueError: return False return True def export_created_at(export: Any) -> datetime: created_at = getattr(export, "created_at", None) if isinstance(created_at, datetime): return created_at return datetime.min def select_cleanup_candidates(exports: list[Any], keep_latest: int) -> tuple[list[Any], list[Any]]: if keep_latest < 0: raise ValueError("keep_latest must be greater than or equal to zero") ordered = sorted(exports, key=export_created_at, reverse=True) 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"))) def prune_empty_parents(start_path: Path, storage_root: Path) -> list[str]: pruned: list[str] = [] parent = start_path.resolve().parent stop_at = storage_root.resolve() while parent != stop_at and is_within_storage_root(parent, stop_at): try: parent.rmdir() except OSError: break pruned.append(str(parent)) parent = parent.parent return pruned 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_exports": [], "candidate_files": [], "deleted_files": [], "missing_files": [], "skipped_outside_storage": [], "pruned_dirs": [], "kept_export_ids": [], } with SessionLocal() as db: projects = ( db.query(Project) .filter(Project.name == project_name) .filter(Project.status != "deleted") .order_by(Project.created_at.desc()) .all() ) for project in projects: exports = ( db.query(Export) .filter(Export.project_id == project.id) .order_by(Export.created_at.desc()) .all() ) 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): summary["skipped_outside_storage"].append( {"export_id": str(export.id), "storage_path": str(path)} ) continue if path.exists(): if apply: path.unlink() summary["pruned_dirs"].extend(prune_empty_parents(path, storage_root)) summary["deleted_files"].append(str(path)) else: summary["candidate_exports"].append( { "export_id": str(export.id), "export_type": str(getattr(export, "export_type", "")), "storage_path": str(path), } ) summary["candidate_files"].append(str(path)) else: summary["missing_files"].append({"export_id": str(export.id), "storage_path": str(path)}) if apply: db.delete(export) summary["deleted_export_count"] += 1 if apply: db.commit() return summary def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description=( "Clean old offline demo export artifacts. The script is dry-run by default " "and only targets the explicit GeoIntel demo project unless overridden." ) ) parser.add_argument("--project-name", default=DEMO_PROJECT_NAME, help="Exact project name to clean.") parser.add_argument( "--keep-latest", type=int, 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.") parser.add_argument( "--backup-dir", type=Path, help="Recent checksum-verified release backup mounted read-only in the runtime.", ) parser.add_argument( "--backup-max-age-hours", type=float, default=24.0, help="Maximum age accepted for the required release backup.", ) parser.add_argument( "--confirm", help=f"Exact destructive-maintenance confirmation token: {DELETE_CONFIRMATION}", ) return parser def main() -> int: parser = build_parser() 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") if args.apply: try: require_confirmation(args.confirm, DELETE_CONFIRMATION) if args.backup_dir is None: raise RuntimeError("--backup-dir is required with --apply") verify_current_backup( args.backup_dir, max_age_hours=args.backup_max_age_hours, ) except (RuntimeError, ValueError) as exc: parser.error(str(exc)) 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 if __name__ == "__main__": raise SystemExit(main())