fn_registry/python/functions/infra/overpass_nearby_pois.py

"""Consulta la Overpass API para obtener POIs cercanos a una coordenada."""

import json
import urllib.error
import urllib.parse
import urllib.request

# Mapeo de categoría legible → (key OSM, value OSM)
# None como value significa wildcard (key=*)
_CATEGORY_TAGS: dict[str, tuple[str, str | None]] = {
    "restaurant": ("amenity", "restaurant"),
    "cafe": ("amenity", "cafe"),
    "bar": ("amenity", "bar"),
    "museum": ("tourism", "museum"),
    "monument": ("historic", "monument"),
    "park": ("leisure", "park"),
    "library": ("amenity", "library"),
    "theatre": ("amenity", "theatre"),
    "cinema": ("amenity", "cinema"),
    "gallery": ("tourism", "gallery"),
    "historic": ("historic", None),
    "tourism": ("tourism", None),
    "shop": ("shop", None),
    "hotel": ("tourism", "hotel"),
    "pharmacy": ("amenity", "pharmacy"),
    "hospital": ("amenity", "hospital"),
}

_DEFAULT_CATEGORIES = [
    "restaurant", "cafe", "bar", "museum", "monument", "park",
    "library", "theatre", "cinema", "gallery", "historic", "tourism", "shop",
]

_OVERPASS_ENDPOINT = "https://overpass-api.de/api/interpreter"


def _build_query(lat: float, lon: float, radius_m: int, categories: list[str]) -> str:
    """Construye una query Overpass QL con union de nodos."""
    lines = ["[out:json][timeout:10];", "("]
    for cat in categories:
        if cat not in _CATEGORY_TAGS:
            continue
        key, value = _CATEGORY_TAGS[cat]
        if value is None:
            lines.append(f'  node["{key}"](around:{radius_m},{lat},{lon});')
        else:
            lines.append(f'  node["{key}"="{value}"](around:{radius_m},{lat},{lon});')
    lines.append(");")
    lines.append("out body;")
    return "\n".join(lines)


def _extract_category(tags: dict) -> str:
    """Determina la categoría legible a partir de los tags OSM del nodo."""
    for cat, (key, value) in _CATEGORY_TAGS.items():
        if key not in tags:
            continue
        if value is None or tags[key] == value:
            return cat
    return "unknown"


def overpass_nearby_pois(
    lat: float,
    lon: float,
    radius_m: int = 500,
    categories: list[str] | None = None,
) -> list[dict]:
    """Obtiene POIs cercanos a una coordenada usando la Overpass API.

    Args:
        lat: latitud del punto central.
        lon: longitud del punto central.
        radius_m: radio de busqueda en metros (defecto 500).
        categories: lista de categorias a buscar. Si es None, usa todas las
            categorias comunes definidas internamente.

    Returns:
        lista de dicts con id, lat, lon, name, category y tags de cada POI.

    Raises:
        RuntimeError: si la request falla por error HTTP, timeout u otro error de red.
    """
    active_categories = categories if categories is not None else _DEFAULT_CATEGORIES
    query = _build_query(lat, lon, radius_m, active_categories)

    encoded_query = urllib.parse.quote(query)
    form_body = f"data={encoded_query}".encode("utf-8")
    req = urllib.request.Request(
        _OVERPASS_ENDPOINT,
        data=form_body,
        method="POST",
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )

    try:
        with urllib.request.urlopen(req, timeout=10) as response:
            raw = response.read()
    except urllib.error.HTTPError as e:
        body_preview = e.read(200).decode("utf-8", errors="replace")
        raise RuntimeError(
            f"Overpass API HTTP error {e.code} for ({lat},{lon}) r={radius_m}m: {body_preview}"
        ) from e
    except urllib.error.URLError as e:
        raise RuntimeError(
            f"Overpass API request failed for ({lat},{lon}) r={radius_m}m: {e.reason}"
        ) from e

    try:
        result = json.loads(raw)
    except json.JSONDecodeError as e:
        raise RuntimeError(
            f"Overpass API returned invalid JSON for ({lat},{lon}): {e}"
        ) from e

    pois: list[dict] = []
    for element in result.get("elements", []):
        if element.get("type") != "node":
            continue
        tags = element.get("tags", {})
        category = _extract_category(tags)
        # Nombre: tag "name", o valor del tag de categoría, o "Unknown"
        if "name" in tags:
            name = tags["name"]
        else:
            key = _CATEGORY_TAGS.get(category, (None, None))[0]
            name = tags.get(key, "Unknown") if key else "Unknown"

        pois.append({
            "id": element["id"],
            "lat": element["lat"],
            "lon": element["lon"],
            "name": name,
            "category": category,
            "tags": tags,
        })

    return pois