Update app.py

app.py

@@ -1,162 +1,168 @@
 import subprocess
 import json
+from datetime import datetime, timedelta
 import gradio as gr
 import pandas as pd
-
-def fetch_pesticide_data(url, output_file):
-    """
-    Fetch data from the EU Pesticides Database API using curl.
-    Args:
-        url (str): The API endpoint URL.
-        output_file (str): The file to save the JSON response.
-    Returns:
-        dict: JSON response from the API or an error dictionary.
-    """
-    try:
-        # Command to execute curl and save the output to a file
-        command = [
-            'curl', '-H', 'Content-Type: application/json',
-            '-H', 'Cache-Control: no-cache',
-            '-H', 'User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
-            url,
-            '-o', output_file
+from typing import Dict, List, Tuple, Optional
+import logging
+
+# Configuration du logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class PesticideDataFetcher:
+    """Classe pour gérer la récupération des données sur les pesticides."""
+    
+    BASE_URL = "https://api.datalake.sante.service.ec.europa.eu/sante/pesticides"
+    HEADERS = {
+        'Content-Type': 'application/json',
+        'Cache-Control': 'no-cache',
+        '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 améliorée.
+        """
+        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:
+                return json.load(file)
+                
+        except Exception as e:
+            logger.error(f"Erreur lors de la récupération des données: {str(e)}")
+            return {"error": str(e)}
+
+    @classmethod
+    def get_products(cls) -> 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")
+        return response.get("value", [])
+
+    @classmethod
+    def get_residues(cls, product_id: int) -> List[Dict]:
+        """Récupère les résidus pour un produit spécifique."""
+        url = f"{cls.BASE_URL}/pesticide_residues?format=json&pesticide_residue_lg=FR&api-version=v2.0"
+        response = cls.fetch_data(url, "residues.json")
+        return [r for r in response.get("value", []) if r.get("product_id") == product_id]
+
+    @classmethod
+    def get_mrls(cls, 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")
+        return response.get("value", [])
+
+class PesticideInterface:
+    """Classe pour gérer l'interface utilisateur Gradio."""
+    
+    def __init__(self):
+        self.fetcher = PesticideDataFetcher()
+        self.products = self.fetcher.get_products()
+        self.product_choices = {
+            f"{p['product_name']} (ID: {p['product_id']})": p['product_id']
+            for p in self.products
+        }
+
+    def filter_by_period(self, data: List[Dict], period: str) -> List[Dict]:
+        """Filtre les données selon la période sélectionnée."""
+        if period == "Toutes les dates":
+            return data
+            
+        today = datetime.now()
+        if period == "Dernière semaine":
+            start_date = today - timedelta(days=7)
+        elif period == "Dernier mois":
+            start_date = today - timedelta(days=30)
+        
+        return [
+            item for item in data
+            if datetime.strptime(item.get("entry_into_force_date", ""), "%Y-%m-%d") >= start_date
         ]
 
-        # Execute the curl command
-        subprocess.run(command, check=True)
-
-        # Read the JSON data from the file
-        with open(output_file, 'r') as file:
-            data = json.load(file)
-
-        return data
-
-    except subprocess.CalledProcessError as e:
-        return {"error": f"Curl command failed: {str(e)}"}
-    except Exception as e:
-        return {"error": str(e)}
-
-def fetch_products():
-    """
-    Fetch products from the EU Pesticides Database API.
-    Returns:
-        list: List of product names and IDs.
-    """
-    url = "https://api.datalake.sante.service.ec.europa.eu/sante/pesticides/pesticide_residues_products?format=json&language=EN&api-version=v2.0"
-    output_file = "products.json"
-    products_response = fetch_pesticide_data(url, output_file)
-
-    if "error" in products_response:
-        return []
-
-    # Extract product names and IDs for display
-    product_choices = [
-        f"{product['product_name']} (ID: {product['product_id']})"
-        for product in products_response.get("value", [])
-    ]
-    return product_choices
-
-def fetch_pesticide_residues(product_id):
-    """
-    Fetch pesticide residues for a specific product.
-    Args:
-        product_id (int): The ID of the product.
-    Returns:
-        dict: Pesticide residues data for the product.
-    """
-    url = f"https://api.datalake.sante.service.ec.europa.eu/sante/pesticides/pesticide_residues?format=json&pesticide_residue_lg=EN&api-version=v2.0"
-    output_file = "pesticide_residues.json"
-    residues_response = fetch_pesticide_data(url, output_file)
-
-    if "error" in residues_response:
-        return residues_response
-
-    # Filter residues for the specific product
-    residues = [
-        residue for residue in residues_response.get("value", [])
-        if residue.get("product_id") == product_id
-    ]
-    return residues
-
-def fetch_pesticide_mrls(product_id):
-    """
-    Fetch MRLs for a specific product.
-    Args:
-        product_id (int): The ID of the product.
-    Returns:
-        dict: MRL data for the product.
-    """
-    url = f"https://api.datalake.sante.service.ec.europa.eu/sante/pesticides/pesticide_residues_mrls?format=json&product_id={product_id}&api-version=v2.0"
-    output_file = "pesticide_mrls.json"
-    mrls_response = fetch_pesticide_data(url, output_file)
-
-    if "error" in mrls_response:
-        return mrls_response
-
-    return mrls_response
-
-def create_pesticide_interface():
-    """
-    Create Gradio interface for Pesticide Database.
-    """
-    # Fetch initial products
-    product_choices = fetch_products()
-
-    def get_product_details(product_name, date_filter):
-        """Get details for a specific product."""
-        if not product_name:
-            return {"error": "No product selected"}
-
-        # Extract product ID from the product name
-        product_id = int(product_name.split(" (ID: ")[1].split(")")[0])
-
-        # Fetch pesticide residues
-        residues = fetch_pesticide_residues(product_id)
-        if "error" in residues:
-            return residues
-
-        # Fetch MRLs
-        mrls = fetch_pesticide_mrls(product_id)
-        if "error" in mrls:
-            return mrls
-
-        # Filter MRLs by date if a date filter is provided
-        if date_filter:
-            mrls["value"] = [
-                mrl for mrl in mrls.get("value", [])
-                if mrl.get("entry_into_force_date") == date_filter
-            ]
-
-        # Convert residues and MRLs to DataFrames for display
-        residues_df = pd.DataFrame(residues)
-        mrls_df = pd.DataFrame(mrls.get("value", []))
-
-        return residues_df, mrls_df
-
-    # Gradio Interface
-    with gr.Blocks() as interface:
-        gr.Markdown("## EU Pesticides Database Explorer")
-        gr.Markdown("Fetch and view data from the EU Pesticides Database")
-
-        with gr.Row():
-            product_dropdown = gr.Dropdown(label="Select Product", choices=product_choices)
-            date_filter = gr.Textbox(label="Filter by Entry into Force Date (YYYY-MM-DD)", placeholder="YYYY-MM-DD")
-            fetch_btn = gr.Button("Get Details")
-
-        with gr.Row():
-            residues_table = gr.Dataframe(label="Pesticide Residues")
-            mrls_table = gr.Dataframe(label="MRLs")
-
-        # Get product details
-        fetch_btn.click(
-            fn=get_product_details,
-            inputs=[product_dropdown, date_filter],
-            outputs=[residues_table, mrls_table]
-        )
-
-    return interface
-
-# Launch the interface
+    def get_product_details(self, 
+                          product_name: str, 
+                          period: str) -> Tuple[Optional[pd.DataFrame], Optional[pd.DataFrame]]:
+        """Récupère et formate les détails d'un produit."""
+        try:
+            if not product_name:
+                raise ValueError("Aucun produit sélectionné")
+
+            product_id = self.product_choices[product_name]
+            
+            # Récupération des données
+            residues = self.fetcher.get_residues(product_id)
+            mrls = self.fetcher.get_mrls(product_id)
+            
+            # Filtrage par période
+            filtered_mrls = self.filter_by_period(mrls, period)
+            
+            # Création des DataFrames avec colonnes sélectionnées
+            residues_df = pd.DataFrame(residues)[['substance_name', 'residue_definition', 'notes']]
+            mrls_df = pd.DataFrame(filtered_mrls)[['substance_name', 'mrl_value', 'entry_into_force_date']]
+            
+            # Formatage des DataFrames
+            residues_df.columns = ['Substance', 'Définition du résidu', 'Notes']
+            mrls_df.columns = ['Substance', 'Valeur LMR', 'Date d\'entrée en vigueur']
+            
+            return residues_df, mrls_df
+            
+        except Exception as e:
+            logger.error(f"Erreur lors de la récupération des détails: {str(e)}")
+            return None, None
+
+    def create_interface(self) -> gr.Blocks:
+        """Crée l'interface Gradio avec un design amélioré."""
+        with gr.Blocks(theme=gr.themes.Soft()) as interface:
+            gr.Markdown("# 🌿 Base de données des pesticides de l'UE")
+            
+            with gr.Box():
+                gr.Markdown("### Sélectionnez un produit et une période")
+                with gr.Row():
+                    with gr.Column(scale=2):
+                        product_dropdown = gr.Dropdown(
+                            choices=list(self.product_choices.keys()),
+                            label="Produit",
+                            info="Choisissez un produit dans la liste"
+                        )
+                    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="Sélectionnez la période d'analyse"
+                        )
+
+            with gr.Row():
+                fetch_btn = gr.Button("📊 Analyser les données", variant="primary")
+
+            with gr.Tabs():
+                with gr.TabItem("Résidus de pesticides"):
+                    residues_table = gr.DataFrame(label="")
+                with gr.TabItem("Limites Maximales de Résidus (LMR)"):
+                    mrls_table = gr.DataFrame(label="")
+
+            # Event handler
+            fetch_btn.click(
+                fn=self.get_product_details,
+                inputs=[product_dropdown, period_radio],
+                outputs=[residues_table, mrls_table]
+            )
+
+        return interface
+
+def main():
+    interface = PesticideInterface()
+    app = interface.create_interface()
+    app.launch(share=True)
+
 if __name__ == "__main__":
-    interface = create_pesticide_interface()
-    interface.launch(share=True)
+    main()