scripts/download_bdcavites.py

"""
Downloads BDCavités (BRGM's national underground cavity inventory --
karst sinkholes, quarries, marl pits, natural cavities) for the Eure/
Risle area, via Géorisques' WFS.

Directly relevant to the bétoire investigation: this is an independent,
purpose-built dataset for exactly this phenomenon, unlike inferring it
from Hub'Eau station naming or BD TOPO's provisional karst attribute.

CONFIRMED (fetched GetCapabilities directly): endpoint
https://georisques.gouv.fr/services, WFS 1.1.0, typeName
CAVITE_LOCALISEE ("Cavités souterraines abandonnées d'origine non
minière"), GeoJSON output supported directly. Axis order for this
specific server is NOT separately confirmed -- reusing the same
lon,lat-then-lat,lon retry that scripts/download_bdtopo_hydro.py needed
for a different WFS server, since different servers have behaved
differently on this before and there's no reason to assume this one
won't too.

Usage:
    python -m scripts.download_bdcavites --check
    python -m scripts.download_bdcavites
"""
import argparse
import json
import sys
from pathlib import Path

import requests

WFS_URL = "https://georisques.gouv.fr/services"
TYPE_NAME = "CAVITE_LOCALISEE"

# Same widened bbox as scripts/download_bdtopo_hydro.py's BBOX (updated
# after confirmed real evidence of truncation on the original, tighter
# box) -- keeping both scripts scoped to the same area.
BBOX = (-0.1, 47.7, 2.1, 49.9)  # (min_lon, min_lat, max_lon, max_lat)


def check_typename() -> bool:
    params = {
        "SERVICE": "WFS", "VERSION": "1.1.0", "REQUEST": "GetFeature",
        "TYPENAME": TYPE_NAME, "MAXFEATURES": 1, "OUTPUTFORMAT": "application/json; subtype=geojson; charset=utf-8",
    }
    resp = requests.get(WFS_URL, params=params, timeout=30)
    ok, detail = _parse_response(resp)
    print(f"  {TYPE_NAME}: {'OK' if ok else 'FAILED'}")
    if not ok:
        print(f"    {detail}")
    return ok


def _parse_response(resp: "requests.Response"):
    if resp.status_code != 200:
        return False, f"HTTP {resp.status_code}: {resp.text[:400]}"
    try:
        data = resp.json()
    except ValueError:
        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, ""
    return False, f"Response was JSON but not a FeatureCollection: {str(data)[:400]}"


def fetch_all_pages(bbox_param: str, page_size: int = 1000) -> list:
    all_features = []
    start_index = 0
    while True:
        params = {
            "SERVICE": "WFS", "VERSION": "1.1.0", "REQUEST": "GetFeature",
            "TYPENAME": TYPE_NAME, "BBOX": bbox_param,
            "OUTPUTFORMAT": "application/json; subtype=geojson; charset=utf-8",
            "MAXFEATURES": page_size, "STARTINDEX": start_index,
        }
        resp = requests.get(WFS_URL, params=params, timeout=60)
        ok, detail = _parse_response(resp)
        if not ok:
            print(f"    ERROR at startIndex={start_index}: {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 fetch_with_axis_retry(bbox: tuple, out_path: Path) -> None:
    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"),
    ]
    features = []
    for label, bbox_param in orderings:
        print(f"  trying axis order {label}...")
        features = fetch_all_pages(bbox_param)
        if features:
            print(f"  -> {label} worked ({len(features)} feature(s))")
            break
        print(f"  -> {label} returned 0 features")

    geojson = {"type": "FeatureCollection", "features": features}
    out_path.write_text(json.dumps(geojson))
    if features:
        print(f"Saved {len(features)} feature(s) to {out_path}")
    else:
        print(f"Saved an EMPTY file to {out_path} -- both axis orders returned nothing. "
              f"Run --check first, or this area may genuinely have zero recorded cavities "
              f"(plausible -- BDCavités coverage is built department-by-department and "
              f"isn't uniformly complete everywhere).")


def main() -> None:
    parser = argparse.ArgumentParser(description="Download BDCavités for the Eure/Risle area")
    parser.add_argument("--check", action="store_true")
    parser.add_argument("--output-dir", type=Path, default=Path("datasets/bdcavites"))
    args = parser.parse_args()

    if args.check:
        print("Testing typeName against the live service...")
        ok = check_typename()
        if not ok:
            print("\nFailed -- the typeName or service details may have changed since this "
                  "script was written. Fetch GetCapabilities directly to check:")
            print(f"  {WFS_URL}?SERVICE=WFS&VERSION=1.1.0&REQUEST=GetCapabilities")
        else:
            print("\nOK. Re-run without --check to download.")
        return

    args.output_dir.mkdir(parents=True, exist_ok=True)
    print("Fetching CAVITE_LOCALISEE (BDCavités)...")
    fetch_with_axis_retry(BBOX, args.output_dir / "cavite_localisee.geojson")
    print()
    print("Once downloaded, cross-reference against the amont/aval bétoire stations "
          "(H605641101 at 48.98492,0.78902 and H605641201 at 49.04707,0.79984) the same "
          "way scripts/analyze_bdtopo_hydro.py's check_karst_near_betoire did for the "
          "BD TOPO karst attribute -- this is a genuinely independent second check, "
          "not a re-run of the same one.")


if __name__ == "__main__":
    main()