Update app.py

app.py

@@ -1,8 +1,11 @@
 import logging
-from datetime import datetime
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timedelta
+from typing import Dict, List, Optional
 import gradio as gr
 import pandas as pd
 import requests
+from dataclasses import dataclass
 from tenacity import retry, stop_after_attempt, wait_fixed
 import plotly.express as px
 
@@ -13,148 +16,260 @@ logging.basicConfig(
 )
 logger = logging.getLogger(__name__)
 
+@dataclass
+class SubstanceDetails:
+    name: str
+    status: Optional[str] = None
+    approval_date: Optional[str] = None
+    expiry_date: Optional[str] = None
+
+@dataclass
+class PesticideRecord:
+    substance_name: str
+    mrl_value: float
+    entry_into_force_date: str
+    regulation_number: str
+    regulation_url: str
+    modification_date: Optional[str] = None
+    substance_status: Optional[str] = None
+    approval_date: Optional[str] = None
+    expiry_date: Optional[str] = None
+
 class PesticideDataFetcher:
     BASE_URL = "https://api.datalake.sante.service.ec.europa.eu/sante/pesticides"
     HEADERS = {
-        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
+        'Content-Type': 'application/json',
+        'Cache-Control': 'no-cache',
+        'User-Agent': 'Mozilla/5.0'
     }
 
     def __init__(self):
-        self.session = requests.Session()
-        self.session.headers.update(self.HEADERS)
-        self.substance_cache = {}
-        self.product_cache = {}
-        self.preload_substances()
+        self.session = self._create_session()
+        self._substance_cache: Dict[int, SubstanceDetails] = {}
+        self._product_cache = {}
+        self.preload_substance_names()
+
+    def _create_session(self):
+        session = requests.Session()
+        for header, value in self.HEADERS.items():
+            session.headers[header] = value
+        return session
 
     @retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
-    def fetch_data(self, url: str) -> dict:
+    def fetch_data(self, url: str) -> Dict:
         try:
             response = self.session.get(url, timeout=10)
             response.raise_for_status()
-            return response.json()
+            data = response.json()
+            logger.info(f"Fetched data from {url}: {str(data)[:200]}...")
+            return data
         except requests.RequestException as e:
-            logger.error(f"Échec de la requête {url}: {e}")
+            logger.error(f"Failed to fetch data from {url}: {str(e)}", exc_info=True)
             return {"error": str(e)}
 
-    def preload_substances(self):
+    def preload_substance_names(self):
         url = f"{self.BASE_URL}/active_substances?format=json&api-version=v2.0"
         while url:
-            data = self.fetch_data(url)
-            for item in data.get("value", []):
-                self.substance_cache[item["substance_id"]] = {
-                    "name": item["substance_name"],
-                    "status": item.get("substance_status"),
-                    "approval_date": item.get("approval_date"),
-                    "expiry_date": item.get("expiry_date")
-                }
-            url = data.get("nextLink")
-        logger.info(f"Cache substances prérempli avec {len(self.substance_cache)} entrées")
-
-    def get_substance_name(self, substance_id: int) -> str:
-        return self.substance_cache.get(substance_id, {}).get("name", f"Substance {substance_id}")
-
-    def get_product_mrls(self, product_id: int) -> list:
+            response = self.fetch_data(url)
+            if not response.get("value"):
+                break
+            for item in response["value"]:
+                substance_id = item.get("substanceId")
+                if substance_id:
+                    self._substance_cache[substance_id] = SubstanceDetails(
+                        name=item.get("substanceName", "Nom non trouvé"),
+                        status=item.get("substanceStatus"),
+                        approval_date=item.get("approvalDate"),
+                        expiry_date=item.get("expiryDate")
+                    )
+            url = response.get("nextLink")
+        logger.info(f"Préchargé {len(self._substance_cache)} substances.")
+
+    def get_substance_name_by_id(self, substance_id: int) -> str:
+        if substance_id in self._substance_cache:
+            return self._substance_cache[substance_id].name
+        return f"Substance {substance_id}"
+
+    def get_active_substance_details(self, substance_name: str) -> Optional[SubstanceDetails]:
+        for details in self._substance_cache.values():
+            if details.name.lower() == substance_name.lower():
+                return details
+        return None
+
+    def get_substance_details(self, pesticide_residue_id: int) -> SubstanceDetails:
+        if pesticide_residue_id in self._substance_cache:
+            return self._substance_cache[pesticide_residue_id]
+        return SubstanceDetails(name=f"Substance {pesticide_residue_id}")
+
+    def get_all_substances(self) -> List[str]:
+        return sorted({details.name for details in self._substance_cache.values()})
+
+    def get_products(self) -> List[Dict]:
+        if self._product_cache:
+            return self._product_cache
+        all_products = []
+        base_url = f"{self.BASE_URL}/pesticide_residues_products?format=json&language=FR&api-version=v2.0"
+        url = base_url
+        while url:
+            response = self.fetch_data(url)
+            all_products.extend(response.get("value", []))
+            url = response.get("nextLink")
+        self._product_cache = all_products
+        logger.info(f"Récupéré {len(all_products)} produits au total")
+        return all_products
+
+    def get_mrls(self, product_id: int) -> List[Dict]:
         url = f"{self.BASE_URL}/pesticide_residues_mrls?format=json&product_id={product_id}&api-version=v2.0"
-        return self.fetch_data(url).get("value", [])
+        response = self.fetch_data(url)
+        return response.get("value", [])
 
-class PesticideApp:
+class PesticideInterface:
     def __init__(self):
         self.fetcher = PesticideDataFetcher()
-        self.product_list = self._get_products()
-        self.substance_list = self._get_substances()
+        self.products = self.fetcher.get_products()
+        self.product_choices = {p['productName']: p['productId'] for p in self.products}
+        self.substances = self.fetcher.get_all_substances()
+        self._cache = {}
+        logger.info(f"Initialisé avec {len(self.product_choices)} produits et {len(self.substances)} substances.")
 
-    def _get_products(self) -> dict:
-        if not self.fetcher.product_cache:
-            url = f"{self.fetcher.BASE_URL}/pesticide_residues_products?format=json&language=FR&api-version=v2.0"
-            data = self.fetcher.fetch_data(url)
-            self.fetcher.product_cache = {p['product_name']: p['product_id'] for p in data['value']}
-        return self.fetcher.product_cache
+    def parse_date(self, date_str: str) -> Optional[str]:
+        if not date_str:
+            return None
+        for fmt in ("%Y-%m-%d", "%d/%m/%Y", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%SZ"):
+            try:
+                return datetime.strptime(date_str, fmt).strftime("%Y-%m-%d")
+            except ValueError:
+                continue
+        return None
 
-    def _get_substances(self) -> list:
-        return [s['name'] for s in self.fetcher.substance_cache.values()]
-
-    def format_date(self, date_str: str) -> str:
-        try:
-            return datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%S").strftime("%d/%m/%Y")
-        except (ValueError, TypeError):
-            return "Date non disponible"
+    def filter_by_period(self, data: List[Dict], period: str) -> List[Dict]:
+        today = datetime.now()
+        start_date = {
+            "Dernière semaine": today - timedelta(days=7),
+            "Dernier mois": today - timedelta(days=30),
+            "Prochains 6 mois": today + timedelta(days=180)
+        }.get(period)
+        
+        if period == "Toutes les dates":
+            return data
+        filtered = []
+        for item in data:
+            date_str = item.get("entryIntoForceDate") or item.get("modificationDate")
+            parsed = self.parse_date(date_str)
+            if parsed:
+                item_date = datetime.strptime(parsed, "%Y-%m-%d")
+                if (period == "Prochains 6 mois" and item_date >= today) or (period != "Prochains 6 mois" and item_date >= start_date):
+                    item["parsed_date"] = parsed
+                    filtered.append(item)
+        return filtered
 
-    def search_products(self, query: str) -> pd.DataFrame:
-        filtered = {k: v for k, v in self.product_list.items() if query.lower() in k.lower()}
-        return pd.DataFrame(list(filtered.items()), columns=["Produit", "ID"])
+    def format_regulation_link(self, regulation_url: str, regulation_number: str) -> str:
+        return f'<a href="{regulation_url}" target="_blank">{regulation_number}</a>' if regulation_url else regulation_number
 
-    def get_product_details(self, product_name: str) -> pd.DataFrame:
-        product_id = self.product_list.get(product_name)
+    def get_product_details(self, product_name: str, period: str, show_only_changes: bool) -> pd.DataFrame:
+        if not product_name:
+            return pd.DataFrame({"Message": ["Sélectionnez un produit"]})
+        product_id = self.product_choices.get(product_name)
         if not product_id:
-            return pd.DataFrame([{"erreur": "Produit non trouvé"}])
+            return pd.DataFrame({"Message": ["Produit non trouvé"]})
         
-        mrls = self.fetcher.get_product_mrls(product_id)
-        data = []
-        for mrl in mrls:
-            substance_id = mrl["pesticide_residue_id"]
-            substance_info = self.fetcher.substance_cache.get(substance_id, {})
-            
-            data.append({
-                "Substance": self.fetcher.get_substance_name(substance_id),
-                "Valeur LMR": mrl.get("mrl_value", "N/C"),
-                "Date d'effet": self.format_date(mrl.get("entry_into_force_date")),
-                "Statut": substance_info.get("status", "N/C"),
-                "Date d'approbation": self.format_date(substance_info.get("approval_date"))
-            })
-        return pd.DataFrame(data)
-
-    def create_ui(self) -> gr.Blocks:
-        with gr.Blocks(theme=gr.themes.Default(), title="Base de Données Pesticides UE") as ui:
-            gr.HTML("""
-                <div style="text-align: center; padding: 20px; background: #006633; color: white;">
-                    <h1>;base de Données des Pesticides</h1>
-                    <p>Consultez les LMR et informations sur les substances actives</p>
-                </div>
-            """)
+        cache_key = f"{product_id}_{period}_{show_only_changes}"
+        if cache_key in self._cache:
+            return self._cache[cache_key]
+        
+        mrls = self.fetcher.get_mrls(product_id)
+        mrls = self.filter_by_period(mrls, period)
+        
+        if not mrls:
+            return pd.DataFrame({"Message": ["Aucune donnée"]})
+        
+        processed = []
+        with ThreadPoolExecutor(max_workers=10) as executor:
+            futures = {executor.submit(self.fetcher.get_substance_name_by_id, mrl["pesticideResidueId"]): mrl for mrl in mrls}
+            for future in futures:
+                mrl = futures[future]
+                try:
+                    substance_name = future.result()
+                    mrl_value = mrl.get("mrlValue", "")
+                    formatted_mrl = f"{mrl_value}*" if isinstance(mrl_value, (int, float)) and str(mrl_value).endswith('*') else str(mrl_value)
+                    
+                    processed.append({
+                        "Substance": substance_name,
+                        "Valeur LMR": formatted_mrl,
+                        "Date d'application": self.parse_date(mrl.get("entryIntoForceDate")),
+                        "Date de modification": self.parse_date(mrl.get("modificationDate")),
+                        "Règlement": self.format_regulation_link(
+                            mrl.get("regulationUrl", ""),
+                            mrl.get("regulationNumber") or mrl.get("regulationReference")
+                        ),
+                        "Statut": self.parse_date(self.fetcher.get_active_substance_details(substance_name).status),
+                        "Date d'approbation": self.parse_date(mrl.get("approvalDate")),
+                        "Date d'expiration": self.parse_date(mrl.get("expiryDate"))
+                    })
+                except Exception as e:
+                    logger.error(f"Erreur: {str(e)}")
+        
+        df = pd.DataFrame(processed)
+        if show_only_changes:
+            df = df[df["Date de modification"].notna()]
+        df = df.sort_values("Date d'application", ascending=False)
+        self._cache[cache_key] = df
+        return df
+
+    def create_graph(self, df: pd.DataFrame) -> gr.Plot:
+        return px.scatter(df, x='Date d\'application', y='Valeur LMR', color='Substance', title='LMR par date')
+
+    def export_data(self, df: pd.DataFrame) -> str:
+        df.to_csv("export.csv", index=False)
+        return "export.csv"
+
+    def search_substances(self, query: str) -> pd.DataFrame:
+        if not query:
+            return pd.DataFrame({"Message": ["Entrez une requête"]})
+        matches = [s for s in self.substances if query.lower() in s.lower()]
+        return pd.DataFrame(matches, columns=["Substance"]) if matches else pd.DataFrame({"Message": ["Aucun résultat"]})
+
+    def create_interface(self) -> gr.Blocks:
+        with gr.Blocks() as interface:
+            gr.Markdown("# 🌿 Base de données des pesticides UE")
             
-            with gr.Tab("Recherche par Produit"):
-                product_search = gr.Dropdown(
-                    list(self.product_list.keys()),
-                    label="Sélectionnez un produit",
-                    info="Commencez à taper pour filtrer"
-                )
-                product_results = gr.Dataframe(max_rows=20, interactive=False)
-                product_search.change(
-                    self.get_product_details,
-                    inputs=product_search,
-                    outputs=product_results
-                )
+            with gr.Tabs():
+                with gr.Tab("Recherche par Produit"):
+                    product = gr.Dropdown(sorted(self.product_choices.keys()), label="Produit")
+                    period = gr.Radio(["Dernière semaine", "Dernier mois", "Prochains 6 mois", "Toutes les dates"], label="Période")
+                    show_changes = gr.Checkbox("Afficher les modifications")
+                    
+                    btn = gr.Button("Analyser")
+                    table = gr.Dataframe(max_height=500)
+                    graph = gr.Plot()
+                    export = gr.Button("Exporter")
+                    file = gr.File()
+                    
+                    btn.click(
+                        self.get_product_details,
+                        inputs=[product, period, show_changes],
+                        outputs=table
+                    )
+                    table.change(self.create_graph, inputs=table, outputs=graph)
+                    export.click(self.export_data, inputs=table, outputs=file)
                 
-            with gr.Tab("Recherche par Substance"):
-                substance_search = gr.Textbox(
-                    placeholder="Rechercher une substance...",
-                    label="Nom de la substance"
-                )
-                substance_results = gr.Dataframe(max_rows=10, interactive=False)
-                substance_search.submit(
-                    self.search_substances,
-                    inputs=substance_search,
-                    outputs=substance_results
-                )
-                
-            gr.Markdown("""
-                **Légende**  
-                - LMR : Limite Maximale de Résidus  
-                - Date d'effet : Date d'entrée en vigueur  
-                - Statut : Approbation/Expiration  
-            """)
-            
-            gr.Button("Actualiser les données", variant="secondary").click(
-                lambda: self.fetcher.preload_substances()
+                with gr.Tab("Recherche par Substance"):
+                    search = gr.Textbox()
+                    btn_search = gr.Button("Rechercher")
+                    df_substances = gr.Dataframe(max_height=300)
+                    btn_search.click(self.search_substances, inputs=search, outputs=df_substances)
+                    
+            gr.Markdown(
+                "**Légende**\n"
+                "- LMR : Limite Maximale de Résidus\n"
+                "- Date d'application : Date d'entrée en vigueur\n"
+                "- Règlement : Cliquez pour ouvrir la source"
             )
-            
-        return ui
+        return interface
 
 def main():
-    app = PesticideApp()
-    ui = app.create_ui()
-    ui.launch(share=True)
+    app = PesticideInterface().create_interface()
+    app.launch(share=True)
 
 if __name__ == "__main__":
-    main()
-    
+    main()