#!/usr/bin/env python3
"""Build a compact, file-backed fairway dataset from Geofabrik OSM PBFs.

The importer intentionally uses two streaming passes over every source. The
first pass retains only routable marine/inland ways and their node IDs; the
second retains only coordinates referenced by those ways. Coordinates and
ways shared by neighbouring extracts are merged by their globally unique OSM
IDs. This avoids loading complete OSM node tables into memory and does not
require PostGIS or Docker.
"""

from __future__ import annotations

import argparse
from collections import defaultdict
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path
import re
import sys
import tempfile
from typing import Any

try:
    import osmium
except ImportError as error:
    raise SystemExit(
        "pyosmium fehlt. Installiere es mit "
        "`python3 -m pip install --target .tools/python 'osmium>=4,<5'` "
        "und starte mit `PYTHONPATH=.tools/python`."
    ) from error


KEPT_TAGS = {
    "access",
    "boat",
    "construction",
    "disused",
    "maxdraft",
    "maxdraught",
    "maxheight",
    "maxheight:physical",
    "maxwidth",
    "maxwidth:physical",
    "motor_vehicle",
    "motorboat",
    "name",
    "oneway",
    "proposed",
    "ref",
    "route",
    "seamark:bridge:clearance_height",
    "seamark:bridge:clearance_height_safe",
    "seamark:lock:chamber_width",
    "seamark:restriction:max_draught",
    "seamark:type",
    "ship",
    "waterway",
}
GENERATOR_VERSION = 2


def is_routable(tags: dict[str, str]) -> bool:
    if (
        tags.get("access") in {"no", "private"}
        or tags.get("boat") == "no"
        or tags.get("ship") == "no"
        or tags.get("motorboat") == "no"
        or tags.get("disused") == "yes"
        or "construction" in tags
        or "proposed" in tags
    ):
        return False

    seamark_type = tags.get("seamark:type")
    if seamark_type in {"navigation_line", "recommended_track"}:
        return True
    if tags.get("waterway") in {"fairway", "canal"}:
        return True
    if tags.get("waterway") == "river" and any(
        tags.get(key) in {"yes", "designated", "permissive"}
        for key in ("boat", "ship", "motorboat")
    ):
        return True
    if (
        tags.get("route") == "ferry"
        and tags.get("ship") != "no"
        and tags.get("motor_vehicle") != "no"
    ):
        return True
    return seamark_type == "fairway"


class WayCollector(osmium.SimpleHandler):
    def __init__(self) -> None:
        super().__init__()
        self.ways: list[dict[str, Any]] = []
        self.node_ids: set[int] = set()

    def way(self, way: Any) -> None:
        tags = {tag.k: tag.v for tag in way.tags}
        if not is_routable(tags):
            return

        node_ids = [node.ref for node in way.nodes]
        if len(node_ids) < 2:
            return

        self.node_ids.update(node_ids)
        self.ways.append(
            {
                "id": str(way.id),
                "version": int(getattr(way, "version", 0) or 0),
                "nodes": node_ids,
                "tags": {key: value for key, value in tags.items() if key in KEPT_TAGS},
            }
        )


class NodeCollector(osmium.SimpleHandler):
    def __init__(self, wanted: set[int]) -> None:
        super().__init__()
        self.wanted = wanted
        self.coordinates: dict[int, tuple[int, float, float]] = {}

    def node(self, node: Any) -> None:
        if node.id not in self.wanted or not node.location.valid():
            return
        candidate = (
            int(getattr(node, "version", 0) or 0),
            node.location.lat,
            node.location.lon,
        )
        current = self.coordinates.get(node.id)
        if current is None or candidate > current:
            self.coordinates[node.id] = candidate


def way_bbox(coordinates: list[list[float]]) -> list[float]:
    lats = [coordinate[0] for coordinate in coordinates]
    lons = [coordinate[1] for coordinate in coordinates]
    return [min(lons), min(lats), max(lons), max(lats)]


def source_region(path: Path) -> str:
    name = path.name.lower()
    for suffix in ("-latest.osm.pbf", ".osm.pbf", ".pbf"):
        if name.endswith(suffix):
            name = name[: -len(suffix)]
            break
    region = re.sub(r"[^a-z0-9]+", "-", name).strip("-")
    return region or "unknown"


def adjacent_md5(path: Path) -> str | None:
    checksum_path = Path(f"{path}.md5")
    if not checksum_path.is_file():
        return None
    fields = checksum_path.read_text(encoding="utf-8").split(maxsplit=1)
    if not fields:
        return None
    checksum = fields[0].lower()
    return checksum if re.fullmatch(r"[0-9a-f]{32}", checksum) else None


def candidate_rank(candidate: dict[str, Any]) -> tuple[int, int, str]:
    """Return a deterministic preference for duplicate versions of an OSM way."""

    fingerprint = hashlib.sha256(
        json.dumps(
            {
                "nodes": candidate["nodes"],
                "tags": candidate["tags"],
                "coordinates": candidate["coordinates"],
            },
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
        ).encode("utf-8")
    ).hexdigest()
    return candidate["version"], len(candidate["nodes"]), fingerprint


def build_document(pbf_paths: list[Path]) -> dict[str, Any]:
    sources: list[dict[str, Any]] = []
    wanted_node_ids: set[int] = set()

    for index, path in enumerate(pbf_paths, start=1):
        print(
            f"Pass 1/2 [{index}/{len(pbf_paths)}]: routbare Wege aus {path} lesen",
            flush=True,
        )
        collector = WayCollector()
        processor = osmium.FileProcessor(
            str(path), osmium.osm.WAY
        ).with_filter(osmium.filter.KeyFilter("seamark:type", "waterway", "route"))
        for way in processor:
            collector.way(way)
        wanted_node_ids.update(collector.node_ids)
        stat = path.stat()
        sources.append(
            {
                "path": path,
                "region": source_region(path),
                "collector": collector,
                "metadata": {
                    "region": source_region(path),
                    "file": path.name,
                    "sizeBytes": stat.st_size,
                    "modifiedAt": datetime.fromtimestamp(
                        stat.st_mtime, timezone.utc
                    ).isoformat(),
                    "checksumMd5": adjacent_md5(path),
                    "routableWaysFound": len(collector.ways),
                    "referencedNodes": len(collector.node_ids),
                    "nodeCoordinatesFound": 0,
                    "exportedCandidateWays": 0,
                    "incompleteWaysSkipped": 0,
                    "closedFairwaysSkipped": 0,
                },
            }
        )
        print(
            f"{len(collector.ways)} Wege mit "
            f"{len(collector.node_ids)} referenzierten Knoten gefunden",
            flush=True,
        )

    global_coordinates: dict[int, tuple[int, float, float]] = {}
    coordinate_conflicts = 0
    for index, source in enumerate(sources, start=1):
        print(
            f"Pass 2/2 [{index}/{len(sources)}]: benötigte Knotenkoordinaten "
            f"aus {source['path']} lesen",
            flush=True,
        )
        nodes = NodeCollector(wanted_node_ids)
        if wanted_node_ids:
            processor = osmium.FileProcessor(
                str(source["path"]), osmium.osm.NODE
            ).with_filter(osmium.filter.IdFilter(wanted_node_ids))
            for node in processor:
                nodes.node(node)

        source["metadata"]["nodeCoordinatesFound"] = len(nodes.coordinates)
        for node_id, candidate in nodes.coordinates.items():
            current = global_coordinates.get(node_id)
            if current is not None and current[1:] != candidate[1:]:
                coordinate_conflicts += 1
            if current is None or candidate > current:
                global_coordinates[node_id] = candidate

    candidates_by_id: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for source in sources:
        for way in source["collector"].ways:
            if any(node_id not in global_coordinates for node_id in way["nodes"]):
                source["metadata"]["incompleteWaysSkipped"] += 1
                continue
            coordinates = [
                [global_coordinates[node_id][1], global_coordinates[node_id][2]]
                for node_id in way["nodes"]
            ]
            if (
                way["tags"].get("seamark:type") == "fairway"
                and coordinates[0] == coordinates[-1]
            ):
                source["metadata"]["closedFairwaysSkipped"] += 1
                continue

            source["metadata"]["exportedCandidateWays"] += 1
            candidates_by_id[way["id"]].append(
                {
                    **way,
                    "coordinates": coordinates,
                    "region": source["region"],
                    "sourceFile": source["path"].name,
                }
            )

    exported: list[dict[str, Any]] = []
    duplicate_ways_merged = 0
    for way_id, candidates in candidates_by_id.items():
        winner = max(candidates, key=candidate_rank)
        duplicate_ways_merged += len(candidates) - 1
        exported.append(
            {
                "id": way_id,
                "osmVersion": winner["version"],
                "regions": sorted({candidate["region"] for candidate in candidates}),
                "sourceFiles": sorted(
                    {candidate["sourceFile"] for candidate in candidates}
                ),
                "bbox": way_bbox(winner["coordinates"]),
                "tags": winner["tags"],
                "coordinates": winner["coordinates"],
            }
        )
    exported.sort(key=lambda way: int(way["id"]))

    source_metadata = [source["metadata"] for source in sources]
    regions = list(dict.fromkeys(source["region"] for source in sources))
    source_label = (
        pbf_paths[0].name if len(pbf_paths) == 1 else "+".join(regions)
    )
    modified_at = max(path.stat().st_mtime for path in pbf_paths)
    return {
        "version": 1,
        "generatorVersion": GENERATOR_VERSION,
        "source": source_label,
        "sources": source_metadata,
        "sourceSizeBytes": sum(path.stat().st_size for path in pbf_paths),
        "sourceModifiedAt": datetime.fromtimestamp(
            modified_at, timezone.utc
        ).isoformat(),
        "generatedAt": datetime.now(timezone.utc).isoformat(),
        "incompleteWaysSkipped": sum(
            source["metadata"]["incompleteWaysSkipped"] for source in sources
        ),
        "closedFairwaysSkipped": sum(
            source["metadata"]["closedFairwaysSkipped"] for source in sources
        ),
        "duplicateWaysMerged": duplicate_ways_merged,
        "nodeCoordinateConflicts": coordinate_conflicts,
        "ways": exported,
    }


def write_document(document: dict[str, Any], output: Path) -> None:
    output.parent.mkdir(parents=True, exist_ok=True)
    temporary_path: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            "w",
            encoding="utf-8",
            dir=output.parent,
            prefix=f".{output.name}.",
            suffix=".tmp",
            delete=False,
        ) as temporary:
            json.dump(document, temporary, ensure_ascii=False, separators=(",", ":"))
            temporary.write("\n")
            temporary_path = Path(temporary.name)
        temporary_path.chmod(0o644)
        os.replace(temporary_path, output)
    except BaseException:
        if temporary_path is not None:
            temporary_path.unlink(missing_ok=True)
        raise


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("pbf", type=Path, nargs="+")
    parser.add_argument(
        "--output",
        type=Path,
        default=Path("data/local/germany-netherlands-fairways.json"),
    )
    args = parser.parse_args()

    pbf_paths: list[Path] = []
    seen_paths: set[Path] = set()
    for path in args.pbf:
        if not path.is_file():
            parser.error(f"PBF nicht gefunden: {path}")
        resolved = path.resolve()
        if resolved in seen_paths:
            parser.error(f"PBF doppelt angegeben: {path}")
        seen_paths.add(resolved)
        pbf_paths.append(path)

    document = build_document(pbf_paths)
    write_document(document, args.output)
    print(
        f"{len(document['ways'])} Wege nach {args.output} geschrieben "
        f"({document['duplicateWaysMerged']} Duplikate zusammengeführt; "
        f"{document['incompleteWaysSkipped']} unvollständige Wege verworfen; "
        f"{args.output.stat().st_size / 1024 / 1024:.1f} MiB)",
        flush=True,
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())