Update app.py

app.py

@@ -9,7 +9,6 @@ from dataclasses import dataclass
 from tenacity import retry, stop_after_attempt, wait_fixed
 import plotly.express as px
 
-# Configuration du logging
 logging.basicConfig(
     level=logging.INFO,
     format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
@@ -23,57 +22,35 @@ class SubstanceDetails:
     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 = {
-        'Content-Type': 'application/json',
-        'Cache-Control': 'no-cache',
-        'User-Agent': 'Mozilla/5.0'
+        'User-Agent': 'Mozilla/5.0',
+        'Accept': 'application/json'
     }
 
     def __init__(self):
-        self.session = self._create_session()
+        self.session = requests.Session()
+        self.session.headers.update(self.HEADERS)
         self._substance_cache: Dict[int, SubstanceDetails] = {}
-        self._product_cache = {}
+        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:
         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
+            return response.json()
         except requests.RequestException as e:
-            logger.error(f"Failed to fetch data from {url}: {str(e)}", exc_info=True)
+            logger.error(f"Échec de la requête {url}: {e}")
             return {"error": str(e)}
 
     def preload_substance_names(self):
         url = f"{self.BASE_URL}/active_substances?format=json&api-version=v2.0"
         while url:
-            response = self.fetch_data(url)
-            if not response.get("value"):
-                break
-            for item in response["value"]:
+            data = self.fetch_data(url)
+            for item in data.get("value", []):
                 substance_id = item.get("substanceId")
                 if substance_id:
                     self._substance_cache[substance_id] = SubstanceDetails(
@@ -82,194 +59,99 @@ class PesticideDataFetcher:
                         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()})
+            url = data.get("nextLink")
+        logger.info(f"Cache prérempli avec {len(self._substance_cache)} substances")
 
-    def get_products(self) -> List[Dict]:
+    def get_product_list(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
+        url = f"{self.BASE_URL}/pesticide_residues_products?format=json&language=FR&api-version=v2.0"
+        data = self.fetch_data(url)
+        self._product_cache = data.get("value", [])
+        logger.info(f"Produits récupérés: {len(self._product_cache)}")
+        return self._product_cache
 
     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"
-        response = self.fetch_data(url)
-        return response.get("value", [])
+        return self.fetch_data(url).get("value", [])
 
-class PesticideInterface:
+    def get_substance_name(self, substance_id: int) -> str:
+        if substance_id in self._substance_cache:
+            return self._substance_cache[substance_id].name
+        return f"Substance {substance_id}"
+
+class PesticideApp:
     def __init__(self):
         self.fetcher = PesticideDataFetcher()
-        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 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 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
+        self.product_list = {p['productName']: p['productId'] for p in self.fetcher.get_product_list()}
+        self.substances = [sd.name for sd in self.fetcher._substance_cache.values()]
 
-    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 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 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)
+    def get_product_details(self, product_name: str) -> pd.DataFrame:
+        product_id = self.product_list.get(product_name)
         if not product_id:
-            return pd.DataFrame({"Message": ["Produit non trouvé"]})
-        
-        cache_key = f"{product_id}_{period}_{show_only_changes}"
-        if cache_key in self._cache:
-            return self._cache[cache_key]
+            return pd.DataFrame([{"erreur": "Produit non trouvé"}])
         
         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"
+        data = []
+        for mrl in mrls:
+            substance_id = mrl.get("pesticideResidueId")
+            substance_name = self.fetcher.get_substance_name(substance_id)
+            
+            data.append({
+                "Substance": substance_name,
+                "Valeur LMR": mrl.get("mrlValue", "N/C"),
+                "Date d'effet": self.format_date(mrl.get("entryIntoForceDate")),
+                "Statut": self.fetcher._substance_cache.get(substance_id, {}).get("status", "N/C")
+            })
+        return pd.DataFrame(data)
 
     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")
+        return pd.DataFrame(matches, columns=["Substance"]) if matches else pd.DataFrame([{"Message": "Aucun résultat"}])
+
+    def create_ui(self) -> gr.Blocks:
+        with gr.Blocks(theme=gr.themes.Default()) as ui:
+            gr.HTML("""
+                <div style="text-align: center; padding: 20px; background: #006633; color: white;">
+                    <h1>Base de Données des Pesticides UE</h1>
+                </div>
+            """)
             
-            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 Produit"):
+                product = gr.Dropdown(
+                    label="Sélectionnez un produit",
+                    choices=sorted(self.product_list.keys()),
+                    interactive=True
+                )
+                output = gr.Dataframe(
+                    headers=["Substance", "Valeur LMR", "Date d'effet", "Statut"],
+                    max_height=500,
+                    overflow_scroll=True
+                )
+                product.change(self.get_product_details, product, output)
+                
+            with gr.Tab("Recherche par Substance"):
+                substance_search = gr.Textbox(
+                    label="Rechercher une substance",
+                    placeholder="Entrez un nom de substance..."
+                )
+                substance_results = gr.Dataframe(max_height=300)
+                substance_search.submit(self.search_substances, substance_search, substance_results)
                 
-                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"
+            gr.Button("Actualiser les données", variant="secondary").click(
+                lambda: self.fetcher.preload_substance_names()
             )
-        return interface
+        return ui
 
 def main():
-    app = PesticideInterface().create_interface()
-    app.launch(share=True)
+    app = PesticideApp()
+    app.create_ui().launch(share=True)
 
 if __name__ == "__main__":
     main()