From eed6ee796e1ec51922e6af6c0565b9323477cae3 Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 16 Jul 2026 17:59:00 +0200 Subject: [PATCH] Add governed GRB refresh workflow --- CHANGELOG.md | 14 + backend/README.md | 38 ++ backend/app/api/routes/datasets.py | 17 + backend/app/schemas/__init__.py | 4 + backend/app/schemas/grb_refresh.py | 63 +++ .../app/services/grb_refresh_plan_service.py | 186 +++++++++ .../test_sprint223_governed_grb_refresh.py | 250 ++++++++++++ docs/API_CONTRACTS.md | 24 ++ docs/CODEX_EXECUTION_LOG.md | 33 ++ docs/DATABASE_IMPLEMENTATION_PLAN.md | 7 + docs/TODO.md | 1 + frontend/README.md | 7 + frontend/src/App.tsx | 3 + .../status/SourceFreshnessPanel.tsx | 59 +++ frontend/src/hooks/useSourceFreshness.ts | 32 +- frontend/src/services/api/datasets.ts | 5 + frontend/src/styles/premium.css | 109 +++++ frontend/src/types.ts | 52 +++ scripts/README.md | 36 ++ scripts/manage_grb_refresh.py | 373 ++++++++++++++++++ scripts/run_readiness_check.sh | 1 + 21 files changed, 1313 insertions(+), 1 deletion(-) create mode 100644 backend/app/schemas/grb_refresh.py create mode 100644 backend/app/services/grb_refresh_plan_service.py create mode 100644 backend/tests/test_sprint223_governed_grb_refresh.py create mode 100644 scripts/manage_grb_refresh.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e623c765..863ab60b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,20 @@ # Changelog +## Sprint 223 Governed regional GRB refresh (2026-07-16) + +- Added a canonical, read-only regional GRB refresh plan for buildings, roads, + water and parcels, derived from the official catalog edition and persisted + temporal snapshots. +- Added an explicit operator coordinator with separate `plan`, `stage` and + SHA-256-confirmed `apply` phases. Staging validates every municipality + partition before PostGIS persistence is possible. +- Kept all provider URLs and collections allowlisted, reused the existing + resumable regional operators and DatasetService/VectorFeatureService import + path, and retained every older snapshot. +- Added compact Status UI, deterministic guardrail tests and readiness syntax + coverage without changing migrations or enabling automatic refresh. + ## Sprint 222 Official source edition probes (2026-07-16) - Added explicit, read-only GRB and most-recent orthophoto catalog probes using diff --git a/backend/README.md b/backend/README.md index af97d7d3..b265bfa8 100644 --- a/backend/README.md +++ b/backend/README.md @@ -1469,3 +1469,41 @@ Runtime controls are `SOURCE_CATALOG_PROBE_ENABLED`, `SOURCE_CATALOG_PROBE_CACHE_TTL_SECONDS`. Metadata links remain restricted to the official HTTPS CSW path on `metadata.vlaanderen.be` even when an operator overrides the capabilities endpoint. + +## Governed regional GRB refresh + +`GET /api/v1/projects/{project_id}/datasets/grb-refresh-plan` combines the +explicit official GRB edition probe with the four existing regional snapshot +series. It is read-only: it reports local feature/storage impact and whether +buildings, roads, water and parcels are current, updateable or require review. + +Regional refreshes use a two-phase operator flow inside the all-in-one +container. First stage and validate every source partition without touching +PostGIS: + +```bash +docker exec geointel python /app/scripts/manage_grb_refresh.py stage \ + --project-id 82a85913-c522-45d7-84a1-02b393d89e55 \ + --api-url http://127.0.0.1:8000/api/v1 \ + --confirm-edition 2026-07-15 \ + --layers buildings roads water parcels +``` + +The JSON result gives `plan_path`, exact feature deltas, artifact sizes and +`plan_sha256`. Review that evidence, then apply those exact staged bytes: + +```bash +docker exec geointel python /app/scripts/manage_grb_refresh.py apply \ + --project-id 82a85913-c522-45d7-84a1-02b393d89e55 \ + --api-url http://127.0.0.1:8000/api/v1 \ + --confirm-edition 2026-07-15 \ + --confirm-plan-sha256 SHA256_FROM_STAGE +``` + +Both commands fail when project/scope, official edition, manifests, partition +count or any checksum differs. Interrupted staging is safely resumable because +the existing municipal manifests are reused. Apply imports through +DatasetService/VectorFeatureService, creates new temporal Datasets and retains +all previous snapshots. Do not add `--force` to this coordinator; a source +refetch remains a separate deliberate recovery action in the lower-level +operators. diff --git a/backend/app/api/routes/datasets.py b/backend/app/api/routes/datasets.py index 212c4cff..fe69b91c 100644 --- a/backend/app/api/routes/datasets.py +++ b/backend/app/api/routes/datasets.py @@ -46,6 +46,7 @@ from app.services.vector_feature_service import VectorFeatureService from app.services.dataset_service import DatasetService from app.services.source_freshness_service import SourceFreshnessService from app.services.source_catalog_probe_service import SourceCatalogProbeService +from app.services.grb_refresh_plan_service import GrbRefreshPlanService from app.services.orthophoto_acquisition_service import OrthophotoAcquisitionService from app.services.dhmv_acquisition_service import DhmvAcquisitionService from app.services.terrain_analysis_service import TerrainAnalysisService @@ -266,6 +267,22 @@ def probe_dataset_source_catalogs( return envelope(report.model_dump()) +@router.get("/datasets/grb-refresh-plan", response_model=dict) +def plan_grb_dataset_refresh( + project_id: UUID, + scope: str = Query(default=GrbRefreshPlanService.SCOPE), + refresh_catalog: bool = Query(default=False), + db: Session = Depends(get_db), +): + report = GrbRefreshPlanService.build( + db, + project_id, + scope=scope, + refresh_catalog=refresh_catalog, + ) + return envelope(report.model_dump()) + + @router.get("/datasets/{dataset_id}", response_model=dict) def get_dataset( project_id: UUID, diff --git a/backend/app/schemas/__init__.py b/backend/app/schemas/__init__.py index 9bc80ab4..e37cb822 100644 --- a/backend/app/schemas/__init__.py +++ b/backend/app/schemas/__init__.py @@ -16,6 +16,7 @@ from .source_catalog import ( SourceCatalogProbeReport, SourceCatalogProbeSummary, ) +from .grb_refresh import GrbRefreshLayerPlan, GrbRefreshPlan, GrbRefreshPlanSummary from .detection import ( DetectionListResponse, DetectionModelCapability, @@ -149,6 +150,9 @@ __all__ = [ "SourceCatalogProbeItem", "SourceCatalogProbeReport", "SourceCatalogProbeSummary", + "GrbRefreshLayerPlan", + "GrbRefreshPlan", + "GrbRefreshPlanSummary", "DetectionListResponse", "DetectionModelCapability", "DetectionModelsResponse", diff --git a/backend/app/schemas/grb_refresh.py b/backend/app/schemas/grb_refresh.py new file mode 100644 index 00000000..5d439fee --- /dev/null +++ b/backend/app/schemas/grb_refresh.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +from datetime import date, datetime +from typing import Literal +from uuid import UUID + +from pydantic import BaseModel + + +GrbRefreshLayerStatus = Literal[ + "current", + "update_available", + "not_loaded", + "review_required", + "remote_unavailable", +] + + +class GrbRefreshLayerPlan(BaseModel): + theme: Literal["buildings", "roads", "water", "parcels"] + display_name: str + collections: list[str] + temporal_series_key: str + status: GrbRefreshLayerStatus + local_dataset_id: UUID | None = None + local_source_version: str | None = None + local_observed_at: datetime | None = None + local_imported_at: datetime | None = None + local_feature_count: int | None = None + local_size_bytes: int | None = None + retained_after_refresh: bool = True + action_message: str + + +class GrbRefreshPlanSummary(BaseModel): + layer_count: int + current_count: int + update_available_count: int + not_loaded_count: int + review_required_count: int + remote_unavailable_count: int + new_dataset_count_if_applied: int + retained_dataset_count: int + current_feature_count: int + current_size_bytes: int + + +class GrbRefreshPlan(BaseModel): + project_id: UUID + scope: str + generated_at: datetime + remote_status: str + remote_version: str | None = None + remote_edition_date: date | None = None + catalog_checked_at: datetime | None = None + summary: GrbRefreshPlanSummary + layers: list[GrbRefreshLayerPlan] + execution_mode: Literal["operator_stage_then_apply"] = "operator_stage_then_apply" + staging_required: bool = True + automatic_import: bool = False + destructive_replacement: bool = False + message: str + limitations: list[str] diff --git a/backend/app/services/grb_refresh_plan_service.py b/backend/app/services/grb_refresh_plan_service.py new file mode 100644 index 00000000..b2a8f781 --- /dev/null +++ b/backend/app/services/grb_refresh_plan_service.py @@ -0,0 +1,186 @@ +from __future__ import annotations + +import re +from dataclasses import dataclass +from datetime import date, datetime, timezone +from uuid import UUID + +from sqlalchemy.orm import Session + +from app.core.errors import AppError +from app.models import Dataset, Project +from app.schemas.grb_refresh import GrbRefreshLayerPlan, GrbRefreshPlan, GrbRefreshPlanSummary +from app.services.source_catalog_probe_service import SourceCatalogProbeService + + +@dataclass(frozen=True) +class _LayerDefinition: + theme: str + display_name: str + collections: tuple[str, ...] + + @property + def series_key(self) -> str: + return f"grb:{self.theme}:kempen-transport-region" + + +class GrbRefreshPlanService: + SCOPE = "kempen-transport-region" + LAYERS = ( + _LayerDefinition("buildings", "Gebouwen", ("GBG",)), + _LayerDefinition("roads", "Wegen", ("Wegsegment",)), + _LayerDefinition("water", "Water", ("WTZ", "WLAS", "WGR")), + _LayerDefinition("parcels", "Percelen", ("ADP",)), + ) + _EDITION_DATE = re.compile(r"(? date | None: + match = cls._EDITION_DATE.search(value or "") + if not match: + return None + try: + return date.fromisoformat(match.group(1)) + except ValueError: + return None + + @staticmethod + def _dataset_feature_count(dataset: Dataset) -> int | None: + metadata = dataset.metadata_json if isinstance(dataset.metadata_json, dict) else {} + value = metadata.get("feature_count") + try: + return int(value) if value is not None else None + except (TypeError, ValueError): + return None + + @staticmethod + def _latest_dataset(rows: list[Dataset], series_key: str) -> Dataset | None: + candidates = [ + row + for row in rows + if row.temporal_series_key == series_key and row.status == "ready" + ] + if not candidates: + return None + minimum = datetime.min.replace(tzinfo=timezone.utc) + return max(candidates, key=lambda row: (row.observed_at or row.imported_at or row.created_at or minimum, str(row.id))) + + @classmethod + def build( + cls, + db: Session, + project_id: UUID, + *, + scope: str = SCOPE, + refresh_catalog: bool = False, + now: datetime | None = None, + ) -> GrbRefreshPlan: + if scope != cls.SCOPE: + raise AppError( + code="GRB_REFRESH_SCOPE_UNSUPPORTED", + message=f"Only the governed scope '{cls.SCOPE}' is supported", + status_code=400, + ) + if not db.get(Project, project_id): + raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404) + + generated_at = now or datetime.now(timezone.utc) + catalog = SourceCatalogProbeService.audit_project(db, project_id, force=refresh_catalog) + grb_probe = next((item for item in catalog.items if item.source_name == "grb"), None) + remote_version = grb_probe.remote_version if grb_probe else None + remote_edition_date = cls._parse_edition_date(remote_version) + remote_available = bool(grb_probe and grb_probe.status == "available" and grb_probe.reachable) + rows = ( + db.query(Dataset) + .filter(Dataset.project_id == project_id, Dataset.source_name == "grb") + .all() + ) + + layer_plans: list[GrbRefreshLayerPlan] = [] + for definition in cls.LAYERS: + local = cls._latest_dataset(rows, definition.series_key) + local_date = cls._parse_edition_date(local.source_version if local else None) + if not remote_available: + status = "remote_unavailable" + action = "De officiële catalogus is niet bereikbaar; er wordt geen vernieuwingsbeslissing genomen." + elif remote_edition_date is None: + status = "review_required" + action = "De officiële editie bevat geen herkenbare datum en vereist menselijke beoordeling." + elif local is None: + status = "not_loaded" + action = "Deze laag kan als nieuwe, afzonderlijke GRB-snapshot worden voorbereid." + elif local_date is None: + status = "review_required" + action = "De lokale editie is niet veilig datumvergelijkbaar; controleer de provenance vóór staging." + elif local_date == remote_edition_date: + status = "current" + action = "De lokale snapshot gebruikt dezelfde officiële editie; geen import nodig." + elif local_date < remote_edition_date: + status = "update_available" + action = "Stage eerst alle bronartifacts en controleer exacte aantallen en checksums vóór import." + else: + status = "review_required" + action = "De lokale editie lijkt nieuwer dan de catalogus; automatische terugval is verboden." + + layer_plans.append( + GrbRefreshLayerPlan( + theme=definition.theme, + display_name=definition.display_name, + collections=list(definition.collections), + temporal_series_key=definition.series_key, + status=status, + local_dataset_id=local.id if local else None, + local_source_version=local.source_version if local else None, + local_observed_at=local.observed_at if local else None, + local_imported_at=local.imported_at if local else None, + local_feature_count=cls._dataset_feature_count(local) if local else None, + local_size_bytes=local.size_bytes if local else None, + action_message=action, + ) + ) + + counts = {status: sum(item.status == status for item in layer_plans) for status in ( + "current", "update_available", "not_loaded", "review_required", "remote_unavailable" + )} + actionable = counts["update_available"] + counts["not_loaded"] + summary = GrbRefreshPlanSummary( + layer_count=len(layer_plans), + current_count=counts["current"], + update_available_count=counts["update_available"], + not_loaded_count=counts["not_loaded"], + review_required_count=counts["review_required"], + remote_unavailable_count=counts["remote_unavailable"], + new_dataset_count_if_applied=actionable, + retained_dataset_count=sum(item.local_dataset_id is not None for item in layer_plans), + current_feature_count=sum(item.local_feature_count or 0 for item in layer_plans), + current_size_bytes=sum(item.local_size_bytes or 0 for item in layer_plans), + ) + if actionable: + message = ( + f"{actionable} GRB-laag{' is' if actionable == 1 else 'en zijn'} voorbereidbaar voor editie " + f"{remote_edition_date.isoformat() if remote_edition_date else remote_version}. " + "Staging berekent eerst de exacte impact; import vereist daarna de plan-checksum." + ) + elif counts["current"] == len(layer_plans): + message = "Alle beheerde regionale GRB-lagen gebruiken de officiële cataloguseditie." + else: + message = "Er is menselijke beoordeling nodig voordat een GRB-staging kan starten." + + return GrbRefreshPlan( + project_id=project_id, + scope=scope, + generated_at=generated_at, + remote_status=grb_probe.status if grb_probe else "unavailable", + remote_version=remote_version, + remote_edition_date=remote_edition_date, + catalog_checked_at=grb_probe.checked_at if grb_probe else None, + summary=summary, + layers=layer_plans, + message=message, + limitations=[ + "Dit endpoint is read-only en start geen download, import of databasejob.", + "Staging bewaart bronartifacts buiten PostGIS; apply vereist de exacte staged plan-checksum.", + "Een refresh maakt nieuwe immutable Datasets en verwijdert of overschrijft oude snapshots niet.", + "Exacte feature- en opslagverschillen zijn pas bekend nadat alle regionale partitions staged en gevalideerd zijn.", + ], + ) diff --git a/backend/tests/test_sprint223_governed_grb_refresh.py b/backend/tests/test_sprint223_governed_grb_refresh.py new file mode 100644 index 00000000..65cf022a --- /dev/null +++ b/backend/tests/test_sprint223_governed_grb_refresh.py @@ -0,0 +1,250 @@ +from __future__ import annotations + +import importlib.util +import json +import sys +from datetime import datetime, timezone +from pathlib import Path +from types import SimpleNamespace +import uuid + +import pytest + +from app.core.errors import AppError +from app.models import Dataset +from app.services.grb_refresh_plan_service import GrbRefreshPlanService + + +NOW = datetime(2026, 7, 16, 16, 0, tzinfo=timezone.utc) +PROJECT_ID = uuid.uuid4() + + +class _Query: + def __init__(self, rows: list[Dataset]) -> None: + self.rows = rows + + def filter(self, *_args): + return self + + def all(self) -> list[Dataset]: + return self.rows + + +class _Db: + def __init__(self, rows: list[Dataset]) -> None: + self.rows = rows + + def get(self, _model, identifier): + return SimpleNamespace(id=identifier) + + def query(self, _model): + return _Query(self.rows) + + +def _dataset(theme: str, version: str = "2026-07-14", count: int = 100) -> Dataset: + return Dataset( + id=uuid.uuid4(), + project_id=PROJECT_ID, + name=f"grb_{theme}.geojson", + dataset_type="vector", + source="operator_official_import", + source_name="grb", + dataset_role="reference", + reference_layer_name=theme, + source_version=version, + temporal_series_key=f"grb:{theme}:kempen-transport-region", + observed_at=datetime.fromisoformat(f"{version}T00:00:00+00:00"), + imported_at=NOW, + metadata_json={"feature_count": count}, + size_bytes=1000, + status="ready", + ) + + +def _catalog(status: str = "available", version: str | None = "Toestand 2026-07-15"): + return SimpleNamespace( + items=[ + SimpleNamespace( + source_name="grb", + status=status, + reachable=status == "available", + remote_version=version, + checked_at=NOW, + ) + ] + ) + + +def test_refresh_plan_marks_all_older_immutable_snapshots_as_update_available(monkeypatch) -> None: + rows = [_dataset(theme, count=(index + 1) * 100) for index, theme in enumerate(("buildings", "roads", "water", "parcels"))] + monkeypatch.setattr( + "app.services.grb_refresh_plan_service.SourceCatalogProbeService.audit_project", + lambda *_args, **_kwargs: _catalog(), + ) + + plan = GrbRefreshPlanService.build(_Db(rows), PROJECT_ID, now=NOW) + + assert plan.remote_edition_date.isoformat() == "2026-07-15" + assert plan.summary.update_available_count == 4 + assert plan.summary.new_dataset_count_if_applied == 4 + assert plan.summary.retained_dataset_count == 4 + assert plan.summary.current_feature_count == 1000 + assert plan.summary.current_size_bytes == 4000 + assert all(item.status == "update_available" for item in plan.layers) + assert all(item.retained_after_refresh for item in plan.layers) + assert plan.automatic_import is False + assert plan.destructive_replacement is False + + +@pytest.mark.parametrize( + ("rows", "catalog", "expected"), + [ + ([_dataset("buildings", "2026-07-15")], _catalog(), "current"), + ([], _catalog(), "not_loaded"), + ([_dataset("buildings")], _catalog("unavailable"), "remote_unavailable"), + ([_dataset("buildings")], _catalog("available", "Onbekende toestand"), "review_required"), + ], +) +def test_refresh_plan_status_matrix(monkeypatch, rows, catalog, expected) -> None: + monkeypatch.setattr( + "app.services.grb_refresh_plan_service.SourceCatalogProbeService.audit_project", + lambda *_args, **_kwargs: catalog, + ) + plan = GrbRefreshPlanService.build(_Db(rows), PROJECT_ID, now=NOW) + buildings = next(item for item in plan.layers if item.theme == "buildings") + assert buildings.status == expected + + +def test_refresh_plan_rejects_an_unapproved_scope(monkeypatch) -> None: + monkeypatch.setattr( + "app.services.grb_refresh_plan_service.SourceCatalogProbeService.audit_project", + lambda *_args, **_kwargs: _catalog(), + ) + with pytest.raises(AppError) as raised: + GrbRefreshPlanService.build(_Db([]), PROJECT_ID, scope="mol", now=NOW) + assert raised.value.code == "GRB_REFRESH_SCOPE_UNSUPPORTED" + + +def test_refresh_plan_route_uses_the_canonical_envelope(monkeypatch) -> None: + from app.api.routes import datasets as dataset_routes + + monkeypatch.setattr( + "app.services.grb_refresh_plan_service.SourceCatalogProbeService.audit_project", + lambda *_args, **_kwargs: _catalog(), + ) + response = dataset_routes.plan_grb_dataset_refresh( + project_id=PROJECT_ID, + scope="kempen-transport-region", + refresh_catalog=False, + db=_Db([]), + ) + assert list(response) == ["data"] + assert response["data"]["project_id"] == PROJECT_ID + assert response["data"]["automatic_import"] is False + + +def _load_operator_module(): + root = Path(__file__).resolve().parents[2] + scripts = root / "scripts" + if str(scripts) not in sys.path: + sys.path.insert(0, str(scripts)) + spec = importlib.util.spec_from_file_location("manage_grb_refresh_s223", scripts / "manage_grb_refresh.py") + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_operator_requires_exact_edition_and_plan_hash() -> None: + module = _load_operator_module() + assert module.require_confirmed_edition({"remote_edition_date": "2026-07-15"}, "2026-07-15") == "2026-07-15" + with pytest.raises(RuntimeError, match="confirm-edition"): + module.require_confirmed_edition({"remote_edition_date": "2026-07-15"}, "2026-07-14") + + payload = {"status": "staged", "layers": []} + first = module.canonical_plan_sha256(payload) + assert first == module.canonical_plan_sha256({**payload, "plan_sha256": first}) + assert first != module.canonical_plan_sha256({"status": "staged", "layers": [{"theme": "roads"}]}) + + +def test_operator_validates_every_staged_artifact_and_partition(tmp_path) -> None: + module = _load_operator_module() + manifest_dir = tmp_path / "kempen-transport-region" / "buildings" / "2026-07-15" + partition_dir = manifest_dir / "partitions" + partition_dir.mkdir(parents=True) + artifact = manifest_dir / "buildings.geojson" + partition = partition_dir / "13025_mol.geojson" + artifact.write_text('{"type":"FeatureCollection","features":[]}', encoding="utf-8") + partition.write_text('{"type":"FeatureCollection","features":[]}', encoding="utf-8") + manifest = { + "status": "complete", + "scope": "kempen-transport-region", + "theme": "buildings", + "observed_at": "2026-07-15", + "reference_truncated": False, + "member_count": 1, + "feature_count": 1, + "artifact_filename": artifact.name, + "artifact_sha256": module.sha256_file(artifact), + "artifact_size_bytes": artifact.stat().st_size, + "partitions": [{"filename": partition.name, "sha256": module.sha256_file(partition)}], + } + path = manifest_dir / "regional_buildings_manifest.json" + path.write_text(json.dumps(manifest), encoding="utf-8") + + assert module.validate_manifest( + path, + output_root=tmp_path, + scope="kempen-transport-region", + theme="buildings", + edition="2026-07-15", + )["feature_count"] == 1 + + partition.write_text("changed", encoding="utf-8") + with pytest.raises(RuntimeError, match="partition checksum"): + module.validate_manifest( + path, + output_root=tmp_path, + scope="kempen-transport-region", + theme="buildings", + edition="2026-07-15", + ) + + +def test_operator_builds_only_allowlisted_local_subprocess_commands(tmp_path) -> None: + module = _load_operator_module() + args = SimpleNamespace( + scope="kempen-transport-region", + api_url="http://127.0.0.1:8000/api/v1", + output_root=tmp_path, + request_timeout=180, + api_timeout=180, + batch_size=1000, + page_limit=1000, + max_features_per_member=100000, + max_total_features=1500000, + ) + commands = module.build_operator_commands( + args, + ["buildings", "roads", "water", "parcels"], + "2026-07-15", + fetch_only=True, + ) + flattened = [argument for _label, command in commands for argument in command] + assert len(commands) == 2 + assert "--fetch-only" in flattened + assert "provision_regional_grb_buildings.py" in " ".join(flattened) + assert "provision_regional_grb_context.py" in " ".join(flattened) + assert "--force" not in flattened + + +def test_refresh_api_and_frontend_remain_explicit_only() -> None: + root = Path(__file__).resolve().parents[2] + service = (root / "backend" / "app" / "services" / "grb_refresh_plan_service.py").read_text(encoding="utf-8") + hook = (root / "frontend" / "src" / "hooks" / "useSourceFreshness.ts").read_text(encoding="utf-8") + operator = (root / "scripts" / "manage_grb_refresh.py").read_text(encoding="utf-8") + assert "DatasetService" not in service + assert "VectorFeatureService" not in service + assert "void probeCatalogs(" not in hook + assert 'choices=("plan", "stage", "apply")' in operator + assert "--confirm-plan-sha256" in operator diff --git a/docs/API_CONTRACTS.md b/docs/API_CONTRACTS.md index 744c08dc..1b203cee 100644 --- a/docs/API_CONTRACTS.md +++ b/docs/API_CONTRACTS.md @@ -432,6 +432,30 @@ not fetch vector features, raster pixels or models, create jobs/datasets, write to PostGIS or trigger an import. The normal `source-freshness` endpoint remains local-only and never invokes this probe implicitly. +### GET `/api/v1/projects/{project_id}/datasets/grb-refresh-plan` + +Builds a read-only refresh decision for the governed +`kempen-transport-region` GRB snapshot series. Query parameter +`refresh_catalog=true` explicitly bypasses the catalog cache. The response +maps the official dated edition to `buildings`, `roads`, `water` and `parcels` +and reports each latest local Dataset, source version, observation date, +feature count, artifact size and refresh state. + +Layer state is `current`, `update_available`, `not_loaded`, `review_required` +or `remote_unavailable`. The summary reports how many new immutable Datasets +would be created and how many existing snapshots remain retained. The endpoint +does not fetch GRB features, stage files, create a Job, write to PostGIS or +start an operator process. Exact remote deltas remain unavailable until every +municipality partition has been staged and validated. + +Regional execution uses `scripts/manage_grb_refresh.py` outside the request +cycle. `stage` requires the exact official ISO edition, invokes the existing +regional GRB operators with `--fetch-only`, validates all artifact/partition +checksums and emits a SHA-256-bound plan. `apply` requires that exact plan hash, +revalidates every staged byte and delegates persistence to DatasetService and +VectorFeatureService. It creates new temporal snapshots and never deletes or +overwrites an older Dataset. + ### GET `/api/v1/projects/{project_id}/datasets/{dataset_id}` Return metadata. diff --git a/docs/CODEX_EXECUTION_LOG.md b/docs/CODEX_EXECUTION_LOG.md index 3696182c..b57c4423 100644 --- a/docs/CODEX_EXECUTION_LOG.md +++ b/docs/CODEX_EXECUTION_LOG.md @@ -9339,3 +9339,36 @@ Next: - Extend the same probe pattern only where another official source publishes a stable machine-readable edition. Do not infer release versions from service protocol versions, ETags or HTTP modification dates. + +## Sprint 223 - Governed regional GRB refresh (2026-07-16) + +Implemented: +- Added a canonical read-only refresh plan that maps the official dated GRB + edition onto the immutable regional buildings, roads, water and parcel + temporal series. +- Added explicit per-layer current/update/review states, existing Dataset, + feature/storage impact and non-destructive action guidance. +- Added `manage_grb_refresh.py` with separate plan, fetch-only stage and + checksum-confirmed apply phases. Stage validates all retained artifacts and + municipality partitions; apply revalidates the exact staged plan before + calling the existing regional DatasetService-based operators. +- Added a compact Status surface after the explicit official catalog check. + +Boundaries: +- No migration, scheduler, arbitrary provider URL, browser-triggered heavy + import, direct vector-feature write or automatic Dataset replacement was + introduced. +- Existing GRB snapshots remain immutable historical observations. Exact + remote deltas are reported only after full staging, never guessed from + capabilities metadata. + +Validation so far: +- Backend compile and frontend typecheck passed. +- Source-catalog plus governed-refresh focused suite passed: 20 tests. +- Full `scripts/run_readiness_check.sh` passed: 779 backend tests, 110 + documented route contracts, one Alembic head, frontend typecheck/build and + all packaged smoke checks. + +Next: +- Run the full release gate, deploy to Tower, execute a live read-only plan and + then stage/apply only the exact confirmed official edition. diff --git a/docs/DATABASE_IMPLEMENTATION_PLAN.md b/docs/DATABASE_IMPLEMENTATION_PLAN.md index 89416396..7e7ae0c7 100644 --- a/docs/DATABASE_IMPLEMENTATION_PLAN.md +++ b/docs/DATABASE_IMPLEMENTATION_PLAN.md @@ -267,6 +267,13 @@ evidence audit. External catalogue checks and refresh jobs remain explicit operator actions; they may never overwrite a fixed edition or scenario in place. +Regional GRB refreshes add no lifecycle table. A read-only plan compares the +official dated edition with the latest Dataset in each governed temporal +series. Staging is filesystem evidence only. Checksum-confirmed apply creates a +new Dataset plus DatasetVersion and vector_features through the existing +DatasetService/VectorFeatureService transaction; prior snapshots remain +unchanged and queryable for temporal comparison. + ## Geometry normalization - User-drawn polygons arrive as EPSG:4326. diff --git a/docs/TODO.md b/docs/TODO.md index 08d26d53..edba95e2 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -40,6 +40,7 @@ - [x] Present municipal DHMV/VMM partitions as logical regional layers and analyse cross-boundary rectangles without a municipality prerequisite. - [x] Add a read-only source freshness/version audit with explicit publication policies, local integrity checks, operator CLI and compact Status workspace surface. - [x] Add explicit bounded GRB/orthophoto catalogue release probes using official capabilities and ISO metadata; keep every acquisition separate and operator-controlled. +- [x] Add a governed regional GRB plan -> stage -> checksum-confirmed apply workflow that preserves every previous snapshot. - [ ] Extend catalogue probes only to additional sources that publish a stable machine-readable edition contract; do not add background polling or infer releases from HTTP dates alone. ## Governed source expansion backlog diff --git a/frontend/README.md b/frontend/README.md index 0eb91901..85aa1a79 100644 --- a/frontend/README.md +++ b/frontend/README.md @@ -13,6 +13,13 @@ replaces source data. A separate `Officiële edities controleren` action is the only trigger for bounded GRB/orthophoto catalog reads. It presents official and local editions, layer-contract coverage and honest non-comparable version markers without starting an import or background poll. +After that explicit check, the same surface shows a compact GRB refresh plan +for buildings, roads, water and parcels. It states whether each local regional +snapshot is current, updateable or needs review and shows the current object +count. The UI cannot start the million-feature import: staging and checksum- +confirmed application remain an operator action so a browser request cannot +silently replace regional reference data. Existing snapshots remain visible +as historical observations. The user-facing shell is task based: `Kaart`, `Bronnen`, `Kwaliteit`, `Beeldanalyse`, `Downloads`, `Status` and `Beheer`. Internal benchmark projects, raw dataset metadata, provider capabilities, model registry details and QA evidence remain accessible through labelled advanced disclosures instead of competing with the normal workflow. diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 9d01c1a9..ddbe41d9 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -864,6 +864,9 @@ function App(): JSX.Element { catalogLoading={sourceFreshness.catalogLoading} catalogError={sourceFreshness.catalogError} onProbeCatalogs={(force) => { void sourceFreshness.probeCatalogs(force) }} + grbRefreshPlan={sourceFreshness.grbRefreshPlan} + grbRefreshPlanLoading={sourceFreshness.grbRefreshPlanLoading} + grbRefreshPlanError={sourceFreshness.grbRefreshPlanError} />
diff --git a/frontend/src/components/status/SourceFreshnessPanel.tsx b/frontend/src/components/status/SourceFreshnessPanel.tsx index 2e089ded..6694b920 100644 --- a/frontend/src/components/status/SourceFreshnessPanel.tsx +++ b/frontend/src/components/status/SourceFreshnessPanel.tsx @@ -1,4 +1,7 @@ import type { + GrbRefreshLayerPlan, + GrbRefreshLayerStatus, + GrbRefreshPlan, SourceCatalogComparisonStatus, SourceCatalogProbeItem, SourceCatalogProbeReport, @@ -17,6 +20,9 @@ interface SourceFreshnessPanelProps { catalogLoading: boolean catalogError: string | null onProbeCatalogs: (force: boolean) => void + grbRefreshPlan: GrbRefreshPlan | null + grbRefreshPlanLoading: boolean + grbRefreshPlanError: string | null } function statusLabel(status: SourceFreshnessStatus): string { @@ -46,6 +52,35 @@ function formatDate(value?: string | null): string { return new Intl.DateTimeFormat('nl-BE', { dateStyle: 'medium' }).format(new Date(value)) } +function formatCount(value?: number | null): string { + return value == null ? 'nog niet bekend' : value.toLocaleString('nl-BE') +} + +function grbLayerStatusLabel(status: GrbRefreshLayerStatus): string { + if (status === 'current') return 'actueel' + if (status === 'update_available') return 'update beschikbaar' + if (status === 'not_loaded') return 'nog niet ingeladen' + if (status === 'remote_unavailable') return 'bron niet bereikbaar' + return 'beoordeling nodig' +} + +function GrbRefreshLayerRow({ item, remoteEdition }: { item: GrbRefreshLayerPlan; remoteEdition?: string | null }): JSX.Element { + return ( +
+
+ {item.display_name} + {grbLayerStatusLabel(item.status)} +
+

{item.action_message}

+
+ Lokaal: {item.local_source_version ?? 'niet aanwezig'} + Officieel: {remoteEdition ?? 'onbekend'} + {formatCount(item.local_feature_count)} objecten lokaal +
+
+ ) +} + function integrityIssueCount(item: SourceFreshnessItem): number { return Object.values(item.integrity).reduce((total, value) => total + value, 0) } @@ -102,6 +137,9 @@ export function SourceFreshnessPanel({ catalogLoading, catalogError, onProbeCatalogs, + grbRefreshPlan, + grbRefreshPlanLoading, + grbRefreshPlanError, }: SourceFreshnessPanelProps): JSX.Element { const attentionItems = report?.items.filter((item) => item.status === 'due' || item.status === 'review_required') ?? [] const summary = report?.summary @@ -180,6 +218,27 @@ export function SourceFreshnessPanel({

) : null} + {grbRefreshPlanLoading ?

Veilig vernieuwingsplan wordt opgebouwd...

: null} + {grbRefreshPlanError ?

{grbRefreshPlanError}

: null} + {grbRefreshPlan ? ( +
+
+
+ Veilige GRB-vernieuwing + {grbRefreshPlan.message} +
+ {grbRefreshPlan.summary.update_available_count} updates +
+
+ {grbRefreshPlan.layers.map((item) => ( + + ))} +
+

+ Een serveroperator controleert eerst exacte aantallen en checksums. Bestaande snapshots blijven altijd behouden. +

+
+ ) : null} ) : null} diff --git a/frontend/src/hooks/useSourceFreshness.ts b/frontend/src/hooks/useSourceFreshness.ts index ec3549a1..fb845a95 100644 --- a/frontend/src/hooks/useSourceFreshness.ts +++ b/frontend/src/hooks/useSourceFreshness.ts @@ -1,7 +1,7 @@ import { useCallback, useEffect, useRef, useState } from 'react' import { datasetsApi } from '../services/api' -import type { SourceCatalogProbeReport, SourceFreshnessReport } from '../types' +import type { GrbRefreshPlan, SourceCatalogProbeReport, SourceFreshnessReport } from '../types' export function useSourceFreshness(selectedProjectId: string | null) { const [report, setReport] = useState(null) @@ -12,6 +12,9 @@ export function useSourceFreshness(selectedProjectId: string | null) { const [catalogReport, setCatalogReport] = useState(null) const [catalogLoading, setCatalogLoading] = useState(false) const [catalogError, setCatalogError] = useState(null) + const [grbRefreshPlan, setGrbRefreshPlan] = useState(null) + const [grbRefreshPlanLoading, setGrbRefreshPlanLoading] = useState(false) + const [grbRefreshPlanError, setGrbRefreshPlanError] = useState(null) const refresh = useCallback(async () => { const requestId = ++requestSequence.current @@ -45,10 +48,15 @@ export function useSourceFreshness(selectedProjectId: string | null) { setCatalogReport(null) setCatalogError(null) setCatalogLoading(false) + setGrbRefreshPlan(null) + setGrbRefreshPlanError(null) + setGrbRefreshPlanLoading(false) return } setCatalogLoading(true) setCatalogError(null) + setGrbRefreshPlanLoading(true) + setGrbRefreshPlanError(null) try { const nextReport = await datasetsApi.sourceCatalogProbes(selectedProjectId, force) if (catalogRequestSequence.current === requestId) { @@ -58,9 +66,25 @@ export function useSourceFreshness(selectedProjectId: string | null) { if (catalogRequestSequence.current === requestId) { setCatalogError(caught instanceof Error ? caught.message : 'De officiële broncatalogi konden niet worden gecontroleerd.') } + if (catalogRequestSequence.current === requestId) { + setCatalogLoading(false) + setGrbRefreshPlanLoading(false) + } + return + } + try { + const nextPlan = await datasetsApi.grbRefreshPlan(selectedProjectId) + if (catalogRequestSequence.current === requestId) { + setGrbRefreshPlan(nextPlan) + } + } catch (caught) { + if (catalogRequestSequence.current === requestId) { + setGrbRefreshPlanError(caught instanceof Error ? caught.message : 'Het veilige GRB-vernieuwingsplan kon niet worden opgebouwd.') + } } finally { if (catalogRequestSequence.current === requestId) { setCatalogLoading(false) + setGrbRefreshPlanLoading(false) } } }, [selectedProjectId]) @@ -77,6 +101,9 @@ export function useSourceFreshness(selectedProjectId: string | null) { setCatalogReport(null) setCatalogError(null) setCatalogLoading(false) + setGrbRefreshPlan(null) + setGrbRefreshPlanError(null) + setGrbRefreshPlanLoading(false) }, [selectedProjectId]) return { @@ -88,5 +115,8 @@ export function useSourceFreshness(selectedProjectId: string | null) { catalogLoading, catalogError, probeCatalogs, + grbRefreshPlan, + grbRefreshPlanLoading, + grbRefreshPlanError, } } diff --git a/frontend/src/services/api/datasets.ts b/frontend/src/services/api/datasets.ts index 6c17eab1..06f7e968 100644 --- a/frontend/src/services/api/datasets.ts +++ b/frontend/src/services/api/datasets.ts @@ -29,6 +29,7 @@ import type { ThematicRasterSelectionResponse, SourceFreshnessReport, SourceCatalogProbeReport, + GrbRefreshPlan, } from '../../types' const DATASET_PAGE_SIZE = 200 @@ -66,6 +67,10 @@ export const datasetsApi = { apiGet( `/api/v1/projects/${projectId}/datasets/source-catalog-probes?refresh=${refresh ? 'true' : 'false'}`, ), + grbRefreshPlan: (projectId: string, refreshCatalog = false): Promise => + apiGet( + `/api/v1/projects/${projectId}/datasets/grb-refresh-plan?refresh_catalog=${refreshCatalog ? 'true' : 'false'}`, + ), upload: ( projectId: string, payload: { diff --git a/frontend/src/styles/premium.css b/frontend/src/styles/premium.css index 74d8080e..61d2089b 100644 --- a/frontend/src/styles/premium.css +++ b/frontend/src/styles/premium.css @@ -1041,6 +1041,95 @@ details.ai-lab-model-surface > summary strong { font-size: 0.75rem; } +.grb-refresh-plan { + border-top: 1px solid var(--line); + background: #ffffff; +} + +.grb-refresh-plan-heading { + display: flex; + align-items: center; + justify-content: space-between; + gap: 1rem; + padding: 0.8rem 1rem; +} + +.grb-refresh-plan-heading > div { + display: grid; + min-width: 0; + gap: 0.15rem; +} + +.grb-refresh-plan-heading strong { + color: #23313b; + font-size: 0.84rem; +} + +.grb-refresh-plan-heading span { + color: var(--muted); + font-size: 0.75rem; +} + +.grb-refresh-plan-heading > span { + flex: 0 0 auto; + font-weight: 750; +} + +.grb-refresh-layer-list { + display: grid; + grid-template-columns: repeat(2, minmax(0, 1fr)); + border-top: 1px solid var(--line); +} + +.grb-refresh-layer { + min-width: 0; + padding: 0.78rem 1rem; + border-right: 1px solid var(--line); + border-bottom: 1px solid var(--line); + box-shadow: inset 3px 0 0 #8a98a3; +} + +.grb-refresh-layer:nth-child(2n) { + border-right: 0; +} + +.grb-refresh-layer:nth-last-child(-n + 2) { + border-bottom: 0; +} + +.grb-refresh-layer-update_available, +.grb-refresh-layer-not_loaded { + box-shadow: inset 3px 0 0 #ca7a18; +} + +.grb-refresh-layer-current { + box-shadow: inset 3px 0 0 #2d9272; +} + +.grb-refresh-layer-remote_unavailable, +.grb-refresh-layer-review_required { + box-shadow: inset 3px 0 0 #b84e4e; +} + +.grb-refresh-layer > div:first-child { + display: flex; + align-items: center; + justify-content: space-between; + gap: 0.75rem; +} + +.grb-refresh-layer > div:first-child span { + color: #4d5c66; + font-size: 0.72rem; + font-weight: 750; +} + +.grb-refresh-layer p { + margin: 0.4rem 0; + color: var(--muted); + font-size: 0.75rem; +} + .source-freshness-details > summary { display: flex; cursor: pointer; @@ -1097,6 +1186,26 @@ details.ai-lab-model-surface > summary strong { .source-catalog-row:last-child { border-bottom: 0; } + + .grb-refresh-plan-heading { + align-items: flex-start; + flex-direction: column; + } + + .grb-refresh-layer-list { + grid-template-columns: 1fr; + } + + .grb-refresh-layer, + .grb-refresh-layer:nth-child(2n), + .grb-refresh-layer:nth-last-child(-n + 2) { + border-right: 0; + border-bottom: 1px solid var(--line); + } + + .grb-refresh-layer:last-child { + border-bottom: 0; + } } .workflow-guidance-panel { diff --git a/frontend/src/types.ts b/frontend/src/types.ts index e8c4915e..9163943e 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -755,6 +755,58 @@ export interface SourceCatalogProbeReport { limitations: string[] } +export type GrbRefreshLayerStatus = + | 'current' + | 'update_available' + | 'not_loaded' + | 'review_required' + | 'remote_unavailable' + +export interface GrbRefreshLayerPlan { + theme: 'buildings' | 'roads' | 'water' | 'parcels' + display_name: string + collections: string[] + temporal_series_key: string + status: GrbRefreshLayerStatus + local_dataset_id?: string | null + local_source_version?: string | null + local_observed_at?: string | null + local_imported_at?: string | null + local_feature_count?: number | null + local_size_bytes?: number | null + retained_after_refresh: boolean + action_message: string +} + +export interface GrbRefreshPlan { + project_id: string + scope: string + generated_at: string + remote_status: string + remote_version?: string | null + remote_edition_date?: string | null + catalog_checked_at?: string | null + summary: { + layer_count: number + current_count: number + update_available_count: number + not_loaded_count: number + review_required_count: number + remote_unavailable_count: number + new_dataset_count_if_applied: number + retained_dataset_count: number + current_feature_count: number + current_size_bytes: number + } + layers: GrbRefreshLayerPlan[] + execution_mode: 'operator_stage_then_apply' + staging_required: true + automatic_import: false + destructive_replacement: false + message: string + limitations: string[] +} + export interface GeojsonEnvelopeResponse { data: object } diff --git a/scripts/README.md b/scripts/README.md index b8a30e13..fb0dc960 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -1716,6 +1716,42 @@ JSON output contains `source_freshness` and `catalog_probes`; without it, the original local report shape is unchanged. No flag downloads provider features or imagery and no flag writes a Dataset. +## Governed GRB refresh + +Use the refresh coordinator only inside the GeoIntel container. The default +`plan` action is read-only and prints the canonical decision from the API: + +```bash +docker exec geointel python /app/scripts/manage_grb_refresh.py plan \ + --project-id 82a85913-c522-45d7-84a1-02b393d89e55 \ + --api-url http://127.0.0.1:8000/api/v1 \ + --refresh-catalog +``` + +A refresh is deliberately split into two operator approvals: + +```bash +docker exec geointel python /app/scripts/manage_grb_refresh.py stage \ + --project-id 82a85913-c522-45d7-84a1-02b393d89e55 \ + --api-url http://127.0.0.1:8000/api/v1 \ + --confirm-edition 2026-07-15 + +docker exec geointel python /app/scripts/manage_grb_refresh.py apply \ + --project-id 82a85913-c522-45d7-84a1-02b393d89e55 \ + --api-url http://127.0.0.1:8000/api/v1 \ + --confirm-edition 2026-07-15 \ + --confirm-plan-sha256 SHA256_FROM_STAGE +``` + +`stage` delegates to the existing buildings/context operators in fetch-only +mode, validates every municipality partition and writes +`/app/storage/operator-evidence/grb-refresh///staged-plan.json`. +The plan contains exact feature-count deltas and artifact sizes. `apply` +revalidates the entire plan and refuses any mismatched byte before delegating +to the DatasetService-based persistence path. Existing snapshots are retained. +The coordinator accepts no provider URL, collection name, arbitrary process or +automatic schedule. + ## Tower deployment Push the local branch to Gitea, then rebuild the Unraid/Tower Docker runtime: diff --git a/scripts/manage_grb_refresh.py b/scripts/manage_grb_refresh.py new file mode 100644 index 00000000..cf87b3c1 --- /dev/null +++ b/scripts/manage_grb_refresh.py @@ -0,0 +1,373 @@ +#!/usr/bin/env python3 +"""Plan, stage and explicitly apply a governed regional GRB refresh. + +Planning is read-only. Staging downloads resumable municipality partitions but +does not persist application data. Applying requires the exact staged plan +SHA-256 and delegates persistence to the existing DatasetService-based regional +operators. Existing snapshots are immutable and remain available. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from urllib.error import HTTPError, URLError +from urllib.parse import urlparse +from urllib.request import Request, urlopen + +from geographic_scopes import GEOGRAPHIC_SCOPES + + +DEFAULT_API_URL = "http://127.0.0.1:8000/api/v1" +DEFAULT_SCOPE = "kempen-transport-region" +DEFAULT_OUTPUT_ROOT = Path("/app/storage/operator-data/regional-themes") +DEFAULT_EVIDENCE_ROOT = Path("/app/storage/operator-evidence/grb-refresh") +THEMES = ("buildings", "roads", "water", "parcels") + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Governed GRB refresh: plan, stage, then checksum-confirmed apply.") + parser.add_argument("action", choices=("plan", "stage", "apply")) + parser.add_argument("--project-id", required=True, help="GeoIntel project UUID for the governed scope") + parser.add_argument("--scope", choices=(DEFAULT_SCOPE,), default=DEFAULT_SCOPE) + parser.add_argument("--api-url", default=os.environ.get("GEOINTEL_API_URL", DEFAULT_API_URL)) + parser.add_argument("--layers", nargs="+", default=list(THEMES), help="Subset: buildings roads water parcels") + parser.add_argument("--confirm-edition", help="Exact official ISO edition date required for stage/apply") + parser.add_argument("--confirm-plan-sha256", help="Exact staged plan hash required for apply") + parser.add_argument("--plan-path", type=Path, help="Staged plan path; defaults to the governed evidence location") + parser.add_argument("--output-root", type=Path, default=Path(os.environ.get("GEOINTEL_REGIONAL_THEME_OUTPUT_ROOT", DEFAULT_OUTPUT_ROOT))) + parser.add_argument("--evidence-root", type=Path, default=Path(os.environ.get("GEOINTEL_OPERATOR_EVIDENCE_ROOT", DEFAULT_EVIDENCE_ROOT))) + parser.add_argument("--refresh-catalog", action="store_true", help="Bypass the short catalog probe cache") + parser.add_argument("--request-timeout", type=int, default=180) + parser.add_argument("--api-timeout", type=int, default=180) + parser.add_argument("--batch-size", type=int, default=1000) + parser.add_argument("--page-limit", type=int, default=1000) + parser.add_argument("--max-features-per-member", type=int, default=100000) + parser.add_argument("--max-total-features", type=int, default=1500000) + return parser.parse_args() + + +def selected_themes(values: str | list[str]) -> list[str]: + raw = [values] if isinstance(values, str) else values + requested = {item.strip().lower() for value in raw for item in value.split(",") if item.strip()} + unknown = requested - set(THEMES) + if unknown or not requested: + raise ValueError(f"Unsupported GRB layers: {sorted(unknown)}") + return [theme for theme in THEMES if theme in requested] + + +def api_data(api_url: str, path: str, timeout: int) -> dict[str, Any]: + endpoint = f"{api_url.rstrip('/')}/{path.lstrip('/')}" + request = Request(endpoint, headers={"Accept": "application/json", "User-Agent": "GeoIntel-GRB-refresh/1.0"}) + try: + with urlopen(request, timeout=timeout) as response: + payload = json.load(response) + except HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace") + raise RuntimeError(f"GeoIntel API returned HTTP {exc.code}: {body[-1000:]}") from exc + except URLError as exc: + raise RuntimeError(f"GeoIntel API is unreachable: {exc.reason}") from exc + if not isinstance(payload, dict) or not isinstance(payload.get("data"), dict): + raise RuntimeError("GeoIntel API response is not a canonical data envelope") + return payload["data"] + + +def fetch_refresh_plan(args: argparse.Namespace, *, refresh: bool) -> dict[str, Any]: + query = "true" if refresh else "false" + return api_data( + args.api_url, + f"projects/{args.project_id}/datasets/grb-refresh-plan?scope={args.scope}&refresh_catalog={query}", + args.api_timeout, + ) + + +def validate_project_scope(args: argparse.Namespace) -> None: + project = api_data(args.api_url, f"projects/{args.project_id}", args.api_timeout) + expected_name = GEOGRAPHIC_SCOPES[args.scope].project_name + if project.get("name") != expected_name: + raise RuntimeError( + f"Project {args.project_id} is '{project.get('name')}', but scope {args.scope} requires '{expected_name}'" + ) + + +def require_confirmed_edition(plan: dict[str, Any], confirmation: str | None) -> str: + edition = str(plan.get("remote_edition_date") or "") + if not edition: + raise RuntimeError("The official GRB catalog did not provide a safe ISO edition date") + if confirmation != edition: + raise RuntimeError(f"Explicit --confirm-edition {edition} is required") + return edition + + +def internal_base_url(api_url: str) -> str: + value = api_url.rstrip("/") + if value.endswith("/api/v1"): + value = value[:-7] + parsed = urlparse(value) + if parsed.hostname not in {"127.0.0.1", "localhost", "::1"}: + raise RuntimeError("Stage/apply must run inside GeoIntel against the local backend API") + return value + + +def build_operator_commands( + args: argparse.Namespace, + themes: list[str], + edition: str, + *, + fetch_only: bool, +) -> list[tuple[str, list[str]]]: + scripts_dir = Path(__file__).resolve().parent + common = [ + "--scope", args.scope, + "--observed-date", edition, + "--base-url", internal_base_url(args.api_url), + "--output-root", str(args.output_root), + "--request-timeout", str(args.request_timeout), + "--api-timeout", str(args.api_timeout), + "--batch-size", str(args.batch_size), + "--page-limit", str(args.page_limit), + "--max-features-per-member", str(args.max_features_per_member), + "--max-total-features", str(args.max_total_features), + ] + if fetch_only: + common.append("--fetch-only") + commands: list[tuple[str, list[str]]] = [] + if "buildings" in themes: + commands.append(("buildings", [sys.executable, str(scripts_dir / "provision_regional_grb_buildings.py"), *common])) + context = [theme for theme in themes if theme != "buildings"] + if context: + commands.append( + ( + "context", + [ + sys.executable, + str(scripts_dir / "provision_regional_grb_context.py"), + "--layers", + *context, + *common, + ], + ) + ) + return commands + + +def run_operator(label: str, command: list[str]) -> dict[str, Any]: + print(f"GRB {label}: {'staging' if '--fetch-only' in command else 'applying'}...", file=sys.stderr, flush=True) + completed = subprocess.run(command, check=False, capture_output=True, text=True, encoding="utf-8") + if completed.returncode != 0: + detail = completed.stderr.strip() or completed.stdout.strip() or "operator returned no diagnostics" + raise RuntimeError(f"GRB {label} failed with exit {completed.returncode}: {detail[-3000:]}") + try: + payload = json.loads(completed.stdout) + except json.JSONDecodeError as exc: + raise RuntimeError(f"GRB {label} returned invalid JSON: {completed.stdout[-1000:]}") from exc + if payload.get("status") != "ok": + raise RuntimeError(f"GRB {label} did not report success") + return payload + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(8 * 1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def canonical_plan_sha256(payload: dict[str, Any]) -> str: + content = {key: value for key, value in payload.items() if key != "plan_sha256"} + encoded = json.dumps(content, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def manifest_path(output_root: Path, scope: str, theme: str, edition: str) -> Path: + name = "regional_buildings_manifest.json" if theme == "buildings" else f"regional_{theme}_manifest.json" + return output_root / scope / theme / edition / name + + +def validate_manifest(path: Path, *, output_root: Path, scope: str, theme: str, edition: str) -> dict[str, Any]: + root = output_root.resolve() + resolved = path.resolve() + if not resolved.is_relative_to(root): + raise RuntimeError(f"Manifest is outside the governed output root: {path}") + if not path.is_file(): + raise RuntimeError(f"Staged manifest is missing: {path}") + payload = json.loads(path.read_text(encoding="utf-8")) + if payload.get("status") != "complete" or payload.get("scope") != scope or payload.get("theme") != theme: + raise RuntimeError(f"Staged manifest identity is invalid: {path}") + if payload.get("observed_at") != edition or payload.get("reference_truncated") is not False: + raise RuntimeError(f"Staged manifest edition or completeness is invalid: {path}") + artifact = path.parent / str(payload.get("artifact_filename") or "") + if not artifact.is_file() or sha256_file(artifact) != payload.get("artifact_sha256"): + raise RuntimeError(f"Staged artifact checksum is invalid: {artifact}") + partitions = payload.get("partitions") + if not isinstance(partitions, list) or len(partitions) != int(payload.get("member_count") or 0): + raise RuntimeError(f"Staged partition count is invalid: {path}") + partition_root = path.parent / "partitions" + for item in partitions: + partition = partition_root / str(item.get("filename") or "") + if not partition.is_file() or sha256_file(partition) != item.get("sha256"): + raise RuntimeError(f"Staged partition checksum is invalid: {partition}") + return payload + + +def default_plan_path(args: argparse.Namespace, edition: str) -> Path: + return args.evidence_root / args.scope / edition / "staged-plan.json" + + +def build_staged_plan( + args: argparse.Namespace, + remote_plan: dict[str, Any], + themes: list[str], + edition: str, +) -> dict[str, Any]: + local_by_theme = {item["theme"]: item for item in remote_plan.get("layers", [])} + layers: list[dict[str, Any]] = [] + for theme in themes: + path = manifest_path(args.output_root, args.scope, theme, edition) + manifest = validate_manifest(path, output_root=args.output_root, scope=args.scope, theme=theme, edition=edition) + local = local_by_theme.get(theme, {}) + staged_count = int(manifest["feature_count"]) + local_count = local.get("local_feature_count") + layers.append( + { + "theme": theme, + "collections": manifest.get("grb_collections") or (["GBG"] if theme == "buildings" else []), + "manifest_path": str(path), + "artifact_path": str(path.parent / manifest["artifact_filename"]), + "artifact_sha256": manifest["artifact_sha256"], + "artifact_size_bytes": int(manifest["artifact_size_bytes"]), + "partition_count": len(manifest["partitions"]), + "staged_feature_count": staged_count, + "local_dataset_id": local.get("local_dataset_id"), + "local_source_version": local.get("local_source_version"), + "local_feature_count": local_count, + "feature_count_delta": staged_count - int(local_count) if local_count is not None else None, + "existing_snapshot_retained": True, + } + ) + payload: dict[str, Any] = { + "schema_version": 1, + "status": "staged", + "created_at": datetime.now(timezone.utc).isoformat(), + "project_id": args.project_id, + "scope": args.scope, + "official_remote_version": remote_plan.get("remote_version"), + "edition": edition, + "layers": layers, + "total_staged_feature_count": sum(item["staged_feature_count"] for item in layers), + "total_artifact_size_bytes": sum(item["artifact_size_bytes"] for item in layers), + "automatic_import": False, + "destructive_replacement": False, + "apply_requires_plan_sha256": True, + } + payload["plan_sha256"] = canonical_plan_sha256(payload) + return payload + + +def write_json(path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(f"{path.suffix}.partial") + temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8") + temporary.replace(path) + + +def load_and_validate_staged_plan(args: argparse.Namespace, edition: str) -> tuple[Path, dict[str, Any]]: + path = args.plan_path or default_plan_path(args, edition) + if not path.is_file(): + raise RuntimeError(f"Staged refresh plan is missing: {path}") + payload = json.loads(path.read_text(encoding="utf-8")) + actual_sha = canonical_plan_sha256(payload) + if payload.get("plan_sha256") != actual_sha: + raise RuntimeError("Staged refresh plan checksum is invalid") + if args.confirm_plan_sha256 != actual_sha: + raise RuntimeError(f"Explicit --confirm-plan-sha256 {actual_sha} is required") + if payload.get("status") != "staged" or payload.get("project_id") != args.project_id: + raise RuntimeError("Staged refresh plan project or status is invalid") + if payload.get("scope") != args.scope or payload.get("edition") != edition: + raise RuntimeError("Staged refresh plan scope or edition is invalid") + for layer in payload.get("layers") or []: + manifest = validate_manifest( + Path(layer["manifest_path"]), + output_root=args.output_root, + scope=args.scope, + theme=layer["theme"], + edition=edition, + ) + if manifest.get("artifact_sha256") != layer.get("artifact_sha256"): + raise RuntimeError(f"Staged plan no longer matches {layer['theme']} artifact") + return path, payload + + +def main() -> int: + args = parse_args() + try: + themes = selected_themes(args.layers) + if min(args.request_timeout, args.api_timeout, args.batch_size, args.page_limit, args.max_features_per_member, args.max_total_features) <= 0: + raise ValueError("All timeout, paging, batching and feature safety limits must be positive") + validate_project_scope(args) + remote_plan = fetch_refresh_plan(args, refresh=args.refresh_catalog or args.action == "stage") + if args.action == "plan": + print(json.dumps(remote_plan, ensure_ascii=False, indent=2)) + return 0 + + edition = require_confirmed_edition(remote_plan, args.confirm_edition) + if args.action == "stage": + actionable = { + item["theme"] + for item in remote_plan.get("layers", []) + if item.get("status") in {"update_available", "not_loaded"} + } + blocked = set(themes) - actionable + if blocked: + raise RuntimeError(f"Layers are not safely stageable according to the refresh plan: {sorted(blocked)}") + commands = build_operator_commands(args, themes, edition, fetch_only=True) + operator_results = {label: run_operator(label, command) for label, command in commands} + staged = build_staged_plan(args, remote_plan, themes, edition) + staged["operator_results"] = operator_results + staged["plan_sha256"] = canonical_plan_sha256(staged) + path = args.plan_path or default_plan_path(args, edition) + write_json(path, staged) + print(json.dumps({"status": "staged", "plan_path": str(path), **staged}, ensure_ascii=False, indent=2)) + return 0 + + path, staged = load_and_validate_staged_plan(args, edition) + staged_themes = [item["theme"] for item in staged["layers"]] + commands = build_operator_commands(args, staged_themes, edition, fetch_only=False) + operator_results = {label: run_operator(label, command) for label, command in commands} + final_plan = fetch_refresh_plan(args, refresh=False) + by_theme = {item["theme"]: item for item in final_plan.get("layers", [])} + incomplete = [theme for theme in staged_themes if by_theme.get(theme, {}).get("status") != "current"] + if incomplete: + raise RuntimeError(f"Applied datasets did not become current: {incomplete}") + evidence = { + "schema_version": 1, + "status": "applied", + "applied_at": datetime.now(timezone.utc).isoformat(), + "project_id": args.project_id, + "scope": args.scope, + "edition": edition, + "staged_plan_path": str(path), + "staged_plan_sha256": staged["plan_sha256"], + "operator_results": operator_results, + "resulting_layers": [by_theme[theme] for theme in staged_themes], + "existing_snapshots_retained": True, + } + evidence_path = path.with_name("applied-evidence.json") + write_json(evidence_path, evidence) + print(json.dumps({"status": "applied", "evidence_path": str(evidence_path), **evidence}, ensure_ascii=False, indent=2)) + return 0 + except (OSError, RuntimeError, ValueError, KeyError, json.JSONDecodeError) as exc: + print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False), file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/run_readiness_check.sh b/scripts/run_readiness_check.sh index fe049a09..cdd8f52b 100755 --- a/scripts/run_readiness_check.sh +++ b/scripts/run_readiness_check.sh @@ -66,6 +66,7 @@ ${PYTHON_BIN} -m py_compile scripts/provision_geographic_scope.py ${PYTHON_BIN} -m py_compile scripts/provision_regional_grb_buildings.py ${PYTHON_BIN} -m py_compile scripts/provision_regional_grb_context.py ${PYTHON_BIN} -m py_compile scripts/audit_source_freshness.py +${PYTHON_BIN} -m py_compile scripts/manage_grb_refresh.py ${PYTHON_BIN} -m py_compile scripts/export_operator_yolo_dataset.py ${PYTHON_BIN} -m py_compile scripts/export_operator_yolo_tile_dataset.py ${PYTHON_BIN} -m py_compile scripts/audit_operator_yolo_dataset_quality.py