app.py

import streamlit as st
import pandas as pd
import re
import matplotlib.pyplot as plt
import seaborn as sns
import io
import requests
from io import StringIO
import plotly.express as px
import plotly.graph_objects as go
from collections import Counter
import base64

# Configuration de la page
st.set_page_config(
    page_title="Analyse des Non-Conformités IFS",
    page_icon="📊",
    layout="wide"
)

# Fonction pour extraire les informations de la colonne Description
def extract_info_from_description(desc):
    if not isinstance(desc, str):
        return {}
        
    info = {}
    # Extraire Type
    type_match = re.search(r'Type:\s*(.+?)(?:\n|$)', desc)
    if type_match:
        info['Type'] = type_match.group(1).strip()
    
    # Extraire Category
    category_match = re.search(r'Category:\s*(.+?)(?:\n|$)', desc)
    if category_match:
        info['Category'] = category_match.group(1).strip()
    
    # Extraire Lock date
    lock_date_match = re.search(r'Lock date:\s*(.+?)(?:\n|$)', desc)
    if lock_date_match:
        info['Lock_date'] = lock_date_match.group(1).strip()
    
    # Extraire les descriptions
    en_desc_match = re.search(r'Lock Description \(en\):\s*(.+?)(?:\nLock Description \(other|$)', desc, re.DOTALL)
    if en_desc_match:
        info['Description_en'] = en_desc_match.group(1).strip()
    
    other_desc_match = re.search(r'Lock Description \(other language\):\s*(.+?)(?:\n[A-Za-z]+:|$)', desc, re.DOTALL)
    if other_desc_match:
        info['Description_fr'] = other_desc_match.group(1).strip()
    
    return info

# Fonction pour extraire les numéros d'exigence des descriptions
# Fonction pour extraire les numéros d'exigence des descriptions
def extract_requirement_numbers(text):
    if not isinstance(text, str):
        return []
    
    # Rechercher les formats de points de norme avec plusieurs approches
    # 1. Format avec le mot "requirement" suivi du numéro
    pattern1 = r'requirement\s+(\d+\.\d+\.\d+(?:\.\d+)?)'
    # 2. Format avec juste le numéro mais précédé par un tiret ou un point
    pattern2 = r'[-•:]\s*(\d+\.\d+\.\d+(?:\.\d+)?)'
    # 3. Format standard (numéro au début d'une phrase ou isolé)
    pattern3 = r'\b(\d+\.\d+\.\d+(?:\.\d+)?)\b'
    
    # Combiner les résultats des différents patterns
    matches = []
    matches.extend(re.findall(pattern1, text, re.IGNORECASE))
    matches.extend(re.findall(pattern2, text))
    matches.extend(re.findall(pattern3, text))
    
    # Éliminer les doublons
    matches = list(set(matches))
    
    # Filtrer pour éliminer les formats de date (DD.MM.YY ou DD.MM.YYYY)
    filtered_matches = []
    for match in matches:
        # Vérifier si c'est potentiellement une date
        parts = match.split('.')
        if len(parts) == 3:
            # Si le premier nombre est entre 1-31 et le deuxième entre 1-12, c'est probablement une date
            try:
                day = int(parts[0])
                month = int(parts[1])
                if 1 <= day <= 31 and 1 <= month <= 12 and len(parts[2]) <= 4:
                    # C'est probablement une date, ignorer
                    continue
            except ValueError:
                pass  # Si ce n'est pas un nombre, ce n'est pas une date
        filtered_matches.append(match)
    
    return filtered_matches if filtered_matches else []

# Fonction pour charger les données des non-conformités
@st.cache_data
def load_nonconformity_data(file):
    try:
        df = pd.read_csv(file, encoding='utf-8')
    except:
        try:
            df = pd.read_csv(file, encoding='latin1')
        except:
            df = pd.read_csv(file, encoding='ISO-8859-1')
    
    # Renommer les colonnes si nécessaire
    if len(df.columns) >= 2:
        # Supposer que la première colonne est COID et la seconde contient les descriptions
        df = df.rename(columns={df.columns[0]: 'COID', df.columns[1]: 'Description'})
    
    # Extraire les informations des descriptions
    extracted_data = []
    for idx, row in df.iterrows():
        coid = row['COID']
        desc = row['Description'] if 'Description' in df.columns else None
        
        if desc is not None:
            info = extract_info_from_description(desc)
            info['COID'] = coid
            
            # Extraire les numéros d'exigence
            if 'Description_en' in info:
                info['requirement_numbers_en'] = extract_requirement_numbers(info['Description_en'])
            if 'Description_fr' in info:
                info['requirement_numbers_fr'] = extract_requirement_numbers(info['Description_fr'])
            
            # Combiner les numéros d'exigence
            all_reqs = set()
            if 'requirement_numbers_en' in info:
                all_reqs.update(info['requirement_numbers_en'])
            if 'requirement_numbers_fr' in info:
                all_reqs.update(info['requirement_numbers_fr'])
            
            info['requirement_numbers'] = list(all_reqs)
            
            extracted_data.append(info)
    
    # Créer un nouveau DataFrame avec les informations extraites
    if extracted_data:
        return pd.DataFrame(extracted_data)
    else:
        return df  # Retourner le DataFrame original si aucune donnée n'a été extraite

# Fonction pour charger les données de recommandation IFS depuis GitHub
@st.cache_data
def load_ifs_recommendations():
    url = "https://raw.githubusercontent.com/M00N69/Action-plan/main/Guide%20Checklist_IFS%20Food%20V%208%20-%20CHECKLIST.csv"
    
    try:
        response = requests.get(url)
        response.raise_for_status()  # Raise an exception for HTTP errors
        
        # Use StringIO to create a file-like object from the response content
        csv_data = StringIO(response.text)
        
        # Read the CSV file
        df = pd.read_csv(csv_data, sep=',')
        
        # Clean the NUM_REQ column to create a proper requirement_id column
        df['requirement_id'] = df['NUM_REQ'].str.replace('*', '').str.strip()
        
        # Create a lookup dictionary for quicker reference
        # Format: 'chapter.section.req' -> row data
        lookup_dict = {}
        
        for idx, row in df.iterrows():
            req_id = row['requirement_id']
            if pd.notna(req_id) and isinstance(req_id, str):
                lookup_dict[req_id] = row.to_dict()
                
                # Also add simplified formats (e.g. if NUM_REQ is "KO N°1 1.2.1", add "1.2.1" as well)
                simple_id_match = re.search(r'(\d+\.\d+\.\d+(?:\.\d+)?)', req_id)
                if simple_id_match:
                    simple_id = simple_id_match.group(1)
                    if simple_id != req_id:
                        lookup_dict[simple_id] = row.to_dict()
        
        df.lookup_dict = lookup_dict
        return df
    except Exception as e:
        st.error(f"Erreur lors du chargement des recommandations IFS: {e}")
        return pd.DataFrame()

# Fonction pour trouver les recommandations correspondantes
def find_matching_recommendations(req_numbers, recommendations_df):
    if not hasattr(recommendations_df, 'lookup_dict'):
        # Utiliser la méthode traditionnelle si le dictionnaire de recherche n'existe pas
        matching_recs = recommendations_df[recommendations_df['requirement_id'].isin(req_numbers)]
        return matching_recs
    
    # Utiliser le dictionnaire de recherche pour une recherche plus efficace et flexible
    matching_rows = []
    
    for req in req_numbers:
        # Essayer d'abord le format exact
        if req in recommendations_df.lookup_dict:
            matching_rows.append(recommendations_df.lookup_dict[req])
        else:
            # Essayer différentes variantes du format (par exemple, 4.13.2 pourrait être listé comme 4.13.2 ou 4.13.02)
            req_parts = req.split('.')
            if len(req_parts) >= 3:
                # Essayer avec des zéros ajoutés pour les nombres à un chiffre
                padded_parts = []
                for part in req_parts:
                    try:
                        num = int(part)
                        padded_parts.append(f"{num:02d}")
                    except ValueError:
                        padded_parts.append(part)
                
                padded_req = '.'.join(padded_parts)
                
                if padded_req in recommendations_df.lookup_dict:
                    matching_rows.append(recommendations_df.lookup_dict[padded_req])
                    continue
                
                # Essayer avec des zéros supprimés
                unpadded_parts = []
                for part in req_parts:
                    try:
                        num = int(part)
                        unpadded_parts.append(str(num))
                    except ValueError:
                        unpadded_parts.append(part)
                
                unpadded_req = '.'.join(unpadded_parts)
                
                if unpadded_req in recommendations_df.lookup_dict:
                    matching_rows.append(recommendations_df.lookup_dict[unpadded_req])
                    continue
                
                # Essayer une correspondance partielle
                for key in recommendations_df.lookup_dict:
                    if key.startswith(req) or req.startswith(key):
                        matching_rows.append(recommendations_df.lookup_dict[key])
                        break
    
    if matching_rows:
        return pd.DataFrame(matching_rows)
    else:
        return pd.DataFrame()

# Interface utilisateur
st.title("Analyse des Non-Conformités d'Audit IFS")

# Sélection de la langue
language = st.sidebar.selectbox(
    "Langue / Language",
    ["Français", "English"]
)

# Chargement direct depuis GitHub
st.sidebar.info("Les données sont chargées directement depuis le dépôt GitHub" if language == "Français" else "Data is loaded directly from the GitHub repository")

@st.cache_data
def load_github_nonconformity_data():
    url = "https://raw.githubusercontent.com/M00N69/Action-plan/main/MAJEURES%20.csv"
    try:
        response = requests.get(url)
        response.raise_for_status()
        
        # Utiliser StringIO pour créer un objet de type fichier à partir du contenu de la réponse
        csv_data = StringIO(response.text)
        
        # Lire le fichier CSV
        return load_nonconformity_data(csv_data)
    except Exception as e:
        st.error(f"Erreur lors du chargement des données depuis GitHub: {e}")
        return pd.DataFrame()

# Option pour charger manuellement ou utiliser GitHub
data_source = st.sidebar.radio(
    "Source des données" if language == "Français" else "Data source",
    ["GitHub", "Fichier local" if language == "Français" else "Local file"]
)

nonconformity_df = None
recommendations_df = None

if data_source == "GitHub":
    # Chargement des données
    nonconformity_df = load_github_nonconformity_data()
    recommendations_df = load_ifs_recommendations()
    
    with st.sidebar.expander("À propos des données" if language == "Français" else "About the data"):
        st.write("Source: https://github.com/M00N69/Action-plan/blob/main/MAJEURES%20.csv")
else:
    # Upload du fichier CSV des non-conformités
    uploaded_file = st.sidebar.file_uploader(
        "Importez votre fichier CSV de non-conformités" if language == "Français" else "Upload your non-conformity CSV file",
        type=["csv"]
    )
    
    if uploaded_file is not None:
        # Chargement des données
        nonconformity_df = load_nonconformity_data(uploaded_file)
        recommendations_df = load_ifs_recommendations()
    else:
        st.info(
            "Veuillez importer un fichier CSV contenant des données de non-conformités IFS" 
            if language == "Français" 
            else "Please upload a CSV file containing IFS non-conformity data"
        )
        st.stop()

# Vérifier que les données sont chargées
if nonconformity_df is not None and recommendations_df is not None:
    # Afficher les données brutes pour le débogage
    with st.expander("Aperçu des données (débogage)" if language == "Français" else "Data preview (debugging)"):
        st.write("Colonnes du DataFrame:" if language == "Français" else "DataFrame columns:", nonconformity_df.columns.tolist())
        st.dataframe(nonconformity_df.head())
    
    # Vérifier que nous avons des numéros d'exigence
    has_requirements = 'requirement_numbers' in nonconformity_df.columns
    
    # Affichage des informations générales
    st.subheader("Informations générales" if language == "Français" else "General Information")
    
    col1, col2, col3 = st.columns(3)
    with col1:
        st.metric(
            "Nombre total de non-conformités" if language == "Français" else "Total number of non-conformities",
            len(nonconformity_df)
        )
    
    with col2:
        # Compter le nombre total d'exigences uniques
        all_requirements = []
        if has_requirements:
            for req_list in nonconformity_df['requirement_numbers']:
                if isinstance(req_list, list):
                    all_requirements.extend(req_list)
        
        st.metric(
            "Nombre d'exigences différentes concernées" if language == "Français" else "Number of different requirements concerned",
            len(set(all_requirements)) if all_requirements else 0
        )
    
    with col3:
        # Compter le nombre de COID uniques
        if 'COID' in nonconformity_df.columns:
            num_companies = nonconformity_df['COID'].nunique()
        else:
            num_companies = "N/A"
            
        st.metric(
            "Nombre d'entreprises concernées" if language == "Français" else "Number of companies concerned",
            num_companies
        )
    
    # Analyse des points de norme les plus fréquents
    st.subheader("Points de norme les plus fréquemment concernés" if language == "Français" else "Most frequently concerned standard points")
    
    if has_requirements and all_requirements:
        # Compter la fréquence de chaque exigence
        requirement_counts = Counter(all_requirements)
        
        # Convertir en DataFrame pour la visualisation
        req_freq_df = pd.DataFrame({
            'requirement': list(requirement_counts.keys()),
            'count': list(requirement_counts.values())
        }).sort_values('count', ascending=False)
        
        if not req_freq_df.empty:
            # Visualisation des points de norme les plus fréquents
            fig = px.bar(
                req_freq_df.head(10),
                x='requirement',
                y='count',
                title="Top 10 des points de norme" if language == "Français" else "Top 10 Standard Points",
                labels={'requirement': 'Exigence' if language == 'Français' else 'Requirement', 'count': 'Fréquence' if language == 'Français' else 'Frequency'}
            )
            st.plotly_chart(fig, use_container_width=True)
        
            # Filtrer par point de norme
            st.subheader("Filtrer par point de norme" if language == "Français" else "Filter by standard point")
            
            # Créer une liste de tous les points de norme disponibles
            all_requirements_list = sorted(list(requirement_counts.keys()))
            
            selected_requirement = st.selectbox(
                "Sélectionnez un point de norme" if language == "Français" else "Select a standard point",
                all_requirements_list
            )
            
            if selected_requirement:
                # Filtrer les non-conformités qui contiennent ce point de norme
                filtered_rows = []
                for idx, row in nonconformity_df.iterrows():
                    if 'requirement_numbers' in row and isinstance(row['requirement_numbers'], list) and selected_requirement in row['requirement_numbers']:
                        filtered_rows.append(row)
                
                filtered_df = pd.DataFrame(filtered_rows)
                
                if not filtered_df.empty:
                    st.write(f"{'Non-conformités concernant le point' if language == 'Français' else 'Non-conformities concerning point'} {selected_requirement}: **{len(filtered_df)} cas trouvés**")
                    
                    # Créer un tableau comparatif des entreprises avec la même exigence
                    comp_table = []
                    desc_col_fr = 'Description_fr'
                    desc_col_en = 'Description_en'
                    
                    # Afficher toutes les non-conformités, même sans description détaillée
                    for idx, row in filtered_df.iterrows():
                        desc_fr = row.get(desc_col_fr, '') if pd.notna(row.get(desc_col_fr, '')) else ''
                        desc_en = row.get(desc_col_en, '') if pd.notna(row.get(desc_col_en, '')) else ''
                        
                        # Utiliser la description dans la langue sélectionnée, ou l'autre si non disponible
                        description = desc_fr if language == "Français" and desc_fr else desc_en
                        if not description and desc_fr:
                            description = desc_fr
                        
                        # Mettre en évidence le texte contenant le numéro d'exigence si disponible
                        highlighted_text = description
                        if description:
                            # Trouver les phrases contenant le numéro d'exigence
                            sentences = re.split(r'(?<=[.!?])\s+', description)
                            highlighted_desc = []
                            
                            for sentence in sentences:
                                if selected_requirement in sentence:
                                    highlighted_desc.append(f"<span style='background-color: #FFFF00'>{sentence}</span>")
                                else:
                                    highlighted_desc.append(sentence)
                            
                            highlighted_text = " ".join(highlighted_desc)
                        else:
                            highlighted_text = "<em>Aucune description détaillée disponible</em>"
                        
                        comp_table.append({
                            "COID": row['COID'],
                            "Type": row.get('Type', ''),
                            "Lock_date": row.get('Lock_date', ''),
                            "Description": highlighted_text
                        })
                    
                    # Afficher le tableau comparatif
                    st.subheader("Tableau comparatif des cas" if language == "Français" else "Comparative table of cases")
                    
                    # Créer un DataFrame pour le tableau
                    comp_df = pd.DataFrame(comp_table)
                    
                    if not comp_df.empty:
                        # Afficher chaque ligne du tableau avec la description mise en forme
                        for idx, row in comp_df.iterrows():
                            with st.container():
                                cols = st.columns([1, 1, 1, 5])
                                with cols[0]:
                                    st.write(f"**COID:**<br>{row['COID']}", unsafe_allow_html=True)
                                with cols[1]:
                                    st.write(f"**Type:**<br>{row['Type']}", unsafe_allow_html=True)
                                with cols[2]:
                                    st.write(f"**Date:**<br>{row['Lock_date']}", unsafe_allow_html=True)
                                with cols[3]:
                                    st.write(f"**Description:**<br>{row['Description']}", unsafe_allow_html=True)
                                
                                st.markdown("---")
                    else:
                        st.warning(
                            "Aucune description détaillée n'a été trouvée pour ce point de norme. Le point a été détecté mais les phrases spécifiques ne sont pas disponibles." 
                            if language == "Français" 
                            else "No detailed description was found for this standard point. The point was detected but the specific sentences are not available."
                        )
                    
                    # Trouver et afficher les recommandations correspondantes
                    matching_recs = find_matching_recommendations([selected_requirement], recommendations_df)
                    
                    if not matching_recs.empty:
                        st.subheader("Recommandations IFS" if language == "Français" else "IFS Recommendations")
                        
                        for idx, rec in matching_recs.iterrows():
                            # Créer un affichage plus structuré des recommandations
                            with st.expander(f"Exigence {rec['NUM_REQ']}", expanded=True):
                                st.markdown(f"**{'Description' if language == 'Français' else 'Description'}:**")
                                st.info(rec['IFS Requirements'])
                                
                                if pd.notna(rec['Good practice']):
                                    st.markdown(f"**{'Bonnes pratiques' if language == 'Français' else 'Good practice'}:**")
                                    st.success(rec['Good practice'])
                                
                                if pd.notna(rec['Example questions']):
                                    st.markdown(f"**{('Questions exemple' if language == 'Français' else 'Example questions')}:**")
                                    
                                    # Formater les questions en liste à puces
                                    questions = rec['Example questions'].split('•')
                                    for q in questions:
                                        if q.strip():
                                            st.write(f"• {q.strip()}")
                                
                                if pd.notna(rec['Elements to check']):
                                    st.markdown(f"**{'Éléments à vérifier' if language == 'Français' else 'Elements to check'}:**")
                                    
                                    # Formater les éléments en liste à puces
                                    elements = rec['Elements to check'].split('•')
                                    for e in elements:
                                        if e.strip():
                                            st.write(f"• {e.strip()}")
                                
                                if pd.notna(rec['Example for non-conformities']):
                                    st.markdown(f"**{'Exemples de non-conformités' if language == 'Français' else 'Example for non-conformities'}:**")
                                    st.error(rec['Example for non-conformities'])
                                
                    else:
                        st.info("Aucune recommandation trouvée pour ce point de norme" if language == "Français" else "No recommendations found for this standard point")
                else:
                    st.info("Aucune non-conformité trouvée pour ce point de norme" if language == "Français" else "No non-conformities found for this standard point")
        else:
            st.warning("Aucun point de norme n'a été trouvé dans les données" if language == "Français" else "No standard points were found in the data")
    else:
        st.warning("Aucun point de norme n'a été extrait des descriptions" if language == "Français" else "No standard points were extracted from the descriptions")

    # Export des données
    st.subheader("Exporter les données" if language == "Français" else "Export data")
    
    if st.button("Télécharger l'analyse complète en CSV" if language == "Français" else "Download complete analysis as CSV"):
        # Créer un DataFrame enrichi pour l'export
        export_df = nonconformity_df.copy()
        
        # Convertir les listes en chaînes pour l'export
        if has_requirements:
            export_df['requirement_numbers'] = export_df['requirement_numbers'].apply(
                lambda x: ', '.join(x) if isinstance(x, list) else '')
        
        # Convertir en CSV
        csv = export_df.to_csv(index=False)
        
        # Créer un lien de téléchargement
        b64 = base64.b64encode(csv.encode()).decode()
        href = f'<a href="data:file/csv;base64,{b64}" download="ifs_nonconformities_analysis.csv">Télécharger le CSV</a>'
        st.markdown(href, unsafe_allow_html=True)

# Afficher des informations sur l'application
st.sidebar.markdown("---")
st.sidebar.info(
    "Cette application analyse les non-conformités majeures ou KO attribuées lors des audits IFS et les compare aux recommandations officielles."
    if language == "Français"
    else "This application analyzes major non-conformities or KO attributed during IFS audits and compares them to official recommendations."
)