app.py

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'<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é."""
        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()