#!/usr/bin/env python3 """Read-only storage, provenance and source-family audit for GeoIntel.""" from __future__ import annotations import argparse import json import os import shutil import sys from collections import defaultdict from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Iterable ROOT = Path(__file__).resolve().parents[1] BACKEND_ROOT = ROOT / "backend" if (ROOT / "backend" / "app").is_dir() else ROOT if str(BACKEND_ROOT) not in sys.path: sys.path.insert(0, str(BACKEND_ROOT)) from app.core.config import get_settings from app.db.session import SessionLocal from sqlalchemy import func from app.models import AnalysisRun, Dataset, DatasetVersion, Detection, Export, Job, Project, Segmentation NATIONAL_PROJECT_NAME = "Belgium and North Sea Workbench" PROTECTED_PREFIXES = ( "release-evidence", "operator-evidence", "operator-data", "originals", "uploads", "models", ) CLEANUP_PREFIXES = ( "exports", "previews", "tiles", "masks", "derived", "rasters/derived", ) IGNORED_FILENAMES = frozenset({".gitkeep", "README.md"}) PATH_METADATA_FIELDS = ( "metadata_json", "source_metadata", "provenance_metadata", "parameters_json", "result_json", "properties_json", "provenance_json", "bbox_json", ) @dataclass(frozen=True) class FileRecord: path: Path relative_path: str category: str size_bytes: int modified_at: datetime protected: bool cleanup_eligible: bool def utc_now() -> datetime: return datetime.now(timezone.utc) def _prefix_match(relative_path: str, prefixes: Iterable[str]) -> str | None: normalized = relative_path.strip("/") matches = [ prefix for prefix in prefixes if normalized == prefix or normalized.startswith(f"{prefix}/") ] return max(matches, key=len) if matches else None def classify_relative_path(relative_path: str) -> tuple[str, bool, bool]: protected = _prefix_match(relative_path, PROTECTED_PREFIXES) if protected: return protected, True, False cleanup = _prefix_match(relative_path, CLEANUP_PREFIXES) if cleanup: return cleanup, False, True category = relative_path.split("/", 1)[0] if relative_path else "." return category, True, False def inventory_storage(storage_root: Path) -> tuple[list[FileRecord], list[str]]: root = storage_root.resolve() records: list[FileRecord] = [] skipped_symlinks: list[str] = [] for current, directories, filenames in os.walk(root, followlinks=False): current_path = Path(current) retained_directories: list[str] = [] for directory in directories: child = current_path / directory if child.is_symlink(): skipped_symlinks.append(child.relative_to(root).as_posix()) else: retained_directories.append(directory) directories[:] = retained_directories for filename in filenames: path = current_path / filename if path.is_symlink(): skipped_symlinks.append(path.relative_to(root).as_posix()) continue try: stat = path.stat() except OSError: continue relative = path.relative_to(root).as_posix() category, protected, cleanup_eligible = classify_relative_path(relative) records.append( FileRecord( path=path.resolve(), relative_path=relative, category=category, size_bytes=stat.st_size, modified_at=datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc), protected=protected, cleanup_eligible=cleanup_eligible, ) ) return records, sorted(skipped_symlinks) def _iter_strings(value: Any) -> Iterable[str]: if isinstance(value, str): yield value elif isinstance(value, dict): for nested in value.values(): yield from _iter_strings(nested) elif isinstance(value, (list, tuple)): for nested in value: yield from _iter_strings(nested) def _manifest_key_is_path(key: str) -> bool: exact = { "path", "paths", "file", "files", "filename", "filenames", "artifact", "artifacts", "manifest", "manifests", "storage_path", "source_tile_path", "mask_path", "output_path", } return key in exact or key.endswith(("_path", "_paths", "_file", "_files", "_filename", "_filenames")) def _iter_manifest_path_values(value: Any, parent_key: str = "") -> Iterable[str]: if isinstance(value, dict): for key, nested in value.items(): normalized_key = str(key).strip().lower() if isinstance(nested, str) and _manifest_key_is_path(normalized_key): yield nested else: yield from _iter_manifest_path_values(nested, normalized_key) elif isinstance(value, (list, tuple)): for nested in value: if isinstance(nested, str) and _manifest_key_is_path(parent_key): yield nested else: yield from _iter_manifest_path_values(nested, parent_key) def normalize_storage_reference(value: str | None, storage_root: Path) -> Path | None: if not value: return None stripped = value.strip() if not stripped or "://" in stripped or stripped.startswith("/vsi"): return None candidate = Path(stripped) if not candidate.is_absolute(): normalized = stripped.replace("\\", "/") if normalized.startswith("storage/"): normalized = normalized.removeprefix("storage/") if _prefix_match(normalized, PROTECTED_PREFIXES + CLEANUP_PREFIXES) is None: return None candidate = storage_root / normalized try: resolved = candidate.resolve() resolved.relative_to(storage_root.resolve()) except (OSError, ValueError): return None return resolved def expand_manifest_references(references: set[Path], storage_root: Path) -> set[Path]: """Protect files named by referenced, bounded JSON manifests.""" root = storage_root.resolve() expanded = set(references) for manifest in list(references): if ( manifest.suffix.lower() != ".json" or "manifest" not in manifest.name.lower() or not manifest.is_file() ): continue try: if manifest.stat().st_size > 16 * 1024 * 1024: continue payload = json.loads(manifest.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError): continue for value in _iter_manifest_path_values(payload): normalized = normalize_storage_reference(value, root) if normalized is None and "://" not in value: try: relative = (manifest.parent / value).resolve() relative.relative_to(root) except (OSError, ValueError): continue normalized = relative if normalized is not None: expanded.add(normalized) return expanded def _add_row_references(target: set[Path], row: Any, storage_root: Path, direct_fields: Iterable[str]) -> None: for field in direct_fields: normalized = normalize_storage_reference(getattr(row, field, None), storage_root) if normalized: target.add(normalized) for field in PATH_METADATA_FIELDS: for value in _iter_strings(getattr(row, field, None)): normalized = normalize_storage_reference(value, storage_root) if normalized: target.add(normalized) def query_count(db: Any, model: Any, *conditions: Any) -> int: query = db.query(func.count(model.id)) if conditions: query = query.filter(*conditions) return int(query.scalar() or 0) def query_distinct_nonnull(db: Any, column: Any) -> list[Any]: return db.query(column).filter(column.is_not(None)).distinct().all() def collect_database_state(db: Any, storage_root: Path) -> dict[str, Any]: projects = db.query(Project.id, Project.name, Project.status).all() project_names = {project.id: project.name for project in projects} datasets = db.query( Dataset.id, Dataset.project_id, Dataset.name, Dataset.source, Dataset.source_name, Dataset.storage_path, Dataset.metadata_json, Dataset.source_metadata, Dataset.provenance_metadata, Dataset.source_version, Dataset.imported_at, Dataset.status, ).all() versions = db.query( DatasetVersion.storage_path, DatasetVersion.source_metadata, DatasetVersion.provenance_metadata, ).all() exports = db.query(Export.storage_path, Export.metadata_json).all() # Geometry and feature properties can be very large. The persisted direct # artifact columns are sufficient here and keep the operator audit bounded. detection_tile_paths = query_distinct_nonnull(db, Detection.source_tile_path) segmentation_mask_paths = query_distinct_nonnull(db, Segmentation.mask_path) segmentation_tile_paths = query_distinct_nonnull(db, Segmentation.source_tile_path) detection_count = query_count(db, Detection) segmentation_count = query_count(db, Segmentation) job_count = query_count(db, Job) analysis_run_count = query_count(db, AnalysisRun) references: set[Path] = set() for row in datasets: _add_row_references(references, row, storage_root, ("storage_path",)) for row in versions: _add_row_references(references, row, storage_root, ("storage_path",)) for row in exports: _add_row_references(references, row, storage_root, ("storage_path",)) for row in detection_tile_paths: _add_row_references(references, row, storage_root, ("source_tile_path",)) for row in segmentation_mask_paths: _add_row_references(references, row, storage_root, ("mask_path",)) for row in segmentation_tile_paths: _add_row_references(references, row, storage_root, ("source_tile_path",)) source_families: dict[str, dict[str, dict[str, Any]]] = { "national": {}, "regional": {}, "maritime": {}, } maritime_tokens = ("marine", "maritime", "north_sea", "bathymetry", "rbin", "mdk", "msp") for dataset in datasets: source_name = (dataset.source_name or dataset.source or "unknown").strip().lower() metadata = dataset.source_metadata if isinstance(dataset.source_metadata, dict) else {} zones = metadata.get("coverage_zones") or metadata.get("coverage_zone") or [] if isinstance(zones, str): zones = [zones] families: set[str] = set() project_name = project_names.get(dataset.project_id, "") if project_name == NATIONAL_PROJECT_NAME or "belgium" in zones: families.add("national") else: families.add("regional") searchable = " ".join([source_name, dataset.name.lower(), *(str(zone).lower() for zone in zones)]) if "belgian_north_sea" in zones or any(token in searchable for token in maritime_tokens): families.add("maritime") for family in families: item = source_families[family].setdefault( source_name, { "source_name": source_name, "dataset_count": 0, "ready_count": 0, "latest_imported_at": None, "source_versions": set(), }, ) item["dataset_count"] += 1 item["ready_count"] += int(dataset.status == "ready") if dataset.source_version: item["source_versions"].add(dataset.source_version) if dataset.imported_at and ( item["latest_imported_at"] is None or dataset.imported_at > item["latest_imported_at"] ): item["latest_imported_at"] = dataset.imported_at serialized_families: dict[str, list[dict[str, Any]]] = {} for family, sources in source_families.items(): serialized_families[family] = [] for source in sorted(sources.values(), key=lambda item: item["source_name"]): serialized_families[family].append( { **source, "latest_imported_at": ( source["latest_imported_at"].isoformat() if source["latest_imported_at"] is not None else None ), "source_versions": sorted(source["source_versions"]), } ) failed_cutoff = utc_now() - timedelta(days=7) failed_jobs = query_count( db, Job, Job.status == "failed", Job.created_at < failed_cutoff, ) failed_runs = query_count( db, AnalysisRun, AnalysisRun.status == "failed", AnalysisRun.created_at < failed_cutoff, ) return { "references": references, "counts": { "projects": len(projects), "active_projects": sum(project.status == "active" for project in projects), "archived_projects": sum(project.status == "archived" for project in projects), "datasets": len(datasets), "dataset_versions": len(versions), "exports": len(exports), "detections": detection_count, "segmentations": segmentation_count, "jobs": job_count, "analysis_runs": analysis_run_count, "failed_jobs_older_than_7d": failed_jobs, "failed_runs_older_than_7d": failed_runs, }, "source_families": serialized_families, } def disk_pressure(storage_root: Path) -> dict[str, Any]: usage = shutil.disk_usage(storage_root) free_percent = (usage.free / usage.total * 100) if usage.total else 0.0 # Absolute headroom governs very large Unraid arrays: a low percentage of # tens of terabytes can still leave hundreds of GiB safely available. if usage.free < 10 * 1024**3: status = "critical" elif usage.free < 50 * 1024**3 or (free_percent < 2 and usage.free < 250 * 1024**3): status = "warning" else: status = "ok" return { "status": status, "total_bytes": usage.total, "used_bytes": usage.used, "free_bytes": usage.free, "free_percent": round(free_percent, 2), "acquisition_allowed": status != "critical", } def build_report( storage_root: Path, db: Any, *, minimum_age_days: int = 7, max_candidate_records: int = 500, ) -> tuple[dict[str, Any], list[FileRecord]]: if minimum_age_days < 1: raise ValueError("minimum_age_days must be at least one") root = storage_root.expanduser().resolve() if not root.is_dir(): raise RuntimeError(f"Storage root is not a directory: {root}") records, skipped_symlinks = inventory_storage(root) database = collect_database_state(db, root) direct_references: set[Path] = database.pop("references") references = expand_manifest_references(direct_references, root) manifest_references = references - direct_references cutoff = utc_now() - timedelta(days=minimum_age_days) categories: dict[str, dict[str, Any]] = defaultdict( lambda: {"file_count": 0, "size_bytes": 0, "protected": False, "cleanup_eligible": False} ) for record in records: item = categories[record.category] item["file_count"] += 1 item["size_bytes"] += record.size_bytes item["protected"] = item["protected"] or record.protected item["cleanup_eligible"] = item["cleanup_eligible"] or record.cleanup_eligible existing_paths = {record.path for record in records} missing_direct_references = sorted( path.relative_to(root).as_posix() for path in direct_references if path not in existing_paths and not path.is_dir() ) missing_manifest_references = sorted( path.relative_to(root).as_posix() for path in manifest_references if path not in existing_paths and not path.is_dir() ) referenced_directories = tuple(path for path in references if path.is_dir()) candidates = sorted( ( record for record in records if record.cleanup_eligible and record.path not in references and not any(record.path.is_relative_to(reference) for reference in referenced_directories) and record.modified_at <= cutoff and record.path.name not in IGNORED_FILENAMES ), key=lambda item: (item.modified_at, item.relative_path), ) report = { "schema_version": 1, "generated_at": utc_now().isoformat(), "mode": "read-only", "storage_root": str(root), "minimum_age_days": minimum_age_days, "disk_pressure": disk_pressure(root), "lifecycle": { "raw_and_normalized": ["originals", "uploads", "operator-data"], "derived_and_cache": ["derived", "rasters/derived", "previews", "tiles", "masks"], "exports": ["exports"], "ai_models": ["models"], "immutable_evidence": ["release-evidence", "operator-evidence"], }, "categories": [ {"category": category, **values} for category, values in sorted(categories.items()) ], "database": database, "integrity": { "referenced_path_count": len(references), "direct_database_reference_count": len(direct_references), "missing_referenced_path_count": len(missing_direct_references), "missing_referenced_paths": missing_direct_references[:max_candidate_records], "manifest_artifact_reference_count": len(manifest_references), "missing_manifest_artifact_count": len(missing_manifest_references), "missing_manifest_artifacts": missing_manifest_references[:max_candidate_records], "skipped_symlinks": skipped_symlinks[:max_candidate_records], }, "cleanup": { "candidate_count": len(candidates), "candidate_bytes": sum(item.size_bytes for item in candidates), "candidate_records_truncated": len(candidates) > max_candidate_records, "candidates": [ { "relative_path": item.relative_path, "category": item.category, "size_bytes": item.size_bytes, "modified_at": item.modified_at.isoformat(), } for item in candidates[:max_candidate_records] ], "protected_prefixes": list(PROTECTED_PREFIXES), "eligible_prefixes": list(CLEANUP_PREFIXES), "apply_requires": [ "exact confirmation token", "recent checksum-verified database and SHA-256 storage backup", "explicit maximum delete count", ], }, "limitations": [ "The audit never downloads or refreshes an official source.", "Failed jobs and analysis runs remain provenance and are reported, not deleted.", "Missing non-authoritative manifest intermediates are reported separately from direct database artifacts.", "Unknown storage categories are protected by default.", "No cleanup is scheduled implicitly.", ], } return report, candidates 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-candidate-records", type=int, default=500) parser.add_argument("--output", type=Path) parser.add_argument("--fail-on-pressure", choices=("never", "critical", "warning"), default="critical") parser.add_argument("--fail-on-missing-reference", action="store_true") return parser.parse_args() def main() -> int: args = parse_args() storage_root = args.storage_root or Path(get_settings().storage_root) with SessionLocal() as db: report, _ = build_report( storage_root, db, minimum_age_days=args.minimum_age_days, max_candidate_records=args.max_candidate_records, ) serialized = json.dumps(report, indent=2, sort_keys=True) print(serialized) if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) temporary = args.output.with_suffix(f"{args.output.suffix}.partial") temporary.write_text(serialized + "\n", encoding="utf-8") temporary.replace(args.output) pressure = report["disk_pressure"]["status"] if args.fail_on_pressure == "critical" and pressure == "critical": return 1 if args.fail_on_pressure == "warning" and pressure in {"warning", "critical"}: return 1 if args.fail_on_missing_reference and report["integrity"]["missing_referenced_path_count"]: return 1 return 0 if __name__ == "__main__": raise SystemExit(main())