feat: add official Kempen operating scope

scripts/README.md

@@ -1254,6 +1254,38 @@ The 2013-2025 series is methodologically separate from the historical
 when both exist; it never calculates one continuous trend across those source
 families.
 
+## Official Kempen operational scope
+
+GeoIntel defines its regional `Kempen` workspace as the official Vlaamse
+`Vervoerregio Kempen`: 28 explicitly registered municipalities. This is a
+reproducible policy boundary, not a claim about the wider cultural,
+landscape or historical Kempen.
+
+Prepare and inspect the current VRBG union and all member boundaries without
+changing persistence:
+
+```bash
+docker exec -it geointel python3 /app/scripts/provision_geographic_scope.py \
+  --scope kempen-transport-region --fetch-only
+```
+
+Persist the complete scope foundation:
+
+```bash
+docker exec -it geointel python3 /app/scripts/provision_geographic_scope.py \
+  --scope kempen-transport-region
+```
+
+The command creates or reuses `Kempen Regional Workbench`, the regional Area,
+28 municipality Areas and two VRBG source datasets through the canonical API.
+Artifacts and checksums remain below
+`/app/storage/operator-data/geographic-scopes/kempen-transport-region`.
+Repeat runs are idempotent; `--force` refreshes today's source snapshot.
+
+This command provisions boundaries only. Regional buildings, population,
+land use, roads, water and parcels must be added by bounded source operators;
+missing themes remain unavailable and are never filled with synthetic values.
+
 ## Tower deployment
 
 Push the local branch to Gitea, then rebuild the Unraid/Tower Docker runtime:

scripts/geographic_scopes.py

@@ -0,0 +1,122 @@
+"""Canonical operator scope definitions for GeoIntel.
+
+These definitions describe administrative/policy scopes only. They do not
+claim that a policy boundary equals a cultural or physical landscape region.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class ScopeMember:
+    name: str
+    nis_code: str
+
+
+@dataclass(frozen=True)
+class GeographicScope:
+    key: str
+    display_name: str
+    project_name: str
+    project_region: str
+    area_name: str
+    authority_name: str
+    authority_url: str
+    scope_type: str
+    limitation_message: str
+    members: tuple[ScopeMember, ...]
+
+    @property
+    def nis_codes(self) -> tuple[str, ...]:
+        return tuple(member.nis_code for member in self.members)
+
+
+MOL_SCOPE = GeographicScope(
+    key="mol",
+    display_name="Mol",
+    project_name="Mol Municipality Workbench",
+    project_region="Mol, Kempen",
+    area_name="Gemeente Mol - officiële grens",
+    authority_name="Digitaal Vlaanderen VRBG",
+    authority_url="https://geo.api.vlaanderen.be/VRBG/ogc/features/v1/collections/Refgem/items",
+    scope_type="municipality",
+    limitation_message="Officiële gemeentegrens; dit is geen perceelsgrens.",
+    members=(ScopeMember("Mol", "13025"),),
+)
+
+
+KEMPEN_TRANSPORT_REGION_SCOPE = GeographicScope(
+    key="kempen-transport-region",
+    display_name="Kempen (28 gemeenten)",
+    project_name="Kempen Regional Workbench",
+    project_region="Vervoerregio Kempen, Vlaanderen",
+    area_name="Vervoerregio Kempen - officiële operationele grens",
+    authority_name="Vlaamse overheid - Vervoerregio Kempen",
+    authority_url=(
+        "https://www.vlaanderen.be/mobiliteitsprofessionals/personenvervoer/"
+        "basisbereikbaarheid/mobiliteitsuitdagingen-regionaal-aanpakken/"
+        "vervoerregios/over-de-vervoerregio-kempen"
+    ),
+    scope_type="transport_region",
+    limitation_message=(
+        "Operationele beleidsgrens van de Vlaamse vervoerregio Kempen; "
+        "geen claim over de ruimere culturele, landschappelijke of historische Kempen."
+    ),
+    members=(
+        ScopeMember("Arendonk", "13001"),
+        ScopeMember("Baarle-Hertog", "13002"),
+        ScopeMember("Balen", "13003"),
+        ScopeMember("Beerse", "13004"),
+        ScopeMember("Dessel", "13006"),
+        ScopeMember("Geel", "13008"),
+        ScopeMember("Grobbendonk", "13010"),
+        ScopeMember("Herentals", "13011"),
+        ScopeMember("Herenthout", "13012"),
+        ScopeMember("Herselt", "13013"),
+        ScopeMember("Hoogstraten", "13014"),
+        ScopeMember("Hulshout", "13016"),
+        ScopeMember("Kasterlee", "13017"),
+        ScopeMember("Laakdal", "13053"),
+        ScopeMember("Lille", "13019"),
+        ScopeMember("Meerhout", "13021"),
+        ScopeMember("Merksplas", "13023"),
+        ScopeMember("Mol", "13025"),
+        ScopeMember("Nijlen", "12026"),
+        ScopeMember("Olen", "13029"),
+        ScopeMember("Oud-Turnhout", "13031"),
+        ScopeMember("Ravels", "13035"),
+        ScopeMember("Retie", "13036"),
+        ScopeMember("Rijkevorsel", "13037"),
+        ScopeMember("Turnhout", "13040"),
+        ScopeMember("Vorselaar", "13044"),
+        ScopeMember("Vosselaar", "13046"),
+        ScopeMember("Westerlo", "13049"),
+    ),
+)
+
+
+GEOGRAPHIC_SCOPES = {
+    scope.key: scope
+    for scope in (MOL_SCOPE, KEMPEN_TRANSPORT_REGION_SCOPE)
+}
+
+
+def validate_scope(scope: GeographicScope) -> None:
+    if not scope.key or not scope.project_name or not scope.area_name:
+        raise ValueError("Geographic scope identity fields must not be empty")
+    if len(scope.members) == 0:
+        raise ValueError(f"Geographic scope {scope.key} has no members")
+    names = [member.name.casefold() for member in scope.members]
+    codes = [member.nis_code for member in scope.members]
+    if len(names) != len(set(names)):
+        raise ValueError(f"Geographic scope {scope.key} contains duplicate municipality names")
+    if len(codes) != len(set(codes)):
+        raise ValueError(f"Geographic scope {scope.key} contains duplicate NIS codes")
+    if any(len(code) != 5 or not code.isdigit() for code in codes):
+        raise ValueError(f"Geographic scope {scope.key} contains an invalid NIS code")
+
+
+for _scope in GEOGRAPHIC_SCOPES.values():
+    validate_scope(_scope)

scripts/provision_geographic_scope.py

@@ -0,0 +1,571 @@
+"""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())

scripts/run_readiness_check.sh

@@ -47,6 +47,8 @@ ${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/provision_mol_historical_landuse.py
 ${PYTHON_BIN} -m py_compile scripts/provision_official_landuse_timeseries.py
+${PYTHON_BIN} -m py_compile scripts/geographic_scopes.py
+${PYTHON_BIN} -m py_compile scripts/provision_geographic_scope.py
 ${PYTHON_BIN} -m py_compile scripts/export_operator_yolo_dataset.py
 ${PYTHON_BIN} -m py_compile scripts/export_operator_yolo_tile_dataset.py
 ${PYTHON_BIN} -m py_compile scripts/audit_operator_yolo_dataset_quality.py