import logging 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 # Configuration du logging 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 substance_status: Optional[str] = None approval_date: Optional[str] = None expiry_date: Optional[str] = None 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' } def __init__(self): self.session = self._create_session() self._substance_cache = {} self._product_cache = {} def _create_session(self): """Crée une session pour les requêtes HTTP.""" 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: """Récupère les données depuis l'API avec gestion d'erreurs et retry.""" 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 requests.RequestException as e: logger.error(f"Failed to fetch data from {url}: {str(e)}") return {"error": str(e)} def get_active_substance_details(self, substance_name: str) -> Dict: """Récupère les détails d'une substance active.""" if substance_name in self._substance_cache: return self._substance_cache[substance_name] url = f"{self.BASE_URL}/active_substances?format=json&substance_name={substance_name}&api-version=v2.0" response = self.fetch_data(url) if response and "value" in response and response["value"]: substance_data = response["value"][0] self._substance_cache[substance_name] = { "status": substance_data.get("substance_status"), "approval_date": substance_data.get("approval_date"), "expiry_date": substance_data.get("expiry_date") } return self._substance_cache[substance_name] return {} def get_products(self) -> List[Dict]: """Récupère la liste complète des produits avec pagination.""" 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) if not response or "value" not in response: break all_products.extend(response["value"]) next_link = response.get("nextLink") if next_link: url = next_link else: break 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]: """Récupère les LMR pour un produit spécifique.""" 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, pesticide_residue_id: int) -> Dict: """Récupère les détails d'une substance à partir de son ID.""" url = f"{self.BASE_URL}/pesticide_residues/{pesticide_residue_id}?format=json&api-version=v2.0" response = self.fetch_data(url) if not response or "value" not in response or not response["value"]: logger.warning(f"Pas de détails trouvés pour la substance {pesticide_residue_id}") return {"substance_name": f"Substance {pesticide_residue_id}"} substance_data = response["value"][0] substance_name = substance_data.get("substance_name") if not substance_name: logger.warning(f"Nom de substance non trouvé pour l'ID {pesticide_residue_id}") return {"substance_name": f"Substance {pesticide_residue_id}"} active_url = f"{self.BASE_URL}/active_substances?format=json&substance_name={substance_name}&api-version=v2.0" active_response = self.fetch_data(active_url) details = { "substance_name": substance_name, "status": None, "approval_date": None, "expiry_date": None } if active_response and "value" in active_response and active_response["value"]: active_data = active_response["value"][0] details.update({ "status": active_data.get("substance_status"), "approval_date": active_data.get("approval_date"), "expiry_date": active_data.get("expiry_date") }) return details def get_substance_name_by_id(self, substance_id: int) -> str: """Récupère le nom de la substance à partir de son ID.""" url = f"{self.BASE_URL}/active_substances/{substance_id}?format=json&api-version=v2.0" response = self.fetch_data(url) if response and "value" in response and response["value"]: substance_data = response["value"][0] return substance_data.get("substance_name", f"Substance {substance_id}") return f"Substance {substance_id}" def get_all_substances(self) -> List[str]: """Récupère la liste complète des substances actives avec pagination.""" if self._substance_cache: return list(self._substance_cache.keys()) all_substances = set() base_url = f"{self.BASE_URL}/active_substances?format=json&api-version=v2.0" url = base_url while url: response = self.fetch_data(url) if not response or "value" not in response: break for item in response.get("value", []): substance_name = item.get("substance_name") if substance_name: all_substances.add(substance_name) self._substance_cache[substance_name] = { "status": item.get("substance_status"), "approval_date": item.get("approval_date"), "expiry_date": item.get("expiry_date") } next_link = response.get("nextLink") if next_link: url = next_link else: break logger.info(f"Récupéré {len(all_substances)} substances au total") return sorted(all_substances) 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 = { p['product_name']: p['product_id'] for p in self.products } self.substances = self.fetcher.get_all_substances() self._cache = {} logger.info(f"Initialized interface with {len(self.product_choices)} products and {len(self.substances)} substances.") def parse_date(self, date_str: str) -> Optional[str]: """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", "%Y-%m-%dT%H:%M:%SZ"): try: return datetime.strptime(date_str, fmt).strftime("%Y-%m-%d") except ValueError: continue logger.warning(f"Impossible de parser la date : {date_str}") return None 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() 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 not start_date: return data filtered_data = [] for item in data: 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: item_date = datetime.strptime(parsed_date, "%Y-%m-%d") if (period == "Prochains 6 mois" and item_date <= start_date) or (period != "Prochains 6 mois" and item_date >= 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 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'{regulation_number}' 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é.""" logger.info(f"Récupération des détails pour le produit: {product_name}") try: if not product_name: return pd.DataFrame({"Message": ["Veuillez sélectionner un produit"]}) product_id = self.product_choices.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] mrls = self.fetcher.get_mrls(product_id) if period != "Toutes les dates": mrls = self.filter_by_period(mrls, period) if not mrls: return pd.DataFrame({"Message": ["Aucune donnée trouvée pour la période sélectionnée"]}) processed_mrls = [] with ThreadPoolExecutor(max_workers=10) as executor: futures = { executor.submit(self.fetcher.get_substance_details, mrl["pesticide_residue_id"]): mrl for mrl in mrls } for future in futures: mrl = futures[future] try: substance_details = future.result() logger.info(f"Détails de la substance récupérés: {substance_details}") # Formatage de la valeur LMR mrl_value = mrl.get("mrl_value", "") if isinstance(mrl_value, (int, float)): formatted_mrl = f"{mrl_value}{'*' if str(mrl_value).endswith('*') else ''}" else: formatted_mrl = str(mrl_value) mrl_data = { "Substance": substance_details.get("substance_name", ""), "Valeur LMR": formatted_mrl, "Date d'application": self.parse_date(mrl.get("entry_into_force_date")), "Date de modification": self.parse_date(mrl.get("modification_date")), "Règlement": self.format_regulation_link( mrl.get("regulation_url", ""), mrl.get("regulation_number", "") or mrl.get("regulation_reference", "") ), "Statut": substance_details.get("status", ""), "Date d'approbation": self.parse_date(substance_details.get("approval_date")), "Date d'expiration": self.parse_date(substance_details.get("expiry_date")) } logger.info(f"Données MRL formatées: {mrl_data}") processed_mrls.append(mrl_data) except Exception as e: logger.error(f"Erreur lors du traitement de la substance: {e}") df = pd.DataFrame(processed_mrls) if show_only_changes and "Date de modification" in df.columns: df = df[df["Date de modification"].notna()] df = df.sort_values("Date d'application", ascending=False) columns_order = [ "Substance", "Valeur LMR", "Date d'application", "Date de modification", "Règlement", "Statut", "Date d'approbation", "Date d'expiration" ] df = df[columns_order] self._cache[cache_key] = df return df except Exception as e: logger.error(f"Erreur dans get_product_details: {str(e)}") return pd.DataFrame({"Message": [f"Erreur: {str(e)}"]}) def create_graph(self, df: pd.DataFrame) -> gr.Plot: """Crée un graphique interactif pour les dates d'application.""" fig = px.scatter(df, x='Date d\'application', y='Valeur LMR', color='Substance', title='Dates d\'application des LMR') return fig def export_data(self, df: pd.DataFrame) -> str: """Exporte les données sous forme de fichier CSV.""" csv_file_path = "mrls_data.csv" df.to_csv(csv_file_path, index=False) return csv_file_path def get_substance_details_table(self, substance_name: str) -> pd.DataFrame: """Récupère et formate les détails d'une substance active.""" logger.info(f"Récupération des détails pour la substance: {substance_name}") try: if not substance_name: return pd.DataFrame({"Message": ["Veuillez sélectionner une substance"]}) substance_details = self.fetcher.get_active_substance_details(substance_name) if not substance_details: return pd.DataFrame({"Message": ["Substance non trouvée"]}) df = pd.DataFrame([substance_details]) df = df.rename(columns={ "status": "Statut", "approval_date": "Date d'approbation", "expiry_date": "Date d'expiration" }) df.insert(0, "Substance", substance_name) return df except Exception as e: logger.error(f"Erreur dans get_substance_details_table: {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(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) et les informations sur les substances actives. """) with gr.Row(): 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", "Prochains 6 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") with gr.Row(): mrls_table = gr.Dataframe( headers=["Substance", "Valeur LMR", "Date d'application", "Date de modification", "Règlement", "Statut", "Date d'approbation", "Date d'expiration"], interactive=False ) graph_output = gr.Plot(label="Graphique des Dates d'Application") with gr.Row(): export_btn = gr.Button("Exporter les données", variant="secondary") export_output = gr.File(label="Fichier CSV Exporté") with gr.Row(): substance_dropdown = gr.Dropdown( choices=sorted(self.substances), label="Substance", info="Sélectionnez une substance active" ) substance_table = gr.Dataframe( headers=["Substance", "Statut", "Date d'approbation", "Date d'expiration"], interactive=False ) fetch_btn.click( fn=self.get_product_details, inputs=[product_dropdown, period_radio, show_changes], outputs=[mrls_table] ) mrls_table.change( fn=self.create_graph, inputs=mrls_table, outputs=graph_output ) export_btn.click( fn=self.export_data, inputs=mrls_table, outputs=export_output ) substance_dropdown.change( fn=self.get_substance_details_table, inputs=substance_dropdown, outputs=substance_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) Statut** : État d'approbation de la substance active Date d'approbation** : Date d'approbation de la substance active Date d'expiration** : Date d'expiration de l'approbation """) return interface def main(): interface = PesticideInterface() app = interface.create_interface() app.launch( server_name="0.0.0.0", server_port=7860, share=True ) if __name__ == "__main__": main()