"""Provision official annual Statbel population snapshots for an approved scope. The command joins annual population totals to the matching official statistical-sector geometries, clips the result to an approved geographic scope and imports each year through the existing GeoIntel upload API. It never runs during application startup and it never synthesizes missing population values. Mol remains the backwards-compatible default. The canonical regional operator uses ``--scope kempen-transport-region`` and the persisted official scope boundary produced by ``provision_geographic_scope.py``. """ from __future__ import annotations import argparse import csv from dataclasses import dataclass from hashlib import sha256 import io import json import os import sys import zipfile from datetime import datetime, timezone from pathlib import Path from typing import Any import requests from pyproj import Transformer 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 ( MAX_GEOMETRY_ARCHIVE_BYTES, MAX_POPULATION_ARCHIVE_BYTES, StatbelPreflightError, validate_geometry_source_url, validate_population_source_url, validate_statbel_release, write_manifest, ) MUNICIPALITY_NAME = "Mol" MUNICIPALITY_NIS_CODE = "13025" PROJECT_NAME = "Mol Municipality Workbench" SERIES_KEY = "statbel:population-statistical-sector:mol" ATTRIBUTION = "Bron: Statbel, bevolking per statistische sector, CC BY 4.0" DEFAULT_API_URL = "http://127.0.0.1:8000" DEFAULT_OUTPUT_DIR = Path("/app/storage/operator-data/mol-population-history") DEFAULT_BOUNDARY_PATH = Path("/app/storage/operator-data/mol-municipality/mol_municipality_boundary.geojson") DEFAULT_SCOPE_KEY = "mol" DEFAULT_SCOPE_OUTPUT_ROOT = Path("/app/storage/operator-data/geographic-scopes") DEFAULT_REGIONAL_OUTPUT_ROOT = Path("/app/storage/operator-data/official-population") SECTOR_URL = ( "https://statbel.fgov.be/sites/default/files/files/opendata/Statistische%20sectoren/" "sh_statbel_statistical_sectors_31370_{year}0101.geojson.zip" ) POPULATION_URLS = { 2021: "https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_2021.zip", 2022: "https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_2022.zip", 2023: "https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_2023.zip", 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} @dataclass(frozen=True) class PopulationReleaseConfig: year: int layout: str population_url: str geometry_url: str def resolve_release_config( year: int, *, population_url: str | None = None, geometry_url: str | None = None, layout: str | None = None, ) -> PopulationReleaseConfig: overrides = (population_url, geometry_url, layout) if any(value is not None for value in overrides): if not all(value is not None for value in overrides): raise ValueError("Population URL, geometry URL and population layout must be supplied together") release = PopulationReleaseConfig( year=year, layout=str(layout), population_url=str(population_url), geometry_url=str(geometry_url), ) elif year in POPULATION_URLS: release = PopulationReleaseConfig( year=year, layout=POPULATION_LAYOUTS[year], population_url=POPULATION_URLS[year], geometry_url=SECTOR_URL.format(year=year), ) else: raise ValueError(f"Unsupported population year without an explicit governed release: {year}") validate_population_source_url(release.population_url, release.year, release.layout) validate_geometry_source_url(release.geometry_url, release.year) return release def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Provision official annual Statbel population snapshots.") parser.add_argument("--scope", choices=sorted(GEOGRAPHIC_SCOPES), default=DEFAULT_SCOPE_KEY) parser.add_argument("--base-url", default=os.environ.get("GEOINTEL_INTERNAL_API_URL", DEFAULT_API_URL)) parser.add_argument("--project-name", default=None) parser.add_argument("--area-name", default=None, help="Case-insensitive fragment identifying the persisted Area.") parser.add_argument("--years", default="2021,2022,2023,2024,2025") parser.add_argument( "--population-url", help="Exact official Statbel ZIP URL for one explicitly governed release; requires the other release flags", ) parser.add_argument( "--geometry-url", help="Exact matching official statistical-sector GeoJSON ZIP URL for one explicitly governed release", ) parser.add_argument("--population-layout", choices=("standard", "new")) parser.add_argument("--output-dir", type=Path, default=None) parser.add_argument("--boundary-path", type=Path, default=None) parser.add_argument( "--scope-output-root", type=Path, default=Path(os.environ.get("GEOINTEL_SCOPE_OUTPUT_ROOT", DEFAULT_SCOPE_OUTPUT_ROOT)), ) parser.add_argument("--request-timeout", type=int, default=180) parser.add_argument("--import-timeout", type=int, default=900) parser.add_argument("--force", action="store_true") parser.add_argument("--fetch-only", action="store_true") return parser.parse_args() def build_session() -> requests.Session: retry = Retry( total=5, connect=5, read=5, status=5, backoff_factor=1.0, status_forcelist=(429, 500, 502, 503, 504), allowed_methods=frozenset({"GET"}), raise_on_status=True, ) session = requests.Session() session.headers.update({"User-Agent": "GeoIntel-Official-Population-Operator/1.0"}) adapter = HTTPAdapter(max_retries=retry) session.mount("https://", adapter) session.mount("http://", adapter) return session def download_archive( session: requests.Session, *, url: str, year: int, layout: str | None, max_bytes: int, timeout: int, ) -> bytes: with session.get(url, timeout=timeout, stream=True) as response: response.raise_for_status() if layout is None: validate_geometry_source_url(response.url, year) else: validate_population_source_url(response.url, year, layout) content_length = response.headers.get("Content-Length") if content_length: try: advertised_size = int(content_length) except ValueError as exc: raise RuntimeError(f"Official source returned an invalid Content-Length for {url}") from exc if advertised_size <= 0 or advertised_size > max_bytes: raise RuntimeError(f"Official source archive exceeds the {max_bytes}-byte download limit") chunks: list[bytes] = [] size = 0 for chunk in response.iter_content(chunk_size=1024 * 1024): if not chunk: continue size += len(chunk) if size > max_bytes: raise RuntimeError(f"Official source archive exceeds the {max_bytes}-byte download limit") chunks.append(chunk) if size <= 0: raise RuntimeError(f"Official source archive is empty: {url}") return b"".join(chunks) def response_data(response: requests.Response) -> Any: try: payload = response.json() except ValueError as exc: raise RuntimeError(f"GeoIntel API returned non-JSON ({response.status_code}): {response.text[:300]}") from exc if not response.ok: raise RuntimeError(f"GeoIntel API failed ({response.status_code}): {json.dumps(payload, ensure_ascii=False)[:800]}") if not isinstance(payload, dict) or "data" not in payload: raise RuntimeError("GeoIntel API response does not use the canonical data envelope") return payload["data"] def list_paginated_items(session: requests.Session, url: str, *, timeout: int) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] offset = 0 total: int | None = None while total is None or offset < total: page = response_data(session.get(url, params={"limit": 200, "offset": offset}, timeout=timeout)) page_items = page.get("items") if isinstance(page, dict) else None if not isinstance(page_items, list): raise RuntimeError(f"GeoIntel list response for {url} has no items array") page_total = int(page.get("total", len(page_items))) if total is None: total = page_total elif page_total != total: raise RuntimeError("GeoIntel pagination total changed while reading the population workspace") items.extend(page_items) if not page_items: break offset += len(page_items) if total is not None and len(items) != total: raise RuntimeError(f"GeoIntel list response for {url} returned {len(items)} of {total} items") return items def series_key(scope: GeographicScope) -> str: return f"statbel:population-statistical-sector:{scope.key}" def resolve_output_dir(args: argparse.Namespace, scope: GeographicScope) -> Path: if args.output_dir is not None: return args.output_dir legacy = os.environ.get("MOL_POPULATION_OUTPUT_DIR") if scope.key == "mol" and legacy: return Path(legacy) if scope.key == "mol": return DEFAULT_OUTPUT_DIR return DEFAULT_REGIONAL_OUTPUT_ROOT / scope.key def resolve_boundary_path(args: argparse.Namespace, scope: GeographicScope) -> Path: if args.boundary_path is not None: return args.boundary_path legacy = os.environ.get("MOL_BOUNDARY_PATH") if scope.key == "mol" and legacy: return Path(legacy) if scope.key == "mol" and DEFAULT_BOUNDARY_PATH.exists(): return DEFAULT_BOUNDARY_PATH if scope.key == "belgium": national_scope_dir = args.scope_output_root / "belgium-north-sea" manifest_path = national_scope_dir / "manifest.json" boundary_path = national_scope_dir / "belgium_land_boundary.geojson" if not manifest_path.is_file() or not boundary_path.is_file(): raise RuntimeError( "Official Belgium boundary evidence is missing; " "run provision_belgium_north_sea_scope.py --fetch-only first" ) manifest = json.loads(manifest_path.read_text(encoding="utf-8")) artifact = (manifest.get("artifacts") or {}).get("belgium_land_boundary") or {} if ( manifest.get("scope") != "belgium-and-belgian-north-sea" or artifact.get("sha256") != sha256_path(boundary_path) ): raise RuntimeError("Official Belgium boundary evidence does not match the retained manifest") return boundary_path scope_dir = args.scope_output_root / scope.key manifest_path = scope_dir / f"{scope.key.replace('-', '_')}_scope_manifest.json" if not manifest_path.exists(): raise RuntimeError( f"Official scope manifest is missing at {manifest_path}; run provision_geographic_scope.py --scope {scope.key} first" ) manifest = json.loads(manifest_path.read_text(encoding="utf-8")) if manifest.get("scope_key") != scope.key or manifest.get("status") != "complete": raise RuntimeError(f"Official scope manifest at {manifest_path} is incomplete or belongs to another scope") boundary_path = scope_dir / str(manifest.get("boundary_filename") or "") if not boundary_path.is_file(): raise RuntimeError(f"Official scope boundary referenced by {manifest_path} is missing") return boundary_path def load_boundary(path: Path, scope: GeographicScope): if not path.exists(): raise RuntimeError(f"Boundary for {scope.display_name} is missing at {path}") payload = json.loads(path.read_text(encoding="utf-8")) features = payload.get("features") or [] if len(features) != 1: raise RuntimeError(f"Boundary artifact for {scope.display_name} must contain exactly one feature") boundary = shape(features[0]["geometry"]) if not boundary.is_valid: boundary = make_valid(boundary) if boundary.is_empty or not boundary.is_valid: raise RuntimeError(f"Boundary artifact for {scope.display_name} is invalid") return boundary def zip_member_json(content: bytes) -> dict[str, Any]: with zipfile.ZipFile(io.BytesIO(content)) as archive: member = next((name for name in archive.namelist() if name.lower().endswith(".geojson")), None) if not member: raise RuntimeError("Statbel sector archive contains no GeoJSON file") return json.loads(archive.read(member).decode("utf-8")) def population_rows(content: bytes, scope: GeographicScope) -> dict[str, dict[str, Any]]: with zipfile.ZipFile(io.BytesIO(content)) as archive: member = next((name for name in archive.namelist() if name.lower().endswith((".txt", ".csv"))), None) if not member: raise RuntimeError("Statbel population archive contains no text table") raw = archive.read(member) try: 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]] = {} 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 not scope.all_municipalities and nis_code not in members: continue municipality_name = members.get(nis_code) or str(row.get("TX_DESCR_NL") or nis_code).strip() rows[sector_code] = { "population_total": int(total_raw), "sector_name_nl": row.get("TX_DESCR_SECTOR_NL"), "municipality_name_nl": row.get("TX_DESCR_NL"), "municipality": municipality_name, "nis_code": nis_code, } if not rows: raise RuntimeError(f"Statbel population table contains no usable sectors for {scope.display_name}") return rows def build_snapshot( year: int, sector_payload: dict[str, Any], 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) features: list[dict[str, Any]] = [] missing_population = 0 for source_feature in sector_payload.get("features") or []: properties = source_feature.get("properties") or {} if ( not scope.all_municipalities and str(properties.get("cd_munty_refnis") or "") not in member_codes ): continue sector_code = str(properties.get("cd_sector") or "").strip() population_values = population.get(sector_code) if not population_values: missing_population += 1 continue geometry = transform(transformer.transform, shape(source_feature["geometry"])) if not geometry.is_valid: geometry = make_valid(geometry) geometry = geometry.intersection(boundary) if geometry.is_empty: continue if not geometry.is_valid: geometry = make_valid(geometry) combined = { **properties, **population_values, "source_name": "statbel", "source_feature_id": sector_code, "reference_layer_name": "population", "authority_level": "authoritative", "municipality": population_values["municipality"], "nis_code": population_values["nis_code"], "observation_year": year, "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}", "features": features, "coverage_scope": scope.key, "scope_type": scope.scope_type, "member_count": int(accounting.get("member_count") or len(scope.members)), "member_nis_codes": list(accounting.get("member_nis_codes") or scope.nis_codes), "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: prefix = f"{scope.key.replace('-', '_')}_statbel_population_" available: list[tuple[int, Path]] = [] for path in output_dir.glob(f"{prefix}*.geojson"): suffix = path.stem.removeprefix(prefix) if suffix.isdigit() and int(suffix) < year and path.is_file(): available.append((int(suffix), path)) return max(available, key=lambda item: item[0])[1] if available else None def load_preflight_manifest(path: Path, snapshot: Path, year: int, scope: GeographicScope) -> dict[str, Any]: 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, release: PopulationReleaseConfig | None = None, ) -> tuple[Path, Path, dict[str, Any]]: active_release = release or resolve_release_config(year) if active_release.year != year: raise ValueError("Population release year does not match the staged year") result = validate_statbel_release( year=year, layout=active_release.layout, population_content=population_content, population_url=active_release.population_url, geometry_content=geometry_content, geometry_url=active_release.geometry_url, scope=scope, baseline_snapshot=previous_snapshot_path(output_dir, scope, year), ) raw_dir = output_dir / "raw" / str(year) population_archive_path = raw_dir / Path(unquote(urlsplit(active_release.population_url).path)).name geometry_archive_path = raw_dir / Path(unquote(urlsplit(active_release.geometry_url).path)).name write_bytes_atomic(population_archive_path, population_content) write_bytes_atomic(geometry_archive_path, geometry_content) 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, project_name: str, area_name: str, timeout: int, ): projects = list_paginated_items(session, f"{base_url}/api/v1/projects", timeout=timeout) project = next((item for item in projects if item.get("name") == project_name), None) if not project: raise RuntimeError(f"Project {project_name!r} is missing") project_id = str(project["id"]) areas = list_paginated_items(session, f"{base_url}/api/v1/projects/{project_id}/areas", timeout=timeout) area_fragment = area_name.strip().casefold() matches = [item for item in areas if area_fragment in str(item.get("name") or "").casefold()] if len(matches) != 1: raise RuntimeError(f"Expected one official Area matching {area_name!r}, received {len(matches)}") datasets = list_paginated_items(session, f"{base_url}/api/v1/projects/{project_id}/datasets", timeout=timeout) return project_id, str(matches[0]["id"]), datasets def upload_snapshot( session: requests.Session, base_url: str, project_id: str, area_id: str, year: int, path: Path, timeout: int, scope: GeographicScope, preflight_path: Path, release: PopulationReleaseConfig | None = None, ) -> dict[str, Any]: active_release = release or resolve_release_config(year) if active_release.year != year: raise ValueError("Population release year does not match the upload year") observed_at = f"{year}-01-01T00:00:00Z" preflight = load_preflight_manifest(preflight_path, path, year, scope) accounting = preflight["scope_accounting"] coverage_zones = ( ["belgium", "flanders", "wallonia", "brussels"] if scope.key == "belgium" else ["flanders"] ) source_metadata = { "provider": "Statbel", "authority_level": "authoritative", "theme": "population", "coverage_zones": coverage_zones, "coverage_scope": scope.key, "scope_type": scope.scope_type, "scope_display_name": scope.display_name, "member_count": int(accounting.get("member_count") or len(scope.members)), "member_nis_codes": list(accounting.get("member_nis_codes") or scope.nis_codes), "geometry_clipped_to_area": True, "attribution": ATTRIBUTION, "license": "CC BY 4.0", "temporal_series_label": "Officiële bevolkingscijfers per statistische sector", "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", "label": "Inwoners", "unit": "inwoners", "warning_only_when_estimate": True, "warning": "Bevolking binnen een gedeeltelijke statistische sector is oppervlaktegewogen en blijft een schatting.", }, } provenance_metadata = { "operator_tool": "provision_mol_population_history.py", "operator_explicit_fetch": True, "scope_key": scope.key, "geometry_clipped_to_area": True, "sector_geometry_url": active_release.geometry_url, "population_url": active_release.population_url, "population_layout": preflight["release"]["population_layout"], "preflight_manifest_path": str(preflight_path), "preflight_manifest_sha256": sha256_path(preflight_path), "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: response = session.post( f"{base_url}/api/v1/projects/{project_id}/datasets/upload", data={ "dataset_type": "vector", "source": "operator_official_import", "dataset_role": "reference", "source_name": "statbel", "reference_layer_name": "population", "source_metadata_json": json.dumps(source_metadata, ensure_ascii=False), "provenance_metadata_json": json.dumps(provenance_metadata, ensure_ascii=False), "area_id": area_id, "temporal_series_key": series_key(scope), "observed_at": observed_at, "valid_from": observed_at, "valid_to": f"{year}-12-31T23:59:59Z", "temporal_granularity": "year", "source_version": str(year), }, files={"file": (path.name, handle, "application/geo+json")}, timeout=timeout, ) return response_data(response) def main() -> int: args = parse_args() scope = GEOGRAPHIC_SCOPES[args.scope] project_name = args.project_name or scope.project_name area_name = args.area_name or scope.area_name try: years = sorted({int(value.strip()) for value in args.years.split(",") if value.strip()}) except ValueError: print(json.dumps({"status": "error", "message": "Years must be comma-separated integers"}), file=sys.stderr) return 2 custom_release_requested = any((args.population_url, args.geometry_url, args.population_layout)) if not years or (custom_release_requested and len(years) != 1): message = "An explicit governed release requires exactly one population year" if custom_release_requested else "No years requested" print(json.dumps({"status": "error", "message": message}), file=sys.stderr) return 2 try: releases = { year: resolve_release_config( year, population_url=args.population_url if custom_release_requested else None, geometry_url=args.geometry_url if custom_release_requested else None, layout=args.population_layout if custom_release_requested else None, ) for year in years } except (ValueError, StatbelPreflightError) as exc: print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False), file=sys.stderr) return 2 output_dir = resolve_output_dir(args, scope) output_dir.mkdir(parents=True, exist_ok=True) results: list[dict[str, Any]] = [] try: boundary_path = resolve_boundary_path(args, scope) boundary = load_boundary(boundary_path, scope) prepared: list[dict[str, Any]] = [] with build_session() as source_session: for year in years: release = releases[year] path = snapshot_path(output_dir, scope, year) manifest_path = preflight_manifest_path(output_dir, scope, year) preflight_status = "passed" if args.force or not path.exists(): geometry_content = download_archive( source_session, url=release.geometry_url, year=year, layout=None, max_bytes=MAX_GEOMETRY_ARCHIVE_BYTES, timeout=args.request_timeout, ) population_content = download_archive( source_session, url=release.population_url, year=year, layout=release.layout, max_bytes=MAX_POPULATION_ARCHIVE_BYTES, timeout=args.request_timeout, ) path, manifest_path, _manifest = stage_release( year=year, population_content=population_content, geometry_content=geometry_content, output_dir=output_dir, boundary=boundary, scope=scope, release=release, ) elif manifest_path.is_file(): load_preflight_manifest(manifest_path, path, year, scope) else: preflight_status = "legacy_existing_only" payload = json.loads(path.read_text(encoding="utf-8")) prepared.append( { "year": year, "path": path, "manifest_path": manifest_path if manifest_path.is_file() else None, "feature_count": len(payload.get("features") or []), "member_count": int(payload.get("member_count") or len(scope.members)), "preflight_status": preflight_status, } ) if args.fetch_only: 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: project_id, area_id, existing = locate_workspace( api_session, base_url, project_name, area_name, args.import_timeout, ) for item in prepared: year = int(item["year"]) path = item["path"] observed_at = f"{year}-01-01T00:00:00+00:00" dataset = next( ( item for item in existing if item.get("temporal_series_key") == series_key(scope) and str(item.get("observed_at") or "").startswith(observed_at[:10]) ), None, ) 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, project_id, area_id, year, path, args.import_timeout, scope, item["manifest_path"], releases[year], ) results.append({"year": year, "dataset_id": dataset["id"], "feature_count": dataset.get("feature_count"), "status": "imported"}) except (OSError, RuntimeError, requests.RequestException, ValueError, KeyError, zipfile.BadZipFile) as exc: 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( json.dumps( { "status": "ok", "scope": scope.key, "display_name": scope.display_name, "member_count": max( (int(item.get("member_count") or 0) for item in prepared), default=len(scope.members), ), "series": series_key(scope), "snapshots": results, }, ensure_ascii=False, indent=2, ) ) return 0 if __name__ == "__main__": sys.exit(main())