diff --git a/CHANGELOG.md b/CHANGELOG.md index 69d595b1..90a8b597 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,25 @@ # Changelog +## Sprint 228 Governed Statbel population release promotion (2026-07-17) + +- Added an operator-only `plan -> stage -> review -> apply` coordinator for + future official Statbel population editions. Plan is read-only, stage is + preflight/filesystem-only, review requires a named approval and apply + requires exact plan plus review SHA-256 values. +- Derived population and matching geometry URLs only from the allowlisted + catalog year/layout contract. Current, older, unavailable, ambiguous or + catalog-drifted releases fail closed; the coordinator accepts no arbitrary + provider URL and creates no scheduler or background fetch. +- Extended the existing population provisioner to accept one complete + explicitly governed future release, bound downloads before buffering and + discover the latest retained baseline across dynamically added years. +- Revalidated every source, manifest and snapshot byte immediately before the + canonical DatasetService upload. Dataset discovery now paginates completely, + preserving idempotence in workspaces with more than 200 datasets. +- Added focused release-state, trust-boundary, tamper, review and apply tests, + Docker packaging and readiness compilation without API or migration changes. + ## Sprint 227 Statbel population import compatibility preflight (2026-07-16) - Added a local, fail-closed preflight for staged official Statbel population diff --git a/backend/README.md b/backend/README.md index 982c003b..f3a64206 100644 --- a/backend/README.md +++ b/backend/README.md @@ -1073,6 +1073,46 @@ excluded from spatial metrics. The 2025 REDEGEO contract deliberately compares explicit municipality fields; it does not assume that `CD_SECTOR` still starts with the current `CD_REFNIS` after municipal mergers. +Future official editions use the separate four-phase release coordinator. The +project id must belong to `Kempen Regional Workbench`: + +```bash +docker exec geointel python /app/scripts/manage_statbel_population_release.py plan \ + --project-id \ + --api-url http://127.0.0.1:8000/api/v1 \ + --refresh-catalog + +docker exec geointel python /app/scripts/manage_statbel_population_release.py stage \ + --project-id \ + --api-url http://127.0.0.1:8000/api/v1 \ + --confirm-edition \ + --confirm-layout + +docker exec geointel python /app/scripts/manage_statbel_population_release.py review \ + --project-id \ + --api-url http://127.0.0.1:8000/api/v1 \ + --confirm-edition \ + --confirm-layout \ + --confirm-plan-sha256 \ + --approve --reviewer "" \ + --review-note "Schema, totalen, ZZZZ en geometrieherstel nagekeken" + +docker exec geointel python /app/scripts/manage_statbel_population_release.py apply \ + --project-id \ + --api-url http://127.0.0.1:8000/api/v1 \ + --confirm-edition \ + --confirm-layout \ + --confirm-plan-sha256 \ + --confirm-review-sha256 +``` + +`plan` is read-only and creates no file. `stage` always uses bounded fresh +downloads and `--fetch-only`; `review` imports nothing; `apply` revalidates +the current catalog, plan, review, source archives, preflight manifest and +derived snapshot before using the existing upload API. An already-current +release cannot be staged. A repeated successful apply resolves the existing +Dataset through complete paginated lookup rather than creating a duplicate. + Historical land-use work can be bounded explicitly: ```bash diff --git a/backend/tests/test_sprint228_statbel_release_management.py b/backend/tests/test_sprint228_statbel_release_management.py new file mode 100644 index 00000000..1b2eedcb --- /dev/null +++ b/backend/tests/test_sprint228_statbel_release_management.py @@ -0,0 +1,437 @@ +from __future__ import annotations + +import argparse +from hashlib import sha256 +import importlib.util +import json +from pathlib import Path +import sys + +import pytest + + +ROOT = Path(__file__).resolve().parents[2] +SCRIPTS = ROOT / "scripts" +if str(SCRIPTS) not in sys.path: + sys.path.insert(0, str(SCRIPTS)) + + +def load_script(name: str): + path = SCRIPTS / name + module_name = f"test_{path.stem}_sprint228" + spec = importlib.util.spec_from_file_location(module_name, path) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + spec.loader.exec_module(module) + return module + + +MANAGER = load_script("manage_statbel_population_release.py") +OPERATOR = load_script("provision_mol_population_history.py") + + +def arguments(tmp_path: Path, **overrides) -> argparse.Namespace: + values = { + "action": "plan", + "project_id": "00000000-0000-0000-0000-000000000001", + "scope": "kempen-transport-region", + "api_url": "http://127.0.0.1:8000/api/v1", + "confirm_edition": None, + "confirm_layout": None, + "confirm_plan_sha256": None, + "confirm_review_sha256": None, + "approve": False, + "reviewer": None, + "review_note": "", + "plan_path": None, + "review_path": None, + "output_root": tmp_path / "operator-data" / "regional-timeseries", + "scope_output_root": tmp_path / "operator-data" / "geographic-scopes", + "evidence_root": tmp_path / "operator-evidence" / "statbel-population-refresh", + "refresh_catalog": False, + "request_timeout": 300, + "api_timeout": 180, + "import_timeout": 3600, + } + values.update(overrides) + return argparse.Namespace(**values) + + +def catalog_item(*, remote: str = "2026", local: str | None = "2025", catalog_hash: str = "a" * 64) -> dict: + return { + "source_name": "statbel", + "status": "available", + "reachable": True, + "error_code": None, + "remote_version": remote, + "local_source_version": local, + "remote_title": f"Bevolking per statistische sector {remote} (nieuwe REDEGEO-sectorindeling)", + "message": "De officiele catalogus bevestigt de nieuwe REDEGEO-sectorindeling.", + "matched_layers": ["population_txt_current", "landing_page", "cc_by_4_0"], + "capabilities_sha256": catalog_hash, + "metadata_identifier": f"NodeID{remote}", + "metadata_url": f"https://statbel.fgov.be/nl/open-data/bevolking-statistische-sector-{remote}", + "checked_at": "2026-07-17T08:00:00Z", + } + + +def decision(args: argparse.Namespace, *, remote: str = "2026", local: str | None = "2025") -> dict: + return MANAGER.fetch_release_decision_from_item(args, catalog_item(remote=remote, local=local)) + + +def release_2026(): + return MANAGER.release_from_catalog_item(catalog_item()) + + +def write_staged_artifacts(args: argparse.Namespace, release) -> None: + scope = MANAGER.GEOGRAPHIC_SCOPES[args.scope] + output_dir = MANAGER.population_output_dir(args) + raw_dir = output_dir / "raw" / str(release.year) + raw_dir.mkdir(parents=True, exist_ok=True) + population_path = raw_dir / "OPENDATA_SECTOREN_2026_NEW.zip" + geometry_path = raw_dir / "sh_statbel_statistical_sectors_31370_20260101.geojson.zip" + population_path.write_bytes(b"official population archive") + geometry_path.write_bytes(b"official geometry archive") + snapshot = MANAGER.snapshot_path(output_dir, scope, release.year) + snapshot.write_text('{"type":"FeatureCollection","features":[]}', encoding="utf-8") + manifest = { + "schema_version": 1, + "status": "passed", + "import_eligible": True, + "release": {"year": 2026, "population_layout": "new"}, + "artifacts": { + "population": { + "source_url": release.population_url, + "archive_sha256": sha256(population_path.read_bytes()).hexdigest(), + "archive_size_bytes": population_path.stat().st_size, + "retained_path": str(population_path), + }, + "geometry": { + "source_url": release.geometry_url, + "archive_sha256": sha256(geometry_path.read_bytes()).hexdigest(), + "archive_size_bytes": geometry_path.stat().st_size, + "retained_path": str(geometry_path), + }, + "derived_snapshot": { + "sha256": sha256(snapshot.read_bytes()).hexdigest(), + "size_bytes": snapshot.stat().st_size, + "feature_count": 700, + }, + }, + "scope_accounting": { + "scope_key": scope.key, + "spatial_sector_count": 700, + "spatial_population_total": 510000, + "unlocated_population_total": 300, + "accounted_population_total": 510300, + }, + "national_accounting": {"population_total": 12000000}, + "baseline": {"year": 2025, "annual_change_ratio": 0.007}, + "schemas": {"geometry_repair_count": 2}, + } + manifest_path = MANAGER.preflight_manifest_path(output_dir, scope, release.year) + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + + +def staged_plan(args: argparse.Namespace, release, release_decision: dict) -> tuple[Path, dict]: + write_staged_artifacts(args, release) + result = { + "status": "ok", + "scope": args.scope, + "snapshots": [ + { + "year": release.year, + "status": "prepared", + "preflight_status": "passed", + "feature_count": 700, + } + ], + } + plan = MANAGER.build_staged_plan(args, release_decision, release, result) + path = MANAGER.default_plan_path(args, release.year) + MANAGER.write_json(path, plan) + return path, plan + + +def test_future_release_config_is_strict_and_previous_snapshot_is_discovered(tmp_path: Path) -> None: + release = OPERATOR.resolve_release_config( + 2026, + layout="new", + population_url=( + "https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/" + "OPENDATA_SECTOREN_2026_NEW.zip" + ), + geometry_url=( + "https://statbel.fgov.be/sites/default/files/files/opendata/Statistische%20sectoren/" + "sh_statbel_statistical_sectors_31370_20260101.geojson.zip" + ), + ) + scope = OPERATOR.GEOGRAPHIC_SCOPES["mol"] + baseline = OPERATOR.snapshot_path(tmp_path, scope, 2026) + baseline.write_text("{}", encoding="utf-8") + + assert release.year == 2026 + assert release.layout == "new" + assert OPERATOR.previous_snapshot_path(tmp_path, scope, 2027) == baseline + with pytest.raises(ValueError, match="supplied together"): + OPERATOR.resolve_release_config(2026, layout="new") + + +def test_population_workspace_pagination_reads_every_dataset() -> None: + rows = [{"id": index} for index in range(401)] + + class Response: + ok = True + status_code = 200 + text = "" + + def __init__(self, payload: dict) -> None: + self.payload = payload + + def json(self) -> dict: + return {"data": self.payload} + + class Session: + def get(self, _url: str, *, params: dict, timeout: int): + assert timeout == 30 + offset = int(params["offset"]) + limit = int(params["limit"]) + return Response({"items": rows[offset : offset + limit], "total": len(rows)}) + + assert OPERATOR.list_paginated_items(Session(), "http://backend/datasets", timeout=30) == rows + + +def test_population_archive_download_is_bounded_before_streaming() -> None: + class Response: + url = ( + "https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/" + "OPENDATA_SECTOREN_2026_NEW.zip" + ) + headers = {"Content-Length": str(OPERATOR.MAX_POPULATION_ARCHIVE_BYTES + 1)} + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def raise_for_status(self) -> None: + return None + + def iter_content(self, *, chunk_size: int): + raise AssertionError(f"download should fail before streaming {chunk_size}") + + class Session: + def get(self, *_args, **_kwargs): + return Response() + + with pytest.raises(RuntimeError, match="download limit"): + OPERATOR.download_archive( + Session(), + url=Response.url, + year=2026, + layout="new", + max_bytes=OPERATOR.MAX_POPULATION_ARCHIVE_BYTES, + timeout=30, + ) + + +@pytest.mark.parametrize( + ("remote", "local", "expected"), + [ + ("2026", "2025", "update_available"), + ("2025", "2025", "current"), + ("2026", None, "not_loaded"), + ("2025", "2026", "blocked_remote_older"), + ], +) +def test_catalog_decision_orders_remote_and_local_editions( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + remote: str, + local: str | None, + expected: str, +) -> None: + args = arguments(tmp_path) + monkeypatch.setattr( + MANAGER, + "api_data", + lambda *_args, **_kwargs: {"items": [catalog_item(remote=remote, local=local)]}, + ) + + result = MANAGER.fetch_release_decision(args, refresh=True) + + assert result["status"] == expected + assert result["release"]["year"] == int(remote) + assert result["automatic_download"] is False + assert result["automatic_import"] is False + + +def test_operator_commands_separate_staging_from_apply(tmp_path: Path) -> None: + args = arguments(tmp_path) + release = release_2026() + + stage = MANAGER.build_operator_command(args, release, fetch_only=True) + apply = MANAGER.build_operator_command(args, release, fetch_only=False) + + assert "--force" in stage + assert "--fetch-only" in stage + assert "--force" not in apply + assert "--fetch-only" not in apply + assert stage[stage.index("--population-layout") + 1] == "new" + assert stage[stage.index("--population-url") + 1] == release.population_url + assert stage[stage.index("--geometry-url") + 1] == release.geometry_url + + +def test_staged_plan_and_review_require_exact_hashes(tmp_path: Path) -> None: + args = arguments(tmp_path) + release = release_2026() + release_decision = decision(args) + plan_path, plan = staged_plan(args, release, release_decision) + args.confirm_plan_sha256 = plan["plan_sha256"] + + loaded_path, loaded = MANAGER.load_staged_plan(args, release) + assert loaded_path == plan_path + assert loaded["evidence"]["scope_accounting"]["accounted_population_total"] == 510300 + + args.approve = True + args.reviewer = "GeoIntel operator" + args.review_note = "Schema, totalen en ZZZZ-accounting nagekeken." + review = MANAGER.build_review_evidence(args, plan_path, plan) + review_path = MANAGER.default_review_path(args, release.year) + MANAGER.write_json(review_path, review) + args.confirm_review_sha256 = review["review_sha256"] + + loaded_review_path, loaded_review = MANAGER.load_review_evidence(args, release, plan) + assert loaded_review_path == review_path + assert loaded_review["status"] == "approved" + assert "scope_and_national_accounting" in loaded_review["reviewed_checks"] + + +def test_tampered_source_or_review_is_rejected(tmp_path: Path) -> None: + args = arguments(tmp_path) + release = release_2026() + release_decision = decision(args) + plan_path, plan = staged_plan(args, release, release_decision) + args.confirm_plan_sha256 = plan["plan_sha256"] + population_path = Path(plan["evidence"]["population_archive"]["retained_path"]) + population_path.write_bytes(population_path.read_bytes() + b"tampered") + + with pytest.raises(RuntimeError, match="population archive"): + MANAGER.load_staged_plan(args, release) + + write_staged_artifacts(args, release) + args.approve = True + args.reviewer = "Operator" + review = MANAGER.build_review_evidence(args, plan_path, plan) + review_path = MANAGER.default_review_path(args, release.year) + MANAGER.write_json(review_path, review) + review_payload = json.loads(review_path.read_text(encoding="utf-8")) + review_payload["reviewer"] = "Someone else" + review_path.write_text(json.dumps(review_payload), encoding="utf-8") + args.confirm_review_sha256 = review["review_sha256"] + + with pytest.raises(RuntimeError, match="checksum"): + MANAGER.load_review_evidence(args, release, plan) + + +def test_catalog_drift_and_outside_evidence_path_are_rejected(tmp_path: Path) -> None: + args = arguments(tmp_path) + original = decision(args) + changed = json.loads(json.dumps(original)) + changed["catalog_identity"]["capabilities_sha256"] = "b" * 64 + + with pytest.raises(RuntimeError, match="changed"): + MANAGER.require_catalog_unchanged( + {"release": original["release"], "catalog_identity": original["catalog_identity"]}, + changed, + ) + with pytest.raises(RuntimeError, match="outside"): + MANAGER.governed_evidence_path(args, tmp_path / "outside.json") + + +def test_apply_flow_requires_approved_review_and_writes_evidence( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + args = arguments( + tmp_path, + action="apply", + confirm_edition="2026", + confirm_layout="new", + ) + release = release_2026() + release_decision = decision(args) + plan_path, plan = staged_plan(args, release, release_decision) + args.confirm_plan_sha256 = plan["plan_sha256"] + args.approve = True + args.reviewer = "GeoIntel operator" + review = MANAGER.build_review_evidence(args, plan_path, plan) + review_path = MANAGER.default_review_path(args, release.year) + MANAGER.write_json(review_path, review) + args.confirm_review_sha256 = review["review_sha256"] + final_decision = json.loads(json.dumps(release_decision)) + final_decision["status"] = "current" + final_decision["local_source_version"] = "2026" + decisions = iter((release_decision, final_decision)) + monkeypatch.setattr(MANAGER, "parse_args", lambda: args) + monkeypatch.setattr(MANAGER, "validate_project_scope", lambda _args: None) + monkeypatch.setattr(MANAGER, "fetch_release_decision", lambda *_args, **_kwargs: next(decisions)) + monkeypatch.setattr( + MANAGER, + "run_operator", + lambda *_args, **_kwargs: { + "status": "ok", + "snapshots": [{"year": 2026, "status": "imported", "dataset_id": "dataset-2026", "feature_count": 700}], + }, + ) + + assert MANAGER.main() == 0 + applied_path = plan_path.with_name("applied-evidence.json") + applied = json.loads(applied_path.read_text(encoding="utf-8")) + assert applied["dataset_id"] == "dataset-2026" + assert applied["review_sha256"] == review["review_sha256"] + assert len(applied["applied_evidence_sha256"]) == 64 + + +def test_current_release_cannot_be_staged( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + args = arguments(tmp_path, action="stage", confirm_edition="2025", confirm_layout="new") + current = decision(args, remote="2025", local="2025") + monkeypatch.setattr(MANAGER, "parse_args", lambda: args) + monkeypatch.setattr(MANAGER, "validate_project_scope", lambda _args: None) + monkeypatch.setattr(MANAGER, "fetch_release_decision", lambda *_args, **_kwargs: current) + + assert MANAGER.main() == 1 + assert "not safely stageable: current" in capsys.readouterr().err + + +def test_plan_action_is_read_only_for_current_release( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + args = arguments(tmp_path, action="plan") + current = decision(args, remote="2025", local="2025") + monkeypatch.setattr(MANAGER, "parse_args", lambda: args) + monkeypatch.setattr(MANAGER, "validate_project_scope", lambda _args: None) + monkeypatch.setattr(MANAGER, "fetch_release_decision", lambda *_args, **_kwargs: current) + + assert MANAGER.main() == 0 + output = json.loads(capsys.readouterr().out) + assert output["status"] == "ok" + assert output["decision"]["status"] == "current" + assert not args.evidence_root.exists() + + +def test_release_manager_is_packaged_and_release_checked() -> None: + dockerfile = (ROOT / "deploy/unraid/Dockerfile.all-in-one").read_text(encoding="utf-8") + readiness = (ROOT / "scripts/run_readiness_check.sh").read_text(encoding="utf-8") + + assert "COPY scripts/manage_statbel_population_release.py" in dockerfile + assert "py_compile scripts/manage_statbel_population_release.py" in readiness diff --git a/deploy/unraid/Dockerfile.all-in-one b/deploy/unraid/Dockerfile.all-in-one index d990347c..2f789f98 100644 --- a/deploy/unraid/Dockerfile.all-in-one +++ b/deploy/unraid/Dockerfile.all-in-one @@ -83,6 +83,7 @@ COPY scripts/provision_mol_soil_map.py /app/scripts/provision_mol_soil_map.py COPY scripts/provision_regional_soil_map.py /app/scripts/provision_regional_soil_map.py COPY scripts/provision_mol_population_history.py /app/scripts/provision_mol_population_history.py COPY scripts/statbel_population_preflight.py /app/scripts/statbel_population_preflight.py +COPY scripts/manage_statbel_population_release.py /app/scripts/manage_statbel_population_release.py COPY scripts/provision_mol_historical_landuse.py /app/scripts/provision_mol_historical_landuse.py COPY scripts/provision_regional_historical_landuse.py /app/scripts/provision_regional_historical_landuse.py COPY scripts/provision_official_landuse_timeseries.py /app/scripts/provision_official_landuse_timeseries.py diff --git a/docs/API_CONTRACTS.md b/docs/API_CONTRACTS.md index 37766153..a2e3124a 100644 --- a/docs/API_CONTRACTS.md +++ b/docs/API_CONTRACTS.md @@ -450,6 +450,13 @@ reported only as transition evidence. `service_type=DCAT` uses A newer statistical-sector geometry edition is not interpreted as a newer population release. +Future Statbel execution remains outside the HTTP request cycle in +`scripts/manage_statbel_population_release.py`. It reuses this read-only +catalog response for `plan`, then separates filesystem-only `stage`, named +human `review` and checksum-confirmed `apply`. No additional public endpoint +is introduced. Apply delegates to the existing dataset upload contract and +creates a new immutable annual snapshot only after all evidence is unchanged. + The endpoint accepts no arbitrary URL, feature query, area or layer. It does not fetch vector features, raster pixels or models, create jobs/datasets, write to PostGIS or trigger an import. The normal `source-freshness` endpoint remains diff --git a/docs/CODEX_EXECUTION_LOG.md b/docs/CODEX_EXECUTION_LOG.md index b6eb0f3e..39012ddf 100644 --- a/docs/CODEX_EXECUTION_LOG.md +++ b/docs/CODEX_EXECUTION_LOG.md @@ -1,3 +1,55 @@ +## Sprint 228 - Governed Statbel population release promotion (2026-07-17) + +Implemented: +- Added `scripts/manage_statbel_population_release.py` with four separate + operator actions for the approved Kempen scope. `plan` is read-only; `stage` + requires exact year/layout confirmation and performs bounded fresh download + plus preflight only; `review` requires a named explicit approval; `apply` + requires the exact staged-plan and review-evidence SHA-256 values. +- Derived the only accepted population and matching sector-geometry URLs from + the allowlisted catalog year/layout contract. The coordinator accepts no + arbitrary source URL or process, and current, older, unavailable, ambiguous + or catalog-drifted releases are not stageable. +- Bound catalog identity, source archives, preflight manifest, derived + snapshot, scope/national accounting, ZZZZ totals, baseline trend and geometry + repairs into `staged-plan.json`. Bound the named human decision to that plan + in `review-evidence.json`; successful apply records the immutable Dataset in + `applied-evidence.json` without deleting prior snapshots. +- Extended `provision_mol_population_history.py` with a complete all-or-none + release config for one future year. Existing 2021-2025 arguments remain + compatible. Source downloads now enforce response and streaming byte bounds, + validate final official URLs and discover the latest retained baseline from + actual snapshot files rather than a hardcoded year list. +- Replaced the population operator's 200-row workspace lookup with complete, + total-consistent pagination. This keeps repeated apply idempotent in the live + project with 2,429 Datasets. +- Added Docker/readiness packaging and updated source, API, persistence, + storage and operator documentation. No API route, migration, scheduler, + automatic fetch or frontend behavior changed. + +Validation: +- 15 focused Sprint 228 tests and 38 combined Sprint 194/227/228 tests passed. + Coverage includes release ordering, future URL/layout validation, dynamic + baseline discovery, bounded downloads, >200-row pagination, stage/apply + command separation, plan/review hashes, source/review tampering, evidence-root + confinement, catalog drift, named review, apply evidence and current-edition + refusal. +- Complete readiness passed with 830 backend tests, 110 documented routes, + one Alembic head `202607160001`, frontend typecheck and production build. + Static Alembic SQL, shell syntax, target Ruff and diff checks passed. +- A compatibility run against the live Tower API used the temporary candidate + scripts before deployment. The official Statbel catalog reported + `NodeID6475`, remote/local edition `2025`, layout `new`, catalog hash + `64b17ce059a9c2f936d4b5741b20aed4b178409d8b5be81be86e6fdc9fe7c9d9` and + decision `current`; `plan` wrote no evidence. A deliberate `stage` attempt + for that current edition exited 1 with `not safely stageable: current` and + left the database Dataset count unchanged at 2,429. + +Boundary: +- No newer population edition is currently advertised, so no real stage, + review or apply was executed. Their complete state machine is fixture-tested; + the first future release must still pass all four explicit operator phases. + ## Sprint 227 - Statbel population import compatibility preflight (2026-07-16) Implemented: diff --git a/docs/DATABASE_IMPLEMENTATION_PLAN.md b/docs/DATABASE_IMPLEMENTATION_PLAN.md index 7e7ae0c7..e5b6d707 100644 --- a/docs/DATABASE_IMPLEMENTATION_PLAN.md +++ b/docs/DATABASE_IMPLEMENTATION_PLAN.md @@ -274,6 +274,15 @@ new Dataset plus DatasetVersion and vector_features through the existing DatasetService/VectorFeatureService transaction; prior snapshots remain unchanged and queryable for temporal comparison. +Statbel population release management likewise adds no lifecycle table. +Planning reads the existing source-catalog response. Stage, named review and +their SHA-256-bound evidence are filesystem-only operator artifacts. Apply +revalidates those artifacts and invokes the existing Dataset upload service, +which creates the ordinary annual `datasets`, `dataset_versions` and +`vector_features` records in one established persistence flow. An existing +year remains idempotent and previous annual snapshots are never updated or +deleted. + ## Geometry normalization - User-drawn polygons arrive as EPSG:4326. diff --git a/docs/DATA_SOURCES.md b/docs/DATA_SOURCES.md index 90c7a37b..161ae11d 100644 --- a/docs/DATA_SOURCES.md +++ b/docs/DATA_SOURCES.md @@ -222,6 +222,22 @@ bronarchieven en de afgeleide GeoJSON. Zij geeft alleen technische importgeschiktheid aan; zij vervangt of importeert nooit automatisch een bestaande Dataset. +`scripts/manage_statbel_population_release.py` beheert een toekomstige editie +in vier afzonderlijke operatorstappen. `plan` vergelijkt de nieuwste +allowlisted DCAT-editie read-only met de lokale bronversie. `stage` vereist +expliciete bevestiging van jaar en REDEGEO-layout, haalt de exact uit het +cataloguscontract afgeleide Statbel-URL's begrensd op en voert alleen de +preflight/artefactstaging uit. `review` schrijft pas na een benoemde menselijke +goedkeuring SHA-256-gebonden reviewevidence. `apply` vereist zowel plan- als +reviewhash, controleert catalogusidentiteit en alle bestanden opnieuw en maakt +via de bestaande uploadservice hoogstens een nieuwe immutable Dataset. + +De coordinator accepteert geen vrije provider-URL, start geen achtergrondtaak +en kan een actuele of oudere remote editie niet stagen. Een ontbrekende +bijbehorende sectorgeometrie, gewijzigde catalogus, gewijzigde bronbyte, +onverklaarde totalensprong of onvolledige review blokkeert de flow. Bestaande +snapshots blijven beschikbaar voor historische vergelijking. + ### Mol population history `scripts/provision_mol_population_history.py` imports official Statbel diff --git a/docs/STORAGE_ARCHITECTURE.md b/docs/STORAGE_ARCHITECTURE.md index 44ef0732..402371f5 100644 --- a/docs/STORAGE_ARCHITECTURE.md +++ b/docs/STORAGE_ARCHITECTURE.md @@ -122,6 +122,23 @@ Dataset/DatasetVersion plus PostGIS `vector_features` through the canonical persistence services. Temporary standalone preflight output does not create a database record and may be removed explicitly after operator review. +Governed release-decision evidence is separate from the source artifacts: + +```text +storage/operator-evidence/statbel-population-refresh/{scope}/{year}/ + staged-plan.json + review-evidence.json + applied-evidence.json +``` + +The staged plan binds the current official catalog identity to every retained +source/snapshot/preflight hash and accounting summary. Review evidence binds a +named approval to that exact plan. Applied evidence binds both approvals to +the resulting Dataset id. These JSON files are operator audit artifacts, not +database lifecycle entities. A plan/review path outside the configured +evidence root or a retained archive outside the population output root is +rejected. No evidence file authorizes deletion or in-place replacement. + Waterinfo raw station layers, timeseries responses and checksum manifests live under `storage/operator-data/waterinfo//`. These are immutable source evidence; queryable annual Point snapshots are normal Dataset/vector_feature diff --git a/docs/TODO.md b/docs/TODO.md index 11feff53..eecc0b72 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -44,6 +44,7 @@ - [x] Add a fail-closed ALZ publication probe that distinguishes provisional v1/v2 snapshots from the definitive v3 historical edition and never downloads an archive. - [x] Add a fail-closed Statbel DCAT publication probe that distinguishes population year, sector-geometry year and the 2025 REDEGEO transition without downloading distributions. - [x] Add a fail-closed Statbel population import preflight with archive/schema/CRS/join/total/baseline checks, retained checksums and explicit ZZZZ accounting before any new Dataset import. +- [x] Add an explicit Statbel population plan -> stage -> named review -> checksum-confirmed apply workflow that preserves every prior annual snapshot. - [ ] Extend catalogue probes only to additional sources that publish a stable official edition contract; do not add background polling or infer releases from HTTP dates alone. ## Governed source expansion backlog diff --git a/scripts/README.md b/scripts/README.md index fb0dc960..652c6150 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -1752,6 +1752,43 @@ to the DatasetService-based persistence path. Existing snapshots are retained. The coordinator accepts no provider URL, collection name, arbitrary process or automatic schedule. +## Governed Statbel population release + +Use the population release coordinator only inside the GeoIntel container. +It derives the exact population and matching sector-geometry URLs from the +strict official catalog contract; operators cannot inject another provider +URL. + +```bash +docker exec geointel python /app/scripts/manage_statbel_population_release.py plan \ + --project-id --refresh-catalog + +docker exec geointel python /app/scripts/manage_statbel_population_release.py stage \ + --project-id \ + --confirm-edition --confirm-layout + +docker exec geointel python /app/scripts/manage_statbel_population_release.py review \ + --project-id \ + --confirm-edition --confirm-layout \ + --confirm-plan-sha256 \ + --approve --reviewer "" + +docker exec geointel python /app/scripts/manage_statbel_population_release.py apply \ + --project-id \ + --confirm-edition --confirm-layout \ + --confirm-plan-sha256 \ + --confirm-review-sha256 +``` + +`plan` writes nothing. `stage` downloads within the population/geometry size +bounds, invokes the existing provisioner with `--force --fetch-only` and emits +`staged-plan.json`. `review` requires a named approval and emits +`review-evidence.json`. `apply` refuses catalog drift or any changed source, +manifest, snapshot, plan or review byte, then delegates to the canonical +DatasetService upload. `applied-evidence.json` records the resulting Dataset +without deleting any historical snapshot. Current, older, unavailable or +ambiguous releases are fail-closed. + ## Tower deployment Push the local branch to Gitea, then rebuild the Unraid/Tower Docker runtime: diff --git a/scripts/manage_statbel_population_release.py b/scripts/manage_statbel_population_release.py new file mode 100644 index 00000000..aabf9a5f --- /dev/null +++ b/scripts/manage_statbel_population_release.py @@ -0,0 +1,585 @@ +#!/usr/bin/env python3 +"""Plan, stage, review and explicitly apply one Statbel population edition. + +Planning is read-only. Staging downloads official artifacts and runs the +existing fail-closed preflight without database persistence. Review records an +explicit named approval. Apply requires both exact evidence hashes and reuses +the canonical population DatasetService upload path. +""" + +from __future__ import annotations + +import argparse +from datetime import datetime, timezone +from hashlib import sha256 +import json +import os +from pathlib import Path +import re +import subprocess +import sys +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 +from provision_mol_population_history import ( + PopulationReleaseConfig, + load_preflight_manifest, + preflight_manifest_path, + resolve_release_config, + sha256_path, + snapshot_path, +) + + +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-timeseries") +DEFAULT_SCOPE_OUTPUT_ROOT = Path("/app/storage/operator-data/geographic-scopes") +DEFAULT_EVIDENCE_ROOT = Path("/app/storage/operator-evidence/statbel-population-refresh") +YEAR_PATTERN = re.compile(r"^20[0-9]{2}$") +SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") +ACTIONABLE_STATUSES = {"update_available", "not_loaded"} + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Governed Statbel population release: plan, stage, review, then checksum-confirmed apply." + ) + parser.add_argument("action", choices=("plan", "stage", "review", "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("--confirm-edition", help="Exact official population year required after plan") + parser.add_argument("--confirm-layout", choices=("standard", "new"), help="Exact planned REDEGEO layout") + parser.add_argument("--confirm-plan-sha256", help="Exact staged-plan hash required for review/apply") + parser.add_argument("--confirm-review-sha256", help="Exact review-evidence hash required for apply") + parser.add_argument("--approve", action="store_true", help="Explicitly approve the staged evidence during review") + parser.add_argument("--reviewer", help="Named human/operator approving the staged evidence") + parser.add_argument("--review-note", default="", help="Optional bounded review note") + parser.add_argument("--plan-path", type=Path, help="Override the governed staged-plan path") + parser.add_argument("--review-path", type=Path, help="Override the governed review-evidence path") + parser.add_argument( + "--output-root", + type=Path, + default=Path(os.environ.get("GEOINTEL_REGIONAL_TIMESERIES_OUTPUT_ROOT", DEFAULT_OUTPUT_ROOT)), + ) + parser.add_argument( + "--scope-output-root", + type=Path, + default=Path(os.environ.get("GEOINTEL_SCOPE_OUTPUT_ROOT", DEFAULT_SCOPE_OUTPUT_ROOT)), + ) + parser.add_argument( + "--evidence-root", + type=Path, + default=Path(os.environ.get("GEOINTEL_STATBEL_REFRESH_EVIDENCE_ROOT", DEFAULT_EVIDENCE_ROOT)), + ) + parser.add_argument("--refresh-catalog", action="store_true", help="Bypass the short official-catalog cache") + parser.add_argument("--request-timeout", type=int, default=300) + parser.add_argument("--api-timeout", type=int, default=180) + parser.add_argument("--import-timeout", type=int, default=3600) + return parser.parse_args() + + +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-Statbel-release/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 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 _layout_from_catalog_item(item: dict[str, Any], year: int) -> str: + evidence = f"{item.get('remote_title') or ''} {item.get('message') or ''}".casefold() + if "nieuwe redegeo-sectorindeling" in evidence: + return "new" + if year == 2025: + raise RuntimeError("The official 2025 release is missing explicit new-REDEGEO evidence") + if "actuele sectorindeling" in evidence: + return "standard" + raise RuntimeError("The official Statbel catalog did not expose a recognized population layout") + + +def release_from_catalog_item(item: dict[str, Any]) -> PopulationReleaseConfig: + version = str(item.get("remote_version") or "") + if not YEAR_PATTERN.fullmatch(version): + raise RuntimeError("The official Statbel catalog did not provide one valid population year") + year = int(version) + layout = _layout_from_catalog_item(item, year) + suffix = "_NEW" if layout == "new" else "" + return resolve_release_config( + year, + layout=layout, + population_url=( + "https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/" + f"OPENDATA_SECTOREN_{year}{suffix}.zip" + ), + geometry_url=( + "https://statbel.fgov.be/sites/default/files/files/opendata/Statistische%20sectoren/" + f"sh_statbel_statistical_sectors_31370_{year}0101.geojson.zip" + ), + ) + + +def fetch_release_decision_from_item(args: argparse.Namespace, item: dict[str, Any]) -> dict[str, Any]: + if ( + item.get("source_name") != "statbel" + or item.get("status") != "available" + or item.get("reachable") is not True + or item.get("error_code") + or not SHA256_PATTERN.fullmatch(str(item.get("capabilities_sha256") or "")) + or not item.get("metadata_identifier") + ): + raise RuntimeError("The official Statbel catalog is not safely available for release planning") + required_evidence = {"population_txt_current", "landing_page", "cc_by_4_0"} + if not required_evidence.issubset(set(item.get("matched_layers") or [])): + raise RuntimeError("The official Statbel catalog is missing required release evidence") + release = release_from_catalog_item(item) + local_version = str(item.get("local_source_version") or "") + if local_version and not YEAR_PATTERN.fullmatch(local_version): + status = "blocked_local_version" + elif not local_version: + status = "not_loaded" + elif int(local_version) == release.year: + status = "current" + elif int(local_version) < release.year: + status = "update_available" + else: + status = "blocked_remote_older" + return { + "schema_version": 1, + "status": status, + "project_id": args.project_id, + "scope": args.scope, + "generated_at": datetime.now(timezone.utc).isoformat(), + "local_source_version": local_version or None, + "release": { + "year": release.year, + "layout": release.layout, + "population_url": release.population_url, + "geometry_url": release.geometry_url, + }, + "catalog_identity": { + "metadata_identifier": item["metadata_identifier"], + "metadata_url": item.get("metadata_url"), + "remote_version": str(release.year), + "capabilities_sha256": item["capabilities_sha256"], + "catalog_checked_at": item.get("checked_at"), + }, + "automatic_download": False, + "automatic_import": False, + "destructive_replacement": False, + "next_action": "stage" if status in ACTIONABLE_STATUSES else None, + } + + +def fetch_release_decision(args: argparse.Namespace, *, refresh: bool) -> dict[str, Any]: + query = "true" if refresh else "false" + report = api_data( + args.api_url, + f"projects/{args.project_id}/datasets/source-catalog-probes?refresh={query}", + args.api_timeout, + ) + matches = [item for item in report.get("items") or [] if item.get("source_name") == "statbel"] + if len(matches) != 1: + raise RuntimeError("Source catalog report did not contain exactly one Statbel population contract") + return fetch_release_decision_from_item(args, matches[0]) + + +def require_release_confirmation(args: argparse.Namespace, release: PopulationReleaseConfig) -> None: + if args.confirm_edition != str(release.year): + raise RuntimeError(f"Explicit --confirm-edition {release.year} is required") + if args.confirm_layout != release.layout: + raise RuntimeError(f"Explicit --confirm-layout {release.layout} is required") + + +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 population_output_dir(args: argparse.Namespace) -> Path: + return args.output_root / args.scope / "population" + + +def default_plan_path(args: argparse.Namespace, year: int) -> Path: + return args.evidence_root / args.scope / str(year) / "staged-plan.json" + + +def default_review_path(args: argparse.Namespace, year: int) -> Path: + return args.evidence_root / args.scope / str(year) / "review-evidence.json" + + +def governed_evidence_path(args: argparse.Namespace, path: Path) -> Path: + if not path.resolve().is_relative_to(args.evidence_root.resolve()): + raise RuntimeError(f"Release evidence path is outside the governed evidence root: {path}") + return path + + +def build_operator_command( + args: argparse.Namespace, + release: PopulationReleaseConfig, + *, + fetch_only: bool, +) -> list[str]: + scope = GEOGRAPHIC_SCOPES[args.scope] + command = [ + sys.executable, + str(Path(__file__).resolve().parent / "provision_mol_population_history.py"), + "--scope", + scope.key, + "--base-url", + internal_base_url(args.api_url), + "--project-name", + scope.project_name, + "--area-name", + scope.area_name, + "--years", + str(release.year), + "--population-url", + release.population_url, + "--geometry-url", + release.geometry_url, + "--population-layout", + release.layout, + "--scope-output-root", + str(args.scope_output_root), + "--output-dir", + str(population_output_dir(args)), + "--request-timeout", + str(args.request_timeout), + "--import-timeout", + str(args.import_timeout), + ] + if fetch_only: + command.extend(("--force", "--fetch-only")) + return command + + +def run_operator(command: list[str], *, action: str) -> dict[str, Any]: + print(f"Statbel population: {action}...", 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"Statbel population {action} failed with exit {completed.returncode}: {detail[-3000:]}") + try: + payload = json.loads(completed.stdout) + except json.JSONDecodeError as exc: + raise RuntimeError(f"Statbel population {action} returned invalid JSON") from exc + if payload.get("status") != "ok": + raise RuntimeError(f"Statbel population {action} did not report success") + return payload + + +def canonical_sha256(payload: dict[str, Any], hash_field: str) -> str: + content = {key: value for key, value in payload.items() if key != hash_field} + encoded = json.dumps(content, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + return sha256(encoded).hexdigest() + + +def write_json(path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(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 validate_staged_release( + args: argparse.Namespace, + release: PopulationReleaseConfig, +) -> dict[str, Any]: + scope = GEOGRAPHIC_SCOPES[args.scope] + output_dir = population_output_dir(args) + resolved_output = output_dir.resolve() + snapshot = snapshot_path(output_dir, scope, release.year) + manifest_path = preflight_manifest_path(output_dir, scope, release.year) + for path in (snapshot, manifest_path): + if not path.resolve().is_relative_to(resolved_output): + raise RuntimeError(f"Staged Statbel evidence is outside the governed output root: {path}") + manifest = load_preflight_manifest(manifest_path, snapshot, release.year, scope) + manifest_release = manifest.get("release") or {} + artifacts = manifest.get("artifacts") or {} + if ( + manifest_release.get("population_layout") != release.layout + or (artifacts.get("population") or {}).get("source_url") != release.population_url + or (artifacts.get("geometry") or {}).get("source_url") != release.geometry_url + ): + raise RuntimeError("Staged Statbel manifest no longer matches the planned release identity") + for artifact_name in ("population", "geometry"): + retained = Path(str((artifacts.get(artifact_name) or {}).get("retained_path") or "")) + if not retained.resolve().is_relative_to(resolved_output): + raise RuntimeError(f"Staged {artifact_name} archive is outside the governed output root: {retained}") + return { + "manifest_path": str(manifest_path), + "manifest_sha256": sha256_path(manifest_path), + "snapshot_path": str(snapshot), + "snapshot_sha256": (artifacts.get("derived_snapshot") or {}).get("sha256"), + "snapshot_size_bytes": (artifacts.get("derived_snapshot") or {}).get("size_bytes"), + "feature_count": (artifacts.get("derived_snapshot") or {}).get("feature_count"), + "population_archive": artifacts.get("population"), + "geometry_archive": artifacts.get("geometry"), + "scope_accounting": manifest.get("scope_accounting"), + "national_accounting": manifest.get("national_accounting"), + "baseline": manifest.get("baseline"), + "geometry_repair_count": (manifest.get("schemas") or {}).get("geometry_repair_count"), + } + + +def build_staged_plan( + args: argparse.Namespace, + decision: dict[str, Any], + release: PopulationReleaseConfig, + operator_result: dict[str, Any], +) -> dict[str, Any]: + evidence = validate_staged_release(args, release) + payload: dict[str, Any] = { + "schema_version": 1, + "status": "staged", + "created_at": datetime.now(timezone.utc).isoformat(), + "project_id": args.project_id, + "scope": args.scope, + "local_source_version_before_apply": decision.get("local_source_version"), + "release": decision["release"], + "catalog_identity": decision["catalog_identity"], + "evidence": evidence, + "operator_result": operator_result, + "review_required": True, + "apply_requires_plan_sha256": True, + "apply_requires_review_sha256": True, + "automatic_import": False, + "destructive_replacement": False, + "existing_snapshots_retained": True, + } + payload["plan_sha256"] = canonical_sha256(payload, "plan_sha256") + return payload + + +def _release_from_payload(payload: dict[str, Any]) -> PopulationReleaseConfig: + release = payload.get("release") or {} + year = release.get("year") if isinstance(release, dict) else None + layout = release.get("layout") if isinstance(release, dict) else None + population_url = release.get("population_url") if isinstance(release, dict) else None + geometry_url = release.get("geometry_url") if isinstance(release, dict) else None + if ( + not isinstance(year, int) + or isinstance(year, bool) + or not isinstance(layout, str) + or not isinstance(population_url, str) + or not isinstance(geometry_url, str) + or not all((layout, population_url, geometry_url)) + ): + raise RuntimeError("Release evidence is missing required identity fields") + return resolve_release_config( + year, + layout=layout, + population_url=population_url, + geometry_url=geometry_url, + ) + + +def load_staged_plan( + args: argparse.Namespace, + release: PopulationReleaseConfig, +) -> tuple[Path, dict[str, Any]]: + path = governed_evidence_path(args, args.plan_path or default_plan_path(args, release.year)) + if not path.is_file(): + raise RuntimeError(f"Staged Statbel plan is missing: {path}") + payload = json.loads(path.read_text(encoding="utf-8")) + actual_sha = canonical_sha256(payload, "plan_sha256") + if payload.get("plan_sha256") != actual_sha: + raise RuntimeError("Staged Statbel 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 + or payload.get("scope") != args.scope + or _release_from_payload(payload) != release + ): + raise RuntimeError("Staged Statbel plan identity is invalid") + current_evidence = validate_staged_release(args, release) + if current_evidence != payload.get("evidence"): + raise RuntimeError("Staged Statbel artifacts no longer match the approved plan") + return path, payload + + +def require_catalog_unchanged(plan: dict[str, Any], decision: dict[str, Any]) -> None: + if plan.get("release") != decision.get("release") or plan.get("catalog_identity") != decision.get("catalog_identity"): + raise RuntimeError("The official Statbel release evidence changed; create and review a new staged plan") + + +def build_review_evidence( + args: argparse.Namespace, + plan_path: Path, + plan: dict[str, Any], +) -> dict[str, Any]: + reviewer = str(args.reviewer or "").strip() + note = str(args.review_note or "").strip() + if not args.approve or len(reviewer) < 2 or len(reviewer) > 120: + raise RuntimeError("Review requires --approve and a named --reviewer between 2 and 120 characters") + if len(note) > 1000: + raise RuntimeError("Review note must not exceed 1000 characters") + payload: dict[str, Any] = { + "schema_version": 1, + "status": "approved", + "reviewed_at": datetime.now(timezone.utc).isoformat(), + "reviewer": reviewer, + "review_note": note or None, + "project_id": args.project_id, + "scope": args.scope, + "release": plan["release"], + "staged_plan_path": str(plan_path), + "staged_plan_sha256": plan["plan_sha256"], + "reviewed_checks": [ + "official_catalog_identity", + "source_archive_checksums", + "population_and_geometry_schemas", + "scope_and_national_accounting", + "unlocated_population_accounting", + "baseline_change_limit", + "bounded_geometry_repairs", + "immutable_dataset_apply", + ], + } + payload["review_sha256"] = canonical_sha256(payload, "review_sha256") + return payload + + +def load_review_evidence( + args: argparse.Namespace, + release: PopulationReleaseConfig, + plan: dict[str, Any], +) -> tuple[Path, dict[str, Any]]: + path = governed_evidence_path(args, args.review_path or default_review_path(args, release.year)) + if not path.is_file(): + raise RuntimeError(f"Approved Statbel review evidence is missing: {path}") + payload = json.loads(path.read_text(encoding="utf-8")) + actual_sha = canonical_sha256(payload, "review_sha256") + if payload.get("review_sha256") != actual_sha: + raise RuntimeError("Statbel review evidence checksum is invalid") + if args.confirm_review_sha256 != actual_sha: + raise RuntimeError(f"Explicit --confirm-review-sha256 {actual_sha} is required") + if ( + payload.get("status") != "approved" + or payload.get("project_id") != args.project_id + or payload.get("scope") != args.scope + or payload.get("release") != plan.get("release") + or payload.get("staged_plan_sha256") != plan.get("plan_sha256") + or not str(payload.get("reviewer") or "").strip() + ): + raise RuntimeError("Statbel review evidence does not authorize this staged plan") + return path, payload + + +def _snapshot_result(operator_result: dict[str, Any], year: int) -> dict[str, Any]: + matches = [item for item in operator_result.get("snapshots") or [] if int(item.get("year") or 0) == year] + if len(matches) != 1: + raise RuntimeError("Population operator did not return exactly one result for the approved edition") + return matches[0] + + +def main() -> int: + args = parse_args() + try: + if min(args.request_timeout, args.api_timeout, args.import_timeout) <= 0: + raise ValueError("All timeout safety limits must be positive") + validate_project_scope(args) + refresh = args.refresh_catalog or args.action in {"stage", "review", "apply"} + decision = fetch_release_decision(args, refresh=refresh) + release = _release_from_payload(decision) + + if args.action == "plan": + print(json.dumps({"status": "ok", "action": "plan", "decision": decision}, ensure_ascii=False, indent=2)) + return 0 + + require_release_confirmation(args, release) + if args.action == "stage": + if decision.get("status") not in ACTIONABLE_STATUSES: + raise RuntimeError(f"Statbel release is not safely stageable: {decision.get('status')}") + operator_result = run_operator(build_operator_command(args, release, fetch_only=True), action="staging") + staged_result = _snapshot_result(operator_result, release.year) + if staged_result.get("status") != "prepared" or staged_result.get("preflight_status") != "passed": + raise RuntimeError("Population staging did not produce passed preflight evidence") + plan = build_staged_plan(args, decision, release, operator_result) + plan_path = governed_evidence_path(args, args.plan_path or default_plan_path(args, release.year)) + write_json(plan_path, plan) + print(json.dumps({"status": "staged", "plan_path": str(plan_path), **plan}, ensure_ascii=False, indent=2)) + return 0 + + plan_path, plan = load_staged_plan(args, release) + require_catalog_unchanged(plan, decision) + if args.action == "review": + review = build_review_evidence(args, plan_path, plan) + review_path = governed_evidence_path(args, args.review_path or default_review_path(args, release.year)) + write_json(review_path, review) + print(json.dumps({"status": "approved", "review_path": str(review_path), **review}, ensure_ascii=False, indent=2)) + return 0 + + review_path, review = load_review_evidence(args, release, plan) + operator_result = run_operator(build_operator_command(args, release, fetch_only=False), action="applying") + applied_result = _snapshot_result(operator_result, release.year) + if applied_result.get("status") not in {"imported", "existing"} or not applied_result.get("dataset_id"): + raise RuntimeError("Population apply did not return one persisted immutable Dataset") + final_decision = fetch_release_decision(args, refresh=True) + if final_decision.get("status") != "current" or final_decision.get("local_source_version") != str(release.year): + raise RuntimeError("Applied population Dataset did not become the current local Statbel edition") + evidence = { + "schema_version": 1, + "status": "applied", + "applied_at": datetime.now(timezone.utc).isoformat(), + "project_id": args.project_id, + "scope": args.scope, + "release": plan["release"], + "staged_plan_path": str(plan_path), + "staged_plan_sha256": plan["plan_sha256"], + "review_path": str(review_path), + "review_sha256": review["review_sha256"], + "reviewer": review["reviewer"], + "dataset_id": applied_result["dataset_id"], + "dataset_status": applied_result["status"], + "feature_count": applied_result.get("feature_count"), + "final_catalog_decision": final_decision, + "existing_snapshots_retained": True, + } + evidence["applied_evidence_sha256"] = canonical_sha256(evidence, "applied_evidence_sha256") + evidence_path = plan_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", "action": args.action, "message": str(exc)}, ensure_ascii=False), + file=sys.stderr, + ) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/provision_mol_population_history.py b/scripts/provision_mol_population_history.py index ab807210..96a812f5 100644 --- a/scripts/provision_mol_population_history.py +++ b/scripts/provision_mol_population_history.py @@ -15,6 +15,7 @@ from __future__ import annotations import argparse import csv +from dataclasses import dataclass from hashlib import sha256 import io import json @@ -36,7 +37,11 @@ from urllib3.util.retry import Retry from geographic_scopes import GEOGRAPHIC_SCOPES, GeographicScope from statbel_population_preflight import ( + MAX_GEOMETRY_ARCHIVE_BYTES, + MAX_POPULATION_ARCHIVE_BYTES, StatbelPreflightError, + validate_geometry_source_url, + validate_population_source_url, validate_statbel_release, write_manifest, ) @@ -67,6 +72,45 @@ POPULATION_URLS = { POPULATION_LAYOUTS = {year: ("new" if year == 2025 else "standard") for year in POPULATION_URLS} +@dataclass(frozen=True) +class PopulationReleaseConfig: + year: int + layout: str + population_url: str + geometry_url: str + + +def resolve_release_config( + year: int, + *, + population_url: str | None = None, + geometry_url: str | None = None, + layout: str | None = None, +) -> PopulationReleaseConfig: + overrides = (population_url, geometry_url, layout) + if any(value is not None for value in overrides): + if not all(value is not None for value in overrides): + raise ValueError("Population URL, geometry URL and population layout must be supplied together") + release = PopulationReleaseConfig( + year=year, + layout=str(layout), + population_url=str(population_url), + geometry_url=str(geometry_url), + ) + elif year in POPULATION_URLS: + release = PopulationReleaseConfig( + year=year, + layout=POPULATION_LAYOUTS[year], + population_url=POPULATION_URLS[year], + geometry_url=SECTOR_URL.format(year=year), + ) + else: + raise ValueError(f"Unsupported population year without an explicit governed release: {year}") + validate_population_source_url(release.population_url, release.year, release.layout) + validate_geometry_source_url(release.geometry_url, release.year) + return release + + def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Provision official annual Statbel population snapshots.") parser.add_argument("--scope", choices=sorted(GEOGRAPHIC_SCOPES), default=DEFAULT_SCOPE_KEY) @@ -74,6 +118,15 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--project-name", default=None) parser.add_argument("--area-name", default=None, help="Case-insensitive fragment identifying the persisted Area.") parser.add_argument("--years", default="2021,2022,2023,2024,2025") + parser.add_argument( + "--population-url", + help="Exact official Statbel ZIP URL for one explicitly governed release; requires the other release flags", + ) + parser.add_argument( + "--geometry-url", + help="Exact matching official statistical-sector GeoJSON ZIP URL for one explicitly governed release", + ) + parser.add_argument("--population-layout", choices=("standard", "new")) parser.add_argument("--output-dir", type=Path, default=None) parser.add_argument("--boundary-path", type=Path, default=None) parser.add_argument( @@ -107,6 +160,43 @@ def build_session() -> requests.Session: return session +def download_archive( + session: requests.Session, + *, + url: str, + year: int, + layout: str | None, + max_bytes: int, + timeout: int, +) -> bytes: + with session.get(url, timeout=timeout, stream=True) as response: + response.raise_for_status() + if layout is None: + validate_geometry_source_url(response.url, year) + else: + validate_population_source_url(response.url, year, layout) + content_length = response.headers.get("Content-Length") + if content_length: + try: + advertised_size = int(content_length) + except ValueError as exc: + raise RuntimeError(f"Official source returned an invalid Content-Length for {url}") from exc + if advertised_size <= 0 or advertised_size > max_bytes: + raise RuntimeError(f"Official source archive exceeds the {max_bytes}-byte download limit") + chunks: list[bytes] = [] + size = 0 + for chunk in response.iter_content(chunk_size=1024 * 1024): + if not chunk: + continue + size += len(chunk) + if size > max_bytes: + raise RuntimeError(f"Official source archive exceeds the {max_bytes}-byte download limit") + chunks.append(chunk) + if size <= 0: + raise RuntimeError(f"Official source archive is empty: {url}") + return b"".join(chunks) + + def response_data(response: requests.Response) -> Any: try: payload = response.json() @@ -119,6 +209,29 @@ def response_data(response: requests.Response) -> Any: return payload["data"] +def list_paginated_items(session: requests.Session, url: str, *, timeout: int) -> list[dict[str, Any]]: + items: list[dict[str, Any]] = [] + offset = 0 + total: int | None = None + while total is None or offset < total: + page = response_data(session.get(url, params={"limit": 200, "offset": offset}, timeout=timeout)) + page_items = page.get("items") if isinstance(page, dict) else None + if not isinstance(page_items, list): + raise RuntimeError(f"GeoIntel list response for {url} has no items array") + page_total = int(page.get("total", len(page_items))) + if total is None: + total = page_total + elif page_total != total: + raise RuntimeError("GeoIntel pagination total changed while reading the population workspace") + items.extend(page_items) + if not page_items: + break + offset += len(page_items) + if total is not None and len(items) != total: + raise RuntimeError(f"GeoIntel list response for {url} returned {len(items)} of {total} items") + return items + + def series_key(scope: GeographicScope) -> str: return f"statbel:population-statistical-sector:{scope.key}" @@ -340,9 +453,13 @@ def preflight_manifest_path(output_dir: Path, scope: GeographicScope, year: int) def previous_snapshot_path(output_dir: Path, scope: GeographicScope, year: int) -> Path | None: - candidates = [snapshot_path(output_dir, scope, candidate) for candidate in POPULATION_URLS if candidate < year] - available = [path for path in candidates if path.is_file()] - return max(available, key=lambda path: int(path.stem.rsplit("_", 1)[-1])) if available else None + prefix = f"{scope.key.replace('-', '_')}_statbel_population_" + available: list[tuple[int, Path]] = [] + for path in output_dir.glob(f"{prefix}*.geojson"): + suffix = path.stem.removeprefix(prefix) + if suffix.isdigit() and int(suffix) < year and path.is_file(): + available.append((int(suffix), path)) + return max(available, key=lambda item: item[0])[1] if available else None def load_preflight_manifest(path: Path, snapshot: Path, year: int, scope: GeographicScope) -> dict[str, Any]: @@ -383,23 +500,24 @@ def stage_release( output_dir: Path, boundary, scope: GeographicScope, + release: PopulationReleaseConfig | None = None, ) -> tuple[Path, Path, dict[str, Any]]: - layout = POPULATION_LAYOUTS[year] - population_url = POPULATION_URLS[year] - geometry_url = SECTOR_URL.format(year=year) + active_release = release or resolve_release_config(year) + if active_release.year != year: + raise ValueError("Population release year does not match the staged year") result = validate_statbel_release( year=year, - layout=layout, + layout=active_release.layout, population_content=population_content, - population_url=population_url, + population_url=active_release.population_url, geometry_content=geometry_content, - geometry_url=geometry_url, + geometry_url=active_release.geometry_url, scope=scope, baseline_snapshot=previous_snapshot_path(output_dir, scope, year), ) raw_dir = output_dir / "raw" / str(year) - population_archive_path = raw_dir / Path(unquote(urlsplit(population_url).path)).name - geometry_archive_path = raw_dir / Path(unquote(urlsplit(geometry_url).path)).name + population_archive_path = raw_dir / Path(unquote(urlsplit(active_release.population_url).path)).name + geometry_archive_path = raw_dir / Path(unquote(urlsplit(active_release.geometry_url).path)).name write_bytes_atomic(population_archive_path, population_content) write_bytes_atomic(geometry_archive_path, geometry_content) @@ -444,18 +562,18 @@ def locate_workspace( area_name: str, timeout: int, ): - projects = response_data(session.get(f"{base_url}/api/v1/projects", params={"limit": 200}, timeout=timeout)) - project = next((item for item in projects.get("items") or [] if item.get("name") == project_name), None) + projects = list_paginated_items(session, f"{base_url}/api/v1/projects", timeout=timeout) + project = next((item for item in projects if item.get("name") == project_name), None) if not project: raise RuntimeError(f"Project {project_name!r} is missing") project_id = str(project["id"]) - areas = response_data(session.get(f"{base_url}/api/v1/projects/{project_id}/areas", params={"limit": 200}, timeout=timeout)) + areas = list_paginated_items(session, f"{base_url}/api/v1/projects/{project_id}/areas", timeout=timeout) area_fragment = area_name.strip().casefold() - matches = [item for item in areas.get("items") or [] if area_fragment in str(item.get("name") or "").casefold()] + matches = [item for item in areas if area_fragment in str(item.get("name") or "").casefold()] if len(matches) != 1: raise RuntimeError(f"Expected one official Area matching {area_name!r}, received {len(matches)}") - datasets = response_data(session.get(f"{base_url}/api/v1/projects/{project_id}/datasets", params={"limit": 200}, timeout=timeout)) - return project_id, str(matches[0]["id"]), list(datasets.get("items") or []) + datasets = list_paginated_items(session, f"{base_url}/api/v1/projects/{project_id}/datasets", timeout=timeout) + return project_id, str(matches[0]["id"]), datasets def upload_snapshot( @@ -468,7 +586,11 @@ def upload_snapshot( timeout: int, scope: GeographicScope, preflight_path: Path, + release: PopulationReleaseConfig | None = None, ) -> dict[str, Any]: + active_release = release or resolve_release_config(year) + if active_release.year != year: + raise ValueError("Population release year does not match the upload year") observed_at = f"{year}-01-01T00:00:00Z" preflight = load_preflight_manifest(preflight_path, path, year, scope) accounting = preflight["scope_accounting"] @@ -506,8 +628,8 @@ def upload_snapshot( "operator_explicit_fetch": True, "scope_key": scope.key, "geometry_clipped_to_area": True, - "sector_geometry_url": SECTOR_URL.format(year=year), - "population_url": POPULATION_URLS[year], + "sector_geometry_url": active_release.geometry_url, + "population_url": active_release.population_url, "population_layout": preflight["release"]["population_layout"], "preflight_manifest_path": str(preflight_path), "preflight_manifest_sha256": sha256_path(preflight_path), @@ -551,9 +673,23 @@ def main() -> int: except ValueError: print(json.dumps({"status": "error", "message": "Years must be comma-separated integers"}), file=sys.stderr) return 2 - unsupported = [year for year in years if year not in POPULATION_URLS] - if unsupported or not years: - print(json.dumps({"status": "error", "message": f"Unsupported years: {unsupported}"}), file=sys.stderr) + custom_release_requested = any((args.population_url, args.geometry_url, args.population_layout)) + if not years or (custom_release_requested and len(years) != 1): + message = "An explicit governed release requires exactly one population year" if custom_release_requested else "No years requested" + print(json.dumps({"status": "error", "message": message}), file=sys.stderr) + return 2 + try: + releases = { + year: resolve_release_config( + year, + population_url=args.population_url if custom_release_requested else None, + geometry_url=args.geometry_url if custom_release_requested else None, + layout=args.population_layout if custom_release_requested else None, + ) + for year in years + } + except (ValueError, StatbelPreflightError) as exc: + print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False), file=sys.stderr) return 2 output_dir = resolve_output_dir(args, scope) @@ -565,21 +701,35 @@ def main() -> int: prepared: list[dict[str, Any]] = [] with build_session() as source_session: for year in years: + release = releases[year] path = snapshot_path(output_dir, scope, year) manifest_path = preflight_manifest_path(output_dir, scope, year) preflight_status = "passed" if args.force or not path.exists(): - sectors_response = source_session.get(SECTOR_URL.format(year=year), timeout=args.request_timeout) - sectors_response.raise_for_status() - population_response = source_session.get(POPULATION_URLS[year], timeout=args.request_timeout) - population_response.raise_for_status() + geometry_content = download_archive( + source_session, + url=release.geometry_url, + year=year, + layout=None, + max_bytes=MAX_GEOMETRY_ARCHIVE_BYTES, + timeout=args.request_timeout, + ) + population_content = download_archive( + source_session, + url=release.population_url, + year=year, + layout=release.layout, + max_bytes=MAX_POPULATION_ARCHIVE_BYTES, + timeout=args.request_timeout, + ) path, manifest_path, _manifest = stage_release( year=year, - population_content=population_response.content, - geometry_content=sectors_response.content, + population_content=population_content, + geometry_content=geometry_content, output_dir=output_dir, boundary=boundary, scope=scope, + release=release, ) elif manifest_path.is_file(): load_preflight_manifest(manifest_path, path, year, scope) @@ -648,6 +798,7 @@ def main() -> int: args.import_timeout, scope, item["manifest_path"], + releases[year], ) results.append({"year": year, "dataset_id": dataset["id"], "feature_count": dataset.get("feature_count"), "status": "imported"}) except (OSError, RuntimeError, requests.RequestException, ValueError, KeyError, zipfile.BadZipFile) as exc: diff --git a/scripts/run_readiness_check.sh b/scripts/run_readiness_check.sh index 56a2f603..c91c3e28 100755 --- a/scripts/run_readiness_check.sh +++ b/scripts/run_readiness_check.sh @@ -46,6 +46,7 @@ ${PYTHON_BIN} -m py_compile scripts/provision_mol_municipality_workspace.py ${PYTHON_BIN} -m py_compile scripts/provision_mol_context_layers.py ${PYTHON_BIN} -m py_compile scripts/provision_mol_population_history.py ${PYTHON_BIN} -m py_compile scripts/statbel_population_preflight.py +${PYTHON_BIN} -m py_compile scripts/manage_statbel_population_release.py ${PYTHON_BIN} -m py_compile scripts/provision_mol_historical_landuse.py ${PYTHON_BIN} -m py_compile scripts/provision_regional_historical_landuse.py ${PYTHON_BIN} -m py_compile scripts/provision_official_landuse_timeseries.py