#!/usr/bin/env python3
"""Build deterministic, quarterly Solarma OSM outing-place packs.

The production path accepts one or more checksum-pinned OSM PBF extracts and
streams them through the PyPI ``osmium`` (pyosmium) binding. A temporary,
disk-backed node-location index bounds memory without materializing filtered
PBF or GeoJSON intermediates. Small OSM XML and GeoJSON inputs are supported
directly so the fixture build and tests run completely offline with Python's
standard library.

The same source manifest and source bytes always produce byte-identical JSON
and gzip outputs. In particular, gzip timestamps and platform identifiers are
fixed instead of inheriting the build machine's clock or OS.
"""

from __future__ import annotations

import argparse
import binascii
import gc
import gzip
import hashlib
import json
import math
import re
import struct
import subprocess
import sys
import tempfile
import threading
import xml.etree.ElementTree as ET
import zlib
from concurrent.futures import Future, ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Iterable, Iterator, Mapping, Sequence


SCHEMA_VERSION = 1
BUILDER_VERSION = "1.2.0"
DEFAULT_GEOHASH_PRECISION = 3
MAX_BUILD_JOBS = 8
DATASET_VERSION_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,63}$")
QUARTER_RE = re.compile(r"^20\d{2}-Q[1-4]$")

ACTIVITY_ORDER = (
    "playground",
    "park",
    "walking",
    "hiking",
    "discGolf",
    "dogPark",
    "bike",
    "court",
    "skate",
    "picnic",
    "water",
    "fishing",
    "viewpoint",
    "birding",
    "golf",
)
ACTIVITY_INDEX = {activity: index for index, activity in enumerate(ACTIVITY_ORDER)}

ATTRIBUTION = {
    "provider": "OpenStreetMap",
    "attribution": "Map data © OpenStreetMap contributors",
    "copyrightUrl": "https://www.openstreetmap.org/copyright",
    "license": "Open Data Commons Open Database License (ODbL) 1.0",
    "licenseId": "ODbL-1.0",
    "licenseUrl": "https://opendatacommons.org/licenses/odbl/1-0/",
    "derivedDatabase": True,
    "notice": (
        "These normalized place packs are a derivative database of "
        "OpenStreetMap and must remain available under ODbL 1.0."
    ),
}

PRIVATE_ACCESS = {"private", "no"}
SUPPORTED_COURT_SPORTS = {"basketball", "tennis", "pickleball"}
TRUE_VALUES = {"yes", "true", "1", "designated"}
FALSE_VALUES = {"no", "false", "0"}

GEOHASH_ALPHABET = "0123456789bcdefghjkmnpqrstuvwxyz"
GEOHASH_BITS = (16, 8, 4, 2, 1)


class BuildError(RuntimeError):
    """A deterministic input or build-contract failure."""


@dataclass(frozen=True)
class RawFeature:
    osm_type: str
    osm_id: int
    latitude: float
    longitude: float
    tags: Mapping[str, str]


@dataclass(frozen=True)
class SourceExtract:
    path: Path
    url: str
    sha256: str


@dataclass(frozen=True)
class SourceManifest:
    dataset_version: str
    quarter: str
    source_timestamp: str
    geohash_precision: int
    extracts: tuple[SourceExtract, ...]


def canonical_json_bytes(value: Any) -> bytes:
    return (
        json.dumps(
            value,
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
            allow_nan=False,
        ).encode("utf-8")
        + b"\n"
    )


def deterministic_gzip(data: bytes, level: int = 9) -> bytes:
    """Return a cross-platform deterministic RFC 1952 gzip member."""

    compressor = zlib.compressobj(level, zlib.DEFLATED, -zlib.MAX_WBITS)
    payload = compressor.compress(data) + compressor.flush()
    # ID1/ID2, deflate, flags=0, mtime=0, XFL=max compression, OS=unknown.
    header = b"\x1f\x8b\x08\x00\x00\x00\x00\x00\x02\xff"
    trailer = struct.pack("<II", binascii.crc32(data) & 0xFFFFFFFF, len(data) & 0xFFFFFFFF)
    return header + payload + trailer


def sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def sha256_file(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 require_string(value: Any, field: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise BuildError(f"{field} must be a non-empty string")
    return value.strip()


def read_source_manifest(path: Path) -> SourceManifest:
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise BuildError(f"cannot read source manifest {path}: {error}") from error
    if not isinstance(raw, dict) or raw.get("schemaVersion") != SCHEMA_VERSION:
        raise BuildError(f"{path}: unsupported source-manifest schemaVersion")

    dataset_version = require_string(raw.get("datasetVersion"), "datasetVersion")
    if not DATASET_VERSION_RE.fullmatch(dataset_version):
        raise BuildError("datasetVersion must use lowercase letters, digits, '.', '_' or '-'")
    quarter = require_string(raw.get("quarter"), "quarter")
    if not QUARTER_RE.fullmatch(quarter):
        raise BuildError("quarter must look like 2026-Q3")
    source_timestamp = require_string(raw.get("sourceTimestamp"), "sourceTimestamp")
    if not re.fullmatch(r"20\d{2}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z", source_timestamp):
        raise BuildError("sourceTimestamp must be a fixed UTC ISO timestamp")
    precision = raw.get("geohashPrecision", DEFAULT_GEOHASH_PRECISION)
    if not isinstance(precision, int) or precision < 1 or precision > 7:
        raise BuildError("geohashPrecision must be an integer from 1 through 7")

    raw_extracts = raw.get("extracts")
    if not isinstance(raw_extracts, list) or not raw_extracts:
        raise BuildError("extracts must contain at least one checksum-pinned input")
    extracts: list[SourceExtract] = []
    for index, item in enumerate(raw_extracts):
        if not isinstance(item, dict):
            raise BuildError(f"extracts[{index}] must be an object")
        relative = require_string(item.get("path"), f"extracts[{index}].path")
        source_path = (path.parent / relative).resolve()
        url = require_string(item.get("url"), f"extracts[{index}].url")
        checksum = require_string(item.get("sha256"), f"extracts[{index}].sha256").lower()
        if not re.fullmatch(r"[0-9a-f]{64}", checksum):
            raise BuildError(f"extracts[{index}].sha256 must be 64 lowercase hex characters")
        if not source_path.is_file():
            raise BuildError(f"source extract does not exist: {source_path}")
        actual = sha256_file(source_path)
        if actual != checksum:
            raise BuildError(
                f"checksum mismatch for {source_path}: expected {checksum}, got {actual}"
            )
        extracts.append(SourceExtract(source_path, url, checksum))

    return SourceManifest(
        dataset_version=dataset_version,
        quarter=quarter,
        source_timestamp=source_timestamp,
        geohash_precision=precision,
        extracts=tuple(extracts),
    )


def local_name(tag: str) -> str:
    return tag.rsplit("}", 1)[-1]


def tags_for(element: ET.Element) -> dict[str, str]:
    result: dict[str, str] = {}
    for child in element:
        if local_name(child.tag) != "tag":
            continue
        key = child.attrib.get("k", "").strip()
        value = child.attrib.get("v", "").strip()
        if key and value:
            result[key] = value
    return result


def coordinate_center(points: Sequence[tuple[float, float]]) -> tuple[float, float] | None:
    if not points:
        return None
    latitudes = [point[0] for point in points if math.isfinite(point[0])]
    longitudes = [point[1] for point in points if math.isfinite(point[1])]
    if len(latitudes) != len(points) or len(longitudes) != len(points):
        return None
    return ((min(latitudes) + max(latitudes)) / 2, (min(longitudes) + max(longitudes)) / 2)


def parse_osm_xml(path: Path) -> Iterator[RawFeature]:
    opener = gzip.open if path.name.lower().endswith(".gz") else open
    try:
        with opener(path, "rb") as handle:
            root = ET.parse(handle).getroot()
    except (OSError, ET.ParseError) as error:
        raise BuildError(f"cannot parse OSM XML {path}: {error}") from error

    nodes: dict[int, tuple[float, float]] = {}
    ways: dict[int, list[int]] = {}
    node_elements: list[ET.Element] = []
    way_elements: list[ET.Element] = []
    relation_elements: list[ET.Element] = []
    for element in root:
        kind = local_name(element.tag)
        try:
            osm_id = int(element.attrib["id"])
        except (KeyError, ValueError):
            continue
        if kind == "node":
            try:
                nodes[osm_id] = (float(element.attrib["lat"]), float(element.attrib["lon"]))
            except (KeyError, ValueError):
                continue
            node_elements.append(element)
        elif kind == "way":
            ways[osm_id] = [
                int(child.attrib["ref"])
                for child in element
                if local_name(child.tag) == "nd" and child.attrib.get("ref", "").lstrip("-").isdigit()
            ]
            way_elements.append(element)
        elif kind == "relation":
            relation_elements.append(element)

    for element in node_elements:
        osm_id = int(element.attrib["id"])
        tags = tags_for(element)
        if tags:
            lat, lon = nodes[osm_id]
            yield RawFeature("node", osm_id, lat, lon, tags)

    for element in way_elements:
        osm_id = int(element.attrib["id"])
        tags = tags_for(element)
        center = coordinate_center([nodes[ref] for ref in ways.get(osm_id, []) if ref in nodes])
        if tags and center is not None:
            yield RawFeature("way", osm_id, center[0], center[1], tags)

    for element in relation_elements:
        try:
            osm_id = int(element.attrib["id"])
        except (KeyError, ValueError):
            continue
        tags = tags_for(element)
        points: list[tuple[float, float]] = []
        for member in element:
            if local_name(member.tag) != "member":
                continue
            try:
                ref = int(member.attrib["ref"])
            except (KeyError, ValueError):
                continue
            if member.attrib.get("type") == "node" and ref in nodes:
                points.append(nodes[ref])
            elif member.attrib.get("type") == "way":
                points.extend(nodes[node_ref] for node_ref in ways.get(ref, []) if node_ref in nodes)
        center = coordinate_center(points)
        if tags and center is not None:
            yield RawFeature("relation", osm_id, center[0], center[1], tags)


def flatten_geojson_coordinates(value: Any) -> Iterator[tuple[float, float]]:
    if not isinstance(value, list):
        return
    if len(value) >= 2 and all(isinstance(item, (int, float)) for item in value[:2]):
        yield (float(value[1]), float(value[0]))
        return
    for child in value:
        yield from flatten_geojson_coordinates(child)


def parse_osm_identity(feature: Mapping[str, Any], properties: Mapping[str, Any]) -> tuple[str, int] | None:
    candidates = (feature.get("id"), properties.get("@id"), properties.get("id"))
    for candidate in candidates:
        if isinstance(candidate, int):
            osm_type = str(properties.get("@type", properties.get("osm_type", "node"))).lower()
            if osm_type in {"node", "way", "relation"}:
                return osm_type, candidate
        if not isinstance(candidate, str):
            continue
        match = re.fullmatch(r"(node|way|relation)[/:](-?\d+)", candidate.lower())
        if match:
            return match.group(1), int(match.group(2))
    return None


def parse_geojson(path: Path) -> Iterator[RawFeature]:
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise BuildError(f"cannot parse GeoJSON {path}: {error}") from error
    if isinstance(raw, dict) and raw.get("type") == "FeatureCollection":
        features = raw.get("features", [])
    elif isinstance(raw, list):
        features = raw
    else:
        raise BuildError(f"{path}: expected a GeoJSON FeatureCollection")
    if not isinstance(features, list):
        raise BuildError(f"{path}: features must be an array")

    for feature in features:
        if not isinstance(feature, dict) or feature.get("type") != "Feature":
            continue
        properties = feature.get("properties")
        geometry = feature.get("geometry")
        if not isinstance(properties, dict) or not isinstance(geometry, dict):
            continue
        identity = parse_osm_identity(feature, properties)
        center = coordinate_center(list(flatten_geojson_coordinates(geometry.get("coordinates"))))
        if identity is None or center is None:
            continue
        tags = {
            str(key): str(value)
            for key, value in properties.items()
            if not str(key).startswith("@") and value is not None and isinstance(value, (str, int, float, bool))
        }
        yield RawFeature(identity[0], identity[1], center[0], center[1], tags)


def split_tag_values(value: str | None) -> set[str]:
    if value is None:
        return set()
    return {part.strip().lower() for part in re.split(r"[;,]", value) if part.strip()}


def activities_for_tags(tags: Mapping[str, str]) -> list[str]:
    leisure = tags.get("leisure", "").lower()
    tourism = tags.get("tourism", "").lower()
    natural = tags.get("natural", "").lower()
    highway = tags.get("highway", "").lower()
    route = tags.get("route", "").lower()
    boundary = tags.get("boundary", "").lower()
    sports = split_tag_values(tags.get("sport"))
    named_path = bool((tags.get("name") or tags.get("official_name") or "").strip())
    activities: set[str] = set()

    if leisure == "playground":
        activities.add("playground")
    if leisure == "park":
        activities.add("park")
    if route == "foot" or (highway in {"footway", "path"} and named_path):
        activities.add("walking")
    if route == "hiking" or highway == "trailhead" or leisure == "nature_reserve":
        activities.add("hiking")
    if leisure == "disc_golf_course":
        activities.add("discGolf")
    if leisure == "dog_park":
        activities.add("dogPark")
    if route == "bicycle":
        activities.add("bike")
    if leisure == "pitch" and sports.intersection(SUPPORTED_COURT_SPORTS):
        activities.add("court")
    if leisure == "skatepark":
        activities.add("skate")
    if tourism == "picnic_site":
        activities.add("picnic")
    if natural == "beach" or leisure == "slipway":
        activities.add("water")
    if leisure == "fishing":
        activities.add("fishing")
    if tourism == "viewpoint":
        activities.add("viewpoint")
    if leisure == "nature_reserve" or boundary == "protected_area":
        activities.add("birding")
    if leisure == "golf_course":
        activities.add("golf")

    return sorted(activities, key=ACTIVITY_INDEX.__getitem__)


def normalized_boolean(value: str | None) -> tuple[bool, str]:
    normalized = (value or "").strip().lower()
    if normalized in TRUE_VALUES:
        return True, "yes"
    if normalized in FALSE_VALUES:
        return False, "no"
    return False, "unknown"


def normalized_name(tags: Mapping[str, str]) -> str | None:
    for key in ("name", "official_name", "short_name"):
        value = tags.get(key, "").strip()
        if value:
            return value
    return None


def normalized_address(tags: Mapping[str, str]) -> str | None:
    street = " ".join(
        part for part in (tags.get("addr:housenumber", "").strip(), tags.get("addr:street", "").strip()) if part
    )
    parts = [
        street,
        tags.get("addr:city", "").strip(),
        tags.get("addr:state", "").strip(),
        tags.get("addr:postcode", "").strip(),
    ]
    compact = ", ".join(part for part in parts if part)
    return compact or None


def normalize_feature(feature: RawFeature) -> dict[str, Any] | None:
    if not (-90 <= feature.latitude <= 90 and -180 <= feature.longitude <= 180):
        return None
    if feature.tags.get("access", "").strip().lower() in PRIVATE_ACCESS:
        return None
    activities = activities_for_tags(feature.tags)
    if not activities:
        return None
    lit, lit_tag = normalized_boolean(feature.tags.get("lit"))
    fee_value = feature.tags.get("fee", "").strip().lower()
    fee = "yes" if fee_value in TRUE_VALUES else "no" if fee_value in FALSE_VALUES else "unknown"
    access = feature.tags.get("access", "").strip().lower() or "unknown"
    selector_tags = {
        key: feature.tags[key]
        for key in ("leisure", "tourism", "natural", "highway", "route", "boundary", "sport")
        if feature.tags.get(key)
    }
    return {
        "id": f"osm:{feature.osm_type}:{feature.osm_id}",
        "name": normalized_name(feature.tags),
        "latitude": round(feature.latitude, 6),
        "longitude": round(feature.longitude, 6),
        "activityIds": activities,
        "lit": lit,
        "litTag": lit_tag,
        "access": access,
        "fee": fee,
        "address": normalized_address(feature.tags),
        "osm": {"type": feature.osm_type, "id": feature.osm_id},
        "tags": selector_tags,
    }


def copy_osmium_tags(tags: Any) -> dict[str, str]:
    """Copy map tags while deliberately excluding OSM edit/contributor metadata."""

    return {str(tag.k): str(tag.v) for tag in tags}


def bounds_from_osmium_way(way: Any) -> tuple[float, float, float, float] | None:
    """Compute a way bounding box while its pyosmium callback view is valid."""

    minimum_latitude = math.inf
    minimum_longitude = math.inf
    maximum_latitude = -math.inf
    maximum_longitude = -math.inf
    found = False
    for node in way.nodes:
        try:
            if not node.location.valid():
                continue
            latitude = float(node.lat)
            longitude = float(node.lon)
        except (AttributeError, RuntimeError, ValueError):
            continue
        if not (math.isfinite(latitude) and math.isfinite(longitude)):
            continue
        minimum_latitude = min(minimum_latitude, latitude)
        minimum_longitude = min(minimum_longitude, longitude)
        maximum_latitude = max(maximum_latitude, latitude)
        maximum_longitude = max(maximum_longitude, longitude)
        found = True
    if not found:
        return None
    return minimum_latitude, minimum_longitude, maximum_latitude, maximum_longitude


def merge_bounds(
    bounds: Iterable[tuple[float, float, float, float]],
) -> tuple[float, float, float, float] | None:
    collected = list(bounds)
    if not collected:
        return None
    return (
        min(item[0] for item in collected),
        min(item[1] for item in collected),
        max(item[2] for item in collected),
        max(item[3] for item in collected),
    )


def center_from_bounds(bounds: tuple[float, float, float, float]) -> tuple[float, float]:
    return ((bounds[0] + bounds[2]) / 2, (bounds[1] + bounds[3]) / 2)


def stream_osm_pbf_in_process(
    path: Path,
    temp_dir: Path,
    consume: Callable[[RawFeature], None],
    cancel_event: threading.Event | None = None,
) -> None:
    """Stream a PBF using pyosmium and a disk-backed node-location index.

    A lightweight first pass records only qualifying relation tags and their
    member IDs. The second pass emits nodes and ways, retaining bounds only for
    members needed to place those relations. No filtered PBF/GeoJSON is made,
    and uid/user/changeset/version/timestamp fields are never copied.
    """

    try:
        import osmium  # type: ignore[import-not-found]
    except ImportError as error:
        raise BuildError(
            f"{path} is PBF; install the pinned PyPI osmium dependency from "
            "tools/osm-place-packs/requirements-pbf.txt"
        ) from error

    relation_specs: dict[int, tuple[dict[str, str], tuple[tuple[str, int], ...]]] = {}
    needed_node_ids: set[int] = set()
    needed_way_ids: set[int] = set()

    def raise_if_cancelled() -> None:
        if cancel_event is not None and cancel_event.is_set():
            raise BuildError("PBF extraction cancelled after another extract failed")

    class RelationPass(osmium.SimpleHandler):
        def relation(self, relation: Any) -> None:
            raise_if_cancelled()
            if not activities_for_tags(relation.tags):
                return
            members = tuple(
                (str(member.type), int(member.ref))
                for member in relation.members
                if member.type in {"n", "w"}
            )
            if not members:
                return
            relation_id = int(relation.id)
            relation_specs[relation_id] = (copy_osmium_tags(relation.tags), members)
            needed_node_ids.update(member_id for member_type, member_id in members if member_type == "n")
            needed_way_ids.update(member_id for member_type, member_id in members if member_type == "w")

    member_node_bounds: dict[int, tuple[float, float, float, float]] = {}
    member_way_bounds: dict[int, tuple[float, float, float, float]] = {}

    class FeaturePass(osmium.SimpleHandler):
        def node(self, node: Any) -> None:
            raise_if_cancelled()
            node_id = int(node.id)
            selected = bool(activities_for_tags(node.tags))
            if not selected and node_id not in needed_node_ids:
                return
            try:
                if not node.location.valid():
                    return
                latitude = float(node.lat)
                longitude = float(node.lon)
            except (AttributeError, RuntimeError, ValueError):
                return
            if node_id in needed_node_ids:
                member_node_bounds[node_id] = (latitude, longitude, latitude, longitude)
            if selected:
                consume(RawFeature("node", node_id, latitude, longitude, copy_osmium_tags(node.tags)))

        def way(self, way: Any) -> None:
            raise_if_cancelled()
            way_id = int(way.id)
            selected = bool(activities_for_tags(way.tags))
            if not selected and way_id not in needed_way_ids:
                return
            bounds = bounds_from_osmium_way(way)
            if bounds is None:
                return
            if way_id in needed_way_ids:
                member_way_bounds[way_id] = bounds
            if selected:
                latitude, longitude = center_from_bounds(bounds)
                consume(RawFeature("way", way_id, latitude, longitude, copy_osmium_tags(way.tags)))

    location_index = (temp_dir / "node-locations.idx").resolve().as_posix()
    location_table: Any | None = None
    location_handler: Any | None = None
    feature_handler: Any | None = None
    try:
        relation_handler = RelationPass()
        relation_handler.apply_file(path, locations=False)
        del relation_handler
        feature_handler = FeaturePass()
        location_table = osmium.index.create_map(f"sparse_file_array,{location_index}")
        location_handler = osmium.NodeLocationsForWays(location_table)
        location_handler.ignore_errors()
        with osmium.io.Reader(path) as reader:
            osmium.apply(reader, location_handler, feature_handler)
        # Force finalization here so Windows releases the index before the
        # enclosing TemporaryDirectory attempts cleanup.
        del feature_handler
        del location_handler
        del location_table
        gc.collect()
    except Exception as error:
        raise BuildError(f"pyosmium failed while streaming {path}: {error}") from error

    for relation_id in sorted(relation_specs):
        tags, members = relation_specs[relation_id]
        relation_bounds = merge_bounds(
            member_node_bounds[member_id]
            if member_type == "n"
            else member_way_bounds[member_id]
            for member_type, member_id in members
            if (member_type == "n" and member_id in member_node_bounds)
            or (member_type == "w" and member_id in member_way_bounds)
        )
        if relation_bounds is None:
            continue
        latitude, longitude = center_from_bounds(relation_bounds)
        consume(RawFeature("relation", relation_id, latitude, longitude, tags))


PBF_WORKER_FLAG = "--internal-stream-pbf-worker"


def raw_feature_to_wire(feature: RawFeature) -> dict[str, Any]:
    return {
        "osmType": feature.osm_type,
        "osmId": feature.osm_id,
        "latitude": feature.latitude,
        "longitude": feature.longitude,
        "tags": feature.tags,
    }


def raw_feature_from_wire(value: Any) -> RawFeature:
    if not isinstance(value, dict):
        raise BuildError("PBF worker emitted a non-object feature")
    osm_type = value.get("osmType")
    osm_id = value.get("osmId")
    latitude = value.get("latitude")
    longitude = value.get("longitude")
    tags = value.get("tags")
    if (
        osm_type not in {"node", "way", "relation"}
        or not isinstance(osm_id, int)
        or isinstance(osm_id, bool)
        or not isinstance(latitude, (int, float))
        or isinstance(latitude, bool)
        or not isinstance(longitude, (int, float))
        or isinstance(longitude, bool)
        or not isinstance(tags, dict)
        or not all(isinstance(key, str) and isinstance(item, str) for key, item in tags.items())
    ):
        raise BuildError("PBF worker emitted an invalid feature")
    return RawFeature(osm_type, osm_id, float(latitude), float(longitude), tags)


def pbf_worker_main(argv: Sequence[str]) -> int:
    """Private subprocess entry point used to release Windows index handles."""

    # Windows pipe streams otherwise inherit a locale code page and fail on
    # perfectly valid non-ASCII OSM names. The parent always decodes UTF-8.
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8", errors="strict", newline="\n")
    if hasattr(sys.stderr, "reconfigure"):
        sys.stderr.reconfigure(encoding="utf-8", errors="backslashreplace", newline="\n")
    if len(argv) != 2:
        print("PBF worker expected input path and temporary directory", file=sys.stderr)
        return 2
    source_path = Path(argv[0]).resolve()
    temp_dir = Path(argv[1]).resolve()
    batch: list[dict[str, Any]] = []

    def flush_batch() -> None:
        if not batch:
            return
        sys.stdout.write(
            json.dumps(batch, ensure_ascii=False, separators=(",", ":"), allow_nan=False) + "\n"
        )
        sys.stdout.flush()
        batch.clear()

    def emit(feature: RawFeature) -> None:
        batch.append(raw_feature_to_wire(feature))
        if len(batch) >= 256:
            flush_batch()

    try:
        stream_osm_pbf_in_process(source_path, temp_dir, emit)
        flush_batch()
    except (BuildError, OSError, ValueError) as error:
        print(f"PBF worker failed: {error}", file=sys.stderr)
        return 2
    return 0


def stream_osm_pbf(
    path: Path,
    temp_dir: Path,
    consume: Callable[[RawFeature], None],
    cancel_event: threading.Event | None = None,
) -> None:
    # Process isolation permits true parallel PBF scans and ensures the
    # file-array index is closed before the parent removes its temp directory.
    # This is essential on Windows, where the handle lasts for the interpreter.
    if cancel_event is not None and cancel_event.is_set():
        raise BuildError("PBF extraction cancelled after another extract failed")
    command = [
        sys.executable,
        str(Path(__file__).resolve()),
        PBF_WORKER_FLAG,
        str(path.resolve()),
        str(temp_dir.resolve()),
    ]
    process = subprocess.Popen(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        encoding="utf-8",
        bufsize=1,
    )
    assert process.stdout is not None
    assert process.stderr is not None

    # Drain stderr concurrently so a verbose native failure cannot fill its
    # pipe and deadlock a worker whose stdout is being consumed by this thread.
    error_parts: list[str] = []
    captured_error_characters = 0
    error_capture_limit = 64 * 1024

    def drain_stderr() -> None:
        nonlocal captured_error_characters
        for chunk in iter(lambda: process.stderr.read(4096), ""):
            remaining = error_capture_limit - captured_error_characters
            if remaining > 0:
                kept = chunk[:remaining]
                error_parts.append(kept)
                captured_error_characters += len(kept)

    stderr_thread = threading.Thread(target=drain_stderr, name="solarma-pbf-stderr", daemon=True)
    stderr_thread.start()

    cancel_watch_stop = threading.Event()

    def watch_for_cancellation() -> None:
        while not cancel_watch_stop.wait(0.25):
            if cancel_event is not None and cancel_event.is_set():
                if process.poll() is None:
                    try:
                        process.terminate()
                    except OSError:
                        pass
                return

    cancel_thread = threading.Thread(
        target=watch_for_cancellation,
        name="solarma-pbf-cancel",
        daemon=True,
    )
    cancel_thread.start()
    return_code: int | None = None
    try:
        for line in process.stdout:
            if cancel_event is not None and cancel_event.is_set():
                raise BuildError("PBF extraction cancelled after another extract failed")
            try:
                batch = json.loads(line)
            except json.JSONDecodeError as error:
                raise BuildError(f"PBF worker emitted invalid JSON: {error}") from error
            if not isinstance(batch, list):
                raise BuildError("PBF worker emitted a non-array batch")
            for value in batch:
                consume(raw_feature_from_wire(value))
        try:
            return_code = process.wait(timeout=30)
        except subprocess.TimeoutExpired as error:
            raise BuildError("PBF worker closed stdout but did not exit") from error
    finally:
        if process.poll() is None:
            try:
                process.terminate()
            except OSError:
                pass
            try:
                process.wait(timeout=10)
            except subprocess.TimeoutExpired:
                process.kill()
                process.wait()
        cancel_watch_stop.set()
        cancel_thread.join()
        stderr_thread.join()
        process.stdout.close()
        process.stderr.close()
    error_output = "".join(error_parts).strip()
    if cancel_event is not None and cancel_event.is_set():
        raise BuildError("PBF extraction cancelled after another extract failed")
    if return_code != 0:
        detail = f": {error_output}" if error_output else ""
        raise BuildError(f"PBF worker exited with status {return_code}{detail}")


def consume_extract(
    path: Path,
    temp_dir: Path,
    consume: Callable[[RawFeature], None],
    cancel_event: threading.Event | None = None,
) -> None:
    lower = path.name.lower()
    if lower.endswith((".osm", ".osm.xml", ".osm.gz", ".xml", ".xml.gz")):
        features = parse_osm_xml(path)
    elif lower.endswith((".geojson", ".json")):
        features = parse_geojson(path)
    elif lower.endswith((".pbf", ".osm.pbf")):
        stream_osm_pbf(path, temp_dir, consume, cancel_event)
        return
    else:
        raise BuildError(f"unsupported extract format: {path}")
    for feature in features:
        if cancel_event is not None and cancel_event.is_set():
            raise BuildError("extract processing cancelled after another extract failed")
        consume(feature)


def encode_geohash(latitude: float, longitude: float, precision: int) -> str:
    if not math.isfinite(latitude) or not math.isfinite(longitude):
        raise BuildError("cannot geohash non-finite coordinates")
    if not 1 <= precision <= 12:
        raise BuildError("geohash precision must be from 1 through 12")
    latitude = min(90.0, max(-90.0, latitude))
    longitude = min(180.0, max(-180.0, longitude))
    lat_interval = [-90.0, 90.0]
    lon_interval = [-180.0, 180.0]
    even = True
    bit = 0
    value = 0
    chars: list[str] = []
    while len(chars) < precision:
        interval = lon_interval if even else lat_interval
        coordinate = longitude if even else latitude
        midpoint = (interval[0] + interval[1]) / 2
        if coordinate >= midpoint:
            value |= GEOHASH_BITS[bit]
            interval[0] = midpoint
        else:
            interval[1] = midpoint
        even = not even
        if bit < 4:
            bit += 1
        else:
            chars.append(GEOHASH_ALPHABET[value])
            bit = 0
            value = 0
    return "".join(chars)


def merge_places(existing: Mapping[str, Any], incoming: Mapping[str, Any]) -> dict[str, Any]:
    """Deterministically merge a boundary duplicate from overlapping extracts."""

    choices = sorted((dict(existing), dict(incoming)), key=canonical_json_bytes)
    winner = choices[0]
    activities = set(existing["activityIds"]) | set(incoming["activityIds"])
    winner["activityIds"] = sorted(activities, key=ACTIVITY_INDEX.__getitem__)
    if not winner.get("name"):
        names = sorted(
            name
            for name in {existing.get("name"), incoming.get("name")}
            if isinstance(name, str) and name
        )
        winner["name"] = names[0] if names else None
    return winner


def collect_extract_places(
    extract: SourceExtract,
    extract_temp: Path,
    cancel_event: threading.Event,
) -> dict[str, dict[str, Any]]:
    local_by_id: dict[str, dict[str, Any]] = {}

    def consume(raw: RawFeature) -> None:
        if cancel_event.is_set():
            raise BuildError("extract processing cancelled after another extract failed")
        place = normalize_feature(raw)
        if place is None:
            return
        prior = local_by_id.get(place["id"])
        local_by_id[place["id"]] = place if prior is None else merge_places(prior, place)

    try:
        consume_extract(extract.path, extract_temp, consume, cancel_event)
    except Exception:
        cancel_event.set()
        raise
    return local_by_id


def collect_places(
    manifest: SourceManifest,
    temp_root: Path | None = None,
    jobs: int = 1,
) -> list[dict[str, Any]]:
    if isinstance(jobs, bool) or not isinstance(jobs, int) or not 1 <= jobs <= MAX_BUILD_JOBS:
        raise BuildError(f"jobs must be an integer from 1 through {MAX_BUILD_JOBS}")
    by_id: dict[str, dict[str, Any]] = {}

    if temp_root is not None and (not temp_root.exists() or not temp_root.is_dir()):
        raise BuildError(f"temporary root must be an existing directory: {temp_root}")
    with tempfile.TemporaryDirectory(prefix="solarma-osm-", dir=temp_root) as temp_name:
        build_temp = Path(temp_name)
        extracts = sorted(manifest.extracts, key=lambda item: str(item.path))
        cancel_event = threading.Event()
        futures: list[Future[dict[str, dict[str, Any]]]] = []
        with ThreadPoolExecutor(
            max_workers=min(jobs, len(extracts)),
            thread_name_prefix="solarma-osm-extract",
        ) as executor:
            future_indexes: dict[Future[dict[str, dict[str, Any]]], int] = {}
            for index, extract in enumerate(extracts):
                extract_temp = build_temp / str(index)
                extract_temp.mkdir()
                future = executor.submit(collect_extract_places, extract, extract_temp, cancel_event)
                futures.append(future)
                future_indexes[future] = index
            ordered_results: list[dict[str, dict[str, Any]] | None] = [None] * len(extracts)
            try:
                for future in as_completed(futures):
                    ordered_results[future_indexes[future]] = future.result()
            except BaseException:
                cancel_event.set()
                for future in futures:
                    future.cancel()
                raise
            for local_by_id in ordered_results:
                assert local_by_id is not None
                for place_id in sorted(local_by_id):
                    place = local_by_id[place_id]
                    prior = by_id.get(place_id)
                    by_id[place_id] = place if prior is None else merge_places(prior, place)
    return sorted(by_id.values(), key=lambda place: place["id"])


def build_place_packs(
    source_manifest_path: Path,
    output_dir: Path,
    temp_root: Path | None = None,
    jobs: int = 1,
) -> dict[str, Any]:
    manifest = read_source_manifest(source_manifest_path.resolve())
    if output_dir.exists():
        raise BuildError(f"output directory already exists: {output_dir}")
    output_dir.mkdir(parents=True)
    packs_dir = output_dir / "packs"
    packs_dir.mkdir()

    places = collect_places(manifest, temp_root, jobs)
    grouped: dict[str, list[dict[str, Any]]] = {}
    for place in places:
        geohash = encode_geohash(place["latitude"], place["longitude"], manifest.geohash_precision)
        grouped.setdefault(geohash, []).append(place)

    descriptors: dict[str, Any] = {}
    deployment_headers: dict[str, dict[str, str]] = {
        "manifest.json": {
            "Cache-Control": "public, max-age=86400, must-revalidate",
            "Content-Type": "application/json; charset=utf-8",
        },
        "attribution.json": {
            "Cache-Control": "public, max-age=7776000",
            "Content-Type": "application/json; charset=utf-8",
        },
    }
    for geohash in sorted(grouped):
        pack = {
            "schemaVersion": SCHEMA_VERSION,
            "datasetVersion": manifest.dataset_version,
            "quarter": manifest.quarter,
            "sourceTimestamp": manifest.source_timestamp,
            "geohash": geohash,
            "attribution": ATTRIBUTION,
            "places": grouped[geohash],
        }
        raw = canonical_json_bytes(pack)
        compressed = deterministic_gzip(raw)
        relative = f"packs/{geohash}.json.gz"
        (output_dir / relative).write_bytes(compressed)
        digest = sha256_bytes(compressed)
        etag = f'"sha256-{digest}"'
        descriptors[geohash] = {
            "path": relative,
            "count": len(grouped[geohash]),
            "contentEncoding": "gzip",
            "contentType": "application/json; charset=utf-8",
            "sha256": digest,
            "etag": etag,
            "compressedBytes": len(compressed),
            "uncompressedBytes": len(raw),
        }
        deployment_headers[relative] = {
            "Cache-Control": "public, max-age=7776000, immutable",
            "Content-Encoding": "gzip",
            "Content-Type": "application/json; charset=utf-8",
            "ETag": etag,
        }

    sources = [
        {"url": extract.url, "sha256": extract.sha256, "fileName": extract.path.name}
        for extract in sorted(manifest.extracts, key=lambda item: (item.url, item.sha256))
    ]
    output_manifest = {
        "schemaVersion": SCHEMA_VERSION,
        "dataset": "solarma-osm-outing-places",
        "datasetVersion": manifest.dataset_version,
        "quarter": manifest.quarter,
        "sourceTimestamp": manifest.source_timestamp,
        "geohashPrecision": manifest.geohash_precision,
        "cacheMaxAgeSeconds": 90 * 24 * 60 * 60,
        "builderVersion": BUILDER_VERSION,
        "attribution": ATTRIBUTION,
        "sources": sources,
        "placeCount": len(places),
        "packs": descriptors,
    }
    (output_dir / "manifest.json").write_bytes(canonical_json_bytes(output_manifest))
    (output_dir / "attribution.json").write_bytes(canonical_json_bytes(ATTRIBUTION))
    (output_dir / "deploy-headers.json").write_bytes(
        canonical_json_bytes({"schemaVersion": SCHEMA_VERSION, "headers": deployment_headers})
    )
    return output_manifest


def parse_args(argv: Sequence[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--source-manifest",
        type=Path,
        required=True,
        help="checksum-pinned quarterly source manifest",
    )
    parser.add_argument(
        "--output",
        type=Path,
        required=True,
        help="new output directory; the builder refuses to overwrite",
    )
    parser.add_argument(
        "--temp-root",
        type=Path,
        help="existing directory for the temporary disk-backed PBF node index",
    )
    parser.add_argument(
        "--jobs",
        type=int,
        default=1,
        choices=range(1, MAX_BUILD_JOBS + 1),
        metavar=f"1-{MAX_BUILD_JOBS}",
        help="independent extracts to process concurrently (production target: 4)",
    )
    return parser.parse_args(argv)


def main(argv: Sequence[str] | None = None) -> int:
    arguments = list(argv if argv is not None else sys.argv[1:])
    if arguments[:1] == [PBF_WORKER_FLAG]:
        return pbf_worker_main(arguments[1:])
    args = parse_args(arguments)
    try:
        manifest = build_place_packs(args.source_manifest, args.output, args.temp_root, args.jobs)
    except (BuildError, OSError) as error:
        print(f"place-pack build failed: {error}", file=sys.stderr)
        return 2
    print(
        f"built {manifest['placeCount']} places in {len(manifest['packs'])} packs "
        f"for {manifest['datasetVersion']} at {args.output}"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())