src/graph/node_features.py

"""
Node feature construction for the surface-connectivity graph.

Builds a per-station feature table by pulling from every available
loader, using each loader's own spatial interpolation/aggregation so a
station gets a value even when it doesn't sit exactly on a raw data
point (between IDPR points, outside a well's radius, off an ERA5 grid
cell, etc).

Loaders used, each skipped gracefully with a clear message if its data
files aren't present (same pattern as generate_plots.py) so one missing
dataset never stops the others:
  - StationElevationsLoader: elevation_m, latitude, longitude (base table)
  - IDPRLoader: infiltration/runoff tendency, via nearest spatial point
  - ADESLoader: groundwater level/depth, via the loader's own
    aggregate_to_stations (radius-averaged, not a graph edge — see the
    earlier decision not to use wells for subsurface edges; this is the
    "use as a station covariate" path instead)
  - SAFRANLoader: ERA5 climate variables, via the loader's own
    nearest/linear interpolation to each station point
  - HydrometricLoader: discharge/water-level summary stats

IMPORTANT — inputs vs. targets: elevation/IDPR/groundwater/climate are
genuine INPUT covariates, and filling them in via spatial interpolation
for any station is expected, ordinary feature engineering. Discharge and
water-level are the model's TARGET variables: they're reported here as
plain summary stats per gauge (prefixed `target_`) and are left NaN for
any station without hydrometric data — never interpolated or guessed.
Interpolating a target would be label leakage, not feature engineering;
don't feed the `target_*` columns back in as model inputs.
"""
from dataclasses import dataclass, field
from pathlib import Path
from typing import List, Optional, Tuple

import numpy as np
import pandas as pd

from ..data.river_graph import assign_basin_id
from ..data.loaders.idpr import IDPRLoader
from ..data.loaders.ades import ADESLoader
from ..data.loaders.station_elevations import StationElevationsLoader
from ..data.loaders.catchment import CatchmentAreaLoader


@dataclass
class NodeFeatureReport:
    """What actually got included, so you can tell at a glance whether a
    feature block is real or silently missing before trusting the table."""
    n_stations: int
    included: List[str] = field(default_factory=list)
    skipped: List[str] = field(default_factory=list)

    def __str__(self) -> str:
        lines = [f"Node features for {self.n_stations} stations:"]
        for name in self.included:
            lines.append(f"  \u2713 {name}")
        for name in self.skipped:
            lines.append(f"  \u2014 {name} (skipped)")
        return "\n".join(lines)


def build_base_station_table(elevations_df: pd.DataFrame, prefix_map=None) -> pd.DataFrame:
    """Base node table: station_code, basin_id, latitude, longitude, elevation_m."""
    df = elevations_df.dropna(subset=["latitude", "longitude"]).copy()
    df["basin_id"] = df["station_code"].apply(lambda c: assign_basin_id(c, prefix_map))
    unmatched = df[df["basin_id"].isna()]
    if not unmatched.empty:
        print(f"build_base_station_table: {len(unmatched)} station(s) matched no basin "
              f"prefix and are excluded: {unmatched['station_code'].tolist()}")
    df = df.dropna(subset=["basin_id"]).copy()
    df["basin_id"] = df["basin_id"].astype(int)
    return df[["station_code", "basin_id", "latitude", "longitude", "elevation_m"]].reset_index(drop=True)


def _haversine_km_vec(lat1: float, lon1: float, lat2, lon2):
    """Vectorized haversine: one (lat1, lon1) point against arrays
    lat2/lon2. Used for the exact narrow-phase distance check after a
    KD-tree coarse prefilter (see add_groundwater_features)."""
    R = 6371.0
    lat1r, lon1r = np.radians(lat1), np.radians(lon1)
    lat2r, lon2r = np.radians(lat2), np.radians(lon2)
    dlat, dlon = lat2r - lat1r, lon2r - lon1r
    a = np.sin(dlat / 2) ** 2 + np.cos(lat1r) * np.cos(lat2r) * np.sin(dlon / 2) ** 2
    return R * 2 * np.arcsin(np.sqrt(a))


def add_idpr_features(nodes_df: pd.DataFrame, idpr_path: Path) -> pd.DataFrame:
    """
    IDPR value per station. Prefers an exact join when the IDPR data
    already has a per-station identifier column (e.g. it was extracted
    at each gauge's coordinates ahead of time, as opposed to being a raw
    spatial point cloud/raster export) — this is both simpler and more
    accurate than nearest-neighbor search, since it isn't a real
    interpolation, it's the actual value at that exact station. Falls
    back to nearest-spatial-point search (KD-tree) for a generic IDPR
    point cloud with no station identifiers.

    NOTE on the fallback path: it does NOT reproject coordinates — if
    IDPR's coordinates are in a different CRS than the station lat/lon,
    results will be wrong. Reproject one side to match the other first
    if needed.
    """
    from scipy.spatial import cKDTree

    loader = IDPRLoader(data_path=idpr_path)
    idpr_df = loader.load()
    value_col = loader._find_value_column(idpr_df)

    id_col = next((c for c in ["station_id", "station_code", "code_station"] if c in idpr_df.columns), None)
    if id_col is not None and set(nodes_df["station_code"]).issubset(set(idpr_df[id_col])):
        merged = nodes_df.merge(
            idpr_df[[id_col, value_col]].rename(columns={id_col: "station_code", value_col: "idpr_value"}),
            on="station_code", how="left",
        )
        merged["idpr_nearest_point_distance"] = 0.0  # exact match, not interpolated
        return merged

    x_col = next((c for c in ["lon", "longitude", "x"] if c in idpr_df.columns), None)
    y_col = next((c for c in ["lat", "latitude", "y"] if c in idpr_df.columns), None)
    if not x_col or not y_col:
        raise ValueError("Could not detect coordinate columns in IDPR data.")

    tree = cKDTree(idpr_df[[x_col, y_col]].values)
    query_coords = nodes_df[["longitude", "latitude"]].values
    dist, idx = tree.query(query_coords)

    out = nodes_df.copy()
    out["idpr_value"] = idpr_df[value_col].values[idx]
    out["idpr_nearest_point_distance"] = dist  # in the IDPR file's own coordinate units
    return out


def add_geology_features(nodes_df: pd.DataFrame, bdcharm_path: Path) -> pd.DataFrame:
    """
    Geological formation class per node, one-hot encoded, from
    scripts/download_bdcharm.py's per-department BD Charm-50 shapefiles
    (datasets/bdcharm50/dept_{027,028,061}/).

    One-hot, not a raw formation code -- same reasoning as
    add_landcover_features: geological formation is a nominal category,
    not an ordered quantity, so a raw code would let build_pyg_graph's
    numeric auto-detection z-score it as if formation codes had a
    meaningful numeric ordering, which they don't.
    """
    import geopandas as gpd
    from shapely.geometry import Point

    base = Path(bdcharm_path)
    shp_files = sorted(base.glob("dept_*/**/*.shp"))
    fgeol_files = [f for f in shp_files if "FGEOL" in f.name.upper()]
    if not fgeol_files:
        raise FileNotFoundError(
            f"No *FGEOL* shapefile found under {base}/dept_*/. Found instead: "
            f"{[f.name for f in shp_files]}. Inspect these directly to find the "
            f"real geological-formations layer if the naming differs from what "
            f"was expected."
        )
    print(f"add_geology_features: using formation layer(s): {[f.name for f in fgeol_files]}")

    polygons = pd.concat([gpd.read_file(f) for f in fgeol_files], ignore_index=True)
    polygons = gpd.GeoDataFrame(polygons, geometry="geometry")

    label_candidates = [c for c in polygons.columns if any(
        kw in c.upper() for kw in ("NOTATION", "CODE", "LEGENDE", "LEG", "LITHO", "FORMATION")
    ) and c.upper() != "GEOMETRY"]
    if not label_candidates:
        raise ValueError(
            f"No plausible lithology/formation column found among {list(polygons.columns)}. "
            f"Inspect the real shapefile's attribute table directly and adjust the "
            f"keyword list in add_geology_features."
        )
    label_col = label_candidates[0]
    print(f"add_geology_features: using attribute column {label_col!r} "
          f"(other candidates considered: {label_candidates[1:]})")

    stations = gpd.GeoDataFrame(
        nodes_df,
        geometry=[Point(lon, lat) for lon, lat in zip(nodes_df["longitude"], nodes_df["latitude"])],
        crs="EPSG:4326",
    )
    if polygons.crs is not None and polygons.crs != stations.crs:
        polygons = polygons.to_crs(stations.crs)

    joined = gpd.sjoin(stations, polygons[[label_col, "geometry"]], how="left", predicate="within")
    joined = joined.drop_duplicates(subset="station_code" if "station_code" in joined.columns else joined.index.name)

    n_matched = joined[label_col].notna().sum()
    print(f"add_geology_features: {n_matched}/{len(nodes_df)} nodes matched to a real formation polygon")

    dummies = pd.get_dummies(joined[label_col], prefix="geology")
    result = pd.concat([nodes_df.reset_index(drop=True), dummies.reset_index(drop=True)], axis=1)
    return result


def add_cavites_features(
    nodes_df: pd.DataFrame, bdcavites_path: Path, max_distance_km: float = 20.0
) -> pd.DataFrame:
    """
    Distance to nearest known cavity/sinkhole + count within
    max_distance_km, from scripts/download_bdcavites.py's
    cavite_localisee.geojson (BRGM's national cavity inventory, via
    Géorisques' WFS).

    Same KD-tree coarse-prefilter + exact-haversine pattern already
    validated for add_groundwater_features -- point data at unknown
    real density, so avoid assuming either a fast-enough naive loop or
    guessing at scale ahead of time.
    """
    import json
    from scipy.spatial import cKDTree

    path = Path(bdcavites_path)
    if not path.exists():
        raise FileNotFoundError(f"{path} not found")

    geojson = json.loads(path.read_text())
    features = geojson.get("features", [])
    if not features:
        print(f"add_cavites_features: {path} has 0 features -- "
              f"every station will get distance=NaN, count=0")
        out = nodes_df.copy()
        out["distance_to_nearest_cavity_km"] = float("nan")
        out["n_cavities_within_20km"] = 0
        return out

    cavity_coords = []
    for f in features:
        geom = f.get("geometry", {})
        coords = geom.get("coordinates")
        if geom.get("type") == "Point" and coords:
            cavity_coords.append((coords[1], coords[0]))  # (lat, lon)
    cavity_lat = np.array([c[0] for c in cavity_coords])
    cavity_lon = np.array([c[1] for c in cavity_coords])
    print(f"add_cavites_features: {len(cavity_coords)} point cavity feature(s) loaded from {path}")

    tree = cKDTree(np.column_stack([cavity_lon, cavity_lat]))
    coarse_radius_deg = (max_distance_km / 111.0) * 1.5  # see add_groundwater_features for rationale

    station_lon = nodes_df["longitude"].values
    station_lat = nodes_df["latitude"].values
    candidate_lists = tree.query_ball_point(np.column_stack([station_lon, station_lat]), r=coarse_radius_deg)

    min_dists, counts = [], []
    for i, candidates in enumerate(candidate_lists):
        if not candidates:
            min_dists.append(float("nan"))
            counts.append(0)
            continue
        cand_idx = np.array(candidates)
        dist = _haversine_km_vec(station_lat[i], station_lon[i], cavity_lat[cand_idx], cavity_lon[cand_idx])
        within = dist <= max_distance_km
        min_dists.append(float(dist.min()) if within.any() else float("nan"))
        counts.append(int(within.sum()))

    out = nodes_df.copy()
    out["distance_to_nearest_cavity_km"] = min_dists
    out[f"n_cavities_within_{int(max_distance_km)}km"] = counts
    return out


def add_ndvi_features(nodes_df: pd.DataFrame, ndvi_path: Path) -> pd.DataFrame:
    """
    ESA WorldCover Sentinel-2 NDVI yearly percentile composite
    (p10/p50/p90) per station, from scripts/fetch_worldcover_ndvi.py.

    NOTE: raw values are VITO's scaled digital numbers, not independently
    confirmed to be literal -1..1 NDVI (no access to their exact scale/
    offset convention in this environment) -- internally consistent
    (p90 > p50 > p10 holds for every real station checked), so the
    relative signal is trustworthy even if the absolute units aren't
    pinned down. Doesn't block use as a model feature: build_pyg_graph
    z-scores every feature anyway, which is invariant to an unknown
    linear scaling. Only matters if literal NDVI units are needed later
    (e.g. for plotting against a textbook NDVI range) -- resolve the
    scale/offset from VITO's product documentation before then.

    Same coverage limitation as add_catchment_features/add_landcover_
    features: exact station_code match, real gauges only for now.
    """
    ndvi_df = pd.read_csv(ndvi_path)[["station_code", "ndvi_p10", "ndvi_p50", "ndvi_p90"]]
    return nodes_df.merge(ndvi_df, on="station_code", how="left")


def add_landcover_features(nodes_df: pd.DataFrame, landcover_path: Path) -> pd.DataFrame:
    """
    ESA WorldCover landcover class per station, one-hot encoded --
    NOT left as the raw integer class code (10=Tree cover, 50=Built-up,
    etc.). Landcover is a nominal category, not an ordered quantity;
    leaving it as a raw integer would let build_pyg_graph's numeric
    auto-detection z-score it as if "Built-up" (50) were meaningfully
    "more" than "Tree cover" (10) in some continuous sense, which isn't
    physically true -- the same class of error as the structural-column
    leakage bug found earlier in this project, just subtler since this
    one IS meant to be a real model input, not excluded metadata.

    From scripts/fetch_worldcover_landcover.py -- currently only
    covers real gauge stations (exact station_code match, same
    limitation as add_catchment_features): non-gauge reach graph nodes
    get NaN/all-zero here until that script is extended to sample the
    full node set, not just the 27 gauges.
    """
    landcover_df = pd.read_csv(landcover_path)[["station_code", "landcover_class_name"]]
    dummies = pd.get_dummies(landcover_df["landcover_class_name"], prefix="landcover")
    landcover_wide = pd.concat([landcover_df[["station_code"]], dummies], axis=1)
    return nodes_df.merge(landcover_wide, on="station_code", how="left")


def add_catchment_features(nodes_df: pd.DataFrame, catchment_path: Path) -> pd.DataFrame:
    """
    Catchment (drainage basin) area per station, in km², via
    CatchmentAreaLoader (see download_catchment_area.py for how this is
    sourced from Hub'Eau's own referentiel — no interpolation involved,
    it's the actual reported value or nothing).

    NOTE: this is genuinely missing (not degraded/approximated) for
    roughly 40% of stations in this project's data — Hub'Eau doesn't
    publish surface_bv for every site (observer stations, some partner-
    network sites, etc). That's real and expected; the missingness flag
    generated by build_pyg_graph's `add_missingness_flags` will capture
    it correctly as long as this column has NaNs, which it does by
    design — don't fill it in here.
    """
    loader = CatchmentAreaLoader(data_path=catchment_path)
    catchment_df = loader.load()[["station_code", "catchment_area_km2"]]
    return nodes_df.merge(catchment_df, on="station_code", how="left")


def add_groundwater_features(
    nodes_df: pd.DataFrame, ades_path: Path, max_distance_km: float = 20.0,
    date_range: Optional[Tuple[str, str]] = None,
) -> pd.DataFrame:
    """
    Radius-averaged groundwater level/depth per station.

    Does NOT use ADESLoader.aggregate_to_stations -- that method loops
    per station and does a full haversine .apply() over the ENTIRE
    groundwater dataframe for each one (O(n_stations * n_readings)).
    At 27 stations against ~272k readings that's slow but tolerable; at
    the reach graph's ~4,500 nodes it's over a billion row-wise Python
    calls, not a proportionally-worse runtime but an unusable one.

    Instead: reduce to each well's most recent reading first (collapses
    ~272k readings to ~100 wells -- we only need each station's latest
    aggregate anyway, same as the old code kept via
    `.sort_values("date").drop_duplicates(...)` after the fact), then a
    KD-tree coarse prefilter narrows each station to a handful of nearby
    candidate wells, with exact haversine distance computed only on
    that small candidate set to get correct radius membership and
    correct averages -- fast without trading away correctness for it.

    date_range: if given, (start, end) date strings (inclusive) -- a
    well's readings are filtered to this window BEFORE picking its
    "most recent" one, so a well with no reading in the window is
    correctly excluded from that call rather than falling back to some
    older reading from outside the period the rest of the graph's
    features are being built for (real wells in this project report on
    wildly different schedules -- see the ground-truth investigation
    that originally found this "most recent" logic needed care).
    """
    from scipy.spatial import cKDTree

    loader = ADESLoader(data_path=ades_path)
    gw_df = loader.load()
    if date_range is not None:
        start, end = date_range
        gw_df = gw_df[(gw_df["date"] >= start) & (gw_df["date"] <= end)]
    latest_per_well = gw_df.sort_values("date").drop_duplicates("code_bss", keep="last").reset_index(drop=True)

    if latest_per_well.empty:
        out = nodes_df.copy()
        out["avg_groundwater_level_m"] = np.nan
        out["avg_groundwater_depth_m"] = np.nan
        out["n_nearby_wells"] = 0
        return out

    well_lat = latest_per_well["lat"].values
    well_lon = latest_per_well["lon"].values
    tree = cKDTree(np.column_stack([well_lon, well_lat]))

    # Coarse degree-radius, deliberately oversized (1.5x safety factor):
    # degrees-to-km varies with latitude and direction (lon degrees are
    # worth fewer km than lat degrees away from the equator), so this
    # only needs to be a safe upper bound, not an accurate radius --
    # the exact haversine pass afterward is what determines real
    # membership and distances.
    coarse_radius_deg = (max_distance_km / 111.0) * 1.5

    station_lon = nodes_df["longitude"].values
    station_lat = nodes_df["latitude"].values
    candidate_lists = tree.query_ball_point(np.column_stack([station_lon, station_lat]), r=coarse_radius_deg)

    avg_levels, avg_depths, n_wells_list = [], [], []
    for i, candidates in enumerate(candidate_lists):
        if not candidates:
            avg_levels.append(np.nan); avg_depths.append(np.nan); n_wells_list.append(0)
            continue
        cand_idx = np.array(candidates)
        dist = _haversine_km_vec(station_lat[i], station_lon[i], well_lat[cand_idx], well_lon[cand_idx])
        within = dist <= max_distance_km
        if not within.any():
            avg_levels.append(np.nan); avg_depths.append(np.nan); n_wells_list.append(0)
            continue
        sel = latest_per_well.iloc[cand_idx[within]]
        avg_levels.append(sel["groundwater_level_m"].mean())
        avg_depths.append(sel["groundwater_depth_m"].mean() if "groundwater_depth_m" in sel.columns else np.nan)
        n_wells_list.append(int(within.sum()))

    out = nodes_df.copy()
    out["avg_groundwater_level_m"] = avg_levels
    out["avg_groundwater_depth_m"] = avg_depths
    out["n_nearby_wells"] = n_wells_list
    return out


def add_safran_features(
    nodes_df: pd.DataFrame, safran_path: Path, date_range: Optional[Tuple[str, str]] = None,
) -> pd.DataFrame:
    """
    Climate summary stats per station, interpolated from ERA5 via
    SAFRANLoader's own station-point interpolation. Requires `xarray`
    and actual era5_*.nc files under safran_path; returns nodes_df
    unchanged (with a clear message) if either is missing.

    date_range: if given, (start, end) date strings (inclusive) --
    filters ERA5 timesteps to this window BEFORE aggregating (mean for
    temp/wind/solar, sum for precip/evap/snow/runoff), so e.g.
    climate_precip_mm becomes "total precip over the window" rather
    than "total precip over the full 1960-2026 record", which wouldn't
    be comparable to a discharge target computed over a specific
    training period.
    """
    try:
        from ..data.loaders.safran import SAFRANLoader
    except ImportError as e:
        print(f"add_safran_features: skipped ({e})")
        return nodes_df
    try:
        import xarray  # noqa: F401
    except ImportError:
        print("add_safran_features: skipped (xarray not installed)")
        return nodes_df

    station_coords = nodes_df.rename(columns={"latitude": "lat", "longitude": "lon"})[
        ["station_code", "lat", "lon"]
    ]
    loader = SAFRANLoader(data_path=safran_path, station_coords=station_coords)
    try:
        df = loader.load()
    except FileNotFoundError as e:
        print(f"add_safran_features: skipped ({e})")
        return nodes_df

    df = loader.convert_units(df)
    if date_range is not None:
        start, end = date_range
        df = df[(df["date"] >= start) & (df["date"] <= end)]
        if df.empty:
            print(f"add_safran_features: skipped (no ERA5 timesteps fall within {date_range})")
            return nodes_df

    agg_spec = {c: "mean" for c in ["temp_C", "wind_speed_ms", "solar_Wm2"] if c in df.columns}
    for c in ["precip_mm", "evap_mm", "snow_mm", "runoff_mm"]:
        if c in df.columns:
            agg_spec[c] = "sum"
    if not agg_spec:
        print("add_safran_features: skipped (no recognized variables after unit conversion)")
        return nodes_df

    summary = df.groupby("station_code").agg(agg_spec)
    summary = summary.add_prefix("climate_")
    return nodes_df.merge(summary, on="station_code", how="left")


def add_hydrometric_target_stats(
    nodes_df: pd.DataFrame, hydrometric_path: Path, date_range: Optional[Tuple[str, str]] = None,
) -> pd.DataFrame:
    """
    Discharge/water-level summary stats per gauge, prefixed `target_`.
    Left-joined so ungauged stations get NaN rather than being dropped —
    that NaN is meaningful (no observations), not a gap to fill in.

    date_range: if given, (start, end) date strings (inclusive) --
    filters observations to this window before computing mean/std/count,
    so the target reflects the same training period the input features
    (climate, groundwater) were built for, not each source's own full
    and differently-shaped historical record.
    """
    from ..data.loaders.hydrometric import HydrometricLoader

    loader = HydrometricLoader(data_path=hydrometric_path)
    try:
        df = loader.load()
    except FileNotFoundError as e:
        print(f"add_hydrometric_target_stats: skipped ({e})")
        return nodes_df

    if date_range is not None:
        start, end = date_range
        df = df[(df["date"] >= start) & (df["date"] <= end)]

    agg_spec = {}
    if "discharge_m3s" in df.columns:
        agg_spec["discharge_m3s"] = ["mean", "std", "count"]
    if "waterlevel_mm" in df.columns:
        agg_spec["waterlevel_mm"] = ["mean", "std", "count"]
    if not agg_spec:
        return nodes_df

    summary = df.groupby("station_code").agg(agg_spec)
    summary.columns = [f"target_{a}_{b}" for a, b in summary.columns]
    return nodes_df.merge(summary, on="station_code", how="left")


def build_node_features(
    station_elevations_path: Optional[Path] = None,
    idpr_path: Optional[Path] = None,
    ades_path: Optional[Path] = None,
    safran_path: Optional[Path] = None,
    hydrometric_path: Optional[Path] = None,
    catchment_path: Optional[Path] = None,
    landcover_path: Optional[Path] = None,
    ndvi_path: Optional[Path] = None,
    bdcavites_path: Optional[Path] = None,
    bdcharm_path: Optional[Path] = None,
    prefix_map=None,
    base_nodes_df: Optional[pd.DataFrame] = None,
    date_range: Optional[Tuple[str, str]] = None,
) -> "tuple[pd.DataFrame, NodeFeatureReport]":
    """
    Orchestrator: build the base station table, then layer on every
    available loader's features. Any path left as None (or pointing to
    missing files) is skipped with a message rather than raising.

    Two ways to get the base table:
      - `station_elevations_path` (original path): builds the 27-gauge
        table via StationElevationsLoader, as before.
      - `base_nodes_df` (for the reach graph): pass an already-built
        node table directly -- e.g. the ~4,500-node reach graph from
        scripts/build_reach_graphs.py (real gauges + real confluences +
        splits/rejoins + virtual infill nodes), which every add_*_features
        function already works against unchanged, since they only ever
        assumed [station_code, latitude, longitude, ...] columns, not a
        specific row count. Extra columns already on base_nodes_df
        (is_gauged, is_confluence, is_split_point, is_rejoin_point,
        braid_id, snap_distance_km) pass through every merge untouched.
        Exactly one of these two must be given.

    date_range: optional (start, end) date strings (inclusive), applied
    to every time-varying source (groundwater, climate, hydrometric
    targets) so all three describe the same period -- otherwise each
    source silently aggregates over its own full, differently-shaped
    historical record (ADES wells reporting from 1972-2026 on wildly
    different schedules, ERA5 spanning 1960-2026, hydrometric records
    with their own per-station date ranges), which makes "climate over
    the period" and "discharge over the period" describe different
    periods without anyone intending that. Static sources (IDPR,
    elevation, catchment area) are unaffected -- they don't vary in time.

    Returns:
        (nodes_df, report)
    """
    if (station_elevations_path is None) == (base_nodes_df is None):
        raise ValueError("Provide exactly one of station_elevations_path or base_nodes_df.")

    if base_nodes_df is not None:
        nodes_df = base_nodes_df.copy()
        report = NodeFeatureReport(n_stations=len(nodes_df), included=["reach graph node table (base)"])
    else:
        nodes_df = build_base_station_table(
            StationElevationsLoader(data_path=station_elevations_path).load(), prefix_map
        )
        report = NodeFeatureReport(n_stations=len(nodes_df), included=["station_elevations (base table)"])

    date_aware = {"ades_groundwater", "safran_climate", "hydrometric_targets"}
    for label, path, fn in [
        ("idpr", idpr_path, add_idpr_features),
        ("catchment_area", catchment_path, add_catchment_features),
        ("landcover", landcover_path, add_landcover_features),
        ("ndvi", ndvi_path, add_ndvi_features),
        ("cavites", bdcavites_path, add_cavites_features),
        ("geology", bdcharm_path, add_geology_features),
        ("ades_groundwater", ades_path, add_groundwater_features),
        ("safran_climate", safran_path, add_safran_features),
        ("hydrometric_targets", hydrometric_path, add_hydrometric_target_stats),
    ]:
        if path is None or not Path(path).exists():
            report.skipped.append(f"{label} (no path given or not found)")
            continue
        try:
            if date_range is not None and label in date_aware:
                nodes_df = fn(nodes_df, path, date_range=date_range)
            else:
                nodes_df = fn(nodes_df, path)
            report.included.append(label if date_range is None or label not in date_aware
                                    else f"{label} (filtered to {date_range[0]}..{date_range[1]})")
        except Exception as e:
            report.skipped.append(f"{label} (error: {e})")

    return nodes_df, report