Rename apptest.py to app.py

app.py

@@ -0,0 +1,249 @@
+import logging
+from datetime import datetime, timedelta
+from typing import Dict, List, Optional, Any
+import pandas as pd
+import requests
+import plotly.express as px
+import streamlit as st
+from tenacity import retry, stop_after_attempt, wait_exponential
+import urllib.parse
+
+# Configuration Streamlit
+st.set_page_config(page_title="Pesticide Data Explorer", page_icon="🌿", layout="wide")
+
+# Configuration logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s",
+    handlers=[logging.FileHandler("pesticide_app.log", encoding="utf-8"), logging.StreamHandler()],
+)
+logger = logging.getLogger(__name__)
+
+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"
+    }
+
+    def __init__(self):
+        self.session = requests.Session()
+        self.session.headers.update(self.HEADERS)
+
+    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
+    def fetch_data(self, url: str, params: Optional[Dict] = None) -> Dict[str, Any]:
+        """Effectue une requête GET avec gestion des erreurs améliorée"""
+        try:
+            response = self.session.get(url, params=params, timeout=15)
+            response.raise_for_status()
+            return response.json()
+        except requests.RequestException as e:
+            logger.error(f"Erreur API : {e}")
+            if "timeout" in str(e).lower():
+                logger.warning("Timeout - tentative de reconnexion...")
+            return {}
+
+    def get_mrls(self, product_ids: Optional[List[int]] = None) -> List[Dict]:
+        """Récupère les LMR pour une liste de produits, gère la pagination."""
+        all_mrls = []
+
+        if not product_ids:
+            #If not product specified, get them all
+            url = f"{self.BASE_URL}/pesticide_residues_mrls?format=json&api-version=v2.0"
+            while url :
+                response = self.fetch_data(url)
+                if response and "value" in response:
+                     all_mrls.extend(response["value"])
+                next_link = response.get("nextLink")
+                if next_link:
+                    url = urllib.parse.urljoin(self.BASE_URL, next_link)
+                else:
+                    url = None
+            return all_mrls
+
+        for product_id in product_ids:
+            url = f"{self.BASE_URL}/pesticide_residues_products/{product_id}/mrls?format=json&language=FR&api-version=v2.0"
+            while url:
+                response = self.fetch_data(url)
+                if response and "value" in response:
+                    all_mrls.extend(response["value"])
+                next_link = response.get("nextLink")
+                if next_link:
+                    url = urllib.parse.urljoin(self.BASE_URL, next_link)
+                else:
+                    url = None  # No more pages, exit the inner loop
+
+        return all_mrls
+
+@st.cache_data(ttl=3600)  # Cache augmented to 1 hour
+def get_products() -> List[Dict]:
+    fetcher = PesticideDataFetcher()
+    url = f"{fetcher.BASE_URL}/pesticide_residues_products?format=json&language=FR&api-version=v2.0"
+
+    all_products = []
+    while url:
+        response = fetcher.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 = urllib.parse.urljoin(fetcher.BASE_URL, next_link)
+        else:
+            url = None
+    return all_products
+
+@st.cache_data(ttl=3600) # Cache augmented to 1 hour and corrected function
+def get_all_substances() -> Dict[int, str]:
+    fetcher = PesticideDataFetcher()
+    url = f"{fetcher.BASE_URL}/active_substances?format=json&api-version=v2.0" # No language needed
+    all_substances = {}  #  Initialize as an empty dict
+    while url:
+        response = fetcher.fetch_data(url)
+        if not response or "value" not in response:
+            break
+        for item in response["value"]:
+            # Corrected Keys. Use get to avoid KeyError.
+            substance_id = item.get("substanceId")
+            substance_name = item.get("substanceName")
+            if substance_id and substance_name:  # Check for None
+                all_substances[substance_id] = substance_name  # int key, str value
+        next_link = response.get("nextLink")
+        if next_link:
+            url = urllib.parse.urljoin(fetcher.BASE_URL, next_link)
+        else:
+            url = None
+    return all_substances
+
+
+class PesticideInterface:
+    def __init__(self):
+        self.fetcher = PesticideDataFetcher()
+        self.products = get_products()
+        self.product_choices = {p.get('productName', 'Unknown'): p.get('productId', None) for p in self.products}
+        # Ensure product_choices only contains valid entries
+        self.product_choices = {k: v for k, v in self.product_choices.items() if k is not None and v is not None}
+
+        self.substances = get_all_substances()
+
+
+    def get_product_details(self, product_names: List[str], future_only: bool = False) -> pd.DataFrame:
+        product_ids = [self.product_choices[name] for name in product_names if name in self.product_choices]
+        all_mrls = self.fetcher.get_mrls(product_ids)
+
+        df = pd.DataFrame(all_mrls)
+        if df.empty:
+            if future_only:
+                st.info("Aucun changement de LMR prévu dans les 6 prochains mois pour les produits sélectionnés.")
+            return df
+
+        logger.info(f"Nombre total d'entrées : {len(df)}")
+        logger.info(f"Colonnes disponibles : {df.columns.tolist()}")
+
+        df["Substance"] = df["pesticideResidueId"].map(self.substances)
+        missing_substances = df[df["Substance"].isna()]["pesticideResidueId"].unique()
+        if len(missing_substances) > 0:
+            logger.warning(f"Substances non trouvées: {missing_substances}")
+        df["Substance"] = df["Substance"].fillna("Inconnu")
+
+
+
+        df["Date d'application"] = pd.to_datetime(df["entryIntoForceDate"], errors="coerce")
+
+        if future_only:
+            now = datetime.now()
+            future_date = now + timedelta(days=180)
+            future_df = df[
+                (df["Date d'application"] > now) &
+                (df["Date d'application"] <= future_date)
+            ]
+
+            if future_df.empty:
+                st.info(f"🔍 Aucun changement de LMR prévu entre le {now.strftime('%d/%m/%Y')} et le {future_date.strftime('%d/%m/%Y')} pour les produits sélectionnés.")
+                return pd.DataFrame()  # Return empty df
+
+            df = future_df
+
+
+        df = df.rename(columns={"mrlValue": "Valeur LMR"})
+        df["Valeur LMR"] = pd.to_numeric(df["Valeur LMR"], errors='coerce')
+        columns = ["Substance", "Valeur LMR", "Date d'application"]
+
+        df = df[columns].sort_values("Date d'application", ascending=False)
+
+        return df
+
+    def create_interface(self):
+        st.title("🌿 Base de données des pesticides de l'UE")
+
+        col1, col2 = st.columns([3, 1])
+        with col1:
+            product_names = st.multiselect(
+                "Sélectionnez un ou plusieurs produits",
+                list(self.product_choices.keys())
+            )
+        with col2:
+            future_only = st.checkbox("Uniquement les 6 prochains mois", value=False)
+
+        if st.button("Afficher les données"):
+            if not product_names:
+                st.warning("Veuillez sélectionner au moins un produit.")
+                return
+
+            df = self.get_product_details(product_names, future_only)
+
+            if df.empty:
+                return  # The info message has already been displayed in get_product_details
+            else:
+                if future_only:
+                    st.markdown("### Changements de LMR prévus dans les 6 prochains mois")
+                else:
+                    st.markdown("### Tableau des LMR")
+
+                df_display = df.copy()
+                df_display["Date d'application"] = df_display["Date d'application"].dt.strftime('%d/%m/%Y')
+
+                def format_value(val):
+                    if pd.isna(val):
+                        return '-'
+                    elif isinstance(val, (int, float)):
+                        return f"{val:.3f}"  # Format numbers to 3 decimal places
+                    return val
+
+
+                # Apply the custom formatting using .style.format
+                styled_df = df_display.style.format({
+                    'Valeur LMR': format_value,
+                })
+
+                 # Add a summary of the changes
+                if not df.empty:
+                    nb_changes = len(df)
+                    st.info(f"📊 {nb_changes} entrée{'s' if nb_changes > 1 else ''} trouvée{'s' if nb_changes > 1 else ''}.")
+
+                st.dataframe(styled_df) # This is better than to_html
+
+                # Create visualizations
+                self.create_visualizations(df)
+
+    def create_visualizations(self, df: pd.DataFrame):
+        """Crée les visualisations des données"""
+        # Graphique d'évolution des LMR
+        if not df.empty:
+            fig1 = px.scatter(
+                df,
+                x="Date d'application",
+                y="Valeur LMR",
+                color="Substance",
+                title="Évolution des LMR dans le temps",
+                hover_data=["Substance"] # Added substance to hover
+            )
+            st.plotly_chart(fig1, use_container_width=True)
+
+def main():
+    interface = PesticideInterface()
+    interface.create_interface()
+
+if __name__ == "__main__":
+    main()