538 lines
20 KiB
Python
538 lines
20 KiB
Python
#!/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 _iter_manifest_path_values(value: Any, parent_key: str = "") -> Iterable[str]:
|
|
path_tokens = ("path", "file", "artifact", "manifest", "output")
|
|
if isinstance(value, dict):
|
|
for key, nested in value.items():
|
|
normalized_key = str(key).strip().lower()
|
|
if isinstance(nested, str) and any(token in normalized_key for token in path_tokens):
|
|
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 any(token in parent_key for token in path_tokens):
|
|
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)
|
|
references: set[Path] = expand_manifest_references(database.pop("references"), root)
|
|
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_references = sorted(
|
|
path.relative_to(root).as_posix()
|
|
for path in 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),
|
|
"missing_referenced_path_count": len(missing_references),
|
|
"missing_referenced_paths": missing_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.",
|
|
"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())
|