""" services/risques_service.py ============================ Données de risques géographiques depuis SQLite : - Sites Seveso : table seveso_sites - Risque inondation : GEORISQUES gaspar API (mis en cache en mémoire) - Radon : table radon_communes - Retrait-Gonflement : table rga_communes Isochrones (voiture / TC) → domain.services.isochrone_service """ from __future__ import annotations import json import logging import httpx import pymysql from fastapi import HTTPException from domain.db import get_db from domain.services.isochrone_service import get_isochrone # re-exported logger = logging.getLogger(__name__) _GEORISQUES_GASPAR_URL = "https://www.georisques.gouv.fr/api/v1/gaspar/risques" _flood_cache: dict[str, dict] = {} _RADON_COLORS: dict[str, str] = {"1": "#52b788", "2": "#f4a261", "3": "#e63946"} _RADON_LABELS: dict[str, str] = { "1": "Classe 1 — Potentiel faible", "2": "Classe 2 — Potentiel moyen", "3": "Classe 3 — Potentiel élevé", } _RGA_COLORS: dict[str, str] = { "0": "#d8f3dc", "1": "#ffe066", "2": "#f4a261", "3": "#e63946", "4": "#9d0208", } _RGA_LABELS: dict[str, str] = { "0": "Risque nul", "1": "Risque faible", "2": "Risque moyen", "3": "Risque fort", "4": "Risque très fort", } __all__ = [ "get_seveso_sites", "get_flood_risk", "get_isochrone", "get_radon_communes_geojson", "get_rga_communes_geojson", ] # --------------------------------------------------------------------------- # Sites Seveso # --------------------------------------------------------------------------- def get_seveso_sites(departement: str = "") -> dict: """ Retourne les sites Seveso depuis la table seveso_sites (SQLite). Filtre par code_departement si fourni. Raises: HTTPException 503: table absente (build_seveso.py n'a pas été exécuté). """ try: with get_db() as conn: try: if departement: rows = conn.execute( "SELECT * FROM seveso_sites WHERE code_departement = ?", (departement,), ).fetchall() else: rows = conn.execute("SELECT * FROM seveso_sites").fetchall() except pymysql.OperationalError as exc: if "no such table" in str(exc): logger.error("Table seveso_sites absente — exécuter build_seveso.py") raise HTTPException( status_code=503, detail="Table seveso_sites absente — exécuter build_seveso.py", ) raise except HTTPException: raise except Exception as exc: logger.error("Erreur lecture seveso_sites | %s", exc, exc_info=True) raise HTTPException(status_code=503, detail=str(exc)) features = [ { "type": "Feature", "geometry": {"type": "Point", "coordinates": [row["longitude"], row["latitude"]]}, "properties": { "nom": row["nom"], "commune": row["commune"], "code_postal": row["code_postal"], "adresse": row["adresse"], "seveso": row["seveso"], "statut": row["statut"], }, } for row in rows ] logger.info("Seveso — dept=%s | %d sites retournés", departement or "France", len(features)) return {"type": "FeatureCollection", "features": features} # --------------------------------------------------------------------------- # Risque inondation # --------------------------------------------------------------------------- async def get_flood_risk(lat: float, lng: float) -> dict: """ Retourne le risque d'inondation GEORISQUES gaspar pour un point. Résultat mis en cache par couple (lat arrondi, lng arrondi). Raises: HTTPException 502/504: erreur ou timeout GEORISQUES. """ key = f"{round(lat, 2)},{round(lng, 2)}" if key in _flood_cache: logger.debug("Flood risk servi depuis le cache | key=%s", key) return _flood_cache[key] logger.info("Requête risque inondation GEORISQUES | lat=%s lng=%s", lat, lng) try: async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get( _GEORISQUES_GASPAR_URL, params={"latlon": f"{lng},{lat}", "rayon": 2000, "page": 1, "page_size": 10}, headers={"User-Agent": "agent-immobilier/1.0"}, ) if not resp.is_success: logger.error( "GEORISQUES gaspar erreur HTTP | status=%d | lat=%s lng=%s", resp.status_code, lat, lng, ) raise HTTPException(status_code=502, detail=f"GEORISQUES {resp.status_code}") data = resp.json() _flood_cache[key] = data logger.debug("Flood risk mis en cache | key=%s", key) return data except HTTPException: raise except httpx.TimeoutException: logger.error("Timeout GEORISQUES gaspar | lat=%s lng=%s", lat, lng) raise HTTPException(status_code=504, detail="GEORISQUES timeout") except Exception as exc: logger.error( "Erreur inattendue GEORISQUES gaspar | type=%s | lat=%s lng=%s | erreur=%s", type(exc).__name__, lat, lng, exc, exc_info=True, ) raise HTTPException(status_code=502, detail=str(exc)) # --------------------------------------------------------------------------- # Radon # --------------------------------------------------------------------------- def get_radon_communes_geojson( min_lng: float, min_lat: float, max_lng: float, max_lat: float, ) -> dict: """ GeoJSON choroplèthe des communes avec classe de potentiel radon (1/2/3). Raises: RuntimeError: si la table radon_communes est absente. """ try: with get_db() as conn: rows = conn.execute( """SELECT code_insee, nom_commune, classe, geom_type, coordinates FROM radon_communes WHERE bbox_min_lng < ? AND bbox_max_lng > ? AND bbox_min_lat < ? AND bbox_max_lat > ?""", [max_lng, min_lng, max_lat, min_lat], ).fetchall() except pymysql.Error as exc: logger.warning("radon_communes DB error: %s", exc) raise RuntimeError( "Table 'radon_communes' introuvable — exécutez build_radon.py" ) from exc features = [] for r in rows: classe = r["classe"] or "" features.append({ "type": "Feature", "geometry": {"type": r["geom_type"], "coordinates": json.loads(r["coordinates"])}, "properties": { "code": r["code_insee"], "nom": r["nom_commune"], "classe": classe, "color": _RADON_COLORS.get(classe, "#adb5bd"), "label": _RADON_LABELS.get(classe, "Données indisponibles"), }, }) logger.debug( "radon_communes | bbox=[%.3f,%.3f,%.3f,%.3f] → %d features", min_lng, min_lat, max_lng, max_lat, len(features), ) return {"type": "FeatureCollection", "features": features} # --------------------------------------------------------------------------- # RGA # --------------------------------------------------------------------------- def get_rga_communes_geojson( min_lng: float, min_lat: float, max_lng: float, max_lat: float, ) -> dict: """ GeoJSON choroplèthe des communes avec exposition RGA (codeExposition 0–4). Raises: RuntimeError: si la table rga_communes est absente. """ try: with get_db() as conn: rows = conn.execute( """SELECT code_insee, nom_commune, code_exposition, geom_type, coordinates FROM rga_communes WHERE bbox_min_lng < ? AND bbox_max_lng > ? AND bbox_min_lat < ? AND bbox_max_lat > ?""", [max_lng, min_lng, max_lat, min_lat], ).fetchall() except pymysql.Error as exc: logger.warning("rga_communes DB error: %s", exc) raise RuntimeError( "Table 'rga_communes' introuvable — exécutez build_rga.py" ) from exc features = [] for r in rows: code_exp = r["code_exposition"] or "" features.append({ "type": "Feature", "geometry": {"type": r["geom_type"], "coordinates": json.loads(r["coordinates"])}, "properties": { "code": r["code_insee"], "nom": r["nom_commune"], "codeExposition": code_exp, "color": _RGA_COLORS.get(code_exp, "#adb5bd"), "label": _RGA_LABELS.get(code_exp, "Données indisponibles"), }, }) logger.debug( "rga_communes | bbox=[%.3f,%.3f,%.3f,%.3f] → %d features", min_lng, min_lat, max_lng, max_lat, len(features), ) return {"type": "FeatureCollection", "features": features}