""" Wrapper pour intégrer Modal ML dans Gradio Ce module fait le pont entre l'interface Gradio et les fonctions Modal ML Avec authentification automatique Modal depuis HuggingFace Space """ import asyncio import json import logging import os from typing import Dict, List, Any, Optional, Tuple from datetime import datetime import pandas as pd # Import Modal pour l'API distante avec authentification try: import modal MODAL_AVAILABLE = True # Configuration automatique de l'authentification Modal # depuis les variables d'environnement HuggingFace Space modal_token_id = os.environ.get("MODAL_TOKEN_ID") modal_token_secret = os.environ.get("MODAL_TOKEN_SECRET") if modal_token_id and modal_token_secret: # Configurer l'authentification Modal pour HuggingFace Space os.environ["MODAL_TOKEN_ID"] = modal_token_id os.environ["MODAL_TOKEN_SECRET"] = modal_token_secret logging.info("🔐 Authentification Modal configurée depuis HuggingFace Space") else: logging.warning("⚠️ Tokens Modal non trouvés dans les secrets HuggingFace Space") # Utiliser l'API Modal distante au lieu des imports locaux # Ceci permet la communication depuis HuggingFace Space APP_NAME = "odoo-lead-analysis-improved" # Références aux fonctions Modal distantes avec authentification try: generate_synthetic_leads = modal.Function.from_name(APP_NAME, "generate_synthetic_leads") train_improved_model = modal.Function.from_name(APP_NAME, "train_improved_model") predict_lead_conversion_improved = modal.Function.from_name(APP_NAME, "predict_lead_conversion_improved") monitor_model_performance = modal.Function.from_name(APP_NAME, "monitor_model_performance") logging.info(f"✅ Connexion Modal établie avec l'app '{APP_NAME}'") except Exception as e: logging.error(f"❌ Erreur connexion Modal: {e}") MODAL_AVAILABLE = False except ImportError: logging.warning("Modal non disponible - fonctionnalités ML désactivées") MODAL_AVAILABLE = False logger = logging.getLogger(__name__) class ModalMLWrapper: """Wrapper pour les fonctions Modal ML distantes""" def __init__(self): self.is_model_trained = False self.model_metadata = None self.reference_data = None self.app_name = APP_NAME def check_modal_availability(self) -> bool: """Vérifie si Modal est disponible""" return MODAL_AVAILABLE and generate_synthetic_leads is not None async def train_model_async(self, num_synthetic_leads: int = 1000) -> Dict[str, Any]: """ Entraîne le modèle ML de manière asynchrone via l'API Modal distante Args: num_synthetic_leads: Nombre de leads synthétiques à générer Returns: Métadonnées du modèle entraîné """ try: if not self.check_modal_availability(): return { "error": "Modal ML non disponible ou app non déployée", "status": "error" } logger.info(f"🚀 Début d'entraînement du modèle ML avec {num_synthetic_leads} leads (API distante)") # 1. Générer les données synthétiques via l'API distante logger.info("📊 Génération des données synthétiques via Modal...") leads_data = await generate_synthetic_leads.remote.aio(num_synthetic_leads) # Sauvegarder les données de référence pour le drift self.reference_data = leads_data[:100] if leads_data else [] # 2. Entraîner le modèle via l'API distante logger.info("🤖 Entraînement du modèle via Modal...") model_metadata = await train_improved_model.remote.aio(leads_data) # Sauvegarder les métadonnées self.model_metadata = model_metadata self.is_model_trained = True logger.info("✅ Modèle entraîné avec succès via Modal") return { "status": "success", "model_metadata": model_metadata, "synthetic_data_count": len(leads_data), "reference_data_count": len(self.reference_data), "training_date": datetime.now().isoformat(), "modal_app": self.app_name } except Exception as e: logger.error(f"❌ Erreur entraînement modèle Modal: {e}") return { "error": f"Erreur Modal: {str(e)}", "status": "error", "modal_app": self.app_name } def train_model(self, num_synthetic_leads: int = 1000) -> Dict[str, Any]: """Version synchrone de l'entraînement Modal distant""" try: # Utiliser la version synchrone de Modal if not self.check_modal_availability(): return { "error": "Modal ML non disponible ou app non déployée", "status": "error" } logger.info(f"🚀 Entraînement synchrone Modal: {num_synthetic_leads} leads") # 1. Génération synchrone leads_data = generate_synthetic_leads.remote(num_synthetic_leads) self.reference_data = leads_data[:100] if leads_data else [] # 2. Entraînement synchrone model_metadata = train_improved_model.remote(leads_data) self.model_metadata = model_metadata self.is_model_trained = True return { "status": "success", "model_metadata": model_metadata, "synthetic_data_count": len(leads_data), "reference_data_count": len(self.reference_data), "training_date": datetime.now().isoformat(), "modal_app": self.app_name } except Exception as e: logger.error(f"❌ Erreur entraînement synchrone Modal: {e}") return { "error": f"Erreur Modal synchrone: {str(e)}", "status": "error", "modal_app": self.app_name } def predict_lead(self, lead_data: Dict[str, Any]) -> Dict[str, Any]: """Prédiction synchrone via Modal distant""" try: if not self.check_modal_availability(): return { "error": "Modal ML non disponible ou app non déployée", "status": "error" } if not self.is_model_trained: return { "error": "Modèle non entraîné. Lancez d'abord l'entraînement.", "status": "error" } logger.info(f"🔮 Prédiction Modal pour: {lead_data.get('name', 'Inconnu')}") # Prédiction via l'API Modal distante prediction = predict_lead_conversion_improved.remote(lead_data) # Ajouter des métadonnées prediction["prediction_date"] = datetime.now().isoformat() prediction["model_version"] = self.model_metadata.get("training_date") if self.model_metadata else None prediction["modal_app"] = self.app_name logger.info(f"✅ Prédiction Modal réussie: {prediction.get('classification', 'N/A')}") return prediction except Exception as e: logger.error(f"❌ Erreur prédiction Modal: {e}") return { "error": f"Erreur Modal prédiction: {str(e)}", "status": "error", "modal_app": self.app_name } async def monitor_performance_async(self, predictions: List[Dict[str, Any]]) -> Dict[str, Any]: """ Monitoring asynchrone des performances Args: predictions: Liste des prédictions à analyser Returns: Résultats du monitoring """ try: if not MODAL_AVAILABLE: return { "error": "Modal ML n'est pas disponible", "status": "error" } if not predictions: return { "error": "Aucune prédiction à analyser", "status": "error" } logger.info(f"📊 Monitoring de {len(predictions)} prédictions") monitoring_results = await monitor_model_performance.remote.aio(predictions) # Ajouter des métadonnées monitoring_results["monitoring_date"] = datetime.now().isoformat() monitoring_results["model_version"] = self.model_metadata.get("training_date") if self.model_metadata else None logger.info(f"✅ Monitoring terminé: {len(monitoring_results.get('performance_alerts', []))} alertes") return monitoring_results except Exception as e: logger.error(f"❌ Erreur monitoring: {e}") return { "error": str(e), "status": "error" } def monitor_performance(self, predictions: List[Dict[str, Any]]) -> Dict[str, Any]: """Version synchrone du monitoring""" try: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) result = loop.run_until_complete(self.monitor_performance_async(predictions)) loop.close() return result except Exception as e: logger.error(f"❌ Erreur monitoring synchrone: {e}") return {"error": str(e), "status": "error"} def get_model_status(self) -> Dict[str, Any]: """Retourne le statut du modèle""" return { "is_trained": self.is_model_trained, "model_metadata": self.model_metadata, "reference_data_count": len(self.reference_data) if self.reference_data else 0, "modal_available": MODAL_AVAILABLE } def format_prediction_for_gradio(self, prediction: Dict[str, Any]) -> str: """Formate la prédiction pour l'affichage Gradio""" if "error" in prediction: return f"❌ Erreur: {prediction['error']}" result = f""" 🎯 **Prédiction pour {prediction.get('lead_name', 'Inconnu')}** 📊 **Résultats principaux:** • Classification: {prediction.get('classification', 'N/A')} • Probabilité de conversion: {prediction.get('conversion_probability', 0):.1%} • Priorité: {prediction.get('priority', 'N/A')} • Confiance: {prediction.get('confidence_score', 0):.1%} 🔍 **Détails avancés:** • Prédiction: {'Oui' if prediction.get('prediction') else 'Non'} • Version du modèle: {prediction.get('model_version', 'N/A')} • Date de prédiction: {prediction.get('prediction_date', 'N/A')} """ # Ajouter l'analyse de drift si disponible drift_analysis = prediction.get('drift_analysis') if drift_analysis: result += "\n🔄 **Analyse de drift:**\n" for feature, analysis in drift_analysis.items(): if isinstance(analysis, dict): if analysis.get('is_outlier') or analysis.get('is_new_category'): result += f"• ⚠️ {feature}: Anomalie détectée\n" else: result += f"• ✅ {feature}: Normal\n" return result def format_monitoring_for_gradio(self, monitoring: Dict[str, Any]) -> str: """Formate les résultats de monitoring pour l'affichage Gradio""" if "error" in monitoring: return f"❌ Erreur monitoring: {monitoring['error']}" stats = monitoring.get('probability_stats', {}) alerts = monitoring.get('performance_alerts', []) distribution = monitoring.get('classification_distribution', {}) result = f""" 📊 **Rapport de monitoring** 📈 **Statistiques des prédictions:** • Nombre total: {monitoring.get('total_predictions', 0)} • Probabilité moyenne: {stats.get('mean', 0):.1%} • Variance: {stats.get('std', 0):.3f} • Min/Max: {stats.get('min', 0):.1%} - {stats.get('max', 0):.1%} 🏷️ **Distribution des classifications:** """ for classification, count in distribution.items(): result += f"• {classification}: {count}\n" if alerts: result += f"\n⚠️ **Alertes ({len(alerts)}):**\n" for alert in alerts: severity_emoji = "🚨" if alert.get('severity') == 'ERROR' else "⚠️" result += f"• {severity_emoji} {alert.get('message', 'Alerte inconnue')}\n" else: result += "\n✅ **Aucune alerte détectée**\n" result += f"\n🕐 **Dernière analyse:** {monitoring.get('monitoring_date', 'N/A')}" return result # Instance globale du wrapper modal_wrapper = ModalMLWrapper() # Fonctions helper pour Gradio def gradio_train_model(num_leads: int = 1000) -> Tuple[str, str]: """Interface Gradio pour l'entraînement""" try: num_leads = max(100, min(5000, int(num_leads))) # Limiter entre 100 et 5000 result = modal_wrapper.train_model(num_leads) if result.get("status") == "success": metadata = result.get("model_metadata", {}) performance = metadata.get("model_performance", {}) success_msg = f""" ✅ **Modèle entraîné avec succès !** 📊 **Données d'entraînement:** • Leads synthétiques générés: {result.get('synthetic_data_count', 0)} • Données de référence: {result.get('reference_data_count', 0)} 🎯 **Performances du modèle:** • Score de test: {performance.get('test_score', 0):.1%} • Validation croisée: {performance.get('cv_mean', 0):.1%} • Score AUC: {performance.get('auc_score', 0):.1%} 🕐 **Date d'entraînement:** {result.get('training_date', 'N/A')} """ return success_msg, "✅ Modèle prêt pour les prédictions" else: error_msg = f"❌ Erreur d'entraînement: {result.get('error', 'Erreur inconnue')}" return error_msg, error_msg except Exception as e: error_msg = f"❌ Erreur: {str(e)}" return error_msg, error_msg def gradio_predict_lead(name: str, industry: str, company_size: str, budget_range: str, urgency: str, source: str, expected_revenue: float, response_time: float) -> str: """Interface Gradio pour la prédiction""" try: if not name.strip(): return "❌ Veuillez entrer un nom de lead" lead_data = { "name": name, "industry": industry, "company_size": company_size, "budget_range": budget_range, "urgency": urgency, "source": source, "expected_revenue": max(0, expected_revenue), "response_time_hours": max(0.1, response_time) } prediction = modal_wrapper.predict_lead(lead_data) return modal_wrapper.format_prediction_for_gradio(prediction) except Exception as e: return f"❌ Erreur de prédiction: {str(e)}" def gradio_get_model_status() -> str: """Interface Gradio pour le statut du modèle""" try: status = modal_wrapper.get_model_status() if not status.get("modal_available"): return """ ❌ **Modal ML non disponible** Pour utiliser les fonctions d'IA avancées, installez Modal: ```bash pip install modal ``` """ if status.get("is_trained"): metadata = status.get("model_metadata", {}) performance = metadata.get("model_performance", {}) return f""" ✅ **Modèle entraîné et prêt** 🎯 **Performances:** • Score de test: {performance.get('test_score', 0):.1%} • Validation croisée: {performance.get('cv_mean', 0):.1%} • Score AUC: {performance.get('auc_score', 0):.1%} 📊 **Données:** • Données de référence: {status.get('reference_data_count', 0)} leads • Date d'entraînement: {metadata.get('training_date', 'N/A')} """ else: return """ ⚠️ **Modèle non entraîné** Lancez d'abord l'entraînement pour utiliser les prédictions avancées. """ except Exception as e: return f"❌ Erreur: {str(e)}"