"""Provision an official geographic scope through the canonical GeoIntel API. The operator fetches current VRBG municipality boundaries, creates one union scope boundary plus a member-boundary artifact, and persists a project, the regional area, all municipality areas and both datasets. It never runs during application startup and never writes directly to PostGIS. """ from __future__ import annotations import argparse import hashlib import json import os import sys 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 GeometryCollection, MultiPolygon, Polygon, mapping, shape from shapely.ops import transform, unary_union from shapely.validation import make_valid from urllib3.util.retry import Retry from geographic_scopes import GEOGRAPHIC_SCOPES, GeographicScope VRBG_ITEMS_URL = "https://geo.api.vlaanderen.be/VRBG/ogc/features/v1/collections/Refgem/items" VRBG_ATTRIBUTION = "Bron: Voorlopig referentiebestand gemeentegrenzen, Digitaal Vlaanderen" DEFAULT_SCOPE_KEY = "kempen-transport-region" DEFAULT_OUTPUT_ROOT = Path("/app/storage/operator-data/geographic-scopes") DEFAULT_API_URL = "http://127.0.0.1:8000" GEOJSON_CRS = {"type": "name", "properties": {"name": "EPSG:4326"}} def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Provision an official GeoIntel geographic scope.") 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( "--output-root", type=Path, default=Path(os.environ.get("GEOINTEL_SCOPE_OUTPUT_ROOT", DEFAULT_OUTPUT_ROOT)), ) parser.add_argument("--request-timeout", type=int, default=180) parser.add_argument("--import-timeout", type=int, default=1800) parser.add_argument("--force", action="store_true", help="Refresh official source artifacts for today's snapshot.") parser.add_argument("--fetch-only", action="store_true", help="Validate and write artifacts without API persistence.") return parser.parse_args() def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def sha256_file(path: Path) -> str: digest = hashlib.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_json_atomic(path: Path, payload: dict[str, Any], *, pretty: bool = False) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(f"{path.suffix}.partial") temporary.write_text( json.dumps( payload, ensure_ascii=False, indent=2 if pretty else None, separators=None if pretty else (",", ":"), sort_keys=pretty, ), encoding="utf-8", ) temporary.replace(path) def normalize_polygonal(geometry): if geometry is None or geometry.is_empty: return None if not geometry.is_valid: geometry = make_valid(geometry) if isinstance(geometry, (Polygon, MultiPolygon)): return geometry if isinstance(geometry, GeometryCollection): polygons = [part for part in geometry.geoms if isinstance(part, (Polygon, MultiPolygon)) and not part.is_empty] if polygons: merged = unary_union(polygons) return merged if isinstance(merged, (Polygon, MultiPolygon)) and not merged.is_empty else None return None def metric_area_km2(geometry) -> float: transformer = Transformer.from_crs(4326, 31370, always_xy=True) return float(transform(transformer.transform, geometry).area / 1_000_000) def build_source_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, ) adapter = HTTPAdapter(max_retries=retry) session = requests.Session() session.headers.update({"User-Agent": "GeoIntel-Geographic-Scope-Operator/1.0"}) session.mount("https://", adapter) session.mount("http://", adapter) return session def fetch_scope_members(session: requests.Session, scope: GeographicScope, timeout: int) -> tuple[list[dict[str, Any]], str]: response = session.get( VRBG_ITEMS_URL, params={"f": "application/geo+json", "limit": "1000"}, timeout=timeout, ) response.raise_for_status() expected = {member.nis_code: member.name for member in scope.members} selected: dict[str, dict[str, Any]] = {} for feature in response.json().get("features") or []: properties = feature.get("properties") or {} nis_code = str(properties.get("NISCODE") or "") if nis_code not in expected: continue if nis_code in selected: raise RuntimeError(f"Official VRBG returned duplicate NIS code {nis_code}") actual_name = str(properties.get("NAAM") or "") if actual_name.casefold() != expected[nis_code].casefold(): raise RuntimeError( f"Official VRBG name drift for {nis_code}: expected {expected[nis_code]!r}, received {actual_name!r}" ) selected[nis_code] = feature missing = [f"{name} ({code})" for code, name in expected.items() if code not in selected] if missing: raise RuntimeError(f"Official VRBG is missing scope members: {', '.join(missing)}") return [selected[member.nis_code] for member in scope.members], response.url def build_scope_payloads( scope: GeographicScope, source_features: list[dict[str, Any]], *, source_url: str, generated_at: str, ) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: if len(source_features) != len(scope.members): raise RuntimeError(f"Expected {len(scope.members)} source boundaries, received {len(source_features)}") member_features: list[dict[str, Any]] = [] member_geometries = [] for member, source_feature in zip(scope.members, source_features, strict=True): properties = dict(source_feature.get("properties") or {}) if str(properties.get("NISCODE") or "") != member.nis_code: raise RuntimeError(f"Scope member order/code mismatch for {member.name}") geometry = normalize_polygonal(shape(source_feature.get("geometry"))) if geometry is None: raise RuntimeError(f"Official boundary for {member.name} is empty, invalid or non-polygonal") member_geometries.append(geometry) properties.update( { "source_name": "vrbg", "source_feature_id": str(source_feature.get("id") or member.nis_code), "layer_type": "municipality_boundary", "authority_level": "authoritative", "coverage_scope": scope.key, "scope_type": scope.scope_type, "municipality": member.name, "nis_code": member.nis_code, "attribution": VRBG_ATTRIBUTION, "source_url": source_url, } ) member_features.append( { "type": "Feature", "id": str(source_feature.get("id") or f"Refgem.{member.nis_code}"), "geometry": mapping(geometry), "properties": properties, } ) boundary = normalize_polygonal(unary_union(member_geometries)) if boundary is None: raise RuntimeError("Union of official scope member boundaries is invalid") member_codes = list(scope.nis_codes) boundary_payload = { "type": "FeatureCollection", "name": f"Official operation boundary - {scope.display_name}", "crs": GEOJSON_CRS, "features": [ { "type": "Feature", "id": f"scope:{scope.key}", "geometry": mapping(boundary), "properties": { "name": scope.area_name, "source_name": "vrbg", "source_feature_id": f"scope:{scope.key}", "layer_type": "regional_boundary", "authority_level": "authoritative", "coverage_scope": scope.key, "scope_type": scope.scope_type, "member_count": len(scope.members), "member_nis_codes": member_codes, "scope_authority": scope.authority_name, "scope_authority_url": scope.authority_url, "scope_limitation": scope.limitation_message, "attribution": VRBG_ATTRIBUTION, "source_url": source_url, }, } ], "source_url": source_url, "scope_authority_url": scope.authority_url, "scope_limitation": scope.limitation_message, "generated_at": generated_at, } members_payload = { "type": "FeatureCollection", "name": f"Official municipality boundaries - {scope.display_name}", "crs": GEOJSON_CRS, "features": member_features, "source_url": source_url, "scope_authority_url": scope.authority_url, "scope_limitation": scope.limitation_message, "generated_at": generated_at, } summary = { "scope_key": scope.key, "scope_type": scope.scope_type, "display_name": scope.display_name, "member_count": len(scope.members), "member_names": [member.name for member in scope.members], "member_nis_codes": member_codes, "area_km2": metric_area_km2(boundary), "wgs84_bbox": list(boundary.bounds), } return boundary_payload, members_payload, summary def artifact_paths(output_dir: Path, scope: GeographicScope, snapshot_date: str) -> tuple[Path, Path, Path]: stem = scope.key.replace("-", "_") return ( output_dir / f"{stem}_boundary_{snapshot_date}.geojson", output_dir / f"{stem}_municipalities_{snapshot_date}.geojson", output_dir / f"{stem}_scope_manifest.json", ) def cached_artifacts(output_dir: Path, scope: GeographicScope) -> tuple[Path, Path, dict[str, Any]] | None: manifest_path = output_dir / f"{scope.key.replace('-', '_')}_scope_manifest.json" if not manifest_path.exists(): return None manifest = json.loads(manifest_path.read_text(encoding="utf-8")) boundary_path = output_dir / str(manifest.get("boundary_filename") or "") members_path = output_dir / str(manifest.get("municipalities_filename") or "") if ( manifest.get("status") == "complete" and manifest.get("scope_key") == scope.key and manifest.get("member_count") == len(scope.members) and boundary_path.is_file() and members_path.is_file() and sha256_file(boundary_path) == manifest.get("boundary_sha256") and sha256_file(members_path) == manifest.get("municipalities_sha256") ): return boundary_path, members_path, manifest return None def prepare_artifacts(args: argparse.Namespace, scope: GeographicScope) -> tuple[Path, Path, dict[str, Any]]: output_dir = args.output_root / scope.key if not args.force: cached = cached_artifacts(output_dir, scope) if cached: return cached generated_at = utc_now() snapshot_date = generated_at[:10] boundary_path, members_path, manifest_path = artifact_paths(output_dir, scope, snapshot_date) with build_source_session() as session: source_features, source_url = fetch_scope_members(session, scope, args.request_timeout) boundary_payload, members_payload, summary = build_scope_payloads( scope, source_features, source_url=source_url, generated_at=generated_at, ) write_json_atomic(boundary_path, boundary_payload) write_json_atomic(members_path, members_payload) manifest = { "schema_version": 1, "status": "complete", "generated_at": generated_at, "observed_at": f"{snapshot_date}T00:00:00Z", "boundary_filename": boundary_path.name, "boundary_sha256": sha256_file(boundary_path), "municipalities_filename": members_path.name, "municipalities_sha256": sha256_file(members_path), "vrbg_source_url": source_url, "vrbg_attribution": VRBG_ATTRIBUTION, "scope_authority_name": scope.authority_name, "scope_authority_url": scope.authority_url, "scope_limitation": scope.limitation_message, **summary, } write_json_atomic(manifest_path, manifest, pretty=True) return boundary_path, members_path, manifest 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[:500]}") from exc if not response.ok: raise RuntimeError(f"GeoIntel API failed ({response.status_code}): {json.dumps(payload, ensure_ascii=False)[:1000]}") 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 True: page = response_data(session.get(url, params={"limit": 200, "offset": offset}, timeout=timeout)) page_items = list(page.get("items") or []) items.extend(page_items) if total is None and page.get("total") is not None: total = int(page["total"]) if not page_items or (total is not None and len(items) >= total) or len(page_items) < 200: break offset += len(page_items) if total is not None and len(items) != total: raise RuntimeError(f"GeoIntel list response returned {len(items)} of {total} records for {url}") return items def find_or_create_project(session: requests.Session, base_url: str, scope: GeographicScope, timeout: int) -> dict[str, Any]: projects = list_paginated_items(session, f"{base_url}/api/v1/projects", timeout=timeout) existing = next((item for item in projects if item.get("name") == scope.project_name), None) if existing: return existing return response_data( session.post( f"{base_url}/api/v1/projects", json={ "name": scope.project_name, "description": ( f"Operational GeoIntel scope for {scope.display_name}, composed from {len(scope.members)} current " f"VRBG municipality boundaries. {scope.limitation_message}" ), "region": scope.project_region, }, timeout=timeout, ) ) def find_or_create_areas( session: requests.Session, base_url: str, project_id: str, scope: GeographicScope, boundary_payload: dict[str, Any], members_payload: dict[str, Any], timeout: int, ) -> tuple[dict[str, Any], list[dict[str, Any]]]: areas = list_paginated_items(session, f"{base_url}/api/v1/projects/{project_id}/areas", timeout=timeout) by_name = {str(item.get("name")): item for item in areas} def ensure(name: str, geometry: dict[str, Any]) -> dict[str, Any]: existing = by_name.get(name) if existing: return existing created = response_data( session.post( f"{base_url}/api/v1/projects/{project_id}/areas", json={"name": name, "crs": "EPSG:4326", "geometry": geometry}, timeout=timeout, ) ) by_name[name] = created return created region_area = ensure(scope.area_name, boundary_payload["features"][0]["geometry"]) member_areas = [ ensure(f"Gemeente {member.name} - officiële grens", feature["geometry"]) for member, feature in zip(scope.members, members_payload["features"], strict=True) ] return region_area, member_areas def upload_dataset( session: requests.Session, *, base_url: str, project_id: str, area_id: str, path: Path, source_metadata: dict[str, Any], provenance_metadata: dict[str, Any], temporal_series_key: str, observed_at: str, timeout: int, ) -> dict[str, Any]: 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": "source", "source_name": "vrbg", "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": temporal_series_key, "observed_at": observed_at, "valid_from": observed_at, "temporal_granularity": "snapshot", "source_version": observed_at[:10], }, files={"file": (path.name, handle, "application/geo+json")}, timeout=timeout, ) return response_data(response) def provision_scope( args: argparse.Namespace, scope: GeographicScope, boundary_path: Path, members_path: Path, manifest: dict[str, Any], ) -> dict[str, Any]: base_url = args.base_url.rstrip("/") boundary_payload = json.loads(boundary_path.read_text(encoding="utf-8")) members_payload = json.loads(members_path.read_text(encoding="utf-8")) with requests.Session() as session: project = find_or_create_project(session, base_url, scope, args.import_timeout) project_id = str(project["id"]) region_area, member_areas = find_or_create_areas( session, base_url, project_id, scope, boundary_payload, members_payload, args.import_timeout, ) datasets = list_paginated_items( session, f"{base_url}/api/v1/projects/{project_id}/datasets", timeout=args.import_timeout, ) source_metadata = { "provider": "Digitaal Vlaanderen", "collection": "VRBG/Refgem", "authority_level": "authoritative", "coverage_scope": scope.key, "scope_type": scope.scope_type, "scope_authority": scope.authority_name, "scope_authority_url": scope.authority_url, "scope_limitation": scope.limitation_message, "member_count": len(scope.members), "member_nis_codes": list(scope.nis_codes), "attribution": VRBG_ATTRIBUTION, } common_provenance = { "operator_tool": "provision_geographic_scope.py", "operator_explicit_fetch": True, "manifest_path": str(args.output_root / scope.key / f"{scope.key.replace('-', '_')}_scope_manifest.json"), "source_url": manifest["vrbg_source_url"], "scope_authority_url": scope.authority_url, } dataset_specs = ( ( boundary_path, "regional_boundary", f"vrbg:scope-boundary:{scope.key}", manifest["boundary_sha256"], ), ( members_path, "municipality_boundaries", f"vrbg:scope-members:{scope.key}", manifest["municipalities_sha256"], ), ) persisted = [] for path, layer_type, series_key, checksum in dataset_specs: existing = next((item for item in datasets if item.get("original_filename") == path.name), None) if existing: persisted_checksum = (existing.get("provenance_metadata") or {}).get("artifact_sha256") if persisted_checksum and persisted_checksum != checksum: raise RuntimeError( f"Immutable scope dataset {path.name} has checksum {persisted_checksum}, " f"but the refreshed source produced {checksum}; use a new observation date instead of overwriting it" ) persisted.append(existing) continue created = upload_dataset( session, base_url=base_url, project_id=project_id, area_id=str(region_area["id"]), path=path, source_metadata={**source_metadata, "layer_type": layer_type}, provenance_metadata={**common_provenance, "artifact_sha256": checksum}, temporal_series_key=series_key, observed_at=manifest["observed_at"], timeout=args.import_timeout, ) persisted.append(created) return { "project_id": project_id, "project_name": project.get("name"), "region_area_id": str(region_area["id"]), "region_area_name": region_area.get("name"), "municipality_area_count": len(member_areas), "boundary_dataset_id": str(persisted[0]["id"]), "municipality_dataset_id": str(persisted[1]["id"]), } def main() -> int: args = parse_args() scope = GEOGRAPHIC_SCOPES[args.scope] try: boundary_path, members_path, manifest = prepare_artifacts(args, scope) workspace = None if args.fetch_only else provision_scope(args, scope, boundary_path, members_path, manifest) except (OSError, RuntimeError, ValueError, KeyError, requests.RequestException) as exc: print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False), file=sys.stderr) return 1 print( json.dumps( { "status": "ok", "mode": "fetch_only" if args.fetch_only else "provisioned", "scope": scope.key, "display_name": scope.display_name, "member_count": manifest["member_count"], "area_km2": manifest["area_km2"], "wgs84_bbox": manifest["wgs84_bbox"], "boundary_path": str(boundary_path), "municipalities_path": str(members_path), "workspace": workspace, "limitation": scope.limitation_message, }, ensure_ascii=False, indent=2, ) ) return 0 if __name__ == "__main__": sys.exit(main())