diff --git a/CHANGELOG.md b/CHANGELOG.md index 48b273f3..69d595b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,26 @@ # Changelog +## Sprint 227 Statbel population import compatibility preflight (2026-07-16) + +- Added a local, fail-closed preflight for staged official Statbel population + and matching statistical-sector archives. It validates exact source + identity, bounded ZIP contents, table/geometry schemas, REDEGEO layout, + EPSG:31370, situation date, geometry validity, scope coverage, joins and + reconciled national/scope totals before a new import is eligible. +- Replaced the historical sector-prefix municipality assumption with explicit + population/geometry municipality-field reconciliation, which supports the + 2025 REDEGEO municipal-merger contract without weakening identity checks. +- Accounted for official `ZZZZ` unlocated population separately from map-ready + sectors and bounded lossless `make_valid` normalization of reported source + topology errors. Both conditions are visible in the evidence manifest. +- Hardened the existing population operator to retain official source ZIPs, + atomically write derived evidence, require a passed manifest for new imports + and revalidate source plus derived SHA-256 values immediately before upload. +- Added negative regression coverage and packaged the preflight in the + all-in-one image and release readiness compile gate. No API, migration, + automatic import or existing Dataset was changed. + ## Sprint 226 Governed Statbel population edition probe (2026-07-16) - Extended the explicit read-only source catalog audit with the official diff --git a/backend/README.md b/backend/README.md index 11a8127a..982c003b 100644 --- a/backend/README.md +++ b/backend/README.md @@ -1043,6 +1043,36 @@ third imports the Departement Omgeving 10 m forest class for 2013, 2016, 2019, API/DatasetService flow and retain fetched artifacts in persistent operator storage. They never run on app startup. +Every newly fetched or `--force` rebuilt Statbel population edition now passes +`statbel_population_preflight.py` before a derived GeoJSON can reach the +upload API. The operator retains both official ZIPs, writes an atomic +preflight manifest and verifies the source and derived SHA-256 values again at +upload time. A passed preflight does not replace an existing Dataset. + +The preflight can also be run without downloads or database mutation against +already staged official artifacts: + +```bash +docker exec geointel python /app/scripts/statbel_population_preflight.py \ + --year 2025 \ + --layout new \ + --scope kempen-transport-region \ + --population-archive /tmp/OPENDATA_SECTOREN_2025_NEW.zip \ + --population-url https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_2025_NEW.zip \ + --geometry-archive /tmp/sh_statbel_statistical_sectors_31370_20250101.geojson.zip \ + --geometry-url https://statbel.fgov.be/sites/default/files/files/opendata/Statistische%20sectoren/sh_statbel_statistical_sectors_31370_20250101.geojson.zip \ + --baseline-snapshot /app/storage/operator-data/regional-timeseries/kempen-transport-region/population/kempen_transport_region_statbel_population_2024.geojson \ + --output /app/storage/operator-evidence/statbel-population/2025-kempen.preflight.json +``` + +The command exits non-zero and emits a stable `error_code` when source +identity, archive safety, schema, CRS, geometry, join, total reconciliation, +scope coverage or the default 5% annualized population-change review limit +fails. `ZZZZ` rows are reconciled as official unlocated population but remain +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. + Historical land-use work can be bounded explicitly: ```bash diff --git a/backend/tests/test_sprint227_statbel_population_preflight.py b/backend/tests/test_sprint227_statbel_population_preflight.py new file mode 100644 index 00000000..2a7607db --- /dev/null +++ b/backend/tests/test_sprint227_statbel_population_preflight.py @@ -0,0 +1,313 @@ +from __future__ import annotations + +import importlib.util +import io +import json +from pathlib import Path +import sys +import zipfile + +import pytest +from shapely.geometry import Polygon + + +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}_sprint227" + 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 + + +PREFLIGHT = load_script("statbel_population_preflight.py") +OPERATOR = load_script("provision_mol_population_history.py") +POPULATION_URL = ( + "https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/" + "OPENDATA_SECTOREN_2025_NEW.zip" +) +GEOMETRY_URL = ( + "https://statbel.fgov.be/sites/default/files/files/opendata/Statistische%20sectoren/" + "sh_statbel_statistical_sectors_31370_20250101.geojson.zip" +) + + +def population_archive( + *, + duplicate: bool = False, + invalid_total: bool = False, + missing_column: bool = False, + unexpected_non_spatial: bool = False, + unsafe_member: bool = False, +) -> bytes: + headers = ["CD_REFNIS", "CD_SECTOR", "TOTAL", "TX_DESCR_SECTOR_NL", "TX_DESCR_NL"] + if missing_column: + headers.remove("TOTAL") + rows = [ + ["13025", "13024A00-", "120", "Mol centrum", "Mol"], + ["13025", "13025A01-", "bad" if invalid_total else "80", "Mol rand", "Mol"], + ["13025", "13025ZZZZ", "3", "Niet te lokaliseren in een sector", "Mol"], + ["13008", "13008A00-", "40", "Geel centrum", "Geel"], + ] + if duplicate: + rows.append(["13025", "13024A00-", "1", "Dubbel", "Mol"]) + if unexpected_non_spatial: + rows.append(["13025", "13025B00-", "2", "Ontbrekende geometrie", "Mol"]) + lines = ["|".join(headers)] + for row in rows: + values = row if not missing_column else [row[0], row[1], row[3], row[4]] + lines.append("|".join(values)) + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w", compression=zipfile.ZIP_DEFLATED) as archive: + archive.writestr("OPENDATA_SECTOREN_2025_NEW.txt", "\n".join(lines)) + if unsafe_member: + archive.writestr("../escape.txt", "unsafe") + return buffer.getvalue() + + +def square_feature( + sector_code: str, + x: float, + *, + date: str = "2025-01-01", + municipality_code: str | None = None, +) -> dict: + coordinates = [[ + [x, 200000], + [x + 100, 200000], + [x + 100, 200100], + [x, 200100], + [x, 200000], + ]] + return { + "type": "Feature", + "properties": { + "cd_sector": sector_code, + "cd_munty_refnis": municipality_code or sector_code[:5], + "dt_situation": date, + "ms_area_ha": 1.0, + "tx_sector_descr_nl": sector_code, + }, + "geometry": {"type": "Polygon", "coordinates": coordinates}, + } + + +def geometry_archive( + *, + crs: str = "urn:ogc:def:crs:EPSG::31370", + date: str = "2025-01-01", + municipality_mismatch: bool = False, + repairable_invalid: bool = False, +) -> bytes: + payload = { + "type": "FeatureCollection", + "name": "sh_statbel_statistical_sectors_31370_20250101", + "crs": {"type": "name", "properties": {"name": crs}}, + "features": [ + square_feature( + "13024A00-", + 150000, + date=date, + municipality_code="13008" if municipality_mismatch else "13025", + ), + square_feature("13025A01-", 150200, date=date), + square_feature("13008A00-", 150400, date=date), + ], + } + if repairable_invalid: + payload["features"][0]["geometry"] = { + "type": "MultiPolygon", + "coordinates": [ + [[[150000, 200000], [150100, 200000], [150100, 200100], [150000, 200100], [150000, 200000]]], + [[[150100, 200000], [150200, 200000], [150200, 200100], [150100, 200100], [150100, 200000]]], + ], + } + payload["features"][0]["properties"]["ms_area_ha"] = 2.0 + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w", compression=zipfile.ZIP_DEFLATED) as archive: + archive.writestr( + "sh_statbel_statistical_sectors_31370_20250101.geojson/" + "sh_statbel_statistical_sectors_31370_20250101.geojson", + json.dumps(payload), + ) + return buffer.getvalue() + + +def baseline_snapshot(path: Path, *, total: int = 198) -> Path: + scope = PREFLIGHT.GEOGRAPHIC_SCOPES["mol"] + path.write_text( + json.dumps( + { + "type": "FeatureCollection", + "observation_year": 2024, + "member_nis_codes": list(scope.nis_codes), + "features": [ + {"type": "Feature", "properties": {"source_feature_id": "13024A00-", "population_total": total - 80}}, + {"type": "Feature", "properties": {"source_feature_id": "13025A01-", "population_total": 80}}, + ], + } + ), + encoding="utf-8", + ) + return path + + +def validate(tmp_path: Path, **overrides): + values = { + "year": 2025, + "layout": "new", + "population_content": population_archive(), + "population_url": POPULATION_URL, + "geometry_content": geometry_archive(), + "geometry_url": GEOMETRY_URL, + "scope": PREFLIGHT.GEOGRAPHIC_SCOPES["mol"], + "baseline_snapshot": baseline_snapshot(tmp_path / "baseline.geojson"), + } + values.update(overrides) + return PREFLIGHT.validate_statbel_release(**values) + + +def test_preflight_reconciles_spatial_and_unlocated_population(tmp_path: Path) -> None: + result = validate(tmp_path) + manifest = result.manifest + + assert manifest["status"] == "passed" + assert manifest["import_eligible"] is True + assert manifest["release"] == { + "year": 2025, + "population_layout": "new", + "geometry_date": "2025-01-01", + "license": "CC BY 4.0", + } + assert manifest["national_accounting"] == { + "population_row_count": 4, + "geometry_feature_count": 3, + "spatial_population_total": 240, + "unlocated_row_count": 1, + "unlocated_population_total": 3, + "population_total": 243, + } + assert manifest["scope_accounting"]["spatial_sector_count"] == 2 + assert manifest["scope_accounting"]["spatial_population_total"] == 200 + assert manifest["scope_accounting"]["unlocated_population_total"] == 3 + assert manifest["scope_accounting"]["accounted_population_total"] == 203 + assert manifest["baseline"]["annual_change_ratio"] == pytest.approx(200 / 198 - 1) + assert len(manifest["artifacts"]["population"]["archive_sha256"]) == 64 + assert len(manifest["schemas"]["geometry_schema_sha256"]) == 64 + + +def test_preflight_reports_bounded_topology_repairs(tmp_path: Path) -> None: + result = validate(tmp_path, geometry_content=geometry_archive(repairable_invalid=True)) + + assert result.manifest["schemas"]["geometry_repair_count"] == 1 + assert result.manifest["schemas"]["geometry_repaired_sector_codes"] == ["13024A00-"] + assert result.geometry.payload["features"][0]["geometry"]["type"] == "Polygon" + + +@pytest.mark.parametrize( + ("population_kwargs", "error_code"), + [ + ({"missing_column": True}, "STATBEL_POPULATION_SCHEMA_MISMATCH"), + ({"duplicate": True}, "STATBEL_POPULATION_DUPLICATE_SECTOR"), + ({"invalid_total": True}, "STATBEL_POPULATION_TOTAL_REJECTED"), + ({"unexpected_non_spatial": True}, "STATBEL_JOIN_GEOMETRY_MISSING"), + ({"unsafe_member": True}, "STATBEL_ARCHIVE_MEMBER_REJECTED"), + ], +) +def test_preflight_rejects_population_contract_breaks( + tmp_path: Path, + population_kwargs: dict, + error_code: str, +) -> None: + with pytest.raises(PREFLIGHT.StatbelPreflightError) as exc_info: + validate(tmp_path, population_content=population_archive(**population_kwargs)) + + assert exc_info.value.code == error_code + + +@pytest.mark.parametrize( + ("geometry_kwargs", "error_code"), + [ + ({"crs": "EPSG:4326"}, "STATBEL_GEOMETRY_CRS_REJECTED"), + ({"date": "2026-01-01"}, "STATBEL_GEOMETRY_DATE_MISMATCH"), + ({"municipality_mismatch": True}, "STATBEL_JOIN_MUNICIPALITY_MISMATCH"), + ], +) +def test_preflight_rejects_geometry_contract_breaks( + tmp_path: Path, + geometry_kwargs: dict, + error_code: str, +) -> None: + with pytest.raises(PREFLIGHT.StatbelPreflightError) as exc_info: + validate(tmp_path, geometry_content=geometry_archive(**geometry_kwargs)) + + assert exc_info.value.code == error_code + + +def test_preflight_rejects_excessive_population_change(tmp_path: Path) -> None: + baseline = baseline_snapshot(tmp_path / "low-baseline.geojson", total=100) + + with pytest.raises(PREFLIGHT.StatbelPreflightError) as exc_info: + validate(tmp_path, baseline_snapshot=baseline) + + assert exc_info.value.code == "STATBEL_POPULATION_CHANGE_REVIEW_REQUIRED" + assert exc_info.value.details["annual_change_ratio"] == pytest.approx(1.0) + + +def test_operator_stages_raw_artifacts_manifest_and_accounted_snapshot(tmp_path: Path) -> None: + scope = OPERATOR.GEOGRAPHIC_SCOPES["mol"] + boundary = Polygon([(-180, -90), (180, -90), (180, 90), (-180, 90)]) + + path, manifest_path, manifest = OPERATOR.stage_release( + year=2025, + population_content=population_archive(), + geometry_content=geometry_archive(), + output_dir=tmp_path, + boundary=boundary, + scope=scope, + ) + + snapshot = json.loads(path.read_text(encoding="utf-8")) + assert len(snapshot["features"]) == 2 + assert snapshot["spatial_population_total"] == 200 + assert snapshot["unlocated_population_total"] == 3 + assert snapshot["accounted_population_total"] == 203 + assert manifest_path.is_file() + assert Path(manifest["artifacts"]["population"]["retained_path"]).is_file() + assert Path(manifest["artifacts"]["geometry"]["retained_path"]).is_file() + assert OPERATOR.load_preflight_manifest(manifest_path, path, 2025, scope)["import_eligible"] is True + + original_snapshot = path.read_bytes() + path.write_bytes(original_snapshot + b"\n") + with pytest.raises(RuntimeError, match="does not authorize"): + OPERATOR.load_preflight_manifest(manifest_path, path, 2025, scope) + + path.write_bytes(original_snapshot) + geometry_path = Path(manifest["artifacts"]["geometry"]["retained_path"]) + geometry_path.write_bytes(geometry_path.read_bytes() + b"tampered") + with pytest.raises(RuntimeError, match="geometry archive"): + OPERATOR.load_preflight_manifest(manifest_path, path, 2025, scope) + + +def test_population_rows_no_longer_silently_skip_invalid_values() -> None: + scope = OPERATOR.GEOGRAPHIC_SCOPES["mol"] + + with pytest.raises(RuntimeError, match="invalid code or TOTAL"): + OPERATOR.population_rows(population_archive(invalid_total=True), scope) + + +def test_preflight_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/statbel_population_preflight.py" in dockerfile + assert "py_compile scripts/statbel_population_preflight.py" in readiness diff --git a/deploy/unraid/Dockerfile.all-in-one b/deploy/unraid/Dockerfile.all-in-one index 347a21e2..d990347c 100644 --- a/deploy/unraid/Dockerfile.all-in-one +++ b/deploy/unraid/Dockerfile.all-in-one @@ -82,6 +82,7 @@ COPY scripts/provision_thematic_rasters.py /app/scripts/provision_thematic_raste 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/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/CODEX_EXECUTION_LOG.md b/docs/CODEX_EXECUTION_LOG.md index 65146bdd..f424626c 100644 --- a/docs/CODEX_EXECUTION_LOG.md +++ b/docs/CODEX_EXECUTION_LOG.md @@ -1,3 +1,64 @@ +## Sprint 227 - Statbel population import compatibility preflight (2026-07-16) + +Implemented: +- Added `scripts/statbel_population_preflight.py` as a local, fail-closed gate + between staged official Statbel archives and the existing population + provisioner. It performs exact source/edition allowlisting, bounded ZIP + inspection, population and geometry schema checks, EPSG:31370 and situation + date validation, geometry validity/area checks, sector joins, explicit + municipality reconciliation, national/scope total reconciliation and an + annualized baseline-change guard. +- Removed the invalid historical assumption that `CD_SECTOR[:5]` always equals + the current `CD_REFNIS`. The 2025 REDEGEO layout can retain old sector codes + after a municipal merger; population and geometry must instead agree on + their explicit current municipality fields. +- Kept official `ZZZZ` population rows in national and scope accounting while + excluding them honestly from map geometry. Bounded `make_valid` repair is + allowed only when the result remains polygonal, valid and area-preserving; + repaired sector codes are recorded in the manifest. +- Hardened `provision_mol_population_history.py` so a new fetch is preflighted + before source retention or derivation, source ZIPs and the derived GeoJSON + are written atomically, and all three SHA-256 values are rechecked before + upload. Existing immutable Datasets remain idempotent; a legacy cached file + cannot create a new Dataset without `--force` restaging evidence. +- Packaged the script in the all-in-one image, added it to readiness compile + checks and documented command, storage layout, evidence and limitations. + No API route, migration, scheduler, background fetch or existing Dataset was + changed. + +Validation: +- Focused preflight/regional-time-series suite: 23 passed. It covers source and + member identity, missing schema, duplicate sectors, invalid totals, archive + traversal, CRS/date mismatch, explicit municipality mismatch, unexpected + non-spatial rows, bounded topology repair, excessive trend change, atomic + staging, source/snapshot tampering and legacy parser strictness. +- Complete readiness passed with 815 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 + repository-wide Ruff audit still reports 17 pre-existing warnings outside + this change; no unrelated refactor was performed. +- Live compatibility-only execution inside the healthy Tower container used + locally staged official archives and made no API/database call: + - 2024 standard layout: 646 spatial Kempen sectors, spatial population + 503,405, 28 `ZZZZ` rows / 276 unlocated inhabitants and accounted total + 503,681 against the retained 2023 baseline. + - 2025 new REDEGEO layout: 733 spatial Kempen sectors, spatial population + 506,473, 26 `ZZZZ` rows / 294 unlocated inhabitants and accounted total + 506,767 against the retained 2024 baseline; annualized change 0.6094%. + - National 2025 accounting reconciled 21,183 population rows and 20,781 + geometries to 11,825,551 inhabitants, including 402 `ZZZZ` rows / 7,654 + unlocated inhabitants. Four official self-intersection cases were repaired + without area change and recorded by sector code. +- The complete real 2025 operator staging path retained both source archives, + generated a 733-feature scoped GeoJSON and authorized source/snapshot hashes + in an isolated `/tmp/statbel-stage` directory. It did not import or replace a + Dataset. + +Boundary: +- A passed preflight proves technical compatibility only. Adding a future + edition to `POPULATION_URLS` and replacing or importing any immutable Dataset + remains an explicit reviewed operator action. + ## Sprint 226 - Governed Statbel population edition probe (2026-07-16) Implemented: diff --git a/docs/DATA_SOURCES.md b/docs/DATA_SOURCES.md index 21f719fa..90c7a37b 100644 --- a/docs/DATA_SOURCES.md +++ b/docs/DATA_SOURCES.md @@ -198,6 +198,30 @@ naar TXT/ZIP en XLSX worden op host, pad, jaar en variant gevalideerd maar niet opgehaald. De afzonderlijke statistische-sectorgeometrie 2026 betekent niet dat er al bevolkingscijfers per sector voor 2026 zijn gepubliceerd. +Een afzonderlijke import-preflight in +`scripts/statbel_population_preflight.py` controleert lokaal gestagede +officiele ZIP-archieven voordat de bevolkingsoperator een nieuwe afgeleide +snapshot mag aanbieden aan DatasetService. De controle is fail-closed voor +bron-URL en editie, ZIP-veiligheid, verplichte kolommen, REDEGEO-layout, +EPSG:31370, situatiedatum, geometrie/sectorjoin, gemeentetoewijzing, nationale +en scopespecifieke totalen en een begrensde vergelijking met de vorige +snapshot. Herstelbare bron-topologiefouten worden uitsluitend met +`make_valid` genormaliseerd wanneer geometriesoort en oppervlakte behouden +blijven; elk herstel staat in het manifest. + +Vanaf de editie 2025 mag de eerste vijf tekens van `CD_SECTOR` niet als actuele +gemeentecode worden gebruikt. Gemeentefusies kunnen een historische +sectorcode onder een nieuwe `CD_REFNIS` plaatsen. GeoIntel valideert daarom de +expliciete gemeentevelden uit de bevolkings- en geometriebron tegen elkaar. +Bevolkingsregels met suffix `ZZZZ` zijn officiele niet-lokaliseerbare totalen: +ze tellen mee in de reconciliatie en het manifest, maar niet in kaartselecties +of ruimtelijke schattingen. + +Een geslaagde preflight maakt een controlespoor met SHA-256 voor beide +bronarchieven en de afgeleide GeoJSON. Zij geeft alleen technische +importgeschiktheid aan; zij vervangt of importeert nooit automatisch een +bestaande Dataset. + ### Mol population history `scripts/provision_mol_population_history.py` imports official Statbel @@ -207,6 +231,12 @@ population-by-statistical-sector tables and matching sector geometries for each year as a separate dataset in `statbel:population-statistical-sector:mol`. +Nieuwe of met `--force` herbouwde edities doorlopen verplicht de +import-preflight. Bestaande historische caches zonder preflightmanifest mogen +alleen worden hergebruikt wanneer de overeenkomende Dataset al bestaat; een +nieuwe import uit zo'n cache wordt geweigerd totdat de officiele bron opnieuw +wordt gestaged en gevalideerd. + Complete sectors use their published population total. A rectangle that cuts through a sector uses an explicitly labelled area-weighted estimate; the source does not justify a more precise intra-sector distribution. diff --git a/docs/STORAGE_ARCHITECTURE.md b/docs/STORAGE_ARCHITECTURE.md index 94491350..44ef0732 100644 --- a/docs/STORAGE_ARCHITECTURE.md +++ b/docs/STORAGE_ARCHITECTURE.md @@ -101,6 +101,27 @@ Do not delete originals automatically. Derived outputs may be cleaned through ex ## Official temporal source artifacts +Statbel population staging stores immutable official source ZIPs below the +selected operator output root: + +```text +/ + raw/{year}/OPENDATA_SECTOREN_{year}[_NEW].zip + raw/{year}/sh_statbel_statistical_sectors_31370_{year}0101.geojson.zip + {scope}_statbel_population_{year}.geojson + {scope}_statbel_population_{year}.preflight.json +``` + +The preflight manifest records official URLs, archive/member size and SHA-256, +schema fingerprints, EPSG:31370 and situation-date evidence, bounded topology +repairs, national/scope join accounting, unlocated `ZZZZ` totals, baseline +change evidence and the final derived-snapshot checksum. The upload operator +revalidates all retained files against that manifest. Raw ZIPs and manifests +are provenance artifacts; queryable population geometry remains an ordinary +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. + 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 64e5b7e1..11feff53 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -43,6 +43,7 @@ - [x] Add a governed regional GRB plan -> stage -> checksum-confirmed apply workflow that preserves every previous snapshot. - [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. - [ ] 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/provision_mol_population_history.py b/scripts/provision_mol_population_history.py index c40dfeee..ab807210 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 hashlib import sha256 import io import json import os @@ -30,9 +31,15 @@ from requests.adapters import HTTPAdapter from shapely.geometry import mapping, shape from shapely.ops import transform from shapely.validation import make_valid +from urllib.parse import unquote, urlsplit from urllib3.util.retry import Retry from geographic_scopes import GEOGRAPHIC_SCOPES, GeographicScope +from statbel_population_preflight import ( + StatbelPreflightError, + validate_statbel_release, + write_manifest, +) MUNICIPALITY_NAME = "Mol" @@ -57,6 +64,7 @@ POPULATION_URLS = { 2024: "https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_2024.zip", 2025: "https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_2025_NEW.zip", } +POPULATION_LAYOUTS = {year: ("new" if year == 2025 else "standard") for year in POPULATION_URLS} def parse_args() -> argparse.Namespace: @@ -182,16 +190,35 @@ def population_rows(content: bytes, scope: GeographicScope) -> dict[str, dict[st text = raw.decode("utf-8-sig") except UnicodeDecodeError: text = raw.decode("cp1252") + required = {"CD_REFNIS", "CD_SECTOR", "TOTAL", "TX_DESCR_SECTOR_NL", "TX_DESCR_NL"} + reader = csv.DictReader(io.StringIO(text), delimiter="|") + missing_columns = sorted(required - set(reader.fieldnames or ())) + if missing_columns: + raise RuntimeError(f"Statbel population table is missing required columns: {', '.join(missing_columns)}") + return scoped_population_rows(list(reader), scope) + + +def scoped_population_rows(source_rows: list[dict[str, Any]], scope: GeographicScope) -> dict[str, dict[str, Any]]: members = {member.nis_code: member.name for member in scope.members} rows: dict[str, dict[str, Any]] = {} - for row in csv.DictReader(io.StringIO(text), delimiter="|"): + seen: set[str] = set() + for row_number, row in enumerate(source_rows, start=2): nis_code = str(row.get("CD_REFNIS") or "").strip() + sector_code = str(row.get("CD_SECTOR") or "").strip().upper() + total_raw = str(row.get("TOTAL") if row.get("TOTAL") is not None else "").strip() + if ( + len(nis_code) != 5 + or not nis_code.isdigit() + or len(sector_code) != 9 + or not sector_code[:5].isdigit() + or not total_raw.isdigit() + ): + raise RuntimeError(f"Statbel population row {row_number} has invalid code or TOTAL values") + if sector_code in seen: + raise RuntimeError(f"Statbel population table contains duplicate sector {sector_code}") + seen.add(sector_code) if nis_code not in members: continue - sector_code = str(row.get("CD_SECTOR") or "").strip() - total_raw = str(row.get("TOTAL") or "").strip() - if not sector_code or not total_raw or not total_raw.isdigit(): - continue rows[sector_code] = { "population_total": int(total_raw), "sector_name_nl": row.get("TX_DESCR_SECTOR_NL"), @@ -210,6 +237,7 @@ def build_snapshot( population: dict[str, dict[str, Any]], boundary, scope: GeographicScope, + preflight_manifest: dict[str, Any] | None = None, ) -> dict[str, Any]: transformer = Transformer.from_crs("EPSG:31370", "EPSG:4326", always_xy=True) member_codes = set(scope.nis_codes) @@ -245,8 +273,20 @@ def build_snapshot( "attribution": ATTRIBUTION, } features.append({"type": "Feature", "id": sector_code, "geometry": mapping(geometry), "properties": combined}) + if missing_population: + raise RuntimeError(f"Statbel geometry has {missing_population} sectors without population rows for {year}") if not features: raise RuntimeError(f"No joined population sectors were produced for {year}") + spatial_population_total = sum(int(feature["properties"]["population_total"]) for feature in features) + accounting = (preflight_manifest or {}).get("scope_accounting") or {} + if accounting: + expected_count = int(accounting.get("spatial_sector_count") or 0) + expected_total = int(accounting.get("spatial_population_total") or -1) + if len(features) != expected_count or spatial_population_total != expected_total: + raise RuntimeError( + f"Derived snapshot accounting differs from the passed Statbel preflight for {year}: " + f"features {len(features)}/{expected_count}, population {spatial_population_total}/{expected_total}" + ) return { "type": "FeatureCollection", "name": f"Statbel population by statistical sector - {scope.display_name} {year}", @@ -258,10 +298,145 @@ def build_snapshot( "geometry_clipped_to_area": True, "observation_year": year, "missing_population_sector_count": missing_population, + "spatial_population_total": spatial_population_total, + "unlocated_population_row_count": int(accounting.get("unlocated_row_count") or 0), + "unlocated_population_total": int(accounting.get("unlocated_population_total") or 0), + "accounted_population_total": int(accounting.get("accounted_population_total") or spatial_population_total), + "population_accounting_limitation": ( + "Statbel ZZZZ rows cannot be mapped and are excluded from spatial selection metrics." + ), "attribution": ATTRIBUTION, } +def sha256_path(path: Path) -> str: + digest = sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def write_bytes_atomic(path: Path, content: bytes) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(path.suffix + ".tmp") + temporary.write_bytes(content) + temporary.replace(path) + + +def write_text_atomic(path: Path, content: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(path.suffix + ".tmp") + temporary.write_text(content, encoding="utf-8") + temporary.replace(path) + + +def snapshot_path(output_dir: Path, scope: GeographicScope, year: int) -> Path: + return output_dir / f"{scope.key.replace('-', '_')}_statbel_population_{year}.geojson" + + +def preflight_manifest_path(output_dir: Path, scope: GeographicScope, year: int) -> Path: + return output_dir / f"{scope.key.replace('-', '_')}_statbel_population_{year}.preflight.json" + + +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 + + +def load_preflight_manifest(path: Path, snapshot: Path, year: int, scope: GeographicScope) -> dict[str, Any]: + try: + manifest = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + raise RuntimeError(f"Statbel preflight manifest is unreadable at {path}") from exc + release = manifest.get("release") or {} + accounting = manifest.get("scope_accounting") or {} + derived = (manifest.get("artifacts") or {}).get("derived_snapshot") or {} + if ( + manifest.get("status") != "passed" + or manifest.get("import_eligible") is not True + or int(release.get("year") or 0) != year + or accounting.get("scope_key") != scope.key + or derived.get("sha256") != sha256_path(snapshot) + ): + raise RuntimeError(f"Statbel preflight manifest at {path} does not authorize the retained snapshot") + for artifact_name in ("population", "geometry"): + artifact = (manifest.get("artifacts") or {}).get(artifact_name) or {} + retained_path = Path(str(artifact.get("retained_path") or "")) + if ( + not retained_path.is_file() + or artifact.get("archive_sha256") != sha256_path(retained_path) + or int(artifact.get("archive_size_bytes") or -1) != retained_path.stat().st_size + ): + raise RuntimeError( + f"Statbel preflight manifest at {path} does not authorize the retained {artifact_name} archive" + ) + return manifest + + +def stage_release( + *, + year: int, + population_content: bytes, + geometry_content: bytes, + output_dir: Path, + boundary, + scope: GeographicScope, +) -> tuple[Path, Path, dict[str, Any]]: + layout = POPULATION_LAYOUTS[year] + population_url = POPULATION_URLS[year] + geometry_url = SECTOR_URL.format(year=year) + result = validate_statbel_release( + year=year, + layout=layout, + population_content=population_content, + population_url=population_url, + geometry_content=geometry_content, + geometry_url=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 + write_bytes_atomic(population_archive_path, population_content) + write_bytes_atomic(geometry_archive_path, geometry_content) + + manifest = dict(result.manifest) + manifest["artifacts"] = { + **manifest["artifacts"], + "population": { + **manifest["artifacts"]["population"], + "retained_path": str(population_archive_path), + }, + "geometry": { + **manifest["artifacts"]["geometry"], + "retained_path": str(geometry_archive_path), + }, + } + population = scoped_population_rows(list(result.population.rows.values()), scope) + path = snapshot_path(output_dir, scope, year) + snapshot = build_snapshot( + year, + result.geometry.payload, + population, + boundary, + scope, + preflight_manifest=manifest, + ) + write_text_atomic(path, json.dumps(snapshot, ensure_ascii=False, separators=(",", ":"))) + manifest["artifacts"]["derived_snapshot"] = { + "retained_path": str(path), + "size_bytes": path.stat().st_size, + "sha256": sha256_path(path), + "feature_count": len(snapshot["features"]), + } + manifest_path = preflight_manifest_path(output_dir, scope, year) + write_manifest(manifest_path, manifest) + return path, manifest_path, manifest + + def locate_workspace( session: requests.Session, base_url: str, @@ -292,8 +467,11 @@ def upload_snapshot( path: Path, timeout: int, scope: GeographicScope, + preflight_path: Path, ) -> dict[str, Any]: observed_at = f"{year}-01-01T00:00:00Z" + preflight = load_preflight_manifest(preflight_path, path, year, scope) + accounting = preflight["scope_accounting"] source_metadata = { "provider": "Statbel", "authority_level": "authoritative", @@ -309,6 +487,11 @@ def upload_snapshot( "observation_date_precision": "year", "identity_stable": False, "identity_limitation": "Statistical-sector codes and boundaries can change between annual editions.", + "population_layout": preflight["release"]["population_layout"], + "population_accounting": accounting, + "spatial_population_limitation": ( + "ZZZZ population rows have no geometry and are excluded from spatial selection metrics." + ), "selection_aggregation": { "method": "area_weighted_sum", "property": "population_total", @@ -325,6 +508,12 @@ def upload_snapshot( "geometry_clipped_to_area": True, "sector_geometry_url": SECTOR_URL.format(year=year), "population_url": POPULATION_URLS[year], + "population_layout": preflight["release"]["population_layout"], + "preflight_manifest_path": str(preflight_path), + "preflight_manifest_sha256": sha256_path(preflight_path), + "population_archive_sha256": preflight["artifacts"]["population"]["archive_sha256"], + "sector_archive_sha256": preflight["artifacts"]["geometry"]["archive_sha256"], + "derived_snapshot_sha256": preflight["artifacts"]["derived_snapshot"]["sha256"], "generated_at": datetime.now(timezone.utc).isoformat(), } with path.open("rb") as handle: @@ -373,28 +562,52 @@ def main() -> int: try: boundary_path = resolve_boundary_path(args, scope) boundary = load_boundary(boundary_path, scope) - prepared: list[tuple[int, Path, int]] = [] + prepared: list[dict[str, Any]] = [] with build_session() as source_session: for year in years: - path = output_dir / f"{scope.key.replace('-', '_')}_statbel_population_{year}.geojson" + 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() - snapshot = build_snapshot( - year, - zip_member_json(sectors_response.content), - population_rows(population_response.content, scope), - boundary, - scope, + path, manifest_path, _manifest = stage_release( + year=year, + population_content=population_response.content, + geometry_content=sectors_response.content, + output_dir=output_dir, + boundary=boundary, + scope=scope, ) - path.write_text(json.dumps(snapshot, ensure_ascii=False, separators=(",", ":")), encoding="utf-8") + elif manifest_path.is_file(): + load_preflight_manifest(manifest_path, path, year, scope) + else: + preflight_status = "legacy_existing_only" payload = json.loads(path.read_text(encoding="utf-8")) - prepared.append((year, path, len(payload.get("features") or []))) + prepared.append( + { + "year": year, + "path": path, + "manifest_path": manifest_path if manifest_path.is_file() else None, + "feature_count": len(payload.get("features") or []), + "preflight_status": preflight_status, + } + ) if args.fetch_only: - results = [{"year": year, "path": str(path), "feature_count": count, "status": "prepared"} for year, path, count in prepared] + results = [ + { + "year": item["year"], + "path": str(item["path"]), + "preflight_manifest_path": str(item["manifest_path"]) if item["manifest_path"] else None, + "preflight_status": item["preflight_status"], + "feature_count": item["feature_count"], + "status": "prepared" if item["preflight_status"] == "passed" else "legacy_cached", + } + for item in prepared + ] else: base_url = args.base_url.rstrip("/") with requests.Session() as api_session: @@ -405,7 +618,9 @@ def main() -> int: area_name, args.import_timeout, ) - for year, path, count in prepared: + for item in prepared: + year = int(item["year"]) + path = item["path"] observed_at = f"{year}-01-01T00:00:00+00:00" dataset = next( ( @@ -419,6 +634,10 @@ def main() -> int: if dataset: results.append({"year": year, "dataset_id": dataset["id"], "feature_count": dataset.get("feature_count"), "status": "existing"}) continue + if item["preflight_status"] != "passed" or item["manifest_path"] is None: + raise RuntimeError( + f"Statbel {year} has only a legacy cached snapshot; rerun with --force to create preflight evidence before import" + ) dataset = upload_snapshot( api_session, base_url, @@ -428,10 +647,14 @@ def main() -> int: path, args.import_timeout, scope, + item["manifest_path"], ) 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: - print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False), file=sys.stderr) + payload = {"status": "error", "message": str(exc)} + if isinstance(exc, StatbelPreflightError): + payload.update({"error_code": exc.code, "details": exc.details}) + print(json.dumps(payload, ensure_ascii=False), file=sys.stderr) return 1 print( diff --git a/scripts/run_readiness_check.sh b/scripts/run_readiness_check.sh index cdd8f52b..56a2f603 100755 --- a/scripts/run_readiness_check.sh +++ b/scripts/run_readiness_check.sh @@ -45,6 +45,7 @@ ${PYTHON_BIN} -m py_compile scripts/prepare_operator_real_data_samples.py ${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/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 diff --git a/scripts/statbel_population_preflight.py b/scripts/statbel_population_preflight.py new file mode 100644 index 00000000..09f8b364 --- /dev/null +++ b/scripts/statbel_population_preflight.py @@ -0,0 +1,814 @@ +"""Fail-closed compatibility preflight for a staged Statbel population release. + +The preflight reads local official ZIP artifacts only. It validates source +identities, archive safety, schemas, CRS, sector joins and population +accounting before an operator may pass derived GeoJSON to DatasetService. +""" + +from __future__ import annotations + +import argparse +import csv +from dataclasses import dataclass +from datetime import datetime, timezone +from hashlib import sha256 +import io +import json +from pathlib import Path, PurePosixPath +import re +import sys +from typing import Any +from urllib.parse import urlsplit +import zipfile + +from shapely.geometry import mapping, shape +from shapely.ops import unary_union +from shapely.validation import make_valid + +from geographic_scopes import GEOGRAPHIC_SCOPES, GeographicScope + + +SCHEMA_VERSION = 1 +DEFAULT_MAX_ANNUAL_CHANGE_RATIO = 0.05 +MAX_ARCHIVE_MEMBERS = 64 +MAX_POPULATION_ARCHIVE_BYTES = 10 * 1024 * 1024 +MAX_POPULATION_UNCOMPRESSED_BYTES = 30 * 1024 * 1024 +MAX_GEOMETRY_ARCHIVE_BYTES = 100 * 1024 * 1024 +MAX_GEOMETRY_UNCOMPRESSED_BYTES = 400 * 1024 * 1024 +MAX_COMPRESSION_RATIO = 100.0 +REQUIRED_POPULATION_FIELDS = ( + "CD_REFNIS", + "CD_SECTOR", + "TOTAL", + "TX_DESCR_SECTOR_NL", + "TX_DESCR_NL", +) +REQUIRED_GEOMETRY_FIELDS = ( + "cd_sector", + "cd_munty_refnis", + "dt_situation", + "ms_area_ha", +) +SECTOR_CODE_PATTERN = re.compile(r"^[0-9]{5}[A-Z0-9-]{4}$") +MUNICIPALITY_CODE_PATTERN = re.compile(r"^[0-9]{5}$") +POPULATION_MEMBER_PATTERN = re.compile( + r"^OPENDATA_SECTOREN_(20[0-9]{2})(?:_(NEW|OLD))?\.(?:txt|csv)$", + re.IGNORECASE, +) +GEOMETRY_MEMBER_PATTERN = re.compile( + r"^sh_statbel_statistical_sectors_31370_(20[0-9]{2})0101\.geojson$", + re.IGNORECASE, +) +POPULATION_SOURCE_PATTERN = re.compile( + r"^/sites/default/files/files/opendata/bevolking/sectoren/" + r"OPENDATA_SECTOREN_(20[0-9]{2})(?:_(NEW|OLD))?\.zip$", + re.IGNORECASE, +) +GEOMETRY_SOURCE_PATTERN = re.compile( + r"^/sites/default/files/files/opendata/Statistische%20sectoren/" + r"sh_statbel_statistical_sectors_31370_(20[0-9]{2})0101\.geojson\.zip$", +) +ALLOWED_CRS_NAMES = {"EPSG:31370", "urn:ogc:def:crs:EPSG::31370"} + + +class StatbelPreflightError(RuntimeError): + def __init__(self, code: str, message: str, *, details: dict[str, Any] | None = None) -> None: + super().__init__(message) + self.code = code + self.message = message + self.details = details or {} + + +@dataclass(frozen=True) +class PopulationArchiveData: + member_name: str + layout: str + columns: tuple[str, ...] + rows: dict[str, dict[str, Any]] + population_total: int + + +@dataclass(frozen=True) +class GeometryArchiveData: + member_name: str + crs: str + property_columns: tuple[str, ...] + payload: dict[str, Any] + municipality_by_sector: dict[str, str] + repaired_sector_codes: tuple[str, ...] + + +@dataclass(frozen=True) +class StatbelPreflightResult: + manifest: dict[str, Any] + population: PopulationArchiveData + geometry: GeometryArchiveData + + +def _fail(code: str, message: str, **details: Any) -> None: + raise StatbelPreflightError(code, message, details=details) + + +def _sha256_bytes(content: bytes) -> str: + return sha256(content).hexdigest() + + +def _schema_fingerprint(values: tuple[str, ...]) -> str: + return sha256(json.dumps(values, ensure_ascii=True, separators=(",", ":")).encode()).hexdigest() + + +def _layout_from_variant(variant: str | None) -> str: + if variant is None: + return "standard" + return variant.lower() + + +def _validate_source_url(url: str, pattern: re.Pattern[str], *, code: str) -> re.Match[str]: + parsed = urlsplit(url) + match = pattern.fullmatch(parsed.path) + if ( + parsed.scheme != "https" + or parsed.hostname != "statbel.fgov.be" + or parsed.port not in {None, 443} + or parsed.username + or parsed.password + or parsed.query + or parsed.fragment + or not match + ): + _fail(code, "Source URL is outside the approved official Statbel release path.", url=url) + return match + + +def validate_population_source_url(url: str, year: int, layout: str) -> None: + match = _validate_source_url(url, POPULATION_SOURCE_PATTERN, code="STATBEL_POPULATION_URL_REJECTED") + url_year = int(match.group(1)) + url_layout = _layout_from_variant(match.group(2)) + if url_year != year or url_layout != layout: + _fail( + "STATBEL_POPULATION_URL_EDITION_MISMATCH", + "Population source URL does not match the requested year and layout.", + expected_year=year, + actual_year=url_year, + expected_layout=layout, + actual_layout=url_layout, + ) + + +def validate_geometry_source_url(url: str, year: int) -> None: + match = _validate_source_url(url, GEOMETRY_SOURCE_PATTERN, code="STATBEL_GEOMETRY_URL_REJECTED") + url_year = int(match.group(1)) + if url_year != year: + _fail( + "STATBEL_GEOMETRY_URL_EDITION_MISMATCH", + "Geometry source URL does not match the requested population year.", + expected_year=year, + actual_year=url_year, + ) + + +def _safe_archive_members( + content: bytes, + *, + compressed_limit: int, + uncompressed_limit: int, + artifact: str, +) -> list[zipfile.ZipInfo]: + if not content or len(content) > compressed_limit: + _fail( + "STATBEL_ARCHIVE_SIZE_REJECTED", + f"{artifact} archive exceeds the bounded compressed size.", + size_bytes=len(content), + limit_bytes=compressed_limit, + ) + try: + with zipfile.ZipFile(io.BytesIO(content)) as archive: + members = archive.infolist() + except zipfile.BadZipFile as exc: + raise StatbelPreflightError("STATBEL_ARCHIVE_INVALID", f"{artifact} archive is not a valid ZIP.") from exc + if not members or len(members) > MAX_ARCHIVE_MEMBERS: + _fail( + "STATBEL_ARCHIVE_MEMBER_COUNT_REJECTED", + f"{artifact} archive has an unexpected member count.", + member_count=len(members), + ) + total_size = 0 + for member in members: + path = PurePosixPath(member.filename.replace("\\", "/")) + if path.is_absolute() or ".." in path.parts or member.flag_bits & 0x1: + _fail( + "STATBEL_ARCHIVE_MEMBER_REJECTED", + f"{artifact} archive contains an unsafe member.", + member=member.filename, + ) + total_size += member.file_size + if member.file_size and member.compress_size == 0: + _fail("STATBEL_ARCHIVE_RATIO_REJECTED", f"{artifact} archive has an invalid compression ratio.") + if member.compress_size and member.file_size / member.compress_size > MAX_COMPRESSION_RATIO: + _fail( + "STATBEL_ARCHIVE_RATIO_REJECTED", + f"{artifact} archive exceeds the allowed compression ratio.", + member=member.filename, + ) + if total_size > uncompressed_limit: + _fail( + "STATBEL_ARCHIVE_UNCOMPRESSED_SIZE_REJECTED", + f"{artifact} archive exceeds the bounded uncompressed size.", + size_bytes=total_size, + limit_bytes=uncompressed_limit, + ) + return members + + +def _decode_population_table(raw: bytes) -> str: + try: + return raw.decode("utf-8-sig") + except UnicodeDecodeError: + try: + return raw.decode("cp1252") + except UnicodeDecodeError as exc: + raise StatbelPreflightError( + "STATBEL_POPULATION_ENCODING_REJECTED", + "Population table is neither UTF-8 nor Windows-1252 text.", + ) from exc + + +def parse_population_archive(content: bytes, year: int, layout: str) -> PopulationArchiveData: + members = _safe_archive_members( + content, + compressed_limit=MAX_POPULATION_ARCHIVE_BYTES, + uncompressed_limit=MAX_POPULATION_UNCOMPRESSED_BYTES, + artifact="Population", + ) + table_members = [member for member in members if member.filename.lower().endswith((".txt", ".csv"))] + if len(table_members) != 1: + _fail( + "STATBEL_POPULATION_MEMBER_AMBIGUOUS", + "Population archive must contain exactly one TXT or CSV table.", + table_member_count=len(table_members), + ) + member = table_members[0] + match = POPULATION_MEMBER_PATTERN.fullmatch(PurePosixPath(member.filename).name) + if not match: + _fail( + "STATBEL_POPULATION_MEMBER_REJECTED", + "Population table filename does not follow the official Statbel edition contract.", + member=member.filename, + ) + member_year = int(match.group(1)) + member_layout = _layout_from_variant(match.group(2)) + if member_year != year or member_layout != layout: + _fail( + "STATBEL_POPULATION_MEMBER_EDITION_MISMATCH", + "Population table member does not match the requested year and layout.", + expected_year=year, + actual_year=member_year, + expected_layout=layout, + actual_layout=member_layout, + ) + if member_layout == "old" or (year == 2025 and member_layout != "new"): + _fail( + "STATBEL_LAYOUT_NOT_CURRENT", + "The selected population layout is transition evidence and is not import eligible.", + year=year, + layout=member_layout, + ) + with zipfile.ZipFile(io.BytesIO(content)) as archive: + text = _decode_population_table(archive.read(member)) + reader = csv.DictReader(io.StringIO(text), delimiter="|") + columns = tuple(reader.fieldnames or ()) + missing_columns = sorted(set(REQUIRED_POPULATION_FIELDS) - set(columns)) + if missing_columns: + _fail( + "STATBEL_POPULATION_SCHEMA_MISMATCH", + "Population table is missing required columns.", + missing_columns=missing_columns, + columns=list(columns), + ) + + rows: dict[str, dict[str, Any]] = {} + population_total = 0 + for row_number, source_row in enumerate(reader, start=2): + row = {str(key): value for key, value in source_row.items() if key is not None} + municipality_code = str(row.get("CD_REFNIS") or "").strip() + sector_code = str(row.get("CD_SECTOR") or "").strip().upper() + total_raw = str(row.get("TOTAL") or "").strip() + if ( + not MUNICIPALITY_CODE_PATTERN.fullmatch(municipality_code) + or not SECTOR_CODE_PATTERN.fullmatch(sector_code) + ): + _fail( + "STATBEL_POPULATION_SECTOR_CODE_REJECTED", + "Population row contains an invalid municipality or sector code.", + row_number=row_number, + municipality_code=municipality_code, + sector_code=sector_code, + ) + if not total_raw.isdigit(): + _fail( + "STATBEL_POPULATION_TOTAL_REJECTED", + "Population TOTAL must be a non-negative integer.", + row_number=row_number, + sector_code=sector_code, + value=total_raw, + ) + if sector_code in rows: + _fail( + "STATBEL_POPULATION_DUPLICATE_SECTOR", + "Population table contains duplicate sector codes.", + sector_code=sector_code, + ) + total = int(total_raw) + rows[sector_code] = {**row, "CD_REFNIS": municipality_code, "CD_SECTOR": sector_code, "TOTAL": total} + population_total += total + if not rows or population_total <= 0: + _fail("STATBEL_POPULATION_EMPTY", "Population table contains no usable population accounting.") + return PopulationArchiveData( + member_name=member.filename, + layout=member_layout, + columns=columns, + rows=rows, + population_total=population_total, + ) + + +def _crs_name(payload: dict[str, Any]) -> str: + crs = payload.get("crs") + if not isinstance(crs, dict): + return "" + properties = crs.get("properties") + return str(properties.get("name") or "") if isinstance(properties, dict) else "" + + +def _polygonal_part(geometry): + if geometry.geom_type in {"Polygon", "MultiPolygon"}: + return geometry + polygonal = [part for part in getattr(geometry, "geoms", ()) if part.geom_type in {"Polygon", "MultiPolygon"}] + return unary_union(polygonal) if polygonal else geometry + + +def parse_geometry_archive(content: bytes, year: int) -> GeometryArchiveData: + members = _safe_archive_members( + content, + compressed_limit=MAX_GEOMETRY_ARCHIVE_BYTES, + uncompressed_limit=MAX_GEOMETRY_UNCOMPRESSED_BYTES, + artifact="Geometry", + ) + geometry_members = [member for member in members if member.filename.lower().endswith(".geojson")] + if len(geometry_members) != 1: + _fail( + "STATBEL_GEOMETRY_MEMBER_AMBIGUOUS", + "Geometry archive must contain exactly one GeoJSON dataset.", + geometry_member_count=len(geometry_members), + ) + member = geometry_members[0] + match = GEOMETRY_MEMBER_PATTERN.fullmatch(PurePosixPath(member.filename).name) + if not match or int(match.group(1)) != year: + _fail( + "STATBEL_GEOMETRY_MEMBER_EDITION_MISMATCH", + "Geometry member does not match the requested January 1 edition.", + expected_year=year, + member=member.filename, + ) + with zipfile.ZipFile(io.BytesIO(content)) as archive: + try: + payload = json.loads(archive.read(member).decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise StatbelPreflightError( + "STATBEL_GEOMETRY_JSON_REJECTED", + "Geometry member is not valid UTF-8 GeoJSON.", + ) from exc + if not isinstance(payload, dict) or payload.get("type") != "FeatureCollection": + _fail("STATBEL_GEOMETRY_TYPE_REJECTED", "Geometry artifact is not a GeoJSON FeatureCollection.") + crs = _crs_name(payload) + if crs not in ALLOWED_CRS_NAMES: + _fail( + "STATBEL_GEOMETRY_CRS_REJECTED", + "Geometry artifact must explicitly declare EPSG:31370.", + actual_crs=crs or None, + ) + features = payload.get("features") + if not isinstance(features, list) or not features: + _fail("STATBEL_GEOMETRY_EMPTY", "Geometry artifact contains no features.") + + municipality_by_sector: dict[str, str] = {} + repaired_sector_codes: list[str] = [] + property_columns: set[str] = set() + expected_date = f"{year}-01-01" + for feature_number, feature in enumerate(features, start=1): + if not isinstance(feature, dict) or feature.get("type") != "Feature": + _fail("STATBEL_GEOMETRY_FEATURE_REJECTED", "Geometry artifact contains an invalid feature.") + properties = feature.get("properties") + if not isinstance(properties, dict): + _fail("STATBEL_GEOMETRY_PROPERTIES_REJECTED", "Geometry feature has no property object.") + property_columns.update(str(key) for key in properties) + missing = sorted(set(REQUIRED_GEOMETRY_FIELDS) - set(properties)) + if missing: + _fail( + "STATBEL_GEOMETRY_SCHEMA_MISMATCH", + "Geometry feature is missing required properties.", + feature_number=feature_number, + missing_columns=missing, + ) + sector_code = str(properties.get("cd_sector") or "").strip().upper() + municipality_code = str(properties.get("cd_munty_refnis") or "").strip() + if ( + not SECTOR_CODE_PATTERN.fullmatch(sector_code) + or not MUNICIPALITY_CODE_PATTERN.fullmatch(municipality_code) + ): + _fail( + "STATBEL_GEOMETRY_SECTOR_CODE_REJECTED", + "Geometry feature contains an invalid municipality or sector code.", + sector_code=sector_code, + municipality_code=municipality_code, + ) + if sector_code in municipality_by_sector: + _fail( + "STATBEL_GEOMETRY_DUPLICATE_SECTOR", + "Geometry artifact contains duplicate sector codes.", + sector_code=sector_code, + ) + if str(properties.get("dt_situation") or "") != expected_date: + _fail( + "STATBEL_GEOMETRY_DATE_MISMATCH", + "Geometry situation date does not match the population reference year.", + sector_code=sector_code, + expected_date=expected_date, + actual_date=properties.get("dt_situation"), + ) + geometry_payload = feature.get("geometry") + if not isinstance(geometry_payload, dict) or geometry_payload.get("type") not in {"Polygon", "MultiPolygon"}: + _fail( + "STATBEL_GEOMETRY_SHAPE_REJECTED", + "Sector geometry must be a Polygon or MultiPolygon.", + sector_code=sector_code, + ) + try: + geometry = shape(geometry_payload) + except (TypeError, ValueError) as exc: + raise StatbelPreflightError( + "STATBEL_GEOMETRY_SHAPE_REJECTED", + f"Sector {sector_code} has unreadable geometry.", + ) from exc + if geometry.is_empty or geometry.area <= 0: + _fail( + "STATBEL_GEOMETRY_INVALID", + "Sector geometry must be non-empty and have positive area.", + sector_code=sector_code, + ) + if not geometry.is_valid: + repaired = _polygonal_part(make_valid(geometry)) + area_delta = abs(repaired.area - geometry.area) + if ( + repaired.is_empty + or not repaired.is_valid + or repaired.geom_type not in {"Polygon", "MultiPolygon"} + or repaired.area <= 0 + or area_delta > max(0.01, geometry.area * 0.000001) + ): + _fail( + "STATBEL_GEOMETRY_REPAIR_REJECTED", + "Invalid sector geometry cannot be repaired without changing its polygonal meaning.", + sector_code=sector_code, + original_geometry_type=geometry.geom_type, + repaired_geometry_type=repaired.geom_type, + original_area=geometry.area, + repaired_area=repaired.area, + ) + geometry = repaired + feature["geometry"] = mapping(geometry) + repaired_sector_codes.append(sector_code) + try: + declared_area_ha = float(properties.get("ms_area_ha")) + except (TypeError, ValueError): + _fail( + "STATBEL_GEOMETRY_AREA_REJECTED", + "Sector ms_area_ha must be numeric.", + sector_code=sector_code, + ) + calculated_area_ha = geometry.area / 10_000 + tolerance = max(0.01, declared_area_ha * 0.001) + if declared_area_ha <= 0 or abs(calculated_area_ha - declared_area_ha) > tolerance: + _fail( + "STATBEL_GEOMETRY_AREA_MISMATCH", + "Declared sector area does not match EPSG:31370 geometry area.", + sector_code=sector_code, + declared_area_ha=declared_area_ha, + calculated_area_ha=calculated_area_ha, + ) + municipality_by_sector[sector_code] = municipality_code + return GeometryArchiveData( + member_name=member.filename, + crs=crs, + property_columns=tuple(sorted(property_columns)), + payload=payload, + municipality_by_sector=municipality_by_sector, + repaired_sector_codes=tuple(repaired_sector_codes), + ) + + +def _validate_scope(scope: GeographicScope, geometry: GeometryArchiveData, population: PopulationArchiveData) -> None: + requested = set(scope.nis_codes) + if not requested or any(not MUNICIPALITY_CODE_PATTERN.fullmatch(value) for value in requested): + _fail("STATBEL_SCOPE_REJECTED", "Approved geographic scope contains invalid NIS codes.") + geometry_codes = set(geometry.municipality_by_sector.values()) + population_codes = {str(row["CD_REFNIS"]) for row in population.rows.values()} + missing = sorted(code for code in requested if code not in geometry_codes or code not in population_codes) + if missing: + _fail( + "STATBEL_SCOPE_COVERAGE_MISSING", + "Candidate release does not cover every municipality in the approved scope.", + missing_nis_codes=missing, + ) + + +def _baseline_summary(path: Path | None, *, year: int, scope: GeographicScope) -> dict[str, Any] | None: + if path is None: + return None + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + raise StatbelPreflightError( + "STATBEL_BASELINE_REJECTED", + "Baseline snapshot is not readable GeoJSON evidence.", + ) from exc + baseline_year = int(payload.get("observation_year") or 0) + if baseline_year <= 0 or baseline_year >= year: + _fail( + "STATBEL_BASELINE_YEAR_REJECTED", + "Baseline observation year must precede the candidate release.", + baseline_year=baseline_year, + candidate_year=year, + ) + if set(str(value) for value in payload.get("member_nis_codes") or []) != set(scope.nis_codes): + _fail("STATBEL_BASELINE_SCOPE_MISMATCH", "Baseline snapshot does not use the same approved scope.") + features = payload.get("features") + if not isinstance(features, list) or not features: + _fail("STATBEL_BASELINE_EMPTY", "Baseline snapshot contains no population features.") + sector_codes: set[str] = set() + spatial_total = 0 + for feature in features: + properties = feature.get("properties") if isinstance(feature, dict) else None + if not isinstance(properties, dict): + _fail("STATBEL_BASELINE_SCHEMA_MISMATCH", "Baseline feature has no property object.") + sector_code = str(properties.get("source_feature_id") or properties.get("cd_sector") or "").strip() + total = properties.get("population_total") + if sector_code in sector_codes or not isinstance(total, int) or total < 0: + _fail("STATBEL_BASELINE_SCHEMA_MISMATCH", "Baseline population evidence is not unique and numeric.") + sector_codes.add(sector_code) + spatial_total += total + return { + "path": str(path), + "year": baseline_year, + "spatial_sector_count": len(sector_codes), + "spatial_population_total": spatial_total, + } + + +def validate_statbel_release( + *, + year: int, + layout: str, + population_content: bytes, + population_url: str, + geometry_content: bytes, + geometry_url: str, + scope: GeographicScope, + baseline_snapshot: Path | None = None, + max_annual_change_ratio: float = DEFAULT_MAX_ANNUAL_CHANGE_RATIO, +) -> StatbelPreflightResult: + if year < 2000 or year > datetime.now(timezone.utc).year + 1: + _fail("STATBEL_YEAR_REJECTED", "Candidate population year is outside the supported review range.", year=year) + if layout not in {"standard", "new"}: + _fail("STATBEL_LAYOUT_REJECTED", "Candidate layout must be standard or new.", layout=layout) + if max_annual_change_ratio <= 0 or max_annual_change_ratio > 0.25: + _fail("STATBEL_CHANGE_LIMIT_REJECTED", "Annual population change limit must be greater than 0 and at most 25%.") + validate_population_source_url(population_url, year, layout) + validate_geometry_source_url(geometry_url, year) + population = parse_population_archive(population_content, year, layout) + geometry = parse_geometry_archive(geometry_content, year) + _validate_scope(scope, geometry, population) + + population_codes = set(population.rows) + geometry_codes = set(geometry.municipality_by_sector) + municipality_mismatches = sorted( + code + for code in population_codes & geometry_codes + if str(population.rows[code]["CD_REFNIS"]) != geometry.municipality_by_sector[code] + ) + if municipality_mismatches: + _fail( + "STATBEL_JOIN_MUNICIPALITY_MISMATCH", + "Population and geometry assign one or more sectors to different reference municipalities.", + count=len(municipality_mismatches), + examples=municipality_mismatches[:10], + ) + geometry_without_population = sorted(geometry_codes - population_codes) + if geometry_without_population: + _fail( + "STATBEL_JOIN_POPULATION_MISSING", + "One or more sector geometries have no population row.", + count=len(geometry_without_population), + examples=geometry_without_population[:10], + ) + population_without_geometry = sorted(population_codes - geometry_codes) + unexpected_non_spatial = [code for code in population_without_geometry if not code.endswith("ZZZZ")] + if unexpected_non_spatial: + _fail( + "STATBEL_JOIN_GEOMETRY_MISSING", + "Population rows without geometry must use the explicit ZZZZ unlocated-sector contract.", + count=len(unexpected_non_spatial), + examples=unexpected_non_spatial[:10], + ) + + national_spatial_total = sum(int(population.rows[code]["TOTAL"]) for code in geometry_codes) + national_unlocated_total = sum(int(population.rows[code]["TOTAL"]) for code in population_without_geometry) + if national_spatial_total + national_unlocated_total != population.population_total: + _fail("STATBEL_TOTAL_RECONCILIATION_FAILED", "National population accounting does not reconcile.") + requested = set(scope.nis_codes) + scope_spatial_codes = { + code for code in geometry_codes if geometry.municipality_by_sector[code] in requested + } + scope_unlocated_codes = { + code for code in population_without_geometry if str(population.rows[code]["CD_REFNIS"]) in requested + } + scope_spatial_total = sum(int(population.rows[code]["TOTAL"]) for code in scope_spatial_codes) + scope_unlocated_total = sum(int(population.rows[code]["TOTAL"]) for code in scope_unlocated_codes) + + baseline = _baseline_summary(baseline_snapshot, year=year, scope=scope) + if baseline: + baseline_total = int(baseline["spatial_population_total"]) + if baseline_total <= 0: + _fail("STATBEL_BASELINE_TOTAL_REJECTED", "Baseline population total must be positive.") + years = year - int(baseline["year"]) + annual_change_ratio = (scope_spatial_total / baseline_total) ** (1 / years) - 1 + baseline["candidate_spatial_population_total"] = scope_spatial_total + baseline["annual_change_ratio"] = annual_change_ratio + baseline["max_annual_change_ratio"] = max_annual_change_ratio + if abs(annual_change_ratio) > max_annual_change_ratio: + _fail( + "STATBEL_POPULATION_CHANGE_REVIEW_REQUIRED", + "Candidate spatial population change exceeds the configured annual review limit.", + baseline_year=baseline["year"], + candidate_year=year, + annual_change_ratio=annual_change_ratio, + max_annual_change_ratio=max_annual_change_ratio, + ) + + manifest = { + "schema_version": SCHEMA_VERSION, + "status": "passed", + "import_eligible": True, + "generated_at": datetime.now(timezone.utc).isoformat(), + "release": { + "year": year, + "population_layout": layout, + "geometry_date": f"{year}-01-01", + "license": "CC BY 4.0", + }, + "artifacts": { + "population": { + "source_url": population_url, + "archive_size_bytes": len(population_content), + "archive_sha256": _sha256_bytes(population_content), + "member": population.member_name, + }, + "geometry": { + "source_url": geometry_url, + "archive_size_bytes": len(geometry_content), + "archive_sha256": _sha256_bytes(geometry_content), + "member": geometry.member_name, + }, + }, + "schemas": { + "population_columns": list(population.columns), + "population_schema_sha256": _schema_fingerprint(population.columns), + "geometry_property_columns": list(geometry.property_columns), + "geometry_schema_sha256": _schema_fingerprint(geometry.property_columns), + "geometry_crs": geometry.crs, + "geometry_repair_count": len(geometry.repaired_sector_codes), + "geometry_repaired_sector_codes": list(geometry.repaired_sector_codes), + }, + "national_accounting": { + "population_row_count": len(population.rows), + "geometry_feature_count": len(geometry_codes), + "spatial_population_total": national_spatial_total, + "unlocated_row_count": len(population_without_geometry), + "unlocated_population_total": national_unlocated_total, + "population_total": population.population_total, + }, + "scope_accounting": { + "scope_key": scope.key, + "scope_display_name": scope.display_name, + "member_count": len(scope.members), + "member_nis_codes": list(scope.nis_codes), + "spatial_sector_count": len(scope_spatial_codes), + "spatial_population_total": scope_spatial_total, + "unlocated_row_count": len(scope_unlocated_codes), + "unlocated_population_total": scope_unlocated_total, + "accounted_population_total": scope_spatial_total + scope_unlocated_total, + }, + "join_accounting": { + "geometry_without_population_count": 0, + "population_without_geometry_count": len(population_without_geometry), + "population_without_geometry_contract": "sector_code_suffix_ZZZZ", + }, + "baseline": baseline, + "checks": [ + "official_source_identity", + "archive_safety", + "population_schema", + "geometry_schema", + "geometry_crs_and_validity", + "sector_join", + "population_total_reconciliation", + "approved_scope_coverage", + "baseline_change_limit" if baseline else "baseline_not_supplied", + ], + "limitations": [ + "ZZZZ population rows have no map geometry and are excluded from spatial selection metrics.", + "Statistical-sector identity and boundaries are not assumed stable across editions.", + "Repairable source topology errors are normalized with make_valid and reported in this manifest.", + "A passed technical preflight is not an instruction to replace an existing Dataset.", + ], + } + return StatbelPreflightResult(manifest=manifest, population=population, geometry=geometry) + + +def write_manifest(path: Path, manifest: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(path.suffix + ".tmp") + temporary.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") + temporary.replace(path) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Validate a staged official Statbel population release.") + parser.add_argument("--year", type=int, required=True) + parser.add_argument("--layout", choices=("standard", "new"), required=True) + parser.add_argument("--scope", choices=sorted(GEOGRAPHIC_SCOPES), default="kempen-transport-region") + parser.add_argument("--population-archive", type=Path, required=True) + parser.add_argument("--population-url", required=True) + parser.add_argument("--geometry-archive", type=Path, required=True) + parser.add_argument("--geometry-url", required=True) + parser.add_argument("--baseline-snapshot", type=Path) + parser.add_argument("--max-annual-change-percent", type=float, default=5.0) + parser.add_argument("--output", type=Path, required=True) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + scope = GEOGRAPHIC_SCOPES[args.scope] + try: + result = validate_statbel_release( + year=args.year, + layout=args.layout, + population_content=args.population_archive.read_bytes(), + population_url=args.population_url, + geometry_content=args.geometry_archive.read_bytes(), + geometry_url=args.geometry_url, + scope=scope, + baseline_snapshot=args.baseline_snapshot, + max_annual_change_ratio=args.max_annual_change_percent / 100, + ) + manifest = dict(result.manifest) + manifest["artifacts"] = { + **manifest["artifacts"], + "population": { + **manifest["artifacts"]["population"], + "retained_path": str(args.population_archive), + }, + "geometry": { + **manifest["artifacts"]["geometry"], + "retained_path": str(args.geometry_archive), + }, + } + write_manifest(args.output, manifest) + except (OSError, StatbelPreflightError) as exc: + if isinstance(exc, StatbelPreflightError): + payload = {"status": "error", "error_code": exc.code, "message": exc.message, "details": exc.details} + else: + payload = {"status": "error", "error_code": "STATBEL_PREFLIGHT_IO_ERROR", "message": str(exc)} + print(json.dumps(payload, ensure_ascii=False), file=sys.stderr) + return 1 + print( + json.dumps( + { + "status": "ok", + "import_eligible": True, + "year": args.year, + "layout": args.layout, + "scope": scope.key, + "manifest_path": str(args.output), + "scope_accounting": manifest["scope_accounting"], + }, + ensure_ascii=False, + indent=2, + ) + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main())