geointel/scripts/provision_regional_grb_context.py

"""Provision regional GRB roads, water and parcel datasets.

Every source layer is fetched in resumable municipality partitions for an
approved geographic scope. Source identities are assigned to exactly one
partition, while retained geometries are clipped only to the complete region.
The resulting artifacts are indexed through DatasetService and
VectorFeatureService; this operator never writes directly to vector_features.
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from dataclasses import dataclass
from datetime import date
from pathlib import Path
from typing import Any, Iterable
from urllib.parse import urlparse
from uuid import UUID

import requests
from shapely.geometry import LineString, MultiLineString, MultiPolygon, Polygon, mapping, shape
from shapely.ops import unary_union
from shapely.validation import make_valid

from geographic_scopes import GEOGRAPHIC_SCOPES, GeographicScope, ScopeMember
from provision_geographic_scope import fetch_scope_members
from provision_regional_grb_buildings import (
    DEFAULT_API_URL,
    DEFAULT_MAX_FEATURES_PER_MEMBER,
    DEFAULT_MAX_TOTAL_FEATURES,
    DEFAULT_OUTPUT_ROOT,
    DEFAULT_PAGE_LIMIT,
    DEFAULT_SCOPE_KEY,
    GEOJSON_CRS,
    GRB_ATTRIBUTION,
    bounds_overlap,
    build_member_geometries,
    build_source_session,
    ensure_backend_path,
    list_paginated_items,
    next_page_url,
    observed_at,
    reusable_manifest,
    safe_slug,
    sha256_file,
    utc_now,
    write_json_atomic,
)


GRB_COLLECTION_URL = "https://geo.api.vlaanderen.be/GRB/ogc/features/v1/collections/{collection}/items"


@dataclass(frozen=True)
class CollectionDefinition:
    name: str
    geometry_dimension: int


@dataclass(frozen=True)
class LayerDefinition:
    key: str
    collections: tuple[CollectionDefinition, ...]
    reference_layer_name: str
    layer_type: str
    geometry_types: tuple[str, ...]
    metric_key: str
    metric_method: str
    metric_label: str
    metric_unit: str
    metric_warning: str | None
    limitation_message: str


LAYERS = (
    LayerDefinition(
        key="roads",
        collections=(CollectionDefinition("Wegsegment", 1),),
        reference_layer_name="roads",
        layer_type="road",
        geometry_types=("LineString", "MultiLineString"),
        metric_key="road_length",
        metric_method="intersection_length",
        metric_label="Totale weglengte",
        metric_unit="km",
        metric_warning="De lengte volgt de GRB-wegsegmenten en zegt niets over rijstroken, verkeersvolume of verhardingsoppervlakte.",
        limitation_message="GRB Wegsegment represents road-network line segments, not traffic volume or routing suitability.",
    ),
    LayerDefinition(
        key="water",
        collections=(
            CollectionDefinition("WTZ", 2),
            CollectionDefinition("WLAS", 1),
            CollectionDefinition("WGR", 1),
        ),
        reference_layer_name="water",
        layer_type="water",
        geometry_types=("LineString", "MultiLineString", "Polygon", "MultiPolygon"),
        metric_key="water_area",
        metric_method="intersection_area",
        metric_label="Wateroppervlakte",
        metric_unit="ha",
        metric_warning="Watervolume is niet berekenbaar zonder betrouwbare diepte- of bathymetrische gegevens. De kaartbron levert alleen oppervlakte- en lijngeometrie.",
        limitation_message="GRB water combines surface-water polygons and water-related line collections; counts are object counts, not water volume.",
    ),
    LayerDefinition(
        key="parcels",
        collections=(CollectionDefinition("ADP", 2),),
        reference_layer_name="parcels",
        layer_type="parcel",
        geometry_types=("Polygon", "MultiPolygon"),
        metric_key="parcel_area",
        metric_method="intersection_area",
        metric_label="Perceeloppervlakte",
        metric_unit="ha",
        metric_warning="GRB-percelen zijn een grafische referentie en vormen geen juridische grensopmeting.",
        limitation_message="GRB ADP is a graphical representation of the presumed cadastral parcel location and is not a legal boundary survey.",
    ),
)
LAYER_BY_KEY = {definition.key: definition for definition in LAYERS}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Provision municipality-partitioned regional GRB context layers.")
    parser.add_argument("--scope", choices=sorted(GEOGRAPHIC_SCOPES), default=DEFAULT_SCOPE_KEY)
    parser.add_argument(
        "--layers",
        nargs="+",
        default=list(LAYER_BY_KEY),
        help="Space- or comma-separated subset: roads water parcels",
    )
    parser.add_argument("--observed-date", type=date.fromisoformat, default=date.today())
    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_REGIONAL_THEME_OUTPUT_ROOT", DEFAULT_OUTPUT_ROOT)),
    )
    parser.add_argument("--page-limit", type=int, default=DEFAULT_PAGE_LIMIT)
    parser.add_argument("--max-features-per-member", type=int, default=DEFAULT_MAX_FEATURES_PER_MEMBER)
    parser.add_argument("--max-total-features", type=int, default=DEFAULT_MAX_TOTAL_FEATURES)
    parser.add_argument("--request-timeout", type=int, default=180)
    parser.add_argument("--api-timeout", type=int, default=180)
    parser.add_argument("--batch-size", type=int, default=1000)
    parser.add_argument("--force", action="store_true", help="Refetch every selected municipality partition.")
    parser.add_argument("--fetch-only", action="store_true", help="Build and validate artifacts without persistence.")
    return parser.parse_args()


def selected_definitions(raw_layers: str | list[str]) -> list[LayerDefinition]:
    values = [raw_layers] if isinstance(raw_layers, str) else raw_layers
    requested = {
        item.strip().lower()
        for value in values
        for item in value.split(",")
        if item.strip()
    }
    unknown = requested - set(LAYER_BY_KEY)
    if unknown or not requested:
        raise ValueError(f"Unsupported layers: {sorted(unknown)}")
    return [definition for definition in LAYERS if definition.key in requested]


def source_session() -> requests.Session:
    session = build_source_session()
    session.headers.update({"User-Agent": "GeoIntel-Regional-GRB-Context-Operator/1.0"})
    return session


def geometry_dimension(geometry) -> int:
    if geometry is None or geometry.is_empty:
        return -1
    if "Polygon" in geometry.geom_type:
        return 2
    if "LineString" in geometry.geom_type or geometry.geom_type == "LinearRing":
        return 1
    if "Point" in geometry.geom_type:
        return 0
    if geometry.geom_type == "GeometryCollection":
        return max((geometry_dimension(part) for part in geometry.geoms), default=-1)
    return -1


def extract_dimension(geometry, expected_dimension: int):
    if geometry is None or geometry.is_empty:
        return None
    if not geometry.is_valid:
        geometry = make_valid(geometry)

    parts: list[Any] = []

    def collect(candidate) -> None:
        if candidate is None or candidate.is_empty:
            return
        if expected_dimension == 2:
            if isinstance(candidate, Polygon):
                parts.append(candidate)
                return
            if isinstance(candidate, MultiPolygon):
                parts.extend(part for part in candidate.geoms if not part.is_empty)
                return
        if expected_dimension == 1:
            if isinstance(candidate, LineString):
                parts.append(candidate)
                return
            if isinstance(candidate, MultiLineString):
                parts.extend(part for part in candidate.geoms if not part.is_empty)
                return
        if hasattr(candidate, "geoms"):
            for part in candidate.geoms:
                collect(part)

    collect(geometry)
    if not parts:
        return None
    normalized = unary_union(parts)
    if normalized.is_empty:
        return None
    if not normalized.is_valid:
        normalized = make_valid(normalized)
    if normalized.is_empty or not normalized.is_valid or geometry_dimension(normalized) != expected_dimension:
        return None
    return normalized


def normalized_geometry(payload: dict[str, Any] | None, expected_dimension: int):
    if not payload:
        return None
    return extract_dimension(shape(payload), expected_dimension)


def geometry_measure(geometry, expected_dimension: int) -> float:
    candidate = extract_dimension(geometry, expected_dimension)
    if candidate is None:
        return 0.0
    if expected_dimension == 2:
        return float(candidate.area)
    if expected_dimension == 1:
        return float(candidate.length)
    return 1.0


def assign_owner_nis(
    source_geometry,
    members: list[tuple[ScopeMember, Any]],
    *,
    expected_dimension: int,
) -> str | None:
    candidates: list[tuple[float, str]] = []
    source_bounds = source_geometry.bounds
    for member, boundary in members:
        if not bounds_overlap(source_bounds, boundary.bounds) or not source_geometry.intersects(boundary):
            continue
        score = geometry_measure(source_geometry.intersection(boundary), expected_dimension)
        if score > 0:
            candidates.append((score, member.nis_code))
    if not candidates:
        return None
    candidates.sort(key=lambda item: (-item[0], item[1]))
    return candidates[0][1]


def iter_collection_pages(
    session: requests.Session,
    collection: CollectionDefinition,
    bounds: tuple[float, float, float, float],
    *,
    page_limit: int,
    timeout: int,
) -> Iterable[tuple[CollectionDefinition, dict[str, Any], str]]:
    params = {
        "f": "application/geo+json",
        "limit": str(page_limit),
        "bbox": ",".join(f"{value:.8f}" for value in bounds),
    }
    url: str | None = GRB_COLLECTION_URL.format(collection=collection.name)
    seen_urls: set[str] = set()
    first_request = True
    while url:
        if url in seen_urls:
            raise RuntimeError(f"GRB pagination loop detected for {collection.name}: {url}")
        seen_urls.add(url)
        response = session.get(url, params=params if first_request else None, timeout=timeout)
        first_request = False
        response.raise_for_status()
        payload = response.json()
        if payload.get("type") != "FeatureCollection":
            raise RuntimeError(f"GRB {collection.name} returned a non-FeatureCollection response")
        yield collection, payload, response.url
        url = next_page_url(payload)


def iter_layer_pages(
    session: requests.Session,
    definition: LayerDefinition,
    bounds: tuple[float, float, float, float],
    *,
    page_limit: int,
    timeout: int,
) -> Iterable[tuple[CollectionDefinition, dict[str, Any], str]]:
    for collection in definition.collections:
        yield from iter_collection_pages(
            session,
            collection,
            bounds,
            page_limit=page_limit,
            timeout=timeout,
        )


def build_partition_features(
    pages: Iterable[tuple[CollectionDefinition, dict[str, Any], str]],
    *,
    definition: LayerDefinition,
    member: ScopeMember,
    members: list[tuple[ScopeMember, Any]],
    regional_boundary,
    scope: GeographicScope,
    max_features: int,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    features: list[dict[str, Any]] = []
    source_urls: list[str] = []
    seen_ids: set[str] = set()
    pages_by_collection: dict[str, int] = {collection.name: 0 for collection in definition.collections}
    features_by_collection: dict[str, int] = {collection.name: 0 for collection in definition.collections}
    geometry_types: dict[str, int] = {}
    bbox_feature_count = 0
    assigned_elsewhere_count = 0
    outside_scope_count = 0
    clipped_to_scope_count = 0
    member_boundary = next(boundary for candidate, boundary in members if candidate.nis_code == member.nis_code)

    for collection, payload, source_url in pages:
        source_urls.append(source_url)
        pages_by_collection[collection.name] += 1
        for source_feature in payload.get("features") or []:
            bbox_feature_count += 1
            raw_id = str(source_feature.get("id") or "").strip()
            if not raw_id:
                raise RuntimeError(
                    f"GRB {collection.name} feature is missing an official source identity"
                )
            feature_id = f"{collection.name}:{raw_id}"
            if feature_id in seen_ids:
                continue
            seen_ids.add(feature_id)

            source_geometry = normalized_geometry(source_feature.get("geometry"), collection.geometry_dimension)
            if source_geometry is None or not source_geometry.intersects(regional_boundary):
                outside_scope_count += 1
                continue
            owner_nis = (
                member.nis_code
                if member_boundary.covers(source_geometry)
                else assign_owner_nis(
                    source_geometry,
                    members,
                    expected_dimension=collection.geometry_dimension,
                )
            )
            if owner_nis != member.nis_code:
                assigned_elsewhere_count += 1
                continue

            clipped = not regional_boundary.covers(source_geometry)
            retained_geometry = source_geometry
            if clipped:
                retained_geometry = extract_dimension(
                    source_geometry.intersection(regional_boundary),
                    collection.geometry_dimension,
                )
                clipped_to_scope_count += 1
            if retained_geometry is None:
                outside_scope_count += 1
                continue
            if len(features) >= max_features:
                raise RuntimeError(
                    f"{member.name} {definition.key} exceeds --max-features-per-member={max_features}; "
                    "refusing truncated output"
                )

            properties = dict(source_feature.get("properties") or {})
            properties.update(
                {
                    "source_name": "grb",
                    "source_collection": collection.name,
                    "source_feature_id": feature_id,
                    "reference_layer_name": definition.reference_layer_name,
                    "layer_type": definition.layer_type,
                    "theme": definition.key,
                    "authority_level": "authoritative",
                    "coverage_scope": scope.key,
                    "scope_type": scope.scope_type,
                    "partition_scope": "municipality",
                    "partition_municipality": member.name,
                    "partition_nis_code": member.nis_code,
                    "partition_assignment": "maximum_same_dimension_intersection",
                    "clipped_to_regional_scope": clipped,
                    "attribution": GRB_ATTRIBUTION,
                }
            )
            features.append(
                {
                    "type": "Feature",
                    "id": feature_id,
                    "geometry": mapping(retained_geometry),
                    "properties": properties,
                }
            )
            features_by_collection[collection.name] += 1
            geometry_types[retained_geometry.geom_type] = geometry_types.get(retained_geometry.geom_type, 0) + 1

    if not features:
        raise RuntimeError(f"No GRB {definition.key} features were assigned to {member.name}")
    return features, {
        "municipality": member.name,
        "nis_code": member.nis_code,
        "pages_fetched": len(source_urls),
        "pages_by_collection": pages_by_collection,
        "source_urls": source_urls,
        "bbox_feature_count": bbox_feature_count,
        "feature_count": len(features),
        "features_by_collection": features_by_collection,
        "geometry_types": geometry_types,
        "assigned_elsewhere_count": assigned_elsewhere_count,
        "outside_scope_count": outside_scope_count,
        "clipped_to_scope_count": clipped_to_scope_count,
        "reference_truncated": False,
    }


def partition_filename(definition: LayerDefinition, member: ScopeMember, observed_date: date) -> str:
    return f"{member.nis_code}_{safe_slug(member.name)}_grb_{definition.key}_{observed_date.isoformat()}.geojson"


def combined_filename(definition: LayerDefinition, scope: GeographicScope, observed_date: date) -> str:
    return f"grb_{definition.key}_{scope.key.replace('-', '_')}_{observed_date.isoformat()}.geojson"


def write_partition(
    path: Path,
    *,
    definition: LayerDefinition,
    scope: GeographicScope,
    member: ScopeMember,
    features: list[dict[str, Any]],
    generated_at: str,
) -> None:
    write_json_atomic(
        path,
        {
            "type": "FeatureCollection",
            "name": f"GRB {definition.key} - {member.name} partition of {scope.display_name}",
            "crs": GEOJSON_CRS,
            "features": features,
            "source": f"Digitaal Vlaanderen GRB OGC API collections {', '.join(item.name for item in definition.collections)}",
            "source_urls": [GRB_COLLECTION_URL.format(collection=item.name) for item in definition.collections],
            "attribution": GRB_ATTRIBUTION,
            "coverage_scope": scope.key,
            "partition_municipality": member.name,
            "partition_nis_code": member.nis_code,
            "generated_at": generated_at,
        },
    )


def write_combined_artifact(
    path: Path,
    *,
    definition: LayerDefinition,
    scope: GeographicScope,
    observed_date: date,
    partition_paths: list[Path],
    expected_feature_count: int,
) -> dict[str, Any]:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(f"{path.suffix}.partial")
    header = {
        "type": "FeatureCollection",
        "name": f"GRB {definition.key} - {scope.display_name}",
        "crs": GEOJSON_CRS,
        "source": f"Digitaal Vlaanderen GRB OGC API collections {', '.join(item.name for item in definition.collections)}",
        "source_urls": [GRB_COLLECTION_URL.format(collection=item.name) for item in definition.collections],
        "attribution": GRB_ATTRIBUTION,
        "coverage_scope": scope.key,
        "scope_type": scope.scope_type,
        "partition_count": len(partition_paths),
        "partition_strategy": "municipality_bbox_maximum_same_dimension_intersection",
        "observed_at": observed_date.isoformat(),
    }
    encoded_header = json.dumps(header, ensure_ascii=False, separators=(",", ":"))
    seen_ids: set[str] = set()
    written = 0
    first = True
    with temporary.open("w", encoding="utf-8", newline="") as output:
        output.write(encoded_header[:-1])
        output.write(',"features":[')
        for partition_path in partition_paths:
            payload = json.loads(partition_path.read_text(encoding="utf-8"))
            if payload.get("type") != "FeatureCollection" or not isinstance(payload.get("features"), list):
                raise RuntimeError(f"Invalid partition artifact: {partition_path}")
            for feature in payload["features"]:
                feature_id = str(feature.get("id") or (feature.get("properties") or {}).get("source_feature_id") or "")
                if not feature_id:
                    raise RuntimeError(f"Partition feature without source identity in {partition_path.name}")
                if feature_id in seen_ids:
                    raise RuntimeError(f"Duplicate regional source feature {feature_id} in {partition_path.name}")
                seen_ids.add(feature_id)
                if not first:
                    output.write(",")
                output.write(json.dumps(feature, ensure_ascii=False, separators=(",", ":")))
                first = False
                written += 1
        output.write("]}")
    if written != expected_feature_count:
        temporary.unlink(missing_ok=True)
        raise RuntimeError(f"Expected {expected_feature_count} combined features, wrote {written}")
    temporary.replace(path)
    return {"feature_count": written, "size_bytes": path.stat().st_size, "sha256": sha256_file(path)}


def prepare_layer_artifacts(
    args: argparse.Namespace,
    scope: GeographicScope,
    definition: LayerDefinition,
) -> tuple[Path, list[Path], Path, dict[str, Any]]:
    observation_dir = args.output_root / scope.key / definition.key / args.observed_date.isoformat()
    partition_dir = observation_dir / "partitions"
    artifact_path = observation_dir / combined_filename(definition, scope, args.observed_date)
    manifest_path = observation_dir / f"regional_{definition.key}_manifest.json"
    partition_dir.mkdir(parents=True, exist_ok=True)

    if not args.force:
        existing = reusable_manifest(manifest_path, artifact_path, partition_dir, len(scope.members))
        if existing:
            paths = [partition_dir / summary["filename"] for summary in existing["partitions"]]
            return artifact_path, paths, manifest_path, existing

    existing_manifest: dict[str, Any] = {}
    if manifest_path.is_file() and not args.force:
        existing_manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    existing_by_nis = {
        str(item.get("nis_code")): item
        for item in existing_manifest.get("partitions") or []
        if isinstance(item, dict)
    }
    generated_at = utc_now()
    with source_session() as session:
        source_features, vrbg_source_url = fetch_scope_members(session, scope, args.request_timeout)
        members, regional_boundary = build_member_geometries(scope, source_features)
        summaries: list[dict[str, Any]] = []
        for member, boundary in members:
            path = partition_dir / partition_filename(definition, member, args.observed_date)
            reusable = existing_by_nis.get(member.nis_code)
            if (
                reusable
                and path.is_file()
                and reusable.get("filename") == path.name
                and reusable.get("sha256") == sha256_file(path)
            ):
                summaries.append(reusable)
                continue
            features, summary = build_partition_features(
                iter_layer_pages(
                    session,
                    definition,
                    boundary.bounds,
                    page_limit=args.page_limit,
                    timeout=args.request_timeout,
                ),
                definition=definition,
                member=member,
                members=members,
                regional_boundary=regional_boundary,
                scope=scope,
                max_features=args.max_features_per_member,
            )
            write_partition(
                path,
                definition=definition,
                scope=scope,
                member=member,
                features=features,
                generated_at=generated_at,
            )
            summary.update({"filename": path.name, "size_bytes": path.stat().st_size, "sha256": sha256_file(path)})
            summaries.append(summary)
            write_json_atomic(
                manifest_path,
                {
                    "schema_version": 1,
                    "status": "in_progress",
                    "scope": scope.key,
                    "theme": definition.key,
                    "observed_at": args.observed_date.isoformat(),
                    "generated_at": generated_at,
                    "vrbg_source_url": vrbg_source_url,
                    "grb_collections": [item.name for item in definition.collections],
                    "partitions": summaries,
                },
                pretty=True,
            )

    total_features = sum(int(summary["feature_count"]) for summary in summaries)
    if total_features > args.max_total_features:
        raise RuntimeError(
            f"Regional {definition.key} count {total_features} exceeds --max-total-features={args.max_total_features}"
        )
    partition_paths = [partition_dir / summary["filename"] for summary in summaries]
    artifact = write_combined_artifact(
        artifact_path,
        definition=definition,
        scope=scope,
        observed_date=args.observed_date,
        partition_paths=partition_paths,
        expected_feature_count=total_features,
    )
    aggregate_geometry_types: dict[str, int] = {}
    for summary in summaries:
        for geometry_type, count in (summary.get("geometry_types") or {}).items():
            aggregate_geometry_types[geometry_type] = aggregate_geometry_types.get(geometry_type, 0) + int(count)
    manifest = {
        "schema_version": 1,
        "status": "complete",
        "scope": scope.key,
        "scope_type": scope.scope_type,
        "scope_authority_url": scope.authority_url,
        "scope_limitation": scope.limitation_message,
        "theme": definition.key,
        "reference_layer_name": definition.reference_layer_name,
        "observed_at": args.observed_date.isoformat(),
        "generated_at": generated_at,
        "member_count": len(scope.members),
        "feature_count": total_features,
        "geometry_types": aggregate_geometry_types,
        "reference_truncated": False,
        "partition_strategy": "municipality_bbox_maximum_same_dimension_intersection",
        "partition_assignment_rule": "largest same-dimension intersection measure; NIS code resolves exact ties",
        "vrbg_source_url": vrbg_source_url,
        "grb_collections": [item.name for item in definition.collections],
        "grb_source_urls": [GRB_COLLECTION_URL.format(collection=item.name) for item in definition.collections],
        "artifact_filename": artifact_path.name,
        "artifact_size_bytes": artifact["size_bytes"],
        "artifact_sha256": artifact["sha256"],
        "bounds_json": {
            "min_x": float(regional_boundary.bounds[0]),
            "min_y": float(regional_boundary.bounds[1]),
            "max_x": float(regional_boundary.bounds[2]),
            "max_y": float(regional_boundary.bounds[3]),
        },
        "limitation_message": definition.limitation_message,
        "partitions": summaries,
        "attribution": GRB_ATTRIBUTION,
    }
    write_json_atomic(manifest_path, manifest, pretty=True)
    return artifact_path, partition_paths, manifest_path, manifest


def provision_dataset(
    args: argparse.Namespace,
    scope: GeographicScope,
    definition: LayerDefinition,
    artifact_path: Path,
    partition_paths: list[Path],
    manifest_path: Path,
    manifest: dict[str, Any],
) -> dict[str, Any]:
    parsed_base = urlparse(args.base_url)
    if parsed_base.hostname not in {"127.0.0.1", "localhost", "::1"}:
        raise RuntimeError("Partitioned service import must run inside the GeoIntel container against its local backend")
    with requests.Session() as session:
        projects = list_paginated_items(session, f"{args.base_url.rstrip('/')}/api/v1/projects", args.api_timeout)
        project = next((item for item in projects if item.get("name") == scope.project_name), None)
        if not project:
            raise RuntimeError("Regional scope project is missing; run provision_geographic_scope.py first")
        project_id = str(project["id"])
        areas = list_paginated_items(
            session,
            f"{args.base_url.rstrip('/')}/api/v1/projects/{project_id}/areas",
            args.api_timeout,
        )
        area = next((item for item in areas if item.get("name") == scope.area_name), None)
        if not area:
            raise RuntimeError("Regional scope Area is missing; run provision_geographic_scope.py first")
        datasets = list_paginated_items(
            session,
            f"{args.base_url.rstrip('/')}/api/v1/projects/{project_id}/datasets",
            args.api_timeout,
        )
    existing = next((item for item in datasets if item.get("original_filename") == artifact_path.name), None)
    if existing:
        persisted_checksum = (existing.get("provenance_metadata") or {}).get("artifact_sha256")
        if persisted_checksum and persisted_checksum != manifest["artifact_sha256"]:
            raise RuntimeError(
                f"Immutable dataset {artifact_path.name} checksum changed; use a new --observed-date for refreshed GRB data"
            )
        return {"dataset_id": str(existing["id"]), "feature_count": existing.get("feature_count"), "reused": True}

    ensure_backend_path()
    from app.db.session import SessionLocal
    from app.services.dataset_service import DatasetService

    metadata_json = {
        "feature_count": manifest["feature_count"],
        "feature_geometry_count": manifest["feature_count"],
        "geometry_types": list(manifest.get("geometry_types") or definition.geometry_types),
        "bounds_json": manifest["bounds_json"],
        "approximate_area_m2": None,
        "invalid_features": 0,
        "z_dimension_feature_count": 0,
        "canonical_storage_dimension": "2D",
        "crs": "EPSG:4326",
        "crs_assumed": False,
        "extracted_at": manifest["generated_at"],
    }
    source_metadata = {
        "provider": "Digitaal Vlaanderen",
        "collections": [item.name for item in definition.collections],
        "authority_level": "authoritative",
        "theme": definition.key,
        "layer_type": f"regional_{definition.key}",
        "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,
        "layer_limitation": definition.limitation_message,
        "member_count": len(scope.members),
        "member_nis_codes": list(scope.nis_codes),
        "geometry_clipped_to_area": True,
        "feature_count": manifest["feature_count"],
        "partition_count": len(partition_paths),
        "partition_strategy": manifest["partition_strategy"],
        "identity_stable": True,
        "identity_scheme": "grb_ogc_feature_id",
        "identity_prefixes": [
            f"{item.name}:{item.name}." for item in definition.collections
        ],
        "selection_aggregation": {
            "metric_key": definition.metric_key,
            "method": definition.metric_method,
            "label": definition.metric_label,
            "unit": definition.metric_unit,
            "is_estimate": False,
            "warning": definition.metric_warning,
        },
        "attribution": GRB_ATTRIBUTION,
    }
    provenance_metadata = {
        "operator_tool": "provision_regional_grb_context.py",
        "operator_explicit_fetch": True,
        "geometry_clipped_to_area": True,
        "manifest_path": str(manifest_path),
        "source_urls": manifest["grb_source_urls"],
        "artifact_sha256": manifest["artifact_sha256"],
        "artifact_size_bytes": manifest["artifact_size_bytes"],
        "partition_checksums": {summary["filename"]: summary["sha256"] for summary in manifest["partitions"]},
        "partition_assignment_rule": manifest["partition_assignment_rule"],
        "reference_truncated": False,
    }
    with SessionLocal() as db:
        dataset = DatasetService.import_partitioned_vector_artifact(
            db,
            project_id=UUID(project_id),
            area_id=UUID(str(area["id"])),
            artifact_path=artifact_path,
            partition_paths=partition_paths,
            original_filename=artifact_path.name,
            source="operator_official_import",
            dataset_role="reference",
            source_name="grb",
            reference_layer_name=definition.reference_layer_name,
            metadata_json=metadata_json,
            source_metadata=source_metadata,
            provenance_metadata=provenance_metadata,
            temporal_series_key=f"grb:{definition.key}:{scope.key}",
            observed_at=observed_at(args.observed_date),
            temporal_granularity="snapshot",
            source_version=args.observed_date.isoformat(),
            batch_size=args.batch_size,
        )
    if dataset.checksum_sha256 != manifest["artifact_sha256"]:
        raise RuntimeError("Managed dataset checksum differs from the retained regional artifact")
    return {"dataset_id": str(dataset.id), "feature_count": dataset.feature_count, "reused": False}


def main() -> int:
    args = parse_args()
    scope = GEOGRAPHIC_SCOPES[args.scope]
    try:
        definitions = selected_definitions(args.layers)
        if args.page_limit <= 0 or args.max_features_per_member <= 0 or args.max_total_features <= 0:
            raise ValueError("Page and feature limits must be positive")
        results: list[dict[str, Any]] = []
        for definition in definitions:
            artifact_path, partition_paths, manifest_path, manifest = prepare_layer_artifacts(args, scope, definition)
            persistence = None if args.fetch_only else provision_dataset(
                args,
                scope,
                definition,
                artifact_path,
                partition_paths,
                manifest_path,
                manifest,
            )
            results.append(
                {
                    "theme": definition.key,
                    "collections": [item.name for item in definition.collections],
                    "feature_count": manifest["feature_count"],
                    "artifact_size_bytes": manifest["artifact_size_bytes"],
                    "artifact_path": str(artifact_path),
                    "manifest_path": str(manifest_path),
                    "reference_truncated": manifest["reference_truncated"],
                    "persistence": persistence,
                }
            )
    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,
                "observed_at": args.observed_date.isoformat(),
                "member_count": len(scope.members),
                "layers": results,
            },
            ensure_ascii=False,
            indent=2,
        )
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())