geointel/scripts/cleanup_storage_artifacts.py

#!/usr/bin/env python3
"""Dry-run-first quarantine for old unreferenced derived/cache artifacts."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
from datetime import datetime, timezone
from pathlib import Path
from uuid import uuid4

from audit_data_operations import build_report
from release_backup_guard import require_confirmation, verify_current_backup

from app.core.config import get_settings
from app.db.session import SessionLocal


CONFIRMATION = "QUARANTINE_STORAGE_ARTIFACTS"


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def write_manifest(path: Path, payload: dict[str, object]) -> None:
    temporary = path.with_suffix(".json.partial")
    temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    temporary.replace(path)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--storage-root", type=Path)
    parser.add_argument("--minimum-age-days", type=int, default=7)
    parser.add_argument("--max-delete", type=int, default=25)
    parser.add_argument("--apply", action="store_true")
    parser.add_argument("--confirm")
    parser.add_argument("--backup-dir", type=Path)
    parser.add_argument("--backup-max-age-hours", type=float, default=24.0)
    parser.add_argument(
        "--quarantine-root",
        type=Path,
        help="Protected destination below the storage root (default: operator-evidence/cleanup-quarantine)",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    if args.max_delete < 0:
        raise SystemExit("--max-delete must be greater than or equal to zero")
    storage_root = (args.storage_root or Path(get_settings().storage_root)).resolve()
    with SessionLocal() as db:
        report, candidates = build_report(
            storage_root,
            db,
            minimum_age_days=args.minimum_age_days,
            max_candidate_records=max(args.max_delete, 500),
        )

    blocked_reason = None
    backup = None
    quarantined: list[dict[str, object]] = []
    quarantine_manifest: Path | None = None
    if args.apply:
        require_confirmation(args.confirm, CONFIRMATION)
        if args.backup_dir is None:
            raise RuntimeError("--backup-dir is required with --apply")
        backup = verify_current_backup(
            args.backup_dir,
            max_age_hours=args.backup_max_age_hours,
        )
        if len(candidates) > args.max_delete:
            blocked_reason = (
                f"candidate_count {len(candidates)} exceeds --max-delete {args.max_delete}; "
                "review the dry run and raise the explicit limit"
            )
        else:
            quarantine_root = (
                args.quarantine_root
                or storage_root / "operator-evidence" / "cleanup-quarantine"
            ).resolve()
            protected_quarantine_root = (
                storage_root / "operator-evidence" / "cleanup-quarantine"
            ).resolve()
            try:
                quarantine_root.relative_to(protected_quarantine_root)
            except ValueError as exc:
                raise RuntimeError(
                    "--quarantine-root must remain below "
                    "operator-evidence/cleanup-quarantine in --storage-root"
                ) from exc
            operation_id = f"cleanup-{datetime.now(timezone.utc):%Y%m%dT%H%M%SZ}-{uuid4().hex[:12]}"
            operation_root = quarantine_root / operation_id
            operation_root.mkdir(parents=True, exist_ok=False)
            quarantine_manifest = operation_root / "manifest.json"
            entries: list[dict[str, object]] = []
            for candidate in candidates:
                if candidate.path.is_symlink():
                    raise RuntimeError(f"Cleanup candidate became a symlink: {candidate.relative_path}")
                try:
                    candidate.path.resolve().relative_to(storage_root)
                except ValueError as exc:
                    raise RuntimeError(
                        f"Cleanup candidate escaped storage: {candidate.relative_path}"
                    ) from exc
                destination = operation_root / "files" / candidate.relative_path
                current_size = candidate.path.stat().st_size
                if current_size != candidate.size_bytes:
                    raise RuntimeError(f"Cleanup candidate changed size: {candidate.relative_path}")
                entries.append(
                    {
                        "relative_path": candidate.relative_path,
                        "size_bytes": current_size,
                        "sha256": sha256(candidate.path),
                        "status": "planned",
                        "quarantine_relative_path": destination.relative_to(storage_root).as_posix(),
                    }
                )
            manifest: dict[str, object] = {
                "schema_version": 1,
                "operation_id": operation_id,
                "created_at": datetime.now(timezone.utc).isoformat(),
                "state": "in_progress",
                "storage_root": str(storage_root),
                "backup_release_id": backup.release_id,
                "entries": entries,
            }
            write_manifest(quarantine_manifest, manifest)
            for candidate, entry in zip(candidates, entries, strict=True):
                destination = storage_root / str(entry["quarantine_relative_path"])
                destination.parent.mkdir(parents=True, exist_ok=True)
                try:
                    os.link(candidate.path, destination, follow_symlinks=False)
                except FileExistsError as exc:
                    raise RuntimeError(f"Quarantine destination already exists: {destination}") from exc
                if (
                    not destination.is_file()
                    or destination.stat().st_size != entry["size_bytes"]
                    or sha256(destination) != entry["sha256"]
                ):
                    destination.unlink(missing_ok=True)
                    raise RuntimeError(f"Quarantine link verification failed: {candidate.relative_path}")
                entry["status"] = "linked"
                write_manifest(quarantine_manifest, manifest)
                candidate.path.unlink()
                entry["status"] = "quarantined"
                quarantined.append(dict(entry))
                write_manifest(quarantine_manifest, manifest)
            manifest["state"] = "complete"
            manifest["completed_at"] = datetime.now(timezone.utc).isoformat()
            write_manifest(quarantine_manifest, manifest)

    payload = {
        "schema_version": 1,
        "mode": "apply" if args.apply else "dry-run",
        "storage_root": str(storage_root),
        "minimum_age_days": args.minimum_age_days,
        "max_delete": args.max_delete,
        "candidate_count": len(candidates),
        "candidate_bytes": sum(item.size_bytes for item in candidates),
        "candidates": [item.relative_path for item in candidates],
        "deleted_count": 0,
        "deleted": [],
        "quarantined_count": len(quarantined),
        "quarantined": quarantined,
        "quarantine_manifest": str(quarantine_manifest) if quarantine_manifest else None,
        "blocked_reason": blocked_reason,
        "protected_prefixes": report["cleanup"]["protected_prefixes"],
        "backup": (
            {
                "release_id": backup.release_id,
                "created_at": backup.created_at.isoformat(),
                "age_hours": round(backup.age_hours, 3),
                "backup_tool_revision": backup.backup_tool_revision,
            }
            if backup
            else None
        ),
    }
    print(json.dumps(payload, indent=2, sort_keys=True))
    return 1 if blocked_reason else 0


if __name__ == "__main__":
    raise SystemExit(main())