From 6438bd418b4ad7e9e870da8f937e3cfe040c3f49 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 18 Jul 2026 07:10:18 +0200 Subject: [PATCH] Scale RC10 audit across persisted detections --- backend/tests/test_rc10_data_operations.py | 7 ++- scripts/audit_data_operations.py | 67 ++++++++++++++-------- 2 files changed, 49 insertions(+), 25 deletions(-) diff --git a/backend/tests/test_rc10_data_operations.py b/backend/tests/test_rc10_data_operations.py index 04e53a26..dc49f508 100644 --- a/backend/tests/test_rc10_data_operations.py +++ b/backend/tests/test_rc10_data_operations.py @@ -146,7 +146,10 @@ def test_storage_audit_only_selects_old_unreferenced_allowlisted_files( assert "release-evidence" in report["cleanup"]["protected_prefixes"] -def test_source_family_report_covers_national_regional_and_maritime(tmp_path: Path) -> None: +def test_source_family_report_covers_national_regional_and_maritime( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: audit = load_script("audit_data_operations.py") national_id = uuid4() regional_id = uuid4() @@ -221,6 +224,8 @@ def test_source_family_report_covers_national_regional_and_maritime(tmp_path: Pa return Query(rows[owner]) raise AssertionError(f"Unexpected query entity: {model!r}") + monkeypatch.setattr(audit, "query_count", lambda _db, model, *_conditions: len(rows[model])) + monkeypatch.setattr(audit, "query_distinct_nonnull", lambda _db, _column: []) state = audit.collect_database_state(Session(), tmp_path) assert {item["source_name"] for item in state["source_families"]["national"]} == { diff --git a/scripts/audit_data_operations.py b/scripts/audit_data_operations.py index 7af79c99..a4aa55b2 100644 --- a/scripts/audit_data_operations.py +++ b/scripts/audit_data_operations.py @@ -22,6 +22,8 @@ if str(BACKEND_ROOT) not in sys.path: 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 @@ -175,6 +177,17 @@ def _add_row_references(target: set[Path], row: Any, storage_root: Path, direct_ 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} @@ -200,16 +213,13 @@ def collect_database_state(db: Any, storage_root: Path) -> dict[str, Any]: 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. - detections = db.query(Detection.source_tile_path).all() - segmentations = db.query( - Segmentation.mask_path, - Segmentation.source_tile_path, - Segmentation.provenance_json, - ).all() - # Job/run JSON is not authoritative artifact persistence. Dataset, Export, - # Detection and Segmentation rows above own every retained path. - jobs = db.query(Job.status, Job.created_at).all() - analysis_runs = db.query(AnalysisRun.status, AnalysisRun.created_at).all() + 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: @@ -218,10 +228,12 @@ def collect_database_state(db: Any, storage_root: Path) -> dict[str, Any]: _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 detections: + 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",)) - for row in segmentations: - _add_row_references(references, row, storage_root, ("mask_path", "source_tile_path")) source_families: dict[str, dict[str, dict[str, Any]]] = { "national": {}, "regional": {}, @@ -280,6 +292,18 @@ def collect_database_state(db: Any, storage_root: Path) -> dict[str, Any]: ) 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": { @@ -289,17 +313,12 @@ def collect_database_state(db: Any, storage_root: Path) -> dict[str, Any]: "datasets": len(datasets), "dataset_versions": len(versions), "exports": len(exports), - "detections": len(detections), - "segmentations": len(segmentations), - "jobs": len(jobs), - "analysis_runs": len(analysis_runs), - "failed_jobs_older_than_7d": sum( - job.status == "failed" and job.created_at and job.created_at < failed_cutoff for job in jobs - ), - "failed_runs_older_than_7d": sum( - run.status == "failed" and run.created_at and run.created_at < failed_cutoff - for run in analysis_runs - ), + "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, }