#!/usr/bin/env python3 """Plan, stage, review and explicitly apply one definitive ALZ edition. Planning reads only the canonical source-catalog report. Staging downloads and prepares one official v3 archive without database persistence. Review records a named approval. Apply requires the exact plan and review hashes and reuses the canonical agricultural DatasetService upload path. """ from __future__ import annotations import argparse from datetime import datetime, timezone from hashlib import sha256 import json import os from pathlib import Path import re import subprocess import sys from typing import Any from urllib.error import HTTPError, URLError from urllib.parse import urlparse from urllib.request import Request, urlopen from geographic_scopes import GEOGRAPHIC_SCOPES from provision_agricultural_parcel_history import ( AgriculturalReleaseConfig, MAX_ARCHIVE_BYTES, OUTPUT_CRS, SOURCE_CRS, STABLE_REQUIRED_FIELDS, artifact_paths, resolve_release_config, reusable_artifact, sha256_file, ) DEFAULT_API_URL = "http://127.0.0.1:8000/api/v1" DEFAULT_SCOPE = "kempen-transport-region" DEFAULT_OUTPUT_ROOT = Path("/app/storage/operator-evidence/agricultural-use-parcels") DEFAULT_EVIDENCE_ROOT = Path("/app/storage/operator-evidence/alz-agriculture-refresh") SOURCE_NAME = "agentschap_landbouw_zeevisserij_agricultural_parcels" REMOTE_VERSION_PATTERN = re.compile(r"^(20[0-9]{2})-v3$") LOCAL_VERSION_PATTERN = re.compile(r"^(20[0-9]{2})-definitive$") SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") PROVISIONAL_VERSION_PATTERN = re.compile(r"\b(20[0-9]{2}-v[12])\b") ACTIONABLE_STATUSES = {"update_available", "not_loaded"} def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Governed definitive ALZ 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("--confirm-edition", help="Exact official definitive edition, for example 2026-v3") 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 the 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("--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( "--output-root", type=Path, default=Path(os.environ.get("GEOINTEL_AGRICULTURE_OUTPUT_ROOT", DEFAULT_OUTPUT_ROOT)), ) parser.add_argument( "--evidence-root", type=Path, default=Path(os.environ.get("GEOINTEL_ALZ_REFRESH_EVIDENCE_ROOT", DEFAULT_EVIDENCE_ROOT)), ) parser.add_argument("--refresh-catalog", action="store_true", help="Bypass the short official-catalog cache") parser.add_argument("--request-timeout", type=int, default=900) parser.add_argument("--api-timeout", type=int, default=180) parser.add_argument("--import-timeout", type=int, default=3600) parser.add_argument("--max-features", type=int, default=250_000) parser.add_argument("--max-archive-mb", type=int, default=250) return parser.parse_args() 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-ALZ-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 validate_project_scope(args: argparse.Namespace) -> None: project = api_data(args.api_url, f"projects/{args.project_id}", args.api_timeout) expected_name = GEOGRAPHIC_SCOPES[args.scope].project_name if project.get("name") != expected_name: raise RuntimeError( f"Project {args.project_id} is '{project.get('name')}', but scope {args.scope} requires '{expected_name}'" ) def parse_catalog_datetime(value: Any) -> datetime: if not isinstance(value, str) or not value.strip(): raise RuntimeError("The official ALZ catalog did not provide a publication date") try: parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00")) except ValueError as exc: raise RuntimeError("The official ALZ catalog publication date is invalid") from exc if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc) def release_from_catalog_item(item: dict[str, Any]) -> AgriculturalReleaseConfig: match = REMOTE_VERSION_PATTERN.fullmatch(str(item.get("remote_version") or "")) if not match: raise RuntimeError("The official ALZ catalog did not provide one definitive YYYY-v3 edition") year = int(match.group(1)) published_at = parse_catalog_datetime(item.get("remote_published_at")) archive_url = ( "https://www.landbouwvlaanderen.be/bestanden/gis/" f"agpa_{year}_{published_at.date().isoformat()}_public.zip" ) return resolve_release_config(year, archive_url=archive_url) def fetch_release_decision_from_item(args: argparse.Namespace, item: dict[str, Any]) -> dict[str, Any]: allowed_degraded = item.get("error_code") == "CATALOG_ALZ_CURRENT_SNAPSHOT_MISSING" if ( item.get("source_name") != SOURCE_NAME or item.get("reachable") is not True or item.get("status") not in {"available", "degraded"} or (item.get("status") == "degraded" and not allowed_degraded) or "definitive_archive" not in set(item.get("matched_layers") or []) or not SHA256_PATTERN.fullmatch(str(item.get("capabilities_sha256") or "")) or item.get("metadata_identifier") != "alz-agricultural-use-parcels" ): raise RuntimeError("The official ALZ catalog is not safely available for release planning") release = release_from_catalog_item(item) local_version = str(item.get("local_source_version") or "") local_match = LOCAL_VERSION_PATTERN.fullmatch(local_version) if local_version and not local_match: status = "blocked_local_version" elif not local_version: status = "not_loaded" elif int(local_match.group(1)) == release.year: status = "current" elif int(local_match.group(1)) < release.year: status = "update_available" else: status = "blocked_remote_older" message = str(item.get("message") or "") provisional_match = PROVISIONAL_VERSION_PATTERN.search(message) return { "schema_version": 1, "status": status, "project_id": args.project_id, "scope": args.scope, "generated_at": datetime.now(timezone.utc).isoformat(), "catalog_checked_at": item.get("checked_at"), "local_source_version": local_version or None, "release": { "year": release.year, "edition": release.definitive_version, "archive_url": release.archive_url, }, "catalog_identity": { "metadata_identifier": item["metadata_identifier"], "metadata_url": item.get("metadata_url"), "remote_version": release.definitive_version, "remote_published_at": item.get("remote_published_at"), "capabilities_sha256": item["capabilities_sha256"], }, "provisional_release": provisional_match.group(1) if provisional_match else None, "provisional_release_importable": False, "automatic_download": False, "automatic_import": False, "destructive_replacement": False, "next_action": "stage" if status in ACTIONABLE_STATUSES else None, } def fetch_release_decision(args: argparse.Namespace, *, refresh: bool) -> dict[str, Any]: query = "true" if refresh else "false" report = api_data( args.api_url, f"projects/{args.project_id}/datasets/source-catalog-probes?refresh={query}", args.api_timeout, ) matches = [item for item in report.get("items") or [] if item.get("source_name") == SOURCE_NAME] if len(matches) != 1: raise RuntimeError("Source catalog report did not contain exactly one ALZ agriculture contract") return fetch_release_decision_from_item(args, matches[0]) def require_release_confirmation(args: argparse.Namespace, release: AgriculturalReleaseConfig) -> None: if args.confirm_edition != release.definitive_version: raise RuntimeError(f"Explicit --confirm-edition {release.definitive_version} is required") def internal_base_url(api_url: str) -> str: value = api_url.rstrip("/") if value.endswith("/api/v1"): value = value[:-7] parsed = urlparse(value) if parsed.hostname not in {"127.0.0.1", "localhost", "::1"}: raise RuntimeError("Stage/apply must run inside GeoIntel against the local backend API") return value def default_plan_path(args: argparse.Namespace, year: int) -> Path: return args.evidence_root / args.scope / str(year) / "staged-plan.json" def default_review_path(args: argparse.Namespace, year: int) -> Path: return args.evidence_root / args.scope / str(year) / "review-evidence.json" def governed_evidence_path(args: argparse.Namespace, path: Path) -> Path: if not path.resolve().is_relative_to(args.evidence_root.resolve()): raise RuntimeError(f"Release evidence path is outside the governed evidence root: {path}") return path def build_operator_command( args: argparse.Namespace, release: AgriculturalReleaseConfig, *, fetch_only: bool, ) -> list[str]: command = [ sys.executable, str(Path(__file__).resolve().parent / "provision_agricultural_parcel_history.py"), "--base-url", internal_base_url(args.api_url), "--scope", args.scope, "--years", str(release.year), "--archive-url", release.archive_url, "--output-root", str(args.output_root), "--request-timeout", str(args.request_timeout), "--import-timeout", str(args.import_timeout), "--max-features", str(args.max_features), "--max-archive-mb", str(args.max_archive_mb), ] if fetch_only: command.extend(("--force", "--fetch-only")) return command def run_operator(command: list[str], *, action: str) -> dict[str, Any]: print(f"ALZ agriculture: {action}...", file=sys.stderr, flush=True) completed = subprocess.run(command, check=False, capture_output=True, text=True, encoding="utf-8") if completed.returncode != 0: detail = completed.stderr.strip() or completed.stdout.strip() or "operator returned no diagnostics" raise RuntimeError(f"ALZ agriculture {action} failed with exit {completed.returncode}: {detail[-3000:]}") try: payload = json.loads(completed.stdout) except json.JSONDecodeError as exc: raise RuntimeError(f"ALZ agriculture {action} returned invalid JSON") from exc if payload.get("status") != "ok": raise RuntimeError(f"ALZ agriculture {action} did not report success") return payload 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 write_json(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".partial") temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8") temporary.replace(path) def _previous_manifest(output_root: Path, scope_key: str, year: int) -> tuple[Path, dict[str, Any]] | None: scope_root = output_root / scope_key if not scope_root.is_dir(): return None previous_years = sorted( (int(path.name) for path in scope_root.iterdir() if path.is_dir() and path.name.isdigit() and int(path.name) < year), reverse=True, ) for previous_year in previous_years: manifests = list((scope_root / str(previous_year)).glob("*.manifest.json")) if len(manifests) != 1: continue payload = json.loads(manifests[0].read_text(encoding="utf-8")) if payload.get("year") == previous_year and payload.get("scope_key") == scope_key: return manifests[0], payload return None def _relative_change(current: float, previous: float) -> float | None: if previous <= 0: return None return round((current - previous) / previous, 8) def validate_staged_release(args: argparse.Namespace, release: AgriculturalReleaseConfig) -> dict[str, Any]: paths = artifact_paths(args.output_root, args.scope, release.year, archive_url=release.archive_url) resolved_output = args.output_root.resolve() for key in ("archive", "artifact", "codelist", "manifest"): if not paths[key].resolve().is_relative_to(resolved_output): raise RuntimeError(f"Staged ALZ {key} is outside the governed output root: {paths[key]}") manifest = reusable_artifact( paths, year=release.year, scope_key=args.scope, archive_url=release.archive_url, ) if manifest is None: raise RuntimeError("Staged ALZ artifacts are incomplete or no longer match their checksums") fields = set(manifest.get("source_fields") or []) if ( manifest.get("source_url") != release.archive_url or manifest.get("source_crs") != SOURCE_CRS or manifest.get("output_crs") != OUTPUT_CRS or not STABLE_REQUIRED_FIELDS.issubset(fields) or int(manifest.get("feature_count") or 0) < 1 or float(manifest.get("clipped_area_ha") or 0) <= 0 or int(manifest.get("source_feature_count") or 0) < int(manifest.get("feature_count") or 0) ): raise RuntimeError("Staged ALZ manifest does not satisfy the definitive release contract") max_archive_bytes = min(args.max_archive_mb * 1024 * 1024, MAX_ARCHIVE_BYTES) if paths["archive"].stat().st_size > max_archive_bytes: raise RuntimeError("Staged ALZ archive exceeds the configured release limit") codelist = json.loads(paths["codelist"].read_text(encoding="utf-8")) crop_entries = codelist.get("crop_entries") if isinstance(codelist, dict) else None if codelist.get("year") != release.year or not isinstance(crop_entries, list) or not crop_entries: raise RuntimeError("Staged ALZ crop-code evidence is incomplete") previous_result = _previous_manifest(args.output_root, args.scope, release.year) baseline = None if previous_result is not None: previous_path, previous = previous_result if not previous_path.resolve().is_relative_to(resolved_output): raise RuntimeError("Previous ALZ baseline manifest is outside the governed output root") current_count = int(manifest["feature_count"]) previous_count = int(previous.get("feature_count") or 0) current_area = float(manifest["clipped_area_ha"]) previous_area = float(previous.get("clipped_area_ha") or 0) baseline = { "year": int(previous["year"]), "manifest_path": str(previous_path), "manifest_sha256": sha256_file(previous_path), "feature_count": previous_count, "clipped_area_ha": previous_area, "feature_count_change_ratio": _relative_change(current_count, previous_count), "clipped_area_change_ratio": _relative_change(current_area, previous_area), } return { "manifest_path": str(paths["manifest"]), "manifest_sha256": sha256_file(paths["manifest"]), "archive_path": str(paths["archive"]), "archive_sha256": manifest["source_archive_sha256"], "archive_size_bytes": manifest["source_archive_size_bytes"], "artifact_path": str(paths["artifact"]), "artifact_sha256": manifest["artifact_sha256"], "crop_code_list_path": str(paths["codelist"]), "crop_code_list_sha256": manifest["crop_code_list_sha256"], "crop_entry_count": len(crop_entries), "crop_code_conflicts": codelist.get("code_title_conflicts") or {}, "source_feature_count": manifest["source_feature_count"], "feature_count": manifest["feature_count"], "clipped_feature_count": manifest["clipped_feature_count"], "clipped_area_ha": manifest["clipped_area_ha"], "source_fields": sorted(fields), "member_nis_codes": manifest.get("member_nis_codes") or [], "baseline": baseline, } def build_staged_plan( args: argparse.Namespace, decision: dict[str, Any], release: AgriculturalReleaseConfig, operator_result: dict[str, Any], ) -> dict[str, Any]: payload: dict[str, Any] = { "schema_version": 1, "status": "staged", "created_at": datetime.now(timezone.utc).isoformat(), "project_id": args.project_id, "scope": args.scope, "local_source_version_before_apply": decision.get("local_source_version"), "release": decision["release"], "catalog_identity": decision["catalog_identity"], "evidence": validate_staged_release(args, release), "operator_result": operator_result, "review_required": True, "apply_requires_plan_sha256": True, "apply_requires_review_sha256": True, "automatic_import": False, "destructive_replacement": False, "existing_snapshots_retained": True, } payload["plan_sha256"] = canonical_sha256(payload, "plan_sha256") return payload def _release_from_payload(payload: dict[str, Any]) -> AgriculturalReleaseConfig: release = payload.get("release") or {} year = release.get("year") if isinstance(release, dict) else None edition = release.get("edition") if isinstance(release, dict) else None archive_url = release.get("archive_url") if isinstance(release, dict) else None if ( not isinstance(year, int) or isinstance(year, bool) or edition != f"{year}-v3" or not isinstance(archive_url, str) ): raise RuntimeError("Release evidence is missing required ALZ identity fields") return resolve_release_config(year, archive_url=archive_url) def load_staged_plan( args: argparse.Namespace, release: AgriculturalReleaseConfig, ) -> tuple[Path, dict[str, Any]]: path = governed_evidence_path(args, args.plan_path or default_plan_path(args, release.year)) if not path.is_file(): raise RuntimeError(f"Staged ALZ 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 ALZ 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 _release_from_payload(payload) != release ): raise RuntimeError("Staged ALZ plan identity is invalid") current_evidence = validate_staged_release(args, release) if current_evidence != payload.get("evidence"): raise RuntimeError("Staged ALZ artifacts no longer match the approved plan") return path, payload def require_catalog_unchanged(plan: dict[str, Any], decision: dict[str, Any]) -> None: if plan.get("release") != decision.get("release") or plan.get("catalog_identity") != decision.get("catalog_identity"): raise RuntimeError("The official ALZ release evidence changed; create and review a new staged plan") def build_review_evidence(args: argparse.Namespace, plan_path: Path, plan: 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, "release": plan["release"], "staged_plan_path": str(plan_path), "staged_plan_sha256": plan["plan_sha256"], "reviewed_checks": [ "official_catalog_identity", "source_archive_checksum", "geopackage_schema_and_crs", "scope_feature_and_area_accounting", "crop_code_list_and_conflicts", "previous_definitive_edition_delta", "provisional_snapshot_exclusion", "immutable_dataset_apply", ], } payload["review_sha256"] = canonical_sha256(payload, "review_sha256") return payload def load_review_evidence( args: argparse.Namespace, release: AgriculturalReleaseConfig, plan: dict[str, Any], ) -> tuple[Path, dict[str, Any]]: path = governed_evidence_path(args, args.review_path or default_review_path(args, release.year)) if not path.is_file(): raise RuntimeError(f"Approved ALZ 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("ALZ 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("release") != plan.get("release") or payload.get("staged_plan_sha256") != plan.get("plan_sha256") or not str(payload.get("reviewer") or "").strip() ): raise RuntimeError("ALZ review evidence does not authorize this staged plan") return path, payload def _release_result(operator_result: dict[str, Any], year: int) -> dict[str, Any]: matches = [item for item in operator_result.get("years") or [] if int(item.get("year") or 0) == year] if len(matches) != 1: raise RuntimeError("Agriculture operator did not return exactly one result for the approved edition") return matches[0] def main() -> int: args = parse_args() try: if min(args.request_timeout, args.api_timeout, args.import_timeout, args.max_features, args.max_archive_mb) <= 0: raise ValueError("All timeout and safety limits must be positive") validate_project_scope(args) refresh = args.refresh_catalog or args.action in {"stage", "review", "apply"} decision = fetch_release_decision(args, refresh=refresh) release = _release_from_payload(decision) if args.action == "plan": print(json.dumps({"status": "ok", "action": "plan", "decision": decision}, ensure_ascii=False, indent=2)) return 0 require_release_confirmation(args, release) if args.action == "stage": if decision.get("status") not in ACTIONABLE_STATUSES: raise RuntimeError(f"ALZ release is not safely stageable: {decision.get('status')}") operator_result = run_operator(build_operator_command(args, release, fetch_only=True), action="staging") staged_result = _release_result(operator_result, release.year) if staged_result.get("status") != "prepared": raise RuntimeError("ALZ staging did not produce prepared release evidence") plan = build_staged_plan(args, decision, release, operator_result) plan_path = governed_evidence_path(args, args.plan_path or default_plan_path(args, release.year)) write_json(plan_path, plan) print(json.dumps({"status": "staged", "plan_path": str(plan_path), **plan}, ensure_ascii=False, indent=2)) return 0 plan_path, plan = load_staged_plan(args, release) require_catalog_unchanged(plan, decision) if args.action == "review": review = build_review_evidence(args, plan_path, plan) review_path = governed_evidence_path(args, args.review_path or default_review_path(args, release.year)) 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, release, plan) operator_result = run_operator(build_operator_command(args, release, fetch_only=False), action="applying") applied_result = _release_result(operator_result, release.year) if applied_result.get("status") not in {"imported", "existing"} or not applied_result.get("dataset_id"): raise RuntimeError("ALZ apply did not return one persisted immutable Dataset") final_decision = fetch_release_decision(args, refresh=True) if ( final_decision.get("status") != "current" or final_decision.get("local_source_version") != f"{release.year}-definitive" ): raise RuntimeError("Applied ALZ Dataset did not become the current local definitive edition") evidence = { "schema_version": 1, "status": "applied", "applied_at": datetime.now(timezone.utc).isoformat(), "project_id": args.project_id, "scope": args.scope, "release": plan["release"], "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": applied_result["dataset_id"], "dataset_status": applied_result["status"], "feature_count": applied_result.get("feature_count"), "final_catalog_decision": final_decision, "existing_snapshots_retained": True, } evidence["applied_evidence_sha256"] = canonical_sha256(evidence, "applied_evidence_sha256") evidence_path = plan_path.with_name("applied-evidence.json") write_json(evidence_path, evidence) 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())