"""Provision all current official Flemish municipality Areas. The municipality inventory is discovered from the official VRBG RefGem collection at runtime. The script uses the existing geographic-scope artifact and API persistence flow; it never writes directly to PostGIS. """ from __future__ import annotations import argparse import json import os import sys from datetime import datetime, timezone from pathlib import Path from typing import Any import requests from geographic_scopes import GeographicScope, ScopeMember from provision_geographic_scope import ( VRBG_ATTRIBUTION, VRBG_ITEMS_URL, build_scope_payloads, build_source_session, provision_scope, sha256_file, write_json_atomic, ) DEFAULT_API_URL = "http://127.0.0.1:8000" DEFAULT_OUTPUT_ROOT = Path("/app/storage/operator-data/geographic-scopes") MIN_EXPECTED_MUNICIPALITIES = 270 MAX_EXPECTED_MUNICIPALITIES = 300 def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Provision the current official Flanders municipality scope.") 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=3600) parser.add_argument("--force", action="store_true") parser.add_argument("--fetch-only", action="store_true") parser.add_argument("--min-municipalities", type=int, default=MIN_EXPECTED_MUNICIPALITIES) parser.add_argument("--max-municipalities", type=int, default=MAX_EXPECTED_MUNICIPALITIES) return parser.parse_args() def discover_flanders_scope( session: requests.Session, *, timeout: int, min_municipalities: int, max_municipalities: int, ) -> tuple[GeographicScope, list[dict[str, Any]], str]: if min_municipalities <= 0 or max_municipalities < min_municipalities: raise ValueError("Municipality count safety limits are invalid") response = session.get( VRBG_ITEMS_URL, params={"f": "application/geo+json", "limit": "1000"}, timeout=timeout, ) response.raise_for_status() payload = response.json() features = payload.get("features") if not isinstance(features, list): raise RuntimeError("Official VRBG response does not contain a feature list") if not min_municipalities <= len(features) <= max_municipalities: raise RuntimeError( f"Official VRBG returned {len(features)} municipalities; expected " f"{min_municipalities}..{max_municipalities}. Refusing an incomplete or broadened scope." ) by_code: dict[str, dict[str, Any]] = {} names: set[str] = set() for feature in features: if not isinstance(feature, dict) or not isinstance(feature.get("geometry"), dict): raise RuntimeError("Official VRBG contains a malformed municipality feature") properties = feature.get("properties") if not isinstance(properties, dict): raise RuntimeError("Official VRBG municipality is missing properties") nis_code = str(properties.get("NISCODE") or "").strip() name = str(properties.get("NAAM") or "").strip() if len(nis_code) != 5 or not nis_code.isdigit() or not name: raise RuntimeError(f"Official VRBG municipality identity is invalid: {nis_code!r} / {name!r}") if nis_code in by_code or name.casefold() in names: raise RuntimeError(f"Official VRBG contains a duplicate municipality: {nis_code} / {name}") by_code[nis_code] = feature names.add(name.casefold()) ordered = [by_code[code] for code in sorted(by_code)] members = tuple( ScopeMember( name=str(feature["properties"]["NAAM"]).strip(), nis_code=str(feature["properties"]["NISCODE"]).strip(), ) for feature in ordered ) scope = GeographicScope( key="flanders", display_name=f"Vlaanderen ({len(members)} gemeenten)", project_name="Flanders Regional Workbench", project_region="Vlaanderen, België", area_name="Vlaanderen - officiële operationele grens", authority_name="Digitaal Vlaanderen VRBG", authority_url=VRBG_ITEMS_URL, scope_type="region", limitation_message=( "Officiële unie van de actuele VRBG-gemeentegrenzen. De operationele regio omvat het Vlaamse " "landgebied; territoriale zee, EEZ en continentaal plat zijn afzonderlijke maritieme scopes." ), members=members, ) return scope, ordered, response.url def prepare_artifacts( args: argparse.Namespace, scope: GeographicScope, features: list[dict[str, Any]], source_url: str, ) -> tuple[Path, Path, dict[str, Any]]: output_dir = args.output_root / scope.key manifest_path = output_dir / "flanders_scope_manifest.json" generated_at = datetime.now(timezone.utc).isoformat() snapshot_date = generated_at[:10] boundary_path = output_dir / f"flanders_boundary_{snapshot_date}.geojson" members_path = output_dir / f"flanders_municipalities_{snapshot_date}.geojson" member_codes = list(scope.nis_codes) if not args.force and manifest_path.is_file(): manifest = json.loads(manifest_path.read_text(encoding="utf-8")) cached_boundary = output_dir / str(manifest.get("boundary_filename") or "") cached_members = output_dir / str(manifest.get("municipalities_filename") or "") if ( manifest.get("status") == "complete" and manifest.get("member_nis_codes") == member_codes and cached_boundary.is_file() and cached_members.is_file() and sha256_file(cached_boundary) == manifest.get("boundary_sha256") and sha256_file(cached_members) == manifest.get("municipalities_sha256") ): return cached_boundary, cached_members, manifest boundary_payload, members_payload, summary = build_scope_payloads( scope, 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 main() -> int: args = parse_args() args.operator_tool = "provision_flanders_geographic_scope.py" try: with build_source_session() as session: scope, features, source_url = discover_flanders_scope( session, timeout=args.request_timeout, min_municipalities=args.min_municipalities, max_municipalities=args.max_municipalities, ) boundary_path, members_path, manifest = prepare_artifacts(args, scope, features, source_url) 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, "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__": raise SystemExit(main())