src/graph/build_reach_graph.py

"""
Reach-based river network graph construction.

Builds the graph from BD TOPO's own tronçon-to-tronçon connectivity
(`lien_vers_noeud_hydrographique_ini` / `_fin`), not from inferring station
order along a single extracted line. This is the NEXT_DOWN-equivalent
approach: each tronçon is a directed edge between two real hydrographic
node IDs, so confluences (multiple tronçons converging on one node) and
branching fall directly out of the data rather than needing to be modeled
separately.

Flow direction, per tronçon, comes from `sens_de_l_ecoulement` where that
field is usably populated, falling back to comparing the Z (altitude)
coordinate at each tronçon's endpoints — higher Z is upstream — since BD
TOPO geometry already carries real altitude at every vertex (see
BDTopoHydroLoader.get_elevation_profile). `check_flow_direction_coverage`
reports how often each source actually applies before you trust either one
blindly; the `sens_de_l_ecoulement` value vocabulary hasn't been verified
against a real full export, so this treats it as unconfirmed until checked.

Node types in the final graph:
  - real gauges (snapped onto the nearest reach)
  - real confluences (any hydrographic node with in-degree >= 2)
  - virtual nodes, inserted along confluence-free stretches longer than
    `virtual_node_spacing_km` so "predict at any point" has reasonable
    resolution without a node at every tronçon vertex

Targets (discharge, water level) are never assigned to confluence or
virtual nodes — see node_features.py's inputs-vs-targets distinction. That
principle doesn't change here; it matters more here, since most nodes in
this graph will be ungauged by construction.
"""
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple

import math
import re
import numpy as np
import pandas as pd
import networkx as nx


def _haversine_km(lat1, lon1, lat2, lon2) -> float:
    R = 6371.0
    lat1, lon1, lat2, lon2 = map(math.radians, [lat1, lon1, lat2, lon2])
    dlat, dlon = lat2 - lat1, lon2 - lon1
    a = math.sin(dlat / 2) ** 2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon / 2) ** 2
    return R * 2 * math.asin(math.sqrt(a))


# River name filter per basin, gathered from BD TOPO catchment-polygon
# toponyms surfaced earlier in this project (cross_check_catchments.py's
# "bdtopo_catchment_name" output) — not a guess, these are real tributary
# names confirmed to exist in this watershed's data. Extend if others turn
# up once the broadened pull actually runs.
TRIBUTARY_NAMES = {
    0: [  # La Eure
        "Eure", "Livier", "Coinon", "Houdouenne", "Blaise", "Avre", "Voise",
        "Drouette", "Maltorne", "Vesgre", "Etang", "Autheuil", "Iton",
    ],
    1: [  # La Risle
        "Risle", "Corru", "Finard", "Sommaire", "Bave", "Charentonne",
        "Croix Blanche", "Bec", "Echauds", "Vieux Bourg",
    ],
}


@dataclass
class ReachGraphReport:
    n_troncons_matched: int
    n_hydro_nodes: int
    n_confluences: int
    n_raw_multi_inflow: int
    flow_direction_from_sens: int
    flow_direction_from_elevation: int
    flow_direction_ambiguous: int
    n_gauges_snapped: int
    n_virtual_nodes_inserted: int
    disconnected_components: int
    largest_component_troncons: int

    def __str__(self) -> str:
        total_dir = (self.flow_direction_from_sens + self.flow_direction_from_elevation
                     + self.flow_direction_ambiguous)
        pct_sens = 100 * self.flow_direction_from_sens / total_dir if total_dir else 0
        return (
            f"Reach graph: {self.n_troncons_matched} tronçons matched, "
            f"{self.n_hydro_nodes} hydrographic nodes\n"
            f"  confluences (different named river actually joining): {self.n_confluences}\n"
            f"  ({self.n_raw_multi_inflow} nodes have in-degree >= 2 total -- the gap between "
            f"the two is same-river fine segmentation, not real branching)\n"
            f"  flow direction: {self.flow_direction_from_sens} from sens_de_l_ecoulement "
            f"({pct_sens:.0f}%), {self.flow_direction_from_elevation} from elevation, "
            f"{self.flow_direction_ambiguous} ambiguous (dropped)\n"
            f"  {self.n_gauges_snapped} real gauges snapped, "
            f"{self.n_virtual_nodes_inserted} virtual nodes inserted\n"
            f"  topology: {self.disconnected_components} component(s), "
            f"largest has {self.largest_component_troncons} tronçons"
            + (" -- single connected network" if self.disconnected_components == 1 else
               " -- see disconnected_components; only the largest is used")
        )


def check_flow_direction_coverage(features: List[dict]) -> Dict[str, int]:
    """
    Report how populated and how varied `sens_de_l_ecoulement` actually is
    in a real export, before any code decides to trust it. Run this first
    against the real data.
    """
    from collections import Counter
    values = Counter()
    for f in features:
        v = f.get("properties", {}).get("sens_de_l_ecoulement")
        values[v if v not in (None, "") else "<empty>"] += 1
    return dict(values)


def load_troncons_for_basin(
    troncon_geojson: dict,
    basin_id: int,
    name_field: str = "cpx_toponyme_de_cours_d_eau",
    extra_names: Optional[List[str]] = None,
    anchor_stations: Optional[pd.DataFrame] = None,
    max_distance_from_anchor_km: float = 20.0,
) -> List[dict]:
    """
    Filter troncon_hydrographique features to a basin's river + known
    tributaries, instead of a single named-river-only match. This is what
    lets branching topology exist at all -- matching only "Risle" or
    "Eure" by name discards every tributary tronçon before the graph is
    even built.

    Name matching alone is NOT sufficient at this bounding-box scale
    (~100x130 km): several tributary names are short/common enough
    ("Bec", "Avre", "Bave", "Corru", "Etang", "Iton") to genuinely match
    an unrelated stream elsewhere in the box that happens to share the
    word, not the actual tributary. Confirmed against real data: a
    name-only pull for these two basins produced 487/214 disconnected
    components and every real gauge snapping 1.5-50 km from the kept
    network -- a name collision pulling in geographically unrelated
    tronçons is the most likely explanation.

    If `anchor_stations` is given (this basin's real gauge coordinates),
    every matched tronçon is additionally required to have at least one
    vertex within `max_distance_from_anchor_km` of some anchor station.
    This can't fully separate "real tributary far from any gauge" from
    "name collision," but it does reject tronçons that are unambiguously
    nowhere near this basin's actual stations -- which the real run's
    50 km snap distances show is happening.
    """
    names = list(TRIBUTARY_NAMES.get(basin_id, [])) + (extra_names or [])
    names_lower = [n.lower() for n in names]
    name_matched = []
    for f in troncon_geojson.get("features", []):
        toponym = str(f.get("properties", {}).get(name_field, "")).lower()
        if any(n in toponym for n in names_lower):
            name_matched.append(f)

    if anchor_stations is None or anchor_stations.empty:
        return name_matched

    anchors = list(zip(anchor_stations["latitude"], anchor_stations["longitude"]))
    matched = []
    for f in name_matched:
        coords = f.get("geometry", {}).get("coordinates", [])
        if not coords:
            continue
        # Cheap pre-check on the tronçon's own endpoints before checking every vertex.
        endpoints = [coords[0], coords[-1]]
        if any(_haversine_km(alat, alon, pt[1], pt[0]) <= max_distance_from_anchor_km
               for pt in endpoints for alat, alon in anchors):
            matched.append(f)
    return matched


def _endpoint_altitudes(coords: List[list]) -> Tuple[Optional[float], Optional[float]]:
    """(start_z, end_z) if the geometry has a Z coordinate, else (None, None)."""
    if not coords or len(coords[0]) < 3:
        return None, None
    return coords[0][2], coords[-1][2]


def normalize_toponym(name: Optional[str]) -> Optional[str]:
    """
    Normalize a river name for comparison: lowercase, strip whitespace,
    drop a leading article ("la "/"le "/"l'"/"les "), drop a "bras de/du/d'"
    prefix (a named secondary channel of a river -- "Bras de la Charentonne"
    is an arm of the Charentonne, not a different river; confirmed against
    real data as a genuine remaining false-confluence source even after
    the topological split-rejoin check, likely because the actual split
    point sits beyond that check's search radius or outside the selected
    component), and drop any parenthetical qualifier ("la Risle (Grand
    Bief)" -> "risle"). BD TOPO's toponym field is not guaranteed to be
    consistently formatted across tronçons for the same physical river --
    comparing raw strings for uniqueness would count formatting variation
    (or a named arm of the same river) as a different river and keep
    inflating the confluence count. Returns None if there's nothing left
    after normalization.
    """
    if not name:
        return None
    n = str(name).strip().lower()
    n = re.sub(r"\([^)]*\)", "", n).strip()  # drop parenthetical qualifiers
    for prefix in ("bras de la ", "bras de l'", "bras du ", "bras de ", "bras d'"):
        if n.startswith(prefix):
            n = n[len(prefix):]
            break
    for article in ("l'", "la ", "le ", "les "):
        if n.startswith(article):
            n = n[len(article):]
            break
    n = n.strip()
    return n or None


def _ancestors_within_km(G: nx.MultiDiGraph, start, max_km: float) -> Dict:
    """Bounded backward BFS (via predecessors) from `start`, tracking
    cumulative upstream distance. Returns {node: distance_km}.

    G is a MultiDiGraph (parallel edges between the same two nodes are
    real -- e.g. two distinct tronçons that both happen to run directly
    between the same pair of hydrographic nodes, exactly the shape a
    short braid takes). Where multiple parallel edges exist between a
    predecessor and n, the shortest is used -- the conservative choice
    for "is there any path within range," not an arbitrary one.
    """
    visited = {start: 0.0}
    frontier = [start]
    while frontier:
        new_frontier = []
        for n in frontier:
            for pred in G.predecessors(n):
                parallel = G.get_edge_data(pred, n)  # {key: data} for every parallel edge
                edge_dist = min((d.get("distance_km", 0.0) or 0.0) for d in parallel.values())
                cum = visited[n] + edge_dist
                if cum <= max_km and (pred not in visited or cum < visited[pred]):
                    visited[pred] = cum
                    new_frontier.append(pred)
        frontier = new_frontier
    return visited


def is_split_rejoin(G: nx.DiGraph, node, max_upstream_km: float = 15.0) -> bool:
    """
    True if `node`'s incoming branches trace back to a common upstream
    node within `max_upstream_km` -- i.e. this is the same water
    splitting into parallel channels and recombining (braiding, an
    anabranch, a named secondary "bras"/"bief"/"noue"), not a tributary
    confluence. No new drainage area or mass enters the system at a
    split-rejoin; treating one as a real confluence would misinform a
    mass-balance physics constraint.

    Doesn't depend on naming conventions at all -- catches a split-rejoin
    even when BD TOPO gives its two channels distinct names that
    `normalize_toponym` has no way to recognize as "the same river."
    """
    in_edges = list(G.in_edges(node))
    if len(in_edges) < 2:
        return False
    branch_sources = [u for u, _ in in_edges]
    ancestor_sets = [set(_ancestors_within_km(G, src, max_upstream_km).keys()) for src in branch_sources]
    common = set.intersection(*ancestor_sets) if ancestor_sets else set()
    common.discard(node)
    return len(common) > 0


def find_split_points(G: nx.DiGraph) -> Set:
    """
    Nodes where a single upstream channel divides into 2+ downstream
    paths -- unambiguous by construction, unlike confluences/rejoins:
    a split has exactly one thing flowing in and multiple things flowing
    out, so it's always the same water partitioning, regardless of what
    the branches get named downstream. No name or topology check needed
    here the way `find_real_confluences` needs one -- out-degree >= 2 is
    a sufficient physical definition on its own.

    Useful as the other half of a mass-conservation pair with
    `find_rejoin_points`: for a genuine split-then-rejoin, flow at the
    split should equal flow at the corresponding rejoin, since no new
    catchment area is added anywhere in between.
    """
    return {n for n in G.nodes if G.out_degree(n) >= 2}


def find_rejoin_points(G: nx.DiGraph, max_upstream_km: float = 15.0) -> Set:
    """
    Nodes where two branches recombine after a common upstream split --
    the companion classification to `find_real_confluences`, made
    explicit and persistable rather than silently discarded. A rejoin
    and a real confluence look identical by raw in-degree; the
    distinguishing test is the same one `find_real_confluences` already
    uses (a common upstream ancestor within `max_upstream_km`), just
    keeping the nodes that test *positively* identifies instead of the
    ones it excludes.

    These matter for a different physics constraint than a real
    confluence: no new mass enters at a rejoin (or its paired split) --
    what should hold is Q_upstream_of_split ~= Q_downstream_of_rejoin,
    not "new tributary inflow contributes here."
    """
    rejoins = set()
    for n in G.nodes:
        if G.in_degree(n) >= 2 and is_split_rejoin(G, n, max_upstream_km):
            rejoins.add(n)
    return rejoins


def pair_splits_and_rejoins(G: nx.DiGraph, max_upstream_km: float = 15.0) -> pd.DataFrame:
    """
    For each rejoin point, identify which upstream split it pairs with
    -- the actual link a mass-conservation constraint needs
    (Q_at_this_split ~= Q_at_this_rejoin), not just "splits exist
    somewhere and rejoins exist somewhere" as two independent facts.

    Returns:
        DataFrame [rejoin_node, split_node, upstream_distance_km,
        branch_nodes] -- one row per rejoin, `branch_nodes` is the list
        of nodes along the shortest path of each branch between the
        split and the rejoin (so edge-level attributes for the braid are
        easy to pull later). A rejoin whose branches share more than one
        common ancestor within range keeps the CLOSEST one (shortest
        upstream distance), since that's the most specific matching split.
    """
    rows = []
    for rejoin in find_rejoin_points(G, max_upstream_km):
        in_edges = list(G.in_edges(rejoin))
        branch_sources = [u for u, _ in in_edges]
        ancestor_sets = [_ancestors_within_km(G, src, max_upstream_km) for src in branch_sources]
        common = set(ancestor_sets[0].keys())
        for s in ancestor_sets[1:]:
            common &= set(s.keys())
        common.discard(rejoin)
        if not common:
            continue
        # closest common ancestor = the actual split point for this rejoin
        split = min(common, key=lambda n: max(a.get(n, float("inf")) for a in ancestor_sets))
        dist = max(a.get(split, 0.0) for a in ancestor_sets)

        branch_nodes = []
        for src in branch_sources:
            try:
                path = nx.shortest_path(G, split, src)
                branch_nodes.append(path)
            except nx.NetworkXNoPath:
                branch_nodes.append([])

        rows.append({
            "rejoin_node": rejoin, "split_node": split,
            "upstream_distance_km": dist, "branch_nodes": branch_nodes,
        })
    return pd.DataFrame(rows)


def find_real_confluences(G: nx.DiGraph, max_upstream_km: float = 15.0) -> Set:
    """
    Nodes where a genuinely different, independently-sourced river joins
    -- not just where in-degree happens to be >= 2, and not a channel
    split that rejoins downstream (see `find_rejoin_points`, which keeps
    that classification rather than discarding it -- splits and rejoins
    carry real mass-conservation information of their own, just a
    different kind than a tributary confluence: no new catchment area
    is added at either point in a braid, whereas a real confluence
    genuinely does add one).

    Two filters, in order:
      1. In-degree >= 2 with more than one distinct (normalized) river
         name -- see `normalize_toponym` for why raw string comparison
         isn't reliable. BD TOPO's fine tronçon segmentation produces
         many same-river multi-inflow nodes with no real branching
         involved (confirmed against real data -- distances as short as
         4.6m at some flagged "confluences").
      2. NOT a rejoin: the branches must not trace back to a common
         upstream node within `max_upstream_km`. A braided or anabranched
         stretch can satisfy filter 1 (different channel names -- e.g. a
         named "bras"/arm) while still being the same water recombining,
         not a real tributary.
    """
    confluences = set()
    for n in G.nodes:
        in_edges = list(G.in_edges(n, data=True))
        if len(in_edges) < 2:
            continue
        toponyms = {normalize_toponym(e[2].get("toponym")) for e in in_edges}
        toponyms.discard(None)
        if len(toponyms) > 1 and not is_split_rejoin(G, n, max_upstream_km):
            confluences.add(n)
    return confluences


def summarize_confluences_by_tributary(G: nx.DiGraph) -> pd.DataFrame:
    """
    For each real confluence, which named tributary is actually joining
    there. This is the direct check against "the number of confluences
    should be explained by the number of tributaries" -- if this table
    has far more rows than TRIBUTARY_NAMES has entries for this basin,
    something is still over-counting (or a tributary itself has real
    internal sub-confluences worth knowing about, which this makes
    visible rather than hiding inside one aggregate number).

    Returns:
        DataFrame [node, main_river, joining_river, joining_river_raw]
        -- one row per real confluence. `joining_river` is normalized;
        `joining_river_raw` keeps the original string(s) seen, so you
        can tell formatting variation apart from a genuinely different
        tributary at a glance.
    """
    rows = []
    for n in find_real_confluences(G):
        in_edges = list(G.in_edges(n, data=True))
        by_norm: Dict[str, List[str]] = {}
        for _, _, data in in_edges:
            raw = data.get("toponym")
            norm = normalize_toponym(raw)
            if norm:
                by_norm.setdefault(norm, []).append(raw)
        if len(by_norm) < 2:
            continue
        sorted_names = sorted(by_norm.items(), key=lambda kv: -len(kv[1]))
        main_river = sorted_names[0][0]
        for joining, raws in sorted_names[1:]:
            rows.append({
                "node": n, "main_river": main_river, "joining_river": joining,
                "joining_river_raw_values_seen": sorted(set(raws)),
            })
    return pd.DataFrame(rows)


def build_node_link_digraph(
    features: List[dict],
    sens_field: str = "sens_de_l_ecoulement",
    sens_downstream_values: Tuple[str, ...] = ("Sens direct", "Direct", "01"),
    sens_upstream_values: Tuple[str, ...] = ("Sens indirect", "Indirect", "02"),
    name_field: str = "cpx_toponyme_de_cours_d_eau",
) -> Tuple[nx.DiGraph, ReachGraphReport]:
    """
    Build a directed graph from tronçon-to-node linkage: one edge per
    tronçon, from its upstream hydrographic node to its downstream one.

    Direction is resolved per tronçon: `sens_de_l_ecoulement` is used if
    its value is recognized (the exact vocabulary is unverified against a
    real full export -- extend `sens_downstream_values`/`sens_upstream_values`
    once `check_flow_direction_coverage` shows what's actually there).
    Otherwise falls back to comparing the tronçon's own endpoint altitudes.
    A tronçon with neither a usable `sens` value nor a usable Z coordinate
    is dropped rather than guessed at, and counted in the report.

    CONFLUENCE DEFINITION: a node has in-degree >= 2 whenever two or more
    tronçons happen to end there -- which turns out NOT to reliably mean
    "a tributary joins here." BD TOPO segments a single continuous
    channel into very short tronçons (confirmed against real data: many
    under 20m), and dense node placement alone produces spurious in-degree
    >= 2 with no real branching involved. A "real" confluence here instead
    requires the incoming tronçon(s) to carry a DIFFERENT river name
    (`cpx_toponyme_de_cours_d_eau`) than the outgoing one -- "la
    Charentonne" joining "la Risle" is a real confluence; two same-named
    "la Risle" fragments both ending at the same node is fine-grained
    segmentation, not a tributary. `report.n_confluences` uses this
    definition; `n_raw_in_degree_ge2` is kept alongside it so the
    difference between the two is visible, not silently discarded.

    Returns:
        (digraph, report) -- digraph nodes are BD TOPO hydrographic node
        IDs; each edge carries the tronçon's geometry, distance_km,
        elevation_drop_m, toponym, basin_id, and cleabs as edge data.
    """
    G = nx.MultiDiGraph()  # parallel edges between the same node pair are real
    # data, not a collision -- two distinct tronçons directly connecting the
    # same two hydrographic nodes is exactly what a short braid looks like.
    # A plain DiGraph silently OVERWRITES the second such edge's data on
    # add_edge rather than keeping both (confirmed as a real, silent data
    # loss bug, not a hypothetical -- caught while testing a synthetic
    # adjacent-split-rejoin case).
    n_sens, n_elev, n_ambiguous = 0, 0, 0

    for f in features:
        props = f.get("properties", {})
        ini = props.get("lien_vers_noeud_hydrographique_ini")
        fin = props.get("lien_vers_noeud_hydrographique_fin")
        if not ini or not fin:
            n_ambiguous += 1
            continue

        coords = f.get("geometry", {}).get("coordinates", [])
        if not coords:
            n_ambiguous += 1
            continue

        sens = props.get(sens_field)
        z_start, z_end = _endpoint_altitudes(coords)

        if sens in sens_downstream_values:
            u, v = ini, fin
            n_sens += 1
        elif sens in sens_upstream_values:
            u, v = fin, ini
            n_sens += 1
        elif z_start is not None and z_end is not None and z_start != z_end:
            u, v = (ini, fin) if z_start > z_end else (fin, ini)
            n_elev += 1
        else:
            n_ambiguous += 1
            continue

        lon1, lat1 = coords[0][0], coords[0][1]
        lon2, lat2 = coords[-1][0], coords[-1][1]
        dist_km = _haversine_km(lat1, lon1, lat2, lon2)
        elev_drop = (z_start - z_end) if (z_start is not None and z_end is not None) else None
        if u != ini:  # direction was flipped relative to raw geometry order
            elev_drop = -elev_drop if elev_drop is not None else None
            coords_ordered = list(reversed(coords))
        else:
            coords_ordered = coords

        G.add_edge(u, v, distance_km=dist_km, elevation_drop_m=elev_drop,
                    coords=coords_ordered, cleabs=props.get("cleabs"),
                    toponym=props.get(name_field))

    components = list(nx.weakly_connected_components(G))
    largest = max((len(c) for c in components), default=0)
    real_confluences = find_real_confluences(G)
    raw_multi_inflow = [n for n in G.nodes if G.in_degree(n) >= 2]

    report = ReachGraphReport(
        n_troncons_matched=len(features), n_hydro_nodes=G.number_of_nodes(),
        n_confluences=len(real_confluences), n_raw_multi_inflow=len(raw_multi_inflow),
        flow_direction_from_sens=n_sens,
        flow_direction_from_elevation=n_elev, flow_direction_ambiguous=n_ambiguous,
        n_gauges_snapped=0, n_virtual_nodes_inserted=0,
        disconnected_components=len(components), largest_component_troncons=largest,
    )
    return G, report


def largest_component(G: nx.DiGraph) -> nx.DiGraph:
    """Keep only the largest weakly-connected component, by tronçon count.
    See `best_component_for_stations` for a criterion that's actually
    validated against real gauges instead -- "largest" has no guarantee
    of being anywhere near the stations we care about, and confirmed
    against real data it wasn't."""
    components = list(nx.weakly_connected_components(G))
    if not components:
        return G
    biggest = max(components, key=len)
    return G.subgraph(biggest).copy()


def best_component_for_stations(
    G: nx.DiGraph, stations_df: pd.DataFrame, search_radius_km: float = 5.0
) -> Tuple[nx.DiGraph, Dict[int, int]]:
    """
    Keep the connected component containing the most real gauge stations,
    rather than the component with the most tronçons. A large-by-count
    component has no guarantee of being the network our actual gauges
    sit on; a component chosen by how many real, trusted station
    coordinates fall near it does.

    A station "belongs" to a component if any of that component's tronçon
    vertices falls within `search_radius_km` of the station.

    Returns:
        (best_component_subgraph, {component_index: n_stations_near_it})
        -- the second value is diagnostic: if the winning component only
        narrowly beats the runner-up, or several components each capture
        a handful of stations, that's a sign the basin's real network is
        still fragmented and no single component is "the" river.
    """
    components = list(nx.weakly_connected_components(G))
    if not components:
        return G, {}

    counts = {}
    for i, comp in enumerate(components):
        subG = G.subgraph(comp)
        n_near = 0
        for _, station in stations_df.iterrows():
            found = False
            for _, _, data in subG.edges(data=True):
                for pt in data["coords"]:
                    if _haversine_km(station["latitude"], station["longitude"], pt[1], pt[0]) <= search_radius_km:
                        found = True
                        break
                if found:
                    break
            if found:
                n_near += 1
        counts[i] = n_near

    best_idx = max(counts, key=counts.get)
    return G.subgraph(components[best_idx]).copy(), counts


def nearest_edge_to_point(G: nx.DiGraph, lat: float, lon: float) -> Tuple[Tuple, Tuple, float, float]:
    """
    Find the (u, v) edge in G whose geometry passes nearest to (lat, lon),
    and how far along that edge (0-1 fraction from u to v) the nearest
    point falls.

    Returns:
        (u, v, fraction, distance_km)
    """
    best = None
    for u, v, data in G.edges(data=True):
        coords = data["coords"]
        for i in range(len(coords) - 1):
            ax, ay = coords[i][0], coords[i][1]
            bx, by = coords[i + 1][0], coords[i + 1][1]
            abx, aby = bx - ax, by - ay
            denom = abx ** 2 + aby ** 2
            t = 0.0 if denom == 0 else np.clip(((lon - ax) * abx + (lat - ay) * aby) / denom, 0.0, 1.0)
            proj_lon, proj_lat = ax + t * abx, ay + t * aby
            dist = _haversine_km(lat, lon, proj_lat, proj_lon)
            if best is None or dist < best[3]:
                # fraction along the WHOLE edge, not just this segment
                seg_count = len(coords) - 1
                frac = (i + t) / seg_count if seg_count else 0.0
                best = (u, v, frac, dist)
    return best


def snap_gauges_to_reach_graph(G: nx.DiGraph, stations_df: pd.DataFrame) -> pd.DataFrame:
    """
    Snap each station onto its nearest edge in the reach graph.

    Args:
        G: from build_node_link_digraph (ideally largest_component'd first).
        stations_df: [station_code, latitude, longitude, elevation_m, ...]

    Returns:
        stations_df with added columns: reach_u, reach_v, reach_fraction, snap_distance_km
    """
    rows = []
    for _, row in stations_df.iterrows():
        u, v, frac, dist = nearest_edge_to_point(G, row["latitude"], row["longitude"])
        rows.append({"station_code": row["station_code"], "reach_u": u, "reach_v": v,
                      "reach_fraction": frac, "snap_distance_km": dist})
    snapped = pd.DataFrame(rows)
    return stations_df.merge(snapped, on="station_code", how="left")


def build_reach_graph_tables(
    G: nx.DiGraph,
    gauges_snapped: pd.DataFrame,
    basin_id: int,
) -> Tuple[pd.DataFrame, pd.DataFrame]:
    """
    Assemble the final nodes_df/edges_df for this basin, combining real
    gauges, real confluences, and virtual nodes into one table in the
    same shape build_pyg_graph/build_pyg_graphs_per_basin already expect
    -- so nothing downstream of graph construction needs to change.

    A gauge station keeps its own real coordinates and elevation (not the
    snapped point's) since that's the actual, correct location of the
    physical instrument; it's associated with the graph via its nearest
    edge, splitting that edge at the gauge's snapped position. Every other
    node in G (confluences, virtual nodes already inserted upstream by
    insert_virtual_nodes) becomes its own graph node using coordinates
    taken directly from the reach geometry.

    Args:
        G: reach digraph, ideally after largest_component() and
            insert_virtual_nodes().
        gauges_snapped: output of snap_gauges_to_reach_graph, run against
            this same G.
        basin_id: 0 (Eure) or 1 (Risle).

    Returns:
        (nodes_df, edges_df) in the same schema as build_surface_edges's
        output: nodes_df has [station_code, basin_id, latitude, longitude,
        elevation_m, is_gauged]; edges_df has [source, target, basin_id,
        distance_km, elevation_drop_m, verified_continuous].
    """
    real_confluences = find_real_confluences(G)  # single authoritative source of truth,
    # computed once here where the actual graph topology is available -- everything
    # downstream (both apps, both scripts) reads the saved is_confluence column
    # rather than each re-deriving it from the flat CSV, which is exactly how the
    # split-rejoin exclusion and the earlier normalization fix drifted out of sync
    # across files before.
    split_points = find_split_points(G)
    rejoin_points = find_rejoin_points(G)
    # Splits and rejoins are NOT discarded as noise -- they carry a real,
    # different mass-conservation constraint than a tributary confluence
    # (Q_at_split ~= Q_at_rejoin, no new catchment area involved, vs. a
    # confluence genuinely adding one). braid_id links a rejoin to its
    # paired split so that constraint is actually constructible later,
    # not just "some splits and rejoins exist somewhere, unpaired."
    braid_pairs = pair_splits_and_rejoins(G)
    rejoin_to_split = dict(zip(braid_pairs["rejoin_node"], braid_pairs["split_node"])) if not braid_pairs.empty else {}

    node_rows = {}
    for n, data in G.nodes(data=True):
        lat, lon, elev = data.get("latitude"), data.get("longitude"), data.get("elevation_m")
        if lat is None or lon is None:
            # Real BD TOPO hydrographic nodes carry no attributes of their
            # own -- only the edges know coordinates. Derive this node's
            # position from any adjacent edge's endpoint nearest to it
            # (an out-edge starts here; an in-edge ends here).
            out_edges = list(G.out_edges(n, data=True))
            in_edges = list(G.in_edges(n, data=True))
            if out_edges:
                p = out_edges[0][2]["coords"][0]
            elif in_edges:
                p = in_edges[0][2]["coords"][-1]
            else:
                continue  # isolated node with no geometry anywhere -- can't place it
            lon, lat = p[0], p[1]
            elev = p[2] if len(p) > 2 else None
        node_rows[n] = {
            "station_code": str(n), "basin_id": basin_id,
            "latitude": lat, "longitude": lon, "elevation_m": elev,
            "is_gauged": False, "is_confluence": n in real_confluences,
            "is_split_point": n in split_points, "is_rejoin_point": n in rejoin_points,
            "braid_id": str(rejoin_to_split[n]) if n in rejoin_to_split else None,
        }

    edge_rows = []
    for u, v, data in G.edges(data=True):
        edge_rows.append({
            "source": str(u), "target": str(v), "basin_id": basin_id,
            "distance_km": data["distance_km"], "elevation_drop_m": data.get("elevation_drop_m"),
            "verified_continuous": True, "cleabs": data.get("cleabs"),
            "toponym": data.get("toponym"),
        })

    # Splice each gauge into its nearest edge: gauge_u -> GAUGE -> gauge_v,
    # replacing the direct u->v edge, so the gauge becomes a real node on
    # the path rather than a disconnected point sitting near the line.
    for _, g in gauges_snapped.dropna(subset=["reach_u", "reach_v"]).iterrows():
        u, v, frac = g["reach_u"], g["reach_v"], g["reach_fraction"]
        code = g["station_code"]
        node_rows[code] = {
            "station_code": code, "basin_id": basin_id,
            "latitude": g["latitude"], "longitude": g["longitude"],
            "elevation_m": g.get("elevation_m"), "is_gauged": True, "is_confluence": False,
            "is_split_point": False, "is_rejoin_point": False, "braid_id": None,
            "snap_distance_km": g.get("snap_distance_km"),
        }

        original = next((e for e in edge_rows if e["source"] == str(u) and e["target"] == str(v)), None)
        if original is None:
            continue
        edge_rows.remove(original)
        d, drop = original["distance_km"], original["elevation_drop_m"]
        edge_rows.append({"source": str(u), "target": code, "basin_id": basin_id,
                           "distance_km": d * frac,
                           "elevation_drop_m": drop * frac if drop is not None else None,
                           "verified_continuous": original["verified_continuous"],
                           "cleabs": original.get("cleabs"), "toponym": original.get("toponym")})
        edge_rows.append({"source": code, "target": str(v), "basin_id": basin_id,
                           "distance_km": d * (1 - frac),
                           "elevation_drop_m": drop * (1 - frac) if drop is not None else None,
                           "verified_continuous": original["verified_continuous"],
                           "cleabs": original.get("cleabs"), "toponym": original.get("toponym")})

    nodes_df = pd.DataFrame(node_rows.values())
    edges_df = pd.DataFrame(edge_rows)
    return nodes_df, edges_df


def insert_virtual_nodes(G: nx.MultiDiGraph, spacing_km: float = 5.0) -> Tuple[nx.MultiDiGraph, int]:
    """
    Split any edge longer than `spacing_km` into evenly-spaced segments by
    inserting virtual node IDs along it, so long confluence-free stretches
    get intermediate nodes instead of one edge covering many kilometers.

    Virtual node IDs are strings like "VIRTUAL::{u}::{v}::{i}::{key}" --
    distinct from real BD TOPO node IDs by construction, so they can
    never collide. The edge `key` is included specifically because two
    parallel tronçons between the same (u, v) pair are real, not a
    collision (see build_node_link_digraph) -- each needs its own
    distinctly-IDed virtual chain, and removing the correct specific
    parallel edge (not an arbitrary one) requires operating on its key,
    not just its endpoints.

    Returns:
        (new_digraph, n_virtual_nodes_inserted)
    """
    G2 = G.copy()
    n_inserted = 0

    for u, v, key, data in list(G.edges(keys=True, data=True)):
        dist = data["distance_km"]
        if dist <= spacing_km:
            continue
        n_segments = max(2, int(np.ceil(dist / spacing_km)))
        coords = data["coords"]
        n_points = len(coords)

        G2.remove_edge(u, v, key)
        prev_node = u
        for i in range(1, n_segments):
            frac = i / n_segments
            idx = min(int(frac * (n_points - 1)), n_points - 2)
            local_t = frac * (n_points - 1) - idx
            p0, p1 = coords[idx], coords[idx + 1]
            lon = p0[0] + local_t * (p1[0] - p0[0])
            lat = p0[1] + local_t * (p1[1] - p0[1])
            z = (p0[2] + local_t * (p1[2] - p0[2])) if len(p0) > 2 else None

            vnode = f"VIRTUAL::{u}::{v}::{key}::{i}"
            G2.add_node(vnode, latitude=lat, longitude=lon, elevation_m=z)
            seg_dist = dist / n_segments
            seg_drop = (data["elevation_drop_m"] / n_segments) if data["elevation_drop_m"] is not None else None
            G2.add_edge(prev_node, vnode, distance_km=seg_dist, elevation_drop_m=seg_drop,
                        coords=[coords[idx], coords[idx + 1]], cleabs=data.get("cleabs"),
                        toponym=data.get("toponym"))
            prev_node = vnode
            n_inserted += 1

        seg_dist = dist / n_segments
        seg_drop = (data["elevation_drop_m"] / n_segments) if data["elevation_drop_m"] is not None else None
        G2.add_edge(prev_node, v, distance_km=seg_dist, elevation_drop_m=seg_drop,
                    coords=coords[-2:], cleabs=data.get("cleabs"), toponym=data.get("toponym"))

    return G2, n_inserted