Jens
faeb58ef6d
GeoIntel release gates / Compile, test, contracts and builds (push) Successful in 1m49s
GeoIntel release gates / Python and npm vulnerability policy (push) Successful in 21s
GeoIntel release gates / Production AI image, SBOM and container scan (push) Successful in 5m39s
GeoIntel release gates / Deploy exact gated revision to Unraid (push) Failing after 58m43s
87 lines
3.5 KiB
Python
87 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Combine completed Belgian portfolio specs with fail-closed provenance."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def combine_specs(paths: list[Path]) -> dict[str, Any]:
|
|
if not paths:
|
|
raise ValueError("At least one portfolio spec is required")
|
|
samples: list[dict[str, Any]] = []
|
|
seen: set[str] = set()
|
|
side_m: float | None = None
|
|
resolution_m: float | None = None
|
|
sources = []
|
|
for path in paths:
|
|
payload = json.loads(path.read_text(encoding="utf-8-sig"))
|
|
status = payload.get("status")
|
|
legacy_unstated = status is None and payload.get("schema_version") == 1 and bool(payload.get("samples"))
|
|
if status != "complete" and not legacy_unstated:
|
|
raise ValueError(f"Portfolio spec is not complete: {path}")
|
|
current_side = float(payload["side_m"]) if payload.get("side_m") is not None else None
|
|
current_resolution = float(payload["resolution_m"]) if payload.get("resolution_m") is not None else None
|
|
if side_m is not None and current_side is not None and current_side != side_m:
|
|
raise ValueError(f"Portfolio side_m differs: {path}")
|
|
if resolution_m is not None and current_resolution is not None and current_resolution != resolution_m:
|
|
raise ValueError(f"Portfolio resolution_m differs: {path}")
|
|
side_m = current_side if current_side is not None else side_m
|
|
resolution_m = current_resolution if current_resolution is not None else resolution_m
|
|
for sample in payload.get("samples") or []:
|
|
slug = str(sample.get("sample_slug") or sample.get("slug") or "")
|
|
if not slug:
|
|
raise ValueError(f"Portfolio sample has no slug: {path}")
|
|
if slug in seen:
|
|
raise ValueError(f"Duplicate portfolio sample slug: {slug}")
|
|
seen.add(slug)
|
|
normalized = dict(sample)
|
|
normalized["sample_slug"] = slug
|
|
normalized.pop("slug", None)
|
|
samples.append(normalized)
|
|
sources.append({
|
|
"path": str(path),
|
|
"sha256": sha256(path),
|
|
"sample_count": len(payload.get("samples") or []),
|
|
"source_status": status or "legacy_unstated",
|
|
})
|
|
if side_m is None or resolution_m is None:
|
|
raise ValueError("Combined portfolio dimensions are unknown")
|
|
return {
|
|
"schema_version": 2,
|
|
"status": "complete",
|
|
"side_m": side_m,
|
|
"resolution_m": resolution_m,
|
|
"sample_count": len(samples),
|
|
"source_specs": sources,
|
|
"samples": samples,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--spec", action="append", type=Path, required=True)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
payload = combine_specs(args.spec)
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(json.dumps({"status": "ok", "sample_count": payload["sample_count"], "output": str(args.output)}))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|