Upload tools/idfm_disruptions_final.py with huggingface_hub

tools/idfm_disruptions_final.py

@@ -0,0 +1,90 @@
+import gradio as gr
+import json
+from PIL import Image
+import numpy as np
+
+# --- User Defined Logic ---
+import requests
+import json
+
+def idfm_disruptions_final(mode: str, token: str, query: str = "") -> str:
+    """Consulte les perturbations IDFM via l'API PRIM."""
+    url = "https://prim.iledefrance-mobilites.fr/marketplace/disruptions_bulk/disruptions/v2"
+    headers = {"apikey": token}
+    
+    try:
+        response = requests.get(url, headers=headers, timeout=10)
+        response.raise_for_status()
+        data = response.json()
+        
+        # L'API peut retourner soit une liste directe, soit un objet avec une clé
+        if isinstance(data, dict) and "disruptions" in data:
+            disruptions = data["disruptions"]
+        elif isinstance(data, list):
+            disruptions = data
+        else:
+            disruptions = []
+            
+    except Exception as e:
+        return f"Erreur API: {str(e)}"
+    
+    if mode == "stats":
+        total = len(disruptions)
+        result = f"Total de perturbations: {total}\n\n"
+        if total > 0:
+            result += "Échantillon des 5 premières:\n"
+            for i, d in enumerate(disruptions[:5], 1):
+                ligne = d.get("impactedObject", {}).get("name", "N/A") if isinstance(d.get("impactedObject"), dict) else "N/A"
+                titre = d.get("title", d.get("message", "Sans titre"))
+                result += f"{i}. {ligne}: {titre}\n"
+        return result
+    
+    elif mode == "find":
+        if not query:
+            return "Erreur: query requis pour le mode find"
+        q_lower = query.lower()
+        matches = []
+        
+        for d in disruptions:
+            # Recherche dans tous les champs texte
+            search_str = json.dumps(d, ensure_ascii=False).lower()
+            if q_lower in search_str:
+                matches.append(d)
+        
+        if not matches:
+            return f"Aucune perturbation trouvée pour '{query}'"
+        
+        result = f"Trouvé {len(matches)} perturbation(s) pour '{query}':\n\n"
+        for i, d in enumerate(matches[:10], 1):
+            ligne = d.get("impactedObject", {}).get("name", "N/A") if isinstance(d.get("impactedObject"), dict) else "N/A"
+            titre = d.get("title", d.get("message", "Sans titre"))
+            severite = d.get("severity", "N/A")
+            result += f"{i}. [{severite}] {ligne}: {titre}\n"
+        
+        if len(matches) > 10:
+            result += f"\n... et {len(matches) - 10} autres"
+        
+        return result
+    
+    elif mode == "list":
+        if not disruptions:
+            return "Aucune perturbation"
+        result = f"Liste des perturbations (10 premières sur {len(disruptions)}):\n\n"
+        for i, d in enumerate(disruptions[:10], 1):
+            ligne = d.get("impactedObject", {}).get("name", "N/A") if isinstance(d.get("impactedObject"), dict) else "N/A"
+            titre = d.get("title", d.get("message", "Sans titre"))
+            result += f"{i}. {ligne}: {titre}\n"
+        return result
+    
+    else:
+        return f"Mode non supporté: {mode}. Utilisez stats, find ou list"
+
+# --- Interface Factory ---
+def create_interface():
+    return gr.Interface(
+        fn=idfm_disruptions_final,
+        inputs=[gr.Textbox(label=k) for k in ['mode', 'token', 'query']],
+        outputs=gr.Textbox(label="Résumé formaté des perturbations"),
+        title="idfm-disruptions-final",
+        description="Auto-generated tool: idfm-disruptions-final"
+    )