scripts/download_bdtopo_hydro.py

"""
Download BD TOPO hydrography (troncon_hydrographique, surface_hydrographique)
for the Eure/Risle area via IGN's Geoplateforme WFS, instead of pulling the
whole national dataset.

VERIFICATION NOTE: I confirmed the WFS endpoint itself is real and live
(https://data.geopf.fr/wfs, WFS 2.0.0) and that a "BDTOPO_V3" namespace
exists on it, by fetching GetCapabilities directly. I could NOT confirm
the exact typeName for the hydrography layers within it — the full
capabilities document is huge (thousands of layers across all IGN
products) and got truncated before reaching them. DEFAULT_TYPENAMES
below is my best guess based on IGN's usual naming convention, NOT a
verified value. This script's first step (check_typenames) tests each
name with a tiny query before attempting the real extraction, and tells
you exactly how to find the right name yourself if my guess is wrong.

Usage:
    python download_bdtopo_hydro.py --check          # verify typeNames first
    python download_bdtopo_hydro.py                  # then run the real extraction
"""
import argparse
import sys
from pathlib import Path

import requests
import pandas as pd

WFS_URL = "https://data.geopf.fr/wfs"

# Best-guess typeNames -- see VERIFICATION NOTE above. Override with
# --troncon-typename / --surface-typename / --catchment-typename if wrong.
# troncon/surface confirmed correct against the real service (see
# CONFIRMED note below); catchment is still an unverified guess.
DEFAULT_TRONCON_TYPENAME = "BDTOPO_V3:troncon_hydrographique"
DEFAULT_SURFACE_TYPENAME = "BDTOPO_V3:surface_hydrographique"
DEFAULT_CATCHMENT_TYPENAME = "BDTOPO_V3:bassin_versant_topographique"

# CONFIRMED: both troncon_hydrographique and surface_hydrographique
# typeNames work against the live service (verified in this project's
# actual run: 30045 tronçons fetched successfully). bassin_versant_
# topographique has NOT been verified the same way -- run --check
# before trusting it.

# Bounding box around the Eure/Risle stations, in WGS84 (matches the
# lon/lat range in your real station_list.csv, with a small margin).
BBOX = (0.3, 48.3, 1.7, 49.5)  # (min_lon, min_lat, max_lon, max_lat)


def find_hydro_typenames(keyword: str = "hydrographique") -> list:
    """
    Fetch GetCapabilities and search for FeatureType <Name> entries
    containing `keyword`. This is the fallback if DEFAULT_*_TYPENAME
    turns out to be wrong -- run this to get the real list.
    """
    print(f"Fetching WFS GetCapabilities and searching for '{keyword}'...")
    resp = requests.get(WFS_URL, params={
        "SERVICE": "WFS", "VERSION": "2.0.0", "REQUEST": "GetCapabilities",
    }, timeout=60)
    resp.raise_for_status()

    import re
    names = re.findall(r"<Name>([^<]*)</Name>", resp.text)
    matches = [n for n in names if keyword.lower() in n.lower()]
    print(f"Found {len(matches)} matching layer(s):")
    for m in matches:
        print(f"  {m}")
    return matches


def check_typename(type_name: str) -> bool:
    """Tiny test query (count=1, no bbox) to confirm a typeName is valid
    before running the real extraction against it."""
    params = {
        "SERVICE": "WFS", "VERSION": "2.0.0", "REQUEST": "GetFeature",
        "TYPENAMES": type_name, "COUNT": 1, "OUTPUTFORMAT": "application/json",
    }
    resp = requests.get(WFS_URL, params=params, timeout=30)
    ok, error_detail = _parse_wfs_response(resp)
    status = "OK" if ok else f"FAILED"
    print(f"  {type_name}: {status}")
    if not ok:
        print(f"    {error_detail}")
    return ok


def _parse_wfs_response(resp: "requests.Response") -> "tuple[bool, str]":
    """
    Check whether a WFS response is a real FeatureCollection or an error
    (WFS ExceptionReport, or a non-JSON/HTML error page). Returns
    (is_valid, detail) -- detail is empty on success, otherwise a
    human-readable explanation of what actually came back, so a failure
    is never silently swallowed into "0 features".
    """
    if resp.status_code != 200:
        return False, f"HTTP {resp.status_code}: {resp.text[:400]}"

    try:
        data = resp.json()
    except ValueError:
        # Not JSON at all -- almost always means an XML ExceptionReport
        # or an HTML error page came back instead.
        return False, f"Response was not JSON (likely a WFS ExceptionReport): {resp.text[:400]}"

    if isinstance(data, dict) and data.get("type") == "FeatureCollection":
        return True, ""

    # Some WFS error responses come back as JSON but without a
    # FeatureCollection shape (e.g. {"exceptionText": [...]})
    return False, f"Response was JSON but not a FeatureCollection: {str(data)[:400]}"


def fetch_features(type_name: str, bbox: tuple, out_path: Path, srs: str = "EPSG:4326") -> None:
    """
    GetFeature request over the given bbox, paginated via startIndex
    since a national layer clipped to a small bbox can still exceed the
    server's per-request feature limit.

    AXIS ORDER: when BBOX specifies its CRS via the URN form
    (urn:ogc:def:crs:EPSG::4326), the OGC spec requires lat,lon axis
    order -- NOT the traditional lon,lat most tools use. Getting this
    backwards doesn't error, it just silently matches zero real-world
    features (a well-known WFS 2.0 gotcha). Rather than guess, this
    tries lon,lat first and automatically retries with lat,lon swapped
    if the first attempt comes back empty, so a silent axis-order bug
    can't hide as "no data in this area" again.
    """
    min_lon, min_lat, max_lon, max_lat = bbox
    orderings = [
        ("lon,lat", f"{min_lon},{min_lat},{max_lon},{max_lat},urn:ogc:def:crs:EPSG::4326"),
        ("lat,lon", f"{min_lat},{min_lon},{max_lat},{max_lon},urn:ogc:def:crs:EPSG::4326"),
    ]

    all_features = []
    for label, bbox_param in orderings:
        print(f"  trying axis order {label}...")
        all_features = _fetch_paginated(type_name, bbox_param)
        if all_features:
            print(f"  -> {label} worked ({len(all_features)} feature(s)). "
                  f"Use this axis order for future queries against this endpoint.")
            break
        print(f"  -> {label} returned 0 features.")

    import json
    geojson = {"type": "FeatureCollection", "features": all_features}
    out_path.write_text(json.dumps(geojson))
    if all_features:
        print(f"Saved {len(all_features)} feature(s) to {out_path}")
    else:
        print(f"Saved an EMPTY file to {out_path} -- both axis orders returned "
              f"nothing real. The typeName is valid (confirmed by --check), so this "
              f"suggests the bbox itself doesn't overlap any features in this layer, "
              f"or another parameter is off. Try a much larger bbox (e.g. all of "
              f"France: -5,41,10,51) as a sanity check.")


def _fetch_paginated(type_name: str, bbox_param: str) -> list:
    """One bbox ordering's worth of paginated GetFeature calls."""
    all_features = []
    start_index = 0
    page_size = 1000

    while True:
        params = {
            "SERVICE": "WFS", "VERSION": "2.0.0", "REQUEST": "GetFeature",
            "TYPENAMES": type_name, "BBOX": bbox_param,
            "OUTPUTFORMAT": "application/json",
            "COUNT": page_size, "STARTINDEX": start_index,
        }
        resp = requests.get(WFS_URL, params=params, timeout=60)
        is_valid, error_detail = _parse_wfs_response(resp)
        if not is_valid:
            print(f"    ERROR at startIndex={start_index}: {error_detail}")
            break
        data = resp.json()
        features = data.get("features", [])
        if not features:
            break
        all_features.extend(features)
        if len(features) < page_size:
            break
        start_index += page_size

    return all_features


def main() -> None:
    parser = argparse.ArgumentParser(description="Download BD TOPO hydrography for Eure/Risle")
    parser.add_argument("--check", action="store_true",
                         help="Only verify typeNames, don't download anything")
    parser.add_argument("--troncon-typename", default=DEFAULT_TRONCON_TYPENAME)
    parser.add_argument("--surface-typename", default=DEFAULT_SURFACE_TYPENAME)
    parser.add_argument("--catchment-typename", default=DEFAULT_CATCHMENT_TYPENAME)
    parser.add_argument("--skip-catchments", action="store_true",
                         help="Skip bassin_versant_topographique (untested typeName)")
    parser.add_argument("--output-dir", type=Path, default=Path("datasets/bdtopo_hydro"))
    args = parser.parse_args()

    if args.check:
        print("=" * 70)
        print("Testing default typeNames...")
        print("=" * 70)
        ok_troncon = check_typename(args.troncon_typename)
        ok_surface = check_typename(args.surface_typename)
        ok_catchment = args.skip_catchments or check_typename(args.catchment_typename)
        if not (ok_troncon and ok_surface and ok_catchment):
            print()
            print("At least one default typeName failed. Searching capabilities "
                  "for the real name(s) instead:")
            find_hydro_typenames("hydrographique")
            find_hydro_typenames("bassin_versant")
            print()
            print("Re-run with --troncon-typename / --surface-typename / "
                  "--catchment-typename set to whatever the search above found, "
                  "then --check again to confirm.")
        else:
            print()
            print("All typeNames OK. Re-run without --check to download the real data.")
        return

    args.output_dir.mkdir(parents=True, exist_ok=True)

    print("Fetching troncon_hydrographique (river centerline geometry + attributes)...")
    fetch_features(args.troncon_typename, BBOX, args.output_dir / "troncon_hydrographique.geojson")

    print()
    print("Fetching surface_hydrographique (includes karst/Nature classification)...")
    fetch_features(args.surface_typename, BBOX, args.output_dir / "surface_hydrographique.geojson")

    if not args.skip_catchments:
        print()
        print("Fetching bassin_versant_topographique (catchment polygons)...")
        print("NOTE: this typeName is unverified -- if it fails, run --check first "
              "to find the real name, or pass --skip-catchments to skip it.")
        fetch_features(args.catchment_typename, BBOX, args.output_dir / "bassin_versant_topographique.geojson")

    print()
    print("Done. Load with: BDTopoHydroLoader(data_path=...)")
    print("To check for the karst-flagged reach near Grosley-sur-Risle/Ajou, use "
          "loader.check_karst_near_point(lat, lon) -- that's the check that would "
          "confirm or deny the bétoire location.")


if __name__ == "__main__":
    main()