"""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, response_data, 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["nis_code"]: 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())