#!/usr/bin/env python3 """Govern one bounded current-orthophoto release from plan through apply. The operator has four deliberately separate actions. Planning is read-only, staging retains source and normalized raster evidence without database writes, review records a named approval, and apply uploads the exact approved GeoTIFF through GeoIntel's canonical DatasetService route. No action refreshes data automatically. """ from __future__ import annotations import argparse from datetime import datetime, time, timezone from hashlib import sha256 import json import math import os from pathlib import Path import re import sys from typing import Any, Callable from urllib.error import HTTPError, URLError from urllib.parse import parse_qs, urlencode, urlparse from urllib.request import Request, urlopen import uuid import orthophoto_release_preflight as preflight DEFAULT_API_URL = "http://127.0.0.1:8000/api/v1" DEFAULT_SCOPE = preflight.DEFAULT_SCOPE DEFAULT_EVIDENCE_ROOT = Path("/app/storage/operator-evidence/orthophoto-release") SOURCE_NAME = preflight.SOURCE_NAME WMS_BASE_URL = f"https://{preflight.WMS_HOST}{preflight.WMS_PATH}" LAYER = "Ortho" OUTPUT_CRS = "EPSG:31370" OUTPUT_RESOLUTION_M = 1.0 MAX_RESPONSE_MB = 32 MAX_PREVIEW_SIDE = 640 SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") SAFE_EDITION_PATTERN = preflight.EDITION_PATTERN ACTIONABLE_STATUSES = {"not_loaded", "update_available"} def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Governed orthophoto release: plan, stage, review, then checksum-confirmed apply." ) parser.add_argument("action", choices=("plan", "stage", "review", "apply")) parser.add_argument("--project-id", required=True, help="GeoIntel project UUID for the governed scope") parser.add_argument("--scope", choices=(DEFAULT_SCOPE,), default=DEFAULT_SCOPE) parser.add_argument("--api-url", default=os.environ.get("GEOINTEL_API_URL", DEFAULT_API_URL)) parser.add_argument( "--bbox", nargs=4, type=float, metavar=("MIN_LON", "MIN_LAT", "MAX_LON", "MAX_LAT"), required=True, help="Exact bounded EPSG:4326 selection; projected sides must be 128-1024 metres", ) parser.add_argument("--area-id", help="Optional existing Area UUID to bind to the imported Dataset") parser.add_argument("--confirm-edition", help="Exact official edition, for example 2025.04") parser.add_argument("--confirm-plan-sha256", help="Exact staged-plan hash required for review/apply") parser.add_argument("--confirm-review-sha256", help="Exact review-evidence hash required for apply") parser.add_argument("--approve", action="store_true", help="Explicitly approve staged evidence during review") parser.add_argument("--reviewer", help="Named human/operator approving the staged evidence") parser.add_argument("--review-note", default="", help="Optional bounded review note") parser.add_argument( "--establish-official-baseline", action="store_true", help="Explicitly replace a non-comparable rolling marker with the first official edition baseline", ) parser.add_argument( "--confirm-local-version", help="Exact non-comparable local marker required with --establish-official-baseline", ) parser.add_argument("--plan-path", type=Path, help="Override the governed staged-plan path") parser.add_argument("--review-path", type=Path, help="Override the governed review-evidence path") parser.add_argument( "--evidence-root", type=Path, default=Path(os.environ.get("GEOINTEL_ORTHOPHOTO_RELEASE_EVIDENCE_ROOT", DEFAULT_EVIDENCE_ROOT)), ) parser.add_argument("--refresh-catalog", action="store_true", help="Bypass the short catalog cache") parser.add_argument("--api-timeout", type=int, default=180) parser.add_argument("--wms-timeout", type=int, default=60) parser.add_argument("--import-timeout", type=int, default=600) parser.add_argument("--max-response-mb", type=int, default=MAX_RESPONSE_MB) return parser.parse_args() def canonical_sha256(payload: dict[str, Any], hash_field: str) -> str: content = {key: value for key, value in payload.items() if key != hash_field} encoded = json.dumps(content, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") return sha256(encoded).hexdigest() def sha256_file(path: Path) -> str: digest = 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_bytes(path: Path, content: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".partial") temporary.write_bytes(content) temporary.replace(path) def write_json(path: Path, payload: dict[str, Any]) -> None: write_bytes( path, (json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode("utf-8"), ) def selection_key(bbox: list[float]) -> str: canonical = json.dumps([round(float(value), 8) for value in bbox], separators=(",", ":")).encode("ascii") return sha256(canonical).hexdigest()[:16] def evidence_directory(args: argparse.Namespace, edition: str) -> Path: return args.evidence_root / args.scope / edition / selection_key(list(args.bbox)) def default_plan_path(args: argparse.Namespace, edition: str) -> Path: return evidence_directory(args, edition) / "staged-plan.json" def default_review_path(args: argparse.Namespace, edition: str) -> Path: return evidence_directory(args, edition) / "review-evidence.json" def governed_path(args: argparse.Namespace, path: Path) -> Path: resolved = path.resolve() if not resolved.is_relative_to(args.evidence_root.resolve()): raise RuntimeError(f"Orthophoto release evidence is outside the governed root: {path}") return path def internal_api_url(api_url: str) -> str: parsed = urlparse(api_url.rstrip("/")) if ( parsed.scheme not in {"http", "https"} or parsed.hostname not in {"127.0.0.1", "localhost", "::1"} or parsed.username or parsed.password or parsed.fragment or parsed.path.rstrip("/") != "/api/v1" ): raise RuntimeError("Stage/apply must run inside GeoIntel against the local /api/v1 backend") return api_url.rstrip("/") def run_preflight(args: argparse.Namespace, *, refresh: bool) -> dict[str, Any]: namespace = argparse.Namespace( project_id=args.project_id, scope=args.scope, api_url=args.api_url, bbox=list(args.bbox), refresh_catalog=refresh, api_timeout=args.api_timeout, wms_timeout=args.wms_timeout, ) return preflight.run_preflight(namespace) def preflight_identity(report: dict[str, Any]) -> dict[str, Any]: release = report["release"] coverage = report["flight_day_coverage"] identity = { "project_id": report["project_id"], "scope": report["scope"], "product": report["product"], "remote_release": { key: release.get(key) for key in ( "remote_edition", "remote_year", "metadata_identifier", "metadata_url", "remote_title", "remote_modified_at", "remote_published_at", "capabilities_url", "capabilities_sha256", ) }, "capabilities": report["capabilities"], "coverage_domain": report["coverage_domain"], "selection": report["selection"], "flight_day_coverage": { key: coverage.get(key) for key in ( "mode", "sample_count", "grid_columns", "grid_rows", "covered_sample_count", "sample_coverage_ratio", "flight_dates", "flight_years", "feature_ids", "sample_evidence_sha256", "claim_boundary", ) }, "flight_year_matches_release": report["flight_year_matches_release"], } identity["preflight_identity_sha256"] = canonical_sha256(identity, "preflight_identity_sha256") return identity def require_edition(args: argparse.Namespace, report: dict[str, Any]) -> str: edition = str(report["release"]["remote_edition"]) if not SAFE_EDITION_PATTERN.fullmatch(edition) or args.confirm_edition != edition: raise RuntimeError(f"Explicit --confirm-edition {edition} is required") return edition def authorize_stage(args: argparse.Namespace, report: dict[str, Any]) -> dict[str, Any]: release = report["release"] status = release["status"] local_version = release.get("local_source_version") if report.get("staging_permitted") and status in ACTIONABLE_STATUSES: return {"mode": "normal_release", "local_source_version": local_version} if status == "blocked_local_version": if not args.establish_official_baseline or args.confirm_local_version != local_version: raise RuntimeError( "The first official baseline requires --establish-official-baseline and exact " f"--confirm-local-version {local_version}" ) if not report.get("flight_year_matches_release"): raise RuntimeError("The bounded flight-day evidence does not match the official edition year") return {"mode": "explicit_legacy_baseline_transition", "local_source_version": local_version} raise RuntimeError(f"Orthophoto release is not safely stageable: {status}") def map_request(report: dict[str, Any]) -> dict[str, Any]: selection = report["selection"] width = max(1, math.ceil(float(selection["width_m"]) / OUTPUT_RESOLUTION_M)) height = max(1, math.ceil(float(selection["height_m"]) / OUTPUT_RESOLUTION_M)) bbox = [float(value) for value in selection["bbox_epsg31370"]] params = { "SERVICE": "WMS", "VERSION": "1.3.0", "REQUEST": "GetMap", "LAYERS": LAYER, "STYLES": "", "FORMAT": "image/tiff", "CRS": OUTPUT_CRS, "BBOX": ",".join(f"{value:.3f}" for value in bbox), "WIDTH": str(width), "HEIGHT": str(height), } request_url = f"{WMS_BASE_URL}?{urlencode(params)}" request_identity = { "url": request_url, "layer": LAYER, "crs": OUTPUT_CRS, "bbox_epsg31370": bbox, "bbox_epsg4326": list(selection["bbox_epsg4326"]), "width": width, "height": height, "resolution_m": OUTPUT_RESOLUTION_M, } request_identity["request_sha256"] = canonical_sha256(request_identity, "request_sha256") return request_identity def validate_map_url(url: str, expected: dict[str, Any]) -> None: parsed = urlparse(url) if ( parsed.scheme.lower() != "https" or parsed.hostname != preflight.WMS_HOST or parsed.port not in (None, 443) or parsed.username or parsed.password or parsed.fragment or parsed.path.rstrip("/").lower() != preflight.WMS_PATH.lower() ): raise RuntimeError("Orthophoto GetMap URL is outside the official allowlist") query = {key.upper(): values for key, values in parse_qs(parsed.query, keep_blank_values=True).items()} required = { "SERVICE": "WMS", "VERSION": "1.3.0", "REQUEST": "GetMap", "LAYERS": LAYER, "STYLES": "", "FORMAT": "image/tiff", "CRS": OUTPUT_CRS, "BBOX": ",".join(f"{value:.3f}" for value in expected["bbox_epsg31370"]), "WIDTH": str(expected["width"]), "HEIGHT": str(expected["height"]), } if set(query) != set(required) or any(query.get(key) != [value] for key, value in required.items()): raise RuntimeError("Orthophoto GetMap URL does not match the approved bounded request") def fetch_map( request_identity: dict[str, Any], *, timeout: int, max_bytes: int, opener: Callable[..., Any] | None = None, ) -> tuple[bytes, str, str]: validate_map_url(request_identity["url"], request_identity) request = Request( request_identity["url"], headers={"Accept": "image/tiff", "User-Agent": "GeoIntel-orthophoto-release/1.0"}, ) fetch = opener or urlopen try: response = fetch(request, timeout=timeout) with response: final_url = response.geturl() validate_map_url(final_url, request_identity) content_type = str(response.headers.get("Content-Type") or "").split(";", 1)[0].strip().lower() declared = response.headers.get("Content-Length") if declared: try: if int(declared) > max_bytes: raise RuntimeError("Official orthophoto response exceeds the configured release limit") except ValueError as exc: raise RuntimeError("Official orthophoto response has an invalid Content-Length") from exc content = response.read(max_bytes + 1) except RuntimeError: raise except HTTPError as exc: raise RuntimeError(f"Official orthophoto WMS returned HTTP {exc.code}") from exc except URLError as exc: raise RuntimeError(f"Official orthophoto WMS is unreachable: {exc.reason}") from exc if len(content) > max_bytes: raise RuntimeError("Official orthophoto response exceeds the configured release limit") if content_type not in {"image/tiff", "image/geotiff", "image/x-geotiff"}: raise RuntimeError(f"Official orthophoto WMS returned unsupported content type {content_type or 'unknown'}") return content, content_type, final_url def normalize_and_preview( raw_content: bytes, request_identity: dict[str, Any], geotiff_path: Path, preview_path: Path, ) -> dict[str, Any]: try: import numpy as np from PIL import Image import rasterio from rasterio.enums import Resampling from rasterio.errors import NotGeoreferencedWarning from rasterio.io import MemoryFile from rasterio.transform import from_bounds import warnings except ImportError as exc: raise RuntimeError("Rasterio, NumPy and Pillow are required for orthophoto release staging") from exc geotiff_path.parent.mkdir(parents=True, exist_ok=True) temporary = geotiff_path.with_suffix(geotiff_path.suffix + ".partial") try: with MemoryFile(raw_content) as source_memory: with warnings.catch_warnings(): warnings.simplefilter("ignore", NotGeoreferencedWarning) with source_memory.open() as source: if ( source.width != request_identity["width"] or source.height != request_identity["height"] or source.count < 3 or any(dtype != "uint8" for dtype in source.dtypes[:3]) ): raise RuntimeError("Official orthophoto response does not match the approved RGB dimensions") rgb = source.read((1, 2, 3)) profile = source.profile.copy() profile.update( driver="GTiff", count=3, dtype="uint8", crs=OUTPUT_CRS, transform=from_bounds( *request_identity["bbox_epsg31370"], source.width, source.height, ), nodata=None, compress="deflate", tiled=False, ) for key in ("blockxsize", "blockysize", "photometric", "interleave"): profile.pop(key, None) with rasterio.open(temporary, "w", **profile) as output: output.write(rgb) output.update_tags( source=f"Digitaal Vlaanderen WMS {LAYER}", source_url=request_identity["url"], attribution="Digitaal Vlaanderen", acquisition="governed_orthophoto_release", ) temporary.replace(geotiff_path) with rasterio.open(geotiff_path) as source: scale = min(1.0, MAX_PREVIEW_SIDE / max(source.width, source.height)) preview_width = max(1, round(source.width * scale)) preview_height = max(1, round(source.height * scale)) preview = source.read( (1, 2, 3), out_shape=(3, preview_height, preview_width), resampling=Resampling.bilinear, ) image = Image.fromarray(np.moveaxis(preview, 0, 2)) preview_temporary = preview_path.with_suffix(preview_path.suffix + ".partial") image.save(preview_temporary, format="PNG", optimize=True) preview_temporary.replace(preview_path) return { "crs": source.crs.to_string() if source.crs else None, "width": source.width, "height": source.height, "band_count": source.count, "dtypes": list(source.dtypes), "bounds": [float(value) for value in source.bounds], "transform": [float(value) for value in source.transform[:6]], "preview_width": preview_width, "preview_height": preview_height, } except RuntimeError: temporary.unlink(missing_ok=True) raise except Exception as exc: temporary.unlink(missing_ok=True) raise RuntimeError(f"Official orthophoto response could not be normalized: {exc}") from exc def stage_artifacts( args: argparse.Namespace, report: dict[str, Any], request_identity: dict[str, Any], *, opener: Callable[..., Any] | None = None, ) -> dict[str, Any]: edition = str(report["release"]["remote_edition"]) directory = governed_path(args, evidence_directory(args, edition)) raw_path = directory / "official-wms-response.tif" geotiff_path = directory / f"orthophoto_{edition}_{selection_key(list(args.bbox))}.tif" preview_path = directory / "review-preview.png" manifest_path = directory / "staged-manifest.json" if directory.exists() and any(directory.iterdir()): raise RuntimeError( f"Governed orthophoto evidence already exists and will not be overwritten: {directory}" ) max_bytes = args.max_response_mb * 1024 * 1024 content, content_type, final_url = fetch_map( request_identity, timeout=args.wms_timeout, max_bytes=max_bytes, opener=opener, ) write_bytes(raw_path, content) raster = normalize_and_preview(content, request_identity, geotiff_path, preview_path) payload: dict[str, Any] = { "schema_version": 1, "status": "staged", "created_at": datetime.now(timezone.utc).isoformat(), "project_id": args.project_id, "scope": args.scope, "edition": edition, "preflight_identity_sha256": preflight_identity(report)["preflight_identity_sha256"], "request": request_identity, "response": { "final_url": final_url, "content_type": content_type, "path": str(raw_path), "size_bytes": len(content), "sha256": sha256(content).hexdigest(), }, "normalized_geotiff": { "path": str(geotiff_path), "size_bytes": geotiff_path.stat().st_size, "sha256": sha256_file(geotiff_path), **raster, }, "review_preview": { "path": str(preview_path), "size_bytes": preview_path.stat().st_size, "sha256": sha256_file(preview_path), "width": raster["preview_width"], "height": raster["preview_height"], }, "pixel_request_count": 1, "datasets_mutated": 0, } payload["manifest_sha256"] = canonical_sha256(payload, "manifest_sha256") write_json(manifest_path, payload) return {"manifest_path": str(manifest_path), **payload} def validate_staged_artifacts(args: argparse.Namespace, manifest_path: Path) -> dict[str, Any]: governed_path(args, manifest_path) if not manifest_path.is_file(): raise RuntimeError(f"Staged orthophoto manifest is missing: {manifest_path}") manifest = json.loads(manifest_path.read_text(encoding="utf-8")) if manifest.get("manifest_sha256") != canonical_sha256(manifest, "manifest_sha256"): raise RuntimeError("Staged orthophoto manifest checksum is invalid") if ( manifest.get("status") != "staged" or manifest.get("project_id") != args.project_id or manifest.get("scope") != args.scope ): raise RuntimeError("Staged orthophoto manifest identity is invalid") for section, expected_content_type in ( ("response", None), ("normalized_geotiff", None), ("review_preview", None), ): evidence = manifest.get(section) if not isinstance(evidence, dict) or not isinstance(evidence.get("path"), str): raise RuntimeError(f"Staged orthophoto {section} evidence is incomplete") path = governed_path(args, Path(evidence["path"])) if ( not path.is_file() or path.stat().st_size != evidence.get("size_bytes") or sha256_file(path) != evidence.get("sha256") ): raise RuntimeError(f"Staged orthophoto {section} bytes no longer match their evidence") raster = manifest["normalized_geotiff"] request_identity = manifest["request"] if ( manifest.get("pixel_request_count") != 1 or manifest.get("datasets_mutated") != 0 or raster.get("crs") != OUTPUT_CRS or raster.get("width") != request_identity.get("width") or raster.get("height") != request_identity.get("height") or raster.get("band_count") != 3 or raster.get("dtypes") != ["uint8", "uint8", "uint8"] ): raise RuntimeError("Staged orthophoto raster no longer satisfies the release contract") validate_map_url(request_identity["url"], request_identity) return manifest def build_staged_plan( args: argparse.Namespace, report: dict[str, Any], authorization: dict[str, Any], staged: dict[str, Any], ) -> dict[str, Any]: identity = preflight_identity(report) payload: dict[str, Any] = { "schema_version": 1, "status": "staged", "created_at": datetime.now(timezone.utc).isoformat(), "project_id": args.project_id, "scope": args.scope, "area_id": args.area_id, "edition": report["release"]["remote_edition"], "local_source_version_before_apply": report["release"].get("local_source_version"), "baseline_authorization": authorization, "preflight_identity": identity, "catalog_checked_at": report["release"].get("catalog_checked_at"), "staged_manifest_path": staged["manifest_path"], "staged_manifest_sha256": staged["manifest_sha256"], "review_required": True, "apply_requires_plan_sha256": True, "apply_requires_review_sha256": True, "automatic_stage": False, "automatic_import": False, "destructive_replacement": False, "existing_snapshots_retained": True, } payload["plan_sha256"] = canonical_sha256(payload, "plan_sha256") return payload def load_staged_plan(args: argparse.Namespace, edition: str) -> tuple[Path, dict[str, Any], dict[str, Any]]: path = governed_path(args, args.plan_path or default_plan_path(args, edition)) if not path.is_file(): raise RuntimeError(f"Staged orthophoto plan is missing: {path}") payload = json.loads(path.read_text(encoding="utf-8")) actual_sha = canonical_sha256(payload, "plan_sha256") if payload.get("plan_sha256") != actual_sha: raise RuntimeError("Staged orthophoto plan checksum is invalid") if args.confirm_plan_sha256 != actual_sha: raise RuntimeError(f"Explicit --confirm-plan-sha256 {actual_sha} is required") if ( payload.get("status") != "staged" or payload.get("project_id") != args.project_id or payload.get("scope") != args.scope or payload.get("area_id") != args.area_id or payload.get("edition") != edition ): raise RuntimeError("Staged orthophoto plan identity is invalid") manifest_path = Path(str(payload.get("staged_manifest_path") or "")) manifest = validate_staged_artifacts(args, manifest_path) if manifest.get("manifest_sha256") != payload.get("staged_manifest_sha256"): raise RuntimeError("Staged orthophoto manifest no longer matches the plan") if manifest.get("preflight_identity_sha256") != payload["preflight_identity"].get("preflight_identity_sha256"): raise RuntimeError("Staged orthophoto preflight identity no longer matches the plan") return path, payload, manifest def require_preflight_unchanged(plan: dict[str, Any], report: dict[str, Any], *, require_local: bool) -> None: if plan.get("preflight_identity") != preflight_identity(report): raise RuntimeError("Official orthophoto evidence changed; create and review a new staged plan") if require_local and report["release"].get("local_source_version") != plan.get("local_source_version_before_apply"): raise RuntimeError("Local orthophoto edition changed after staging; create and review a new plan") def build_review_evidence( args: argparse.Namespace, plan_path: Path, plan: dict[str, Any], manifest: dict[str, Any], ) -> dict[str, Any]: reviewer = str(args.reviewer or "").strip() note = str(args.review_note or "").strip() if not args.approve or len(reviewer) < 2 or len(reviewer) > 120: raise RuntimeError("Review requires --approve and a named --reviewer between 2 and 120 characters") if len(note) > 1000: raise RuntimeError("Review note must not exceed 1000 characters") payload: dict[str, Any] = { "schema_version": 1, "status": "approved", "reviewed_at": datetime.now(timezone.utc).isoformat(), "reviewer": reviewer, "review_note": note or None, "project_id": args.project_id, "scope": args.scope, "edition": plan["edition"], "staged_plan_path": str(plan_path), "staged_plan_sha256": plan["plan_sha256"], "staged_manifest_sha256": manifest["manifest_sha256"], "review_preview_path": manifest["review_preview"]["path"], "review_preview_sha256": manifest["review_preview"]["sha256"], "reviewed_checks": [ "official_catalog_and_service_identity", "wcs_native_domain_and_rgb_contract", "bounded_flight_day_evidence", "exact_getmap_request_and_source_checksum", "normalized_epsg31370_rgb_geotiff", "visual_preview", "temporal_and_spatial_provenance", "immutable_dataset_apply", ], } payload["review_sha256"] = canonical_sha256(payload, "review_sha256") return payload def load_review_evidence( args: argparse.Namespace, edition: str, plan: dict[str, Any], manifest: dict[str, Any], ) -> tuple[Path, dict[str, Any]]: path = governed_path(args, args.review_path or default_review_path(args, edition)) if not path.is_file(): raise RuntimeError(f"Approved orthophoto review evidence is missing: {path}") payload = json.loads(path.read_text(encoding="utf-8")) actual_sha = canonical_sha256(payload, "review_sha256") if payload.get("review_sha256") != actual_sha: raise RuntimeError("Orthophoto review evidence checksum is invalid") if args.confirm_review_sha256 != actual_sha: raise RuntimeError(f"Explicit --confirm-review-sha256 {actual_sha} is required") if ( payload.get("status") != "approved" or payload.get("project_id") != args.project_id or payload.get("scope") != args.scope or payload.get("edition") != edition or payload.get("staged_plan_sha256") != plan.get("plan_sha256") or payload.get("staged_manifest_sha256") != manifest.get("manifest_sha256") or payload.get("review_preview_sha256") != manifest["review_preview"].get("sha256") or not str(payload.get("reviewer") or "").strip() ): raise RuntimeError("Orthophoto review evidence does not authorize this staged plan") return path, payload def parse_flight_dates(values: list[str]) -> list[datetime]: parsed: list[datetime] = [] for value in values: candidate = str(value).strip() result = None for date_format in ("%d/%m/%Y", "%Y-%m-%d", "%d-%m-%Y"): try: result = datetime.strptime(candidate, date_format).date() break except ValueError: continue if result is None: raise RuntimeError(f"Official flight date is not safely parseable: {candidate}") parsed.append(datetime.combine(result, time.min, tzinfo=timezone.utc)) if not parsed: raise RuntimeError("Official flight-day evidence has no dates") return sorted(set(parsed)) def api_data(api_url: str, path: str, timeout: int) -> dict[str, Any]: endpoint = f"{api_url.rstrip('/')}/{path.lstrip('/')}" request = Request(endpoint, headers={"Accept": "application/json", "User-Agent": "GeoIntel-orthophoto-release/1.0"}) try: with urlopen(request, timeout=timeout) as response: payload = json.load(response) except HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") raise RuntimeError(f"GeoIntel API returned HTTP {exc.code}: {body[-1000:]}") from exc except URLError as exc: raise RuntimeError(f"GeoIntel API is unreachable: {exc.reason}") from exc if not isinstance(payload, dict) or not isinstance(payload.get("data"), dict): raise RuntimeError("GeoIntel API response is not a canonical data envelope") return payload["data"] def project_datasets(args: argparse.Namespace) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] offset = 0 total: int | None = None while total is None or offset < total: page = api_data( args.api_url, f"projects/{args.project_id}/datasets?limit=200&offset={offset}", args.api_timeout, ) page_items = list(page.get("items") or []) page_total = int(page.get("total") or 0) if total is None: total = page_total elif total != page_total: raise RuntimeError("Dataset pagination total changed while checking orthophoto idempotence") items.extend(page_items) if not page_items: break offset += len(page_items) if total is not None and len(items) != total: raise RuntimeError(f"Dataset pagination returned {len(items)} of {total} items") return items def matching_dataset(args: argparse.Namespace, plan: dict[str, Any], manifest: dict[str, Any]) -> dict[str, Any] | None: matches = [] for dataset in project_datasets(args): provenance = dataset.get("provenance_metadata") or {} if ( dataset.get("source_name") == SOURCE_NAME and dataset.get("source_version") == plan["edition"] and dataset.get("status") == "ready" and provenance.get("release_plan_sha256") == plan["plan_sha256"] and provenance.get("normalized_geotiff_sha256") == manifest["normalized_geotiff"]["sha256"] ): matches.append(dataset) if len(matches) > 1: raise RuntimeError("More than one Dataset matches the exact approved orthophoto release") return matches[0] if matches else None def multipart_body(fields: dict[str, str], filename: str, content: bytes) -> tuple[bytes, str]: boundary = f"geointel-{uuid.uuid4().hex}" chunks: list[bytes] = [] for name, value in fields.items(): chunks.extend( ( f"--{boundary}\r\n".encode("ascii"), f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode("ascii"), value.encode("utf-8"), b"\r\n", ) ) safe_filename = filename.replace('"', "_").replace("\r", "_").replace("\n", "_") chunks.extend( ( f"--{boundary}\r\n".encode("ascii"), f'Content-Disposition: form-data; name="file"; filename="{safe_filename}"\r\n'.encode("ascii"), b"Content-Type: image/tiff\r\n\r\n", content, b"\r\n", f"--{boundary}--\r\n".encode("ascii"), ) ) return b"".join(chunks), boundary def upload_approved_dataset( args: argparse.Namespace, plan: dict[str, Any], manifest: dict[str, Any], review: dict[str, Any], ) -> dict[str, Any]: internal_api_url(args.api_url) geotiff_path = Path(manifest["normalized_geotiff"]["path"]) dates = parse_flight_dates(plan["preflight_identity"]["flight_day_coverage"]["flight_dates"]) observed_at = dates[-1].isoformat() source_metadata = { "provider": SOURCE_NAME, "service": "WMS", "service_version": "1.3.0", "product_key": preflight.PRODUCT_KEY, "official_edition": plan["edition"], "metadata_identifier": preflight.METADATA_IDENTIFIER, "metadata_url": plan["preflight_identity"]["remote_release"]["metadata_url"], "catalog_url": preflight.CATALOG_URL, "layer": LAYER, "native_crs": OUTPUT_CRS, "native_resolution_m": plan["preflight_identity"]["coverage_domain"]["native_resolution_m"], "requested_resolution_m": OUTPUT_RESOLUTION_M, "band_count": 3, "color_mode": "rgb", "authority_level": "authoritative", "flight_dates": [value.isoformat() for value in dates], "flight_date_evidence_mode": "official_queryable_flight_day_grid", "attribution": "Digitaal Vlaanderen", "license_note": "Gebruik volgens het gebruiksrecht geografische webdiensten van Digitaal Vlaanderen.", "limitation_message": plan["preflight_identity"]["flight_day_coverage"]["claim_boundary"], } provenance_metadata = { "operator_tool": "manage_orthophoto_release.py", "operator_explicit_apply": True, "release_plan_sha256": plan["plan_sha256"], "release_plan_path": str(args.plan_path or default_plan_path(args, plan["edition"])), "review_sha256": review["review_sha256"], "review_evidence_path": str(args.review_path or default_review_path(args, plan["edition"])), "reviewer": review["reviewer"], "staged_manifest_sha256": manifest["manifest_sha256"], "source_response_sha256": manifest["response"]["sha256"], "normalized_geotiff_sha256": manifest["normalized_geotiff"]["sha256"], "review_preview_sha256": manifest["review_preview"]["sha256"], "preflight_identity_sha256": plan["preflight_identity"]["preflight_identity_sha256"], "request": manifest["request"], "response_content_type": manifest["response"]["content_type"], "source_response_path": manifest["response"]["path"], "staged_manifest_path": plan["staged_manifest_path"], "review_preview_path": manifest["review_preview"]["path"], "baseline_authorization": plan["baseline_authorization"], "existing_snapshots_retained": True, } fields = { "dataset_type": "raster", "source": "operator_official_import", "dataset_role": "source", "source_name": SOURCE_NAME, "source_metadata_json": json.dumps(source_metadata, ensure_ascii=False), "provenance_metadata_json": json.dumps(provenance_metadata, ensure_ascii=False), "temporal_series_key": f"digitaal-vlaanderen:orthophoto:{selection_key(list(args.bbox))}", "observed_at": observed_at, "valid_from": dates[0].isoformat(), "valid_to": dates[-1].isoformat(), "temporal_granularity": "snapshot", "source_version": plan["edition"], } if args.area_id: fields["area_id"] = args.area_id body, boundary = multipart_body(fields, geotiff_path.name, geotiff_path.read_bytes()) request = Request( f"{args.api_url}/projects/{args.project_id}/datasets/upload", data=body, method="POST", headers={ "Accept": "application/json", "Content-Type": f"multipart/form-data; boundary={boundary}", "User-Agent": "GeoIntel-orthophoto-release/1.0", }, ) try: with urlopen(request, timeout=args.import_timeout) as response: payload = json.load(response) except HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") raise RuntimeError(f"GeoIntel orthophoto upload returned HTTP {exc.code}: {detail[-1500:]}") from exc except URLError as exc: raise RuntimeError(f"GeoIntel orthophoto upload is unreachable: {exc.reason}") from exc if not isinstance(payload, dict) or not isinstance(payload.get("data"), dict): raise RuntimeError("GeoIntel orthophoto upload did not return a canonical data envelope") dataset = payload["data"] if ( dataset.get("status") != "ready" or dataset.get("source_name") != SOURCE_NAME or dataset.get("source_version") != plan["edition"] or dataset.get("checksum_sha256") != manifest["normalized_geotiff"]["sha256"] ): raise RuntimeError("GeoIntel did not persist the exact approved orthophoto Dataset") return dataset def applied_evidence( args: argparse.Namespace, plan_path: Path, plan: dict[str, Any], review_path: Path, review: dict[str, Any], dataset: dict[str, Any], final_report: dict[str, Any], *, reused: bool, ) -> tuple[Path, dict[str, Any]]: path = plan_path.with_name("applied-evidence.json") if path.is_file(): existing = json.loads(path.read_text(encoding="utf-8")) if existing.get("applied_evidence_sha256") != canonical_sha256(existing, "applied_evidence_sha256"): raise RuntimeError("Existing orthophoto applied evidence checksum is invalid") if ( existing.get("project_id") != args.project_id or existing.get("scope") != args.scope or existing.get("edition") != plan["edition"] or existing.get("staged_plan_sha256") != plan["plan_sha256"] or existing.get("review_sha256") != review["review_sha256"] or existing.get("dataset_id") != dataset["id"] or existing.get("dataset_checksum_sha256") != dataset["checksum_sha256"] ): raise RuntimeError("Existing orthophoto applied evidence does not match this approved Dataset") return path, existing payload: dict[str, Any] = { "schema_version": 1, "status": "applied", "applied_at": datetime.now(timezone.utc).isoformat(), "project_id": args.project_id, "scope": args.scope, "edition": plan["edition"], "staged_plan_path": str(plan_path), "staged_plan_sha256": plan["plan_sha256"], "review_path": str(review_path), "review_sha256": review["review_sha256"], "reviewer": review["reviewer"], "dataset_id": dataset["id"], "dataset_checksum_sha256": dataset["checksum_sha256"], "dataset_status": "existing" if reused else "imported", "final_local_source_version": final_report["release"].get("local_source_version"), "final_comparison_status": final_report["release"].get("comparison_status"), "existing_snapshots_retained": True, } payload["applied_evidence_sha256"] = canonical_sha256(payload, "applied_evidence_sha256") write_json(path, payload) return path, payload def main() -> int: args = parse_args() try: if min(args.api_timeout, args.wms_timeout, args.import_timeout, args.max_response_mb) <= 0: raise ValueError("All timeout and response limits must be positive") if args.max_response_mb > MAX_RESPONSE_MB: raise ValueError(f"--max-response-mb may not exceed {MAX_RESPONSE_MB}") if args.action in {"stage", "apply"}: internal_api_url(args.api_url) report = run_preflight(args, refresh=args.refresh_catalog or args.action != "plan") if args.action == "plan": output = { "status": "ok", "action": "plan", "preflight": report, "preflight_identity": preflight_identity(report), "baseline_transition_required": report["release"]["status"] == "blocked_local_version", "automatic_stage": False, "automatic_import": False, } print(json.dumps(output, ensure_ascii=False, indent=2)) return 0 edition = require_edition(args, report) if args.action == "stage": authorization = authorize_stage(args, report) plan_path = governed_path(args, args.plan_path or default_plan_path(args, edition)) if plan_path.exists(): raise RuntimeError(f"Governed orthophoto plan already exists and will not be overwritten: {plan_path}") request_identity = map_request(report) staged = stage_artifacts(args, report, request_identity) plan = build_staged_plan(args, report, authorization, staged) write_json(plan_path, plan) print( json.dumps( {"status": "staged", "plan_path": str(plan_path), "preview_path": staged["review_preview"]["path"], **plan}, ensure_ascii=False, indent=2, ) ) return 0 plan_path, plan, manifest = load_staged_plan(args, edition) require_preflight_unchanged(plan, report, require_local=args.action == "review") if args.action == "review": review = build_review_evidence(args, plan_path, plan, manifest) review_path = governed_path(args, args.review_path or default_review_path(args, edition)) if review_path.exists(): raise RuntimeError( f"Governed orthophoto review already exists and will not be overwritten: {review_path}" ) write_json(review_path, review) print(json.dumps({"status": "approved", "review_path": str(review_path), **review}, ensure_ascii=False, indent=2)) return 0 review_path, review = load_review_evidence(args, edition, plan, manifest) existing = matching_dataset(args, plan, manifest) if existing is None: if report["release"].get("local_source_version") != plan.get("local_source_version_before_apply"): raise RuntimeError("Local orthophoto edition changed after review; create and review a new plan") dataset = upload_approved_dataset(args, plan, manifest, review) reused = False else: dataset = existing reused = True final_report = run_preflight(args, refresh=True) require_preflight_unchanged(plan, final_report, require_local=False) if ( final_report["release"].get("status") != "current" or final_report["release"].get("local_source_version") != edition or final_report["release"].get("comparison_status") != "same" ): raise RuntimeError("Applied orthophoto Dataset did not become the current official local edition") evidence_path, evidence = applied_evidence( args, plan_path, plan, review_path, review, dataset, final_report, reused=reused, ) print(json.dumps({"status": "applied", "evidence_path": str(evidence_path), **evidence}, ensure_ascii=False, indent=2)) return 0 except (OSError, RuntimeError, ValueError, KeyError, json.JSONDecodeError) as exc: print(json.dumps({"status": "error", "action": args.action, "message": str(exc)}, ensure_ascii=False), file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())