flowdeck/app/services/geocoding.py

"""FlowDeck — Geocoding Service (vue Map §13.4).

Interface de fournisseur unique : search / resolve / reverse.
Implementations : ``none`` (defaut — saisie manuelle uniquement, aucun
appel externe), ``nominatim`` (OpenStreetMap, User-Agent identifie,
<= 1 req/s, cache obligatoire), ``custom`` (endpoint compatible Nominatim).

Le cache SQLite (migration 45, table ``geocode_cache``) est consulte
avant tout appel. Les journaux ne contiennent jamais d'adresses en clair
au-dela de compteurs (le texte reste dans le cache, purgeable).
"""
from __future__ import annotations

import hashlib
import json
import logging
import time
import urllib.parse
import urllib.request
from typing import Any

logger = logging.getLogger(__name__)

_last_call_ts = 0.0


def _norm(query: str) -> str:
    return " ".join((query or "").lower().split())


def query_hash(query: str, provider: str) -> str:
    return hashlib.sha256(f"{provider}:{_norm(query)}".encode()).hexdigest()


def get_settings(conn=None) -> dict:
    """Lit la config geocodeur (instance). Defauts surs : provider=none."""
    try:
        from app.config import settings as _s  # type: ignore
        provider = getattr(_s, "GEOCODER_PROVIDER", None) or "none"
        endpoint = getattr(_s, "GEOCODER_ENDPOINT", None) or "https://nominatim.openstreetmap.org"
        return {"provider": provider, "endpoint": endpoint.rstrip("/")}
    except Exception:
        return {"provider": "none", "endpoint": "https://nominatim.openstreetmap.org"}


def cache_get(conn, qhash: str) -> dict | None:
    try:
        row = conn.execute(
            "SELECT result_json, lat, lng, hit_count FROM geocode_cache WHERE query_hash=?",
            (qhash,),
        ).fetchone()
    except Exception:
        return None
    if not row:
        return None
    try:
        conn.execute("UPDATE geocode_cache SET hit_count = hit_count + 1 WHERE query_hash=?", (qhash,))
    except Exception:
        pass
    try:
        data = json.loads(row["result_json"]) if row["result_json"] else {}
    except (json.JSONDecodeError, TypeError):
        data = {}
    data.setdefault("lat", row["lat"])
    data.setdefault("lng", row["lng"])
    return data


def cache_put(conn, qhash: str, provider: str, query_text: str, result: dict) -> None:
    try:
        conn.execute(
            """INSERT OR REPLACE INTO geocode_cache
               (query_hash, provider, query_text, result_json, lat, lng, created_at, hit_count)
               VALUES (?, ?, ?, ?, ?, ?, datetime('now'), COALESCE(
                 (SELECT hit_count FROM geocode_cache WHERE query_hash=?), 0))""",
            (qhash, provider, query_text[:300], json.dumps(result),
             result.get("lat"), result.get("lng"), qhash),
        )
    except Exception:
        pass


def _http_json(url: str, params: dict, timeout: float = 3.0) -> Any:
    global _last_call_ts
    # Politesse Nominatim : <= 1 req/s.
    wait = 1.0 - (time.time() - _last_call_ts)
    if wait > 0:
        time.sleep(wait)
    full = url + "?" + urllib.parse.urlencode(params)
    req = urllib.request.Request(full, headers={
        "User-Agent": "FlowDeck/1.0 (self-hosted; contact: admin)",
        "Accept": "application/json",
    })
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        _last_call_ts = time.time()
        return json.loads(resp.read().decode("utf-8", "replace"))


def search(conn, query: str, limit: int = 5) -> list[dict]:
    """Autocompletion d'adresses. Vide quand provider=none."""
    cfg = get_settings(conn)
    provider = cfg["provider"]
    if not query or len(query.strip()) < 3 or provider == "none":
        return []
    qhash = query_hash(f"search:{query}:{limit}", provider)
    cached = cache_get(conn, qhash)
    if cached and isinstance(cached.get("suggestions"), list):
        return cached["suggestions"][:limit]
    base = cfg["endpoint"]
    try:
        if provider in ("nominatim", "custom"):
            data = _http_json(f"{base}/search", {
                "q": query, "format": "jsonv2", "limit": limit,
                "addressdetails": 1,
            })
            suggestions = [{
                "name": (d.get("display_name") or "").split(",")[0].strip(),
                "address": d.get("display_name", ""),
                "lat": float(d["lat"]), "lng": float(d["lon"]),
                "provider": provider, "place_id": str(d.get("place_id", "")),
            } for d in (data or []) if d.get("lat") and d.get("lon")]
        else:
            suggestions = []
    except Exception as exc:
        logger.warning("geocode search failed: %s", type(exc).__name__)
        return []
    try:
        cache_put(conn, qhash, provider, query, {"suggestions": suggestions})
        conn.commit()
    except Exception:
        pass
    return suggestions


def resolve(conn, query: str) -> dict | None:
    """Adresse -> {lat, lng, ...}. Cache d'abord, None si echec."""
    cfg = get_settings(conn)
    provider = cfg["provider"]
    if not query or provider == "none":
        return None
    qhash = query_hash(f"resolve:{query}", provider)
    cached = cache_get(conn, qhash)
    if cached and cached.get("lat") is not None:
        return cached
    results = search(conn, query, limit=1)
    if not results:
        return None
    first = dict(results[0])
    try:
        cache_put(conn, qhash, provider, query, first)
        conn.commit()
    except Exception:
        pass
    return first