161 lines
5.1 KiB
Python
161 lines
5.1 KiB
Python
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))
|
|
|
|
from app.core.config import get_settings
|
|
from app.db.session import SessionLocal
|
|
from app.models import Export, Project
|
|
|
|
|
|
DEMO_PROJECT_NAME = "GeoIntel Demo - Building QA"
|
|
|
|
|
|
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 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) -> dict[str, Any]:
|
|
settings = get_settings()
|
|
storage_root = Path(settings.storage_root).resolve()
|
|
summary: dict[str, Any] = {
|
|
"dry_run": not apply,
|
|
"project_name": project_name,
|
|
"storage_root": str(storage_root),
|
|
"projects": [],
|
|
"matched_export_count": 0,
|
|
"selected_export_count": 0,
|
|
"deleted_export_count": 0,
|
|
"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()
|
|
)
|
|
kept, candidates = select_cleanup_candidates(exports, keep_latest)
|
|
summary["projects"].append(str(project.id))
|
|
summary["matched_export_count"] += len(exports)
|
|
summary["selected_export_count"] += len(candidates)
|
|
summary["kept_export_ids"].extend(str(export.id) for export in kept)
|
|
|
|
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["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("--apply", action="store_true", help="Delete selected export rows and files.")
|
|
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")
|
|
|
|
summary = cleanup_demo_exports(
|
|
project_name=args.project_name,
|
|
keep_latest=args.keep_latest,
|
|
apply=args.apply,
|
|
)
|
|
print(json.dumps(summary, indent=2, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|