geointel/scripts/provision_geographic_scope.py

"""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": getattr(args, "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())