Update app.py

app.py

@@ -5,11 +5,26 @@ import gradio as gr
 import pandas as pd
 from typing import Dict, List, Optional
 import logging
+from dataclasses import dataclass
+from concurrent.futures import ThreadPoolExecutor
 
 # Configuration du logging
-logging.basicConfig(level=logging.INFO)
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
 logger = logging.getLogger(__name__)
 
+@dataclass
+class PesticideRecord:
+    """Structure de données pour les enregistrements de pesticides."""
+    substance_name: str
+    mrl_value: float
+    entry_into_force_date: str
+    regulation_number: str
+    regulation_url: str
+    modification_date: Optional[str] = None
+
 class PesticideDataFetcher:
     """Classe pour gérer la récupération des données sur les pesticides."""
     
@@ -20,38 +35,51 @@ class PesticideDataFetcher:
         'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
     }
 
-    @staticmethod
-    def fetch_data(url: str, output_file: str) -> Dict:
-        """Récupère les données depuis l'API avec gestion d'erreurs."""
-        try:
-            command = ['curl'] + [
-                item for header, value in PesticideDataFetcher.HEADERS.items()
-                for item in ['-H', f'{header}: {value}']
-            ] + [url, '-o', output_file]
-            
-            subprocess.run(command, check=True, capture_output=True)
-            with open(output_file, 'r') as file:
-                data = json.load(file)
+    def __init__(self):
+        self.session = self._create_session()
+
+    def _create_session(self):
+        """Crée une session pour les requêtes HTTP."""
+        import requests
+        session = requests.Session()
+        for header, value in self.HEADERS.items():
+            session.headers[header] = value
+        return session
+
+    def fetch_data(self, url: str) -> Dict:
+        """Récupère les données depuis l'API avec gestion d'erreurs et retry."""
+        max_retries = 3
+        for attempt in range(max_retries):
+            try:
+                response = self.session.get(url, timeout=10)
+                response.raise_for_status()
+                data = response.json()
                 logger.info(f"Fetched data from {url}: {str(data)[:200]}...")
                 return data
-        except Exception as e:
-            logger.error(f"Erreur lors de la récupération des données : {str(e)}")
-            return {"error": str(e)}
+            except Exception as e:
+                logger.warning(f"Tentative {attempt + 1}/{max_retries} échouée: {str(e)}")
+                if attempt == max_retries - 1:
+                    logger.error(f"Échec de la récupération des données: {str(e)}")
+                    return {"error": str(e)}
+        return {"error": "Max retries exceeded"}
 
-    @classmethod
-    def get_products(cls) -> List[Dict]:
+    def get_products(self) -> List[Dict]:
         """Récupère la liste des produits."""
-        url = f"{cls.BASE_URL}/pesticide_residues_products?format=json&language=FR&api-version=v2.0"
-        response = cls.fetch_data(url, "products.json")
+        url = f"{self.BASE_URL}/pesticide_residues_products?format=json&language=FR&api-version=v2.0"
+        response = self.fetch_data(url)
         return response.get("value", [])
 
-    @classmethod
-    def get_mrls(cls, product_id: int) -> List[Dict]:
+    def get_mrls(self, product_id: int) -> List[Dict]:
         """Récupère les LMR pour un produit spécifique."""
-        url = f"{cls.BASE_URL}/pesticide_residues_mrls?format=json&product_id={product_id}&api-version=v2.0"
-        response = cls.fetch_data(url, "mrls.json")
+        url = f"{self.BASE_URL}/pesticide_residues_mrls?format=json&product_id={product_id}&api-version=v2.0"
+        response = self.fetch_data(url)
         return response.get("value", [])
 
+    def get_substance_details(self, substance_id: int) -> Dict:
+        """Récupère les détails d'une substance."""
+        url = f"{self.BASE_URL}/substances/{substance_id}?format=json&api-version=v2.0"
+        return self.fetch_data(url)
+
 class PesticideInterface:
     """Classe pour gérer l'interface utilisateur Gradio."""
 
@@ -61,11 +89,15 @@ class PesticideInterface:
         self.product_choices = {
             p['product_name']: p['product_id'] for p in self.products
         }
+        self._cache = {}
         logger.info(f"Initialized interface with {len(self.product_choices)} products.")
 
     def parse_date(self, date_str: str) -> Optional[str]:
-        """Convertit une date au format 'YYYY-MM-DD', détecte automatiquement le format."""
-        for fmt in ("%Y-%m-%d", "%d/%m/%Y"):
+        """Convertit une date au format 'YYYY-MM-DD'."""
+        if not date_str:
+            return None
+            
+        for fmt in ("%Y-%m-%d", "%d/%m/%Y", "%Y-%m-%dT%H:%M:%S"):
             try:
                 return datetime.strptime(date_str, fmt).strftime("%Y-%m-%d")
             except ValueError:
@@ -79,112 +111,174 @@ class PesticideInterface:
             return data
             
         today = datetime.now()
-        start_date = today - timedelta(days=7) if period == "Dernière semaine" else today - timedelta(days=30)
+        start_date = {
+            "Dernière semaine": today - timedelta(days=7),
+            "Dernier mois": today - timedelta(days=30)
+        }.get(period)
         
+        if not start_date:
+            return data
+
         filtered_data = []
         for item in data:
-            raw_date = item.get("entry_into_force_date")
-            if raw_date:
-                corrected_date = self.parse_date(raw_date)
-                if corrected_date and datetime.strptime(corrected_date, "%Y-%m-%d") >= start_date:
-                    item["entry_into_force_date"] = corrected_date
+            date_str = item.get("entry_into_force_date") or item.get("modification_date")
+            if date_str:
+                parsed_date = self.parse_date(date_str)
+                if parsed_date and datetime.strptime(parsed_date, "%Y-%m-%d") >= start_date:
+                    item["parsed_date"] = parsed_date
                     filtered_data.append(item)
+
         logger.info(f"Filtered {len(data)} items to {len(filtered_data)} items for period {period}.")
         return filtered_data
 
-    def get_product_details(self, product_name: str, period: str) -> pd.DataFrame:
+    def format_regulation_link(self, regulation_url: str, regulation_number: str) -> str:
+        """Formate un lien de règlement en HTML cliquable."""
+        if not regulation_url:
+            return regulation_number
+        return f'<a href="{regulation_url}" target="_blank">{regulation_number}</a>'
+
+    def get_product_details(self, product_name: str, period: str, show_only_changes: bool) -> pd.DataFrame:
         """Récupère et formate les détails des MRLs pour un produit donné."""
         try:
-            logger.info(f"Getting details for product: {product_name}, period: {period}")
-            
             if not product_name:
-                logger.warning("No product selected.")
-                return pd.DataFrame(columns=["Message"])
+                return pd.DataFrame({"Message": ["Veuillez sélectionner un produit"]})
 
             product_id = self.product_choices.get(product_name)
             if not product_id:
-                logger.error(f"Product ID not found for {product_name}.")
-                return pd.DataFrame(columns=["Message"], data=[["Produit introuvable."]])
+                return pd.DataFrame({"Message": ["Produit non trouvé"]})
+
+            # Utiliser le cache si disponible
+            cache_key = f"{product_id}_{period}_{show_only_changes}"
+            if cache_key in self._cache:
+                return self._cache[cache_key]
 
             # Récupérer les MRLs
             mrls = self.fetcher.get_mrls(product_id)
-            filtered_mrls = self.filter_by_period(mrls, period)
-
-            if not filtered_mrls:
-                logger.warning("No MRLs found after filtering.")
-                return pd.DataFrame(columns=["Message"], data=[["Aucune donnée trouvée."]])
-
-            # Créer un DataFrame
-            mrls_df = pd.DataFrame(filtered_mrls)
+            
+            # Filtrer par période si nécessaire
+            if period != "Toutes les dates":
+                mrls = self.filter_by_period(mrls, period)
 
-            # Colonnes disponibles
-            available_columns = set(mrls_df.columns)
-            logger.info(f"Available columns: {available_columns}")
+            if not mrls:
+                return pd.DataFrame({"Message": ["Aucune donnée trouvée pour la période sélectionnée"]})
 
-            # Colonnes attendues
-            expected_columns = {
+            # Créer le DataFrame
+            df = pd.DataFrame(mrls)
+            
+            # Formater les colonnes
+            columns_mapping = {
                 "substance_name": "Substance",
                 "mrl_value": "Valeur LMR",
-                "entry_into_force_date": "Date d'entrée en vigueur",
+                "entry_into_force_date": "Date d'application",
                 "regulation_number": "Règlement",
-                "regulation_url": "Lien"
+                "modification_date": "Date de modification"
             }
-            display_columns = {key: value for key, value in expected_columns.items() if key in available_columns}
+            
+            df = df.rename(columns=columns_mapping)
+            
+            # Formater les dates
+            for date_col in ["Date d'application", "Date de modification"]:
+                if date_col in df.columns:
+                    df[date_col] = df[date_col].apply(self.parse_date)
 
-            # Ajouter une colonne par défaut si `substance_name` est manquant
-            if "substance_name" not in available_columns:
-                mrls_df["substance_name"] = "Nom de la substance indisponible"
-                display_columns["substance_name"] = "Substance"
+            # Formater les valeurs LMR
+            if "Valeur LMR" in df.columns:
+                df["Valeur LMR"] = df["Valeur LMR"].apply(
+                    lambda x: f"{x}*" if str(x).endswith('*') else x
+                )
 
-            if not display_columns:
-                logger.warning("No relevant columns available to display.")
-                return pd.DataFrame(columns=["Message"], data=[["Colonnes insuffisantes."]])
+            # Ajouter les liens cliquables pour les règlements
+            if "regulation_url" in df.columns:
+                df["Règlement"] = df.apply(
+                    lambda row: self.format_regulation_link(
+                        row.get("regulation_url", ""), 
+                        row["Règlement"]
+                    ),
+                    axis=1
+                )
 
-            mrls_df = mrls_df[list(display_columns.keys())].rename(columns=display_columns)
+            # Filtrer pour ne montrer que les changements si demandé
+            if show_only_changes:
+                df = df[df["Date de modification"].notna()]
 
-            # Convertir les URLs en liens cliquables
-            if "Lien" in mrls_df.columns:
-                mrls_df["Lien"] = mrls_df["Lien"].apply(lambda x: f"[Lien texte]({x})" if pd.notna(x) else "")
+            # Trier par date d'application décroissante
+            df = df.sort_values("Date d'application", ascending=False)
+
+            # Mettre en cache
+            self._cache[cache_key] = df
+
+            return df
 
-            return mrls_df
-            
         except Exception as e:
-            logger.error(f"Error in get_product_details: {str(e)}")
-            return pd.DataFrame(columns=["Message"], data=[[f"Erreur : {str(e)}"]])
+            logger.error(f"Erreur dans get_product_details: {str(e)}")
+            return pd.DataFrame({"Message": [f"Erreur: {str(e)}"]})
 
     def create_interface(self) -> gr.Blocks:
         """Crée l'interface Gradio avec un design amélioré."""
-        with gr.Blocks() as interface:
-            gr.Markdown("# 🌿 Base de données des pesticides de l'UE")
+        with gr.Blocks(
+            theme=gr.themes.Soft(
+                primary_hue="green",
+                secondary_hue="blue"
+            )
+        ) as interface:
+            gr.Markdown("""
+                # 🌿 Base de données des pesticides de l'UE
+                Consultez les Limites Maximales de Résidus (LMR) pour les produits agricoles.
+            """)
             
             with gr.Row():
-                product_dropdown = gr.Dropdown(
-                    choices=list(self.product_choices.keys()),
-                    label="Produit",
-                    info="Choisissez un produit dans la liste"
-                )
-                period_radio = gr.Radio(
-                    choices=["Dernière semaine", "Dernier mois", "Toutes les dates"],
-                    value="Toutes les dates",
-                    label="Période",
-                    info="Sélectionnez la période d'analyse"
-                )
+                with gr.Column(scale=2):
+                    product_dropdown = gr.Dropdown(
+                        choices=sorted(list(self.product_choices.keys())),
+                        label="Produit",
+                        info="Sélectionnez un produit agricole"
+                    )
+                with gr.Column(scale=1):
+                    period_radio = gr.Radio(
+                        choices=["Dernière semaine", "Dernier mois", "Toutes les dates"],
+                        value="Toutes les dates",
+                        label="Période",
+                        info="Filtrer par période"
+                    )
+                    show_changes = gr.Checkbox(
+                        label="Afficher uniquement les modifications récentes",
+                        info="Cochez pour voir uniquement les LMR qui ont été modifiées"
+                    )
+
+            with gr.Row():
+                fetch_btn = gr.Button("📊 Analyser les données", variant="primary")
 
-            fetch_btn = gr.Button("📊 Analyser les données")
-            mrls_table = gr.DataFrame()
+            with gr.Row():
+                mrls_table = gr.DataFrame(
+                    headers=["Substance", "Valeur LMR", "Date d'application", 
+                            "Date de modification", "Règlement"],
+                    interactive=False
+                )
 
             fetch_btn.click(
                 fn=self.get_product_details,
-                inputs=[product_dropdown, period_radio],
+                inputs=[product_dropdown, period_radio, show_changes],
                 outputs=[mrls_table]
             )
 
+            gr.Markdown("""
+                ### Légende
+                - **Valeur LMR** : Limite Maximale de Résidus autorisée
+                - **Date d'application** : Date d'entrée en vigueur de la LMR
+                - **Date de modification** : Date de la dernière modification
+                - **Règlement** : Référence du règlement européen (cliquez pour accéder au texte)
+            """)
+
         return interface
 
 def main():
     interface = PesticideInterface()
     app = interface.create_interface()
-    app.launch(server_name="0.0.0.0", server_port=7860)
+    app.launch(
+        server_name="0.0.0.0", 
+        server_port=7860,
+        share=True
+    )
 
 if __name__ == "__main__":
     main()