"""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())