"""Provision one queryable GRB building dataset for an approved regional scope. Source requests are partitioned by official municipality boundaries and can be resumed per partition. The retained partition artifacts are then indexed as one Dataset through DatasetService and VectorFeatureService. This operator never runs during application startup and never writes directly to vector_features. """ from __future__ import annotations import argparse import hashlib import json import os import sys import unicodedata from datetime import date, datetime, time, timezone from pathlib import Path from typing import Any, Iterable from urllib.parse import urlparse from uuid import UUID import requests from requests.adapters import HTTPAdapter from shapely.geometry import GeometryCollection, MultiPolygon, Polygon, mapping, shape from shapely.ops import unary_union from shapely.validation import make_valid from urllib3.util.retry import Retry from geographic_scopes import GEOGRAPHIC_SCOPES, GeographicScope, ScopeMember from provision_geographic_scope import fetch_scope_members GRB_GBG_ITEMS_URL = "https://geo.api.vlaanderen.be/GRB/ogc/features/v1/collections/GBG/items" GRB_ATTRIBUTION = "Bron: Grootschalig Referentie Bestand Vlaanderen, Digitaal Vlaanderen" DEFAULT_SCOPE_KEY = "kempen-transport-region" DEFAULT_OUTPUT_ROOT = Path("/app/storage/operator-data/regional-themes") DEFAULT_API_URL = "http://127.0.0.1:8000" DEFAULT_PAGE_LIMIT = 1000 DEFAULT_MAX_FEATURES_PER_MEMBER = 100000 DEFAULT_MAX_TOTAL_FEATURES = 1500000 GEOJSON_CRS = {"type": "name", "properties": {"name": "EPSG:4326"}} def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Provision municipality-partitioned regional GRB buildings.") parser.add_argument("--scope", choices=sorted(GEOGRAPHIC_SCOPES), default=DEFAULT_SCOPE_KEY) 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 municipality partition for this date.") parser.add_argument("--fetch-only", action="store_true", help="Build and validate artifacts without persistence.") return parser.parse_args() def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def observed_at(value: date) -> datetime: return datetime.combine(value, time.min, tzinfo=timezone.utc) def safe_slug(value: str) -> str: normalized = unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii") return "-".join(part for part in "".join(char.lower() if char.isalnum() else " " for char in normalized).split()) def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(8 * 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 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-Regional-GRB-Buildings-Operator/1.0"}) session.mount("https://", adapter) session.mount("http://", adapter) return session def next_page_url(payload: dict[str, Any]) -> str | None: links = payload.get("links") or [] for link in links: if link.get("rel") == "next" and "geo+json" in str(link.get("type", "")).lower(): return str(link["href"]) for link in links: if link.get("rel") == "next" and link.get("href"): return str(link["href"]) return None def iter_grb_pages( session: requests.Session, bounds: tuple[float, float, float, float], *, page_limit: int, timeout: int, ) -> Iterable[tuple[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_GBG_ITEMS_URL seen_urls: set[str] = set() first_request = True while url: if url in seen_urls: raise RuntimeError(f"GRB pagination loop detected: {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("GRB returned a non-FeatureCollection response") yield payload, response.url url = next_page_url(payload) def build_member_geometries( scope: GeographicScope, source_features: list[dict[str, Any]], ) -> tuple[list[tuple[ScopeMember, Any]], Any]: if len(source_features) != len(scope.members): raise RuntimeError(f"Expected {len(scope.members)} scope boundaries, received {len(source_features)}") members: list[tuple[ScopeMember, Any]] = [] for member, source_feature in zip(scope.members, source_features, strict=True): properties = source_feature.get("properties") or {} if str(properties.get("NISCODE") or "") != member.nis_code: raise RuntimeError(f"VRBG scope member mismatch for {member.name}") geometry = normalize_polygonal(shape(source_feature.get("geometry"))) if geometry is None: raise RuntimeError(f"Invalid official boundary for {member.name}") members.append((member, geometry)) regional_boundary = normalize_polygonal(unary_union([geometry for _, geometry in members])) if regional_boundary is None: raise RuntimeError("The regional scope union is invalid") return members, regional_boundary def bounds_overlap(left: tuple[float, float, float, float], right: tuple[float, float, float, float]) -> bool: return left[0] <= right[2] and left[2] >= right[0] and left[1] <= right[3] and left[3] >= right[1] def assign_owner_nis(source_geometry, members: list[tuple[ScopeMember, Any]]) -> 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 intersection = source_geometry.intersection(boundary) if not intersection.is_empty and intersection.area > 0: candidates.append((float(intersection.area), member.nis_code)) if not candidates: return None candidates.sort(key=lambda item: (-item[0], item[1])) return candidates[0][1] def build_partition_features( pages: Iterable[tuple[dict[str, Any], str]], *, 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() 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 payload, source_url in pages: source_urls.append(source_url) for source_feature in payload.get("features") or []: bbox_feature_count += 1 feature_id = str(source_feature.get("id") or "") if not feature_id: feature_id = hashlib.sha256( json.dumps(source_feature.get("geometry"), sort_keys=True).encode("utf-8") ).hexdigest() if feature_id in seen_ids: continue seen_ids.add(feature_id) source_geometry = normalize_polygonal(shape(source_feature.get("geometry"))) 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) ) if owner_nis != member.nis_code: assigned_elsewhere_count += 1 continue clipped_geometry = source_geometry if not source_geometry.within(regional_boundary): clipped_geometry = normalize_polygonal(source_geometry.intersection(regional_boundary)) clipped_to_scope_count += 1 if clipped_geometry is None: outside_scope_count += 1 continue if len(features) >= max_features: raise RuntimeError( f"{member.name} exceeds --max-features-per-member={max_features}; refusing truncated output" ) properties = dict(source_feature.get("properties") or {}) properties.update( { "source_name": "grb", "source_feature_id": feature_id, "reference_layer_name": "buildings", "layer_type": "building", "theme": "buildings", "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_boundary_intersection", "clipped_to_regional_scope": clipped_geometry is not source_geometry, "attribution": GRB_ATTRIBUTION, } ) features.append( { "type": "Feature", "id": feature_id, "geometry": mapping(clipped_geometry), "properties": properties, } ) if not features: raise RuntimeError(f"No GRB buildings were assigned to {member.name}") return features, { "municipality": member.name, "nis_code": member.nis_code, "pages_fetched": len(source_urls), "source_urls": source_urls, "bbox_feature_count": bbox_feature_count, "feature_count": len(features), "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(member: ScopeMember, observed_date: date) -> str: return f"{member.nis_code}_{safe_slug(member.name)}_grb_buildings_{observed_date.isoformat()}.geojson" def combined_filename(scope: GeographicScope, observed_date: date) -> str: return f"grb_buildings_{scope.key.replace('-', '_')}_{observed_date.isoformat()}.geojson" def write_partition( path: Path, *, scope: GeographicScope, member: ScopeMember, features: list[dict[str, Any]], generated_at: str, source_url: str, ) -> None: write_json_atomic( path, { "type": "FeatureCollection", "name": f"GRB buildings - {member.name} partition of {scope.display_name}", "crs": GEOJSON_CRS, "features": features, "source": "Digitaal Vlaanderen GRB OGC API Features collection GBG", "source_url": source_url, "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, *, 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 buildings - {scope.display_name}", "crs": GEOJSON_CRS, "source": "Digitaal Vlaanderen GRB OGC API Features collection GBG", "source_url": GRB_GBG_ITEMS_URL, "attribution": GRB_ATTRIBUTION, "coverage_scope": scope.key, "scope_type": scope.scope_type, "partition_count": len(partition_paths), "partition_strategy": "municipality_bbox_maximum_boundary_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 reusable_manifest( manifest_path: Path, artifact_path: Path, partition_dir: Path, expected_member_count: int, ) -> dict[str, Any] | None: if not manifest_path.is_file() or not artifact_path.is_file(): return None manifest = json.loads(manifest_path.read_text(encoding="utf-8")) if manifest.get("status") != "complete" or len(manifest.get("partitions") or []) != expected_member_count: return None if sha256_file(artifact_path) != manifest.get("artifact_sha256"): return None for summary in manifest["partitions"]: path = partition_dir / str(summary.get("filename") or "") if not path.is_file() or sha256_file(path) != summary.get("sha256"): return None return manifest def prepare_artifacts( args: argparse.Namespace, scope: GeographicScope, ) -> tuple[Path, list[Path], Path, dict[str, Any]]: if args.page_limit <= 0 or args.max_features_per_member <= 0 or args.max_total_features <= 0: raise RuntimeError("Page and feature limits must be positive") observation_dir = args.output_root / scope.key / "buildings" / args.observed_date.isoformat() partition_dir = observation_dir / "partitions" artifact_path = observation_dir / combined_filename(scope, args.observed_date) manifest_path = observation_dir / "regional_buildings_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 build_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(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_grb_pages( session, boundary.bounds, page_limit=args.page_limit, timeout=args.request_timeout, ), member=member, members=members, regional_boundary=regional_boundary, scope=scope, max_features=args.max_features_per_member, ) write_partition( path, scope=scope, member=member, features=features, generated_at=generated_at, source_url=summary["source_urls"][0], ) summary.update({"filename": path.name, "size_bytes": path.stat().st_size, "sha256": sha256_file(path)}) summaries.append(summary) progress = { "schema_version": 1, "status": "in_progress", "scope": scope.key, "theme": "buildings", "observed_at": args.observed_date.isoformat(), "generated_at": generated_at, "vrbg_source_url": vrbg_source_url, "grb_source_url": GRB_GBG_ITEMS_URL, "partitions": summaries, } write_json_atomic(manifest_path, progress, pretty=True) total_features = sum(int(summary["feature_count"]) for summary in summaries) if total_features > args.max_total_features: raise RuntimeError( f"Regional building 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, scope=scope, observed_date=args.observed_date, partition_paths=partition_paths, expected_feature_count=total_features, ) 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": "buildings", "observed_at": args.observed_date.isoformat(), "generated_at": generated_at, "member_count": len(scope.members), "feature_count": total_features, "reference_truncated": False, "partition_strategy": "municipality_bbox_maximum_boundary_intersection", "partition_assignment_rule": "largest geometry intersection; NIS code resolves exact ties", "vrbg_source_url": vrbg_source_url, "grb_source_url": GRB_GBG_ITEMS_URL, "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]), }, "partitions": summaries, "attribution": GRB_ATTRIBUTION, } write_json_atomic(manifest_path, manifest, pretty=True) return artifact_path, partition_paths, manifest_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 request failed ({response.status_code}): {json.dumps(payload)[: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 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) total = int(page.get("total") or len(items)) if not page_items or len(items) >= total: return items offset += len(page_items) def ensure_backend_path() -> None: repository_root = Path(__file__).resolve().parents[1] backend_root = repository_root / "backend" if (repository_root / "backend" / "app").is_dir() else repository_root if str(backend_root) not in sys.path: sys.path.insert(0, str(backend_root)) def provision_dataset( args: argparse.Namespace, scope: GeographicScope, 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 partition_checksums = { summary["nis_code"]: summary["sha256"] for summary in manifest["partitions"] } metadata_json = { "feature_count": manifest["feature_count"], "feature_geometry_count": manifest["feature_count"], "geometry_types": ["MultiPolygon", "Polygon"], "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", "collection": "GRB/GBG", "authority_level": "authoritative", "theme": "buildings", "layer_type": "regional_buildings", "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), "geometry_clipped_to_area": True, "feature_count": manifest["feature_count"], "partition_count": len(partition_paths), "partition_strategy": manifest["partition_strategy"], "selection_aggregation": { "metric_key": "footprint_area", "method": "intersection_area", "label": "Bebouwde grondoppervlakte", "unit": "ha", "is_estimate": False, "warning": "Dit is de grondoppervlakte van gebouwcontouren, niet de totale vloeroppervlakte of het gebouwvolume.", }, "attribution": GRB_ATTRIBUTION, } provenance_metadata = { "operator_tool": "provision_regional_grb_buildings.py", "operator_explicit_fetch": True, "geometry_clipped_to_area": True, "manifest_path": str(manifest_path), "source_url": GRB_GBG_ITEMS_URL, "artifact_sha256": manifest["artifact_sha256"], "artifact_size_bytes": manifest["artifact_size_bytes"], "partition_checksums": partition_checksums, "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="buildings", metadata_json=metadata_json, source_metadata=source_metadata, provenance_metadata=provenance_metadata, temporal_series_key=f"grb:buildings:{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: artifact_path, partition_paths, manifest_path, manifest = prepare_artifacts(args, scope) persistence = None if args.fetch_only else provision_dataset( args, scope, artifact_path, partition_paths, manifest_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, "theme": "buildings", "observed_at": args.observed_date.isoformat(), "member_count": manifest["member_count"], "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, }, ensure_ascii=False, indent=2, ) ) return 0 if __name__ == "__main__": sys.exit(main())