planet/scripts/collect_earth_boundary_sources.py

#!/usr/bin/env python3
"""Collect versioned source files for the Earth boundary tile pipeline.

This is an offline ingestion helper, not a runtime service. It downloads or
copies declared source packages, verifies optional sha256 checksums, and writes
a manifest that the tile builder can reference during audited rebuilds.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import shutil
import urllib.request
from pathlib import Path
from typing import Any


DEFAULT_CONFIG = Path("config/earth-boundary-sources.example.json")
DEFAULT_OUTPUT = Path("data/earth-boundary-sources")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG)
    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
    return parser.parse_args()


def load_config(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as f:
        payload = json.load(f)
    if not isinstance(payload.get("sources"), list):
        raise ValueError(f"{path} must contain a sources array")
    return payload


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def safe_filename(source: dict[str, Any]) -> str:
    name = str(source.get("name") or source.get("id") or "source")
    filename = source.get("filename")
    if filename:
        return Path(str(filename)).name
    suffix = Path(str(source.get("url") or source.get("path") or name)).suffix
    cleaned = "".join(ch if ch.isalnum() or ch in ("-", "_") else "-" for ch in name).strip("-")
    return f"{cleaned or 'source'}{suffix or '.dat'}"


def collect_source(source: dict[str, Any], output_dir: Path) -> dict[str, Any]:
    source_id = source.get("id")
    if not source_id:
        raise ValueError("each source requires an id")
    target = output_dir / safe_filename(source)
    target.parent.mkdir(parents=True, exist_ok=True)

    if source.get("url"):
        request = urllib.request.Request(
            str(source["url"]),
            headers={"User-Agent": "planet-earth-boundary-source-collector/1.0"},
        )
        with urllib.request.urlopen(request, timeout=60) as response, target.open("wb") as f:
            shutil.copyfileobj(response, f)
    elif source.get("path"):
        shutil.copy2(Path(str(source["path"])).expanduser(), target)
    else:
        raise ValueError(f"source {source_id} requires url or path")

    actual_sha256 = sha256_file(target)
    expected_sha256 = source.get("sha256")
    if expected_sha256 and actual_sha256.lower() != str(expected_sha256).lower():
        target.unlink(missing_ok=True)
        raise ValueError(
            f"source {source_id} sha256 mismatch: expected {expected_sha256}, got {actual_sha256}"
        )

    return {
        "id": source_id,
        "name": source.get("name") or source_id,
        "kind": source.get("kind") or "unknown",
        "license": source.get("license"),
        "pov": source.get("pov"),
        "path": str(target),
        "sha256": actual_sha256,
        "source": source.get("url") or source.get("path"),
        "notes": source.get("notes"),
    }


def collect_policy(config: dict[str, Any]) -> dict[str, Any] | None:
    policy = config.get("policy") or {}
    policy_path_value = policy.get("povPolicyPath")
    if not policy_path_value:
        return None

    policy_path = Path(str(policy_path_value)).expanduser()
    if not policy_path.exists():
        raise FileNotFoundError(f"POV policy file not found: {policy_path}")

    with policy_path.open("r", encoding="utf-8") as f:
        policy_payload = json.load(f)

    return {
        "profile": policy_payload.get("profile") or policy.get("profile"),
        "path": str(policy_path),
        "sha256": sha256_file(policy_path),
        "schema": policy_payload.get("schema"),
        "ruleCount": len(policy_payload.get("rules") or []),
        "productionTileFormat": policy.get("productionTileFormat"),
        "debugTileFormat": policy.get("debugTileFormat"),
    }


def main() -> None:
    args = parse_args()
    config = load_config(args.config)
    args.output.mkdir(parents=True, exist_ok=True)
    collected = [collect_source(source, args.output) for source in config["sources"]]
    pov_policy = collect_policy(config)
    manifest = {
        "schema": "planet-earth-boundary-sources/v1",
        "sources": collected,
        "povPolicy": pov_policy,
        "policy": config.get("policy", {}),
    }
    manifest_path = args.output / "manifest.json"
    with manifest_path.open("w", encoding="utf-8") as f:
        json.dump(manifest, f, ensure_ascii=False, indent=2)
        f.write("\n")
    print(json.dumps({"manifest": str(manifest_path), "sources": len(collected)}, sort_keys=True))


if __name__ == "__main__":
    main()