from __future__ import annotations from enum import Enum from pydantic import BaseModel, ConfigDict, Field, field_validator from domain.tools.analysis import Orientation, TypeBien QUESTION_MAX_CHARS = 2000 ERREURS_COMMUNES = { 400: {"description": "Requête invalide (paramètre manquant, adresse introuvable, injection détectée…)"}, 401: {"description": "Header X-API-Key manquant"}, 403: {"description": "Clé API invalide"}, 422: {"description": "Erreur de validation Pydantic"}, 500: {"description": "Erreur interne du serveur"}, 503: {"description": "Service temporairement indisponible"}, } OPENAPI_TAGS = [ { "name": "Agent", "description": ( "Agent ReAct immobilier — questions en langage naturel. " "Utilise un pipeline **Classify → ReAct → LLM** avec accès aux données DVF 2025 " "et à l'Observatoire des Loyers (Toulouse)." ), }, { "name": "Analyse", "description": ( "Estimation immobilière **sans LLM** basée sur les ventes DVF individuelles 2021-2025 (Nord-59). " "Algorithme : 20 ventes les plus proches (lat/lon parcelle ou centroïde commune) · " "chaque prix/m² est projeté au marché 2025 par composition des évolutions annuelles de `prix_evolution` · " "pondération (1/distance) × récence · ajustements état/orientation/terrain." ), }, { "name": "Monitoring", "description": "État du service, métriques de performance et coûts LLM agrégés.", }, { "name": "Streaming", "description": ( "Stream SSE des logs Python en temps réel. " "Consommable depuis le dashboard ou tout client `EventSource`." ), }, ] class DemandeQuestion(BaseModel): model_config = ConfigDict(json_schema_extra={ "example": {"question": "Quel est le prix au m² à Lille pour un appartement ?"} }) question: str = Field( ..., min_length=1, max_length=QUESTION_MAX_CHARS, description=f"Question en langage naturel (max {QUESTION_MAX_CHARS} caractères)", examples=["Quel est le prix au m² à Lille ?", "Quelle est la rentabilité d'un T2 à Roubaix ?"], ) @field_validator("question") @classmethod def question_non_vide(cls, v: str) -> str: v = v.strip() if not v: raise ValueError("La question ne peut pas être vide.") return v class ReponseQuestion(BaseModel): model_config = ConfigDict(json_schema_extra={ "example": { "reponse": "Le prix moyen au m² à Lille est de **3 200 €/m²** pour les appartements...", "duree_ms": 1842.5, } }) reponse: str = Field(description="Réponse générée par l'agent (Markdown)") duree_ms: float = Field(description="Durée de traitement en millisecondes") class EtatService(BaseModel): model_config = ConfigDict(json_schema_extra={ "example": {"statut": "ok", "modele": "gpt-4o"} }) statut: str = Field(description="État du service : `ok` / `dégradé`") modele: str = Field(description="Identifiant du modèle LLM utilisé") class ErreurAPI(BaseModel): code: str = Field(description="Identifiant machine de l'erreur") detail: str = Field(description="Description lisible de l'erreur") hint: str | None = Field(default=None, description="Conseil pour résoudre le problème") class StatutBien(str, Enum): """État de rénovation du bien — du plus dégradé au plus haut de gamme.""" terrible = "terrible" bad = "bad" light_renovation = "light renovation" medium = "medium" renovated = "renovated" good = "good" premium = "premium" premium_plus = "premium plus" class DemandeAnalysis(BaseModel): model_config = ConfigDict(json_schema_extra={ "example": { "adresse": "7 avenue nelson mandela", "code_postal": 59290, "ville": "wasquehal", "surface_m2": 65.0, "type_bien": "appartement", "statut": "medium", "orientation": "sud", "taille_terrain": 0.0, } }) adresse: str = Field( ..., min_length=2, max_length=300, description="Numéro et nom de la voie (sans code postal ni ville)", examples=["7 avenue nelson mandela", "2 rue de la Paix", "15 place du Général de Gaulle"], ) code_postal: int = Field( ..., ge=1000, le=99999, description="Code postal à 5 chiffres", examples=[59290, 59000, 59800], ) ville: str = Field( ..., min_length=1, max_length=100, description="Nom de la commune", examples=["wasquehal", "Lille", "Roubaix"], ) surface_m2: float = Field( ..., gt=0, le=2000, description="Surface habitable en m²", examples=[45.0, 75.0, 120.0], ) type_bien: TypeBien = Field( ..., description="Type de bien : `maison` ou `appartement`", ) statut: StatutBien = Field( ..., description=( "État de rénovation du bien. " "Coefficients : `terrible` ×0.50 · `bad` ×0.70 · " "`light renovation` ×0.85 · `medium` ×0.95 · `renovated` ×1.00 · " "`good` ×1.10 · `premium` ×1.15 · `premium plus` ×1.30" ), ) orientation: Orientation = Field( ..., description=( "Orientation principale du bien. " "Coefficients : `sud` ×1.10 · `sud-est`/`sud-ouest` ×1.05 · " "`est`/`ouest` ×1.00 · `nord-est`/`nord-ouest` ×0.95 · `nord` ×0.90" ), ) taille_terrain: float = Field( default=0.0, ge=0, le=50000, description=( "Surface du terrain en m² (0 pour un appartement sans terrain). " "Coefficients : 0 m² ×1.00 · ≤100 ×1.03 · ≤300 ×1.08 · " "≤600 ×1.15 · ≤1000 ×1.25 · >1000 ×1.35" ), examples=[0.0, 150.0, 400.0], ) class TransactionRef(BaseModel): model_config = ConfigDict(json_schema_extra={ "example": { "commune": "Wasquehal", "code_postal": "59290", "type_local": "Appartement", "annee": 2023, "date": "2023-06-15", "surface_m2": 65.0, "prix_total": 201500.0, "prix_m2_vente": 3100.0, "evol_marche_pct": 4.8, "prix_m2_ajuste": 3249.0, "distance_km": 0.3, "google_maps_url": "https://www.google.com/maps/search/?api=1&query=...", } }) commune: str = Field(description="Nom de la commune") code_postal: str = Field(description="Code postal (5 chiffres)") type_local: str = Field(description="`Appartement` ou `Maison`") annee: int = Field(description="Année de la vente DVF") date: str = Field(description="Date de la mutation (YYYY-MM-DD)") surface_m2: float = Field(description="Surface du bien vendu (m²)") prix_total: float = Field(description="Prix de vente DVF (€)") prix_m2_vente: float = Field(description="Prix au m² au moment de la vente (€/m²)") evol_marche_pct: float = Field(description="Évolution cumulée du marché appliquée depuis la vente jusqu'à 2025 (%)") prix_m2_ajuste: float = Field(description="Prix/m² de la vente projeté au marché 2025 (€/m²)") distance_km: float = Field(description="Distance au bien estimé (km)") google_maps_url: str = Field(description="Lien Google Maps vers l'adresse de la vente") class EstimationBien(BaseModel): model_config = ConfigDict(json_schema_extra={ "example": { "prix_m2_marche": 3402.0, "coef_renovation": 1.00, "coef_orientation": 1.10, "coef_terrain": 1.08, "coef_total": 1.188, "prix_m2_estime": 4041.6, "prix_total_estime": 363744, } }) prix_m2_marche: float = Field(description="Prix/m² marché pondéré (distance × récence) des 20 ventes DVF les plus proches, projeté au marché 2025 (€/m²)") coef_renovation: float = Field(description="Coefficient état du bien (`renovated`=1.00)") coef_orientation: float = Field(description="Coefficient orientation (`est`/`ouest`=1.00)") coef_terrain: float = Field(description="Coefficient surface terrain (0 m²=1.00)") coef_total: float = Field(description="Coefficient global = renovation × orientation × terrain") prix_m2_estime: float = Field(description="Prix/m² estimé = prix_m2_marche × coef_total (€/m²)") prix_total_estime: int = Field(description="Estimation du prix total du bien (€)") class ReponseAnalysis(BaseModel): model_config = ConfigDict(json_schema_extra={ "example": { "adresse": "7 avenue nelson mandela", "code_postal": 59290, "ville": "wasquehal", "surface_m2": 65.0, "type_bien": "appartement", "statut": "medium", "orientation": "sud", "taille_terrain": 0.0, "google_maps_url": "https://www.google.com/maps/search/?api=1&query=7+avenue+nelson+mandela%2C+59290+wasquehal", "estimation": { "prix_m2_marche": 3402.0, "coef_renovation": 1.00, "coef_orientation": 1.10, "coef_terrain": 1.08, "coef_total": 1.188, "prix_m2_estime": 4041.6, "prix_total_estime": 363744, }, "transactions_reference": [], } }) adresse: str code_postal: int ville: str surface_m2: float type_bien: str statut: str orientation: str taille_terrain: float google_maps_url: str = Field(description="Lien Google Maps cliquable vers l'adresse estimée") estimation: EstimationBien transactions_reference: list[TransactionRef] = Field(description="20 ventes DVF les plus proches (2021-2025), triées par distance, avec le prix/m² projeté au marché 2025") class Metriques(BaseModel): model_config = ConfigDict(json_schema_extra={ "example": { "total_requetes": 42, "requetes_en_erreur": 1, "taux_erreur_pct": 2.38, "duree_moyenne_ms": 1250.0, "duree_max_ms": 3800.0, "tokens_prompt_total": 85000, "tokens_completion_total": 12000, "cout_total_usd": 0.42, "cout_par_requete_usd": 0.01, "roi_economies_usd": 3.5, "roi_pct": 833.0, "derniere_requete": None, } }) total_requetes: int requetes_en_erreur: int taux_erreur_pct: float = Field(description="Taux d'erreur en pourcentage") duree_moyenne_ms: float = Field(description="Latence moyenne (ms)") duree_max_ms: float = Field(description="Latence maximale (ms)") tokens_prompt_total: int = Field(description="Total tokens de prompt consommés") tokens_completion_total: int = Field(description="Total tokens de complétion consommés") cout_total_usd: float = Field(description="Coût LLM total estimé (USD)") cout_par_requete_usd: float = Field(description="Coût moyen par requête (USD)") roi_economies_usd: float = Field(description="Économies réalisées grâce au cache (USD)") roi_pct: float = Field(description="ROI du cache en pourcentage") derniere_requete: dict | None = Field(description="Métadonnées de la dernière requête")