Code/app/plan_d_action/interface.py

#!/usr/bin/env python3
#
import networkx as nx

"""
Script pour générer un rapport factorisé des vulnérabilités critiques
suivant la structure définie dans Remarques.md.
"""

import uuid

import streamlit as st
from networkx.drawing.nx_agraph import write_dot

from app.plan_d_action import (
    JOBS,
    exporter_graphe_filtre,
    extraire_chemins_selon_criteres,
    extraire_liens_filtres,
    extraire_niveaux,
    initialiser_interface,
    niveau_labels,
    preparer_graphe,
    remplacer_par_badge,
    selectionner_minerais,
    selectionner_noeuds,
)
from batch_ia import (
    calculate_vulnerabilities,
    extract_data_from_graph,
    generate_report,
    load_config,
    parse_graphs,
    write_report,
)
from utils.translations import _
from utils.widgets import html_expander

inverse_niveau_labels = {v: k for k, v in niveau_labels.items()}


def interface_plan_d_action(G_temp: nx.DiGraph) -> None:
    """Interface pour planifier l'action de sélection des minerais.

    Args:
        G_temp (nx.DiGraph): Le graphe temporaire à analyser.

    Returns:
        None: Modifie le state du Streamlit avec les données nécessaires
            pour la génération du rapport factorisé des vulnérabilités critiques.
    """
    if "sel_prod" not in st.session_state:
        st.session_state.sel_prod = None
    if "sel_comp" not in st.session_state:
        st.session_state.sel_comp = None
    if "sel_miner" not in st.session_state:
        st.session_state.sel_miner = None

    if "plan_d_action" not in st.session_state:
        st.session_state["plan_d_action"] = 0
    if "g_md_done" not in st.session_state:
        st.session_state["g_md_done"] = False
    if "uuid" not in st.session_state:
        st.session_state["uuid"] = str(uuid.uuid4())[:8]
        st.session_state["G_dot"] = f"{JOBS}/{st.session_state["uuid"]}.dot"
        st.session_state["G_md"] = f"{JOBS}/{st.session_state["uuid"]}.md"

    if st.session_state["plan_d_action"] == 0:
        st.markdown(f"# {str(_('pages.plan_d_action.title'))}")
        html_expander(f"{str(_('pages.plan_d_action.help'))}", content="\n".join(_("pages.plan_d_action.help_content")), open_by_default=False, details_class="details_introduction")
        # Préparation du graphe
        G_temp, niveaux_temp = preparer_graphe(G_temp)

        # Sélection des niveaux
        niveau_depart = 0
        niveau_arrivee = 99
        # Sélection fine des noeuds
        noeuds_depart, noeuds_arrivee = selectionner_noeuds(G_temp, niveaux_temp, niveau_depart)
        # Sélection des minerais si nécessaire
        if noeuds_depart:
            minerais = selectionner_minerais(G_temp, noeuds_depart)
            # Étape 1 : Extraction des niveaux des nœuds
            niveaux = extraire_niveaux(G_temp)
            # Étape 2 : Extraction des chemins selon les critères
            chemins = extraire_chemins_selon_criteres(G_temp, niveaux, niveau_depart, noeuds_depart, noeuds_arrivee, minerais)
            niveaux_speciaux = [1000, 1001, 1002, 1010, 1011, 1012]
            # Extraction des liens sans filtrage
            liens_chemins = extraire_liens_filtres(chemins, niveaux, niveau_depart, niveau_arrivee, niveaux_speciaux)
        else:
            liens_chemins = None

        if liens_chemins:
            G_final = exporter_graphe_filtre(G_temp, liens_chemins)
            st.session_state["G_final"] = G_final
            # formulaire ou sélection
            if st.button(str(_("pages.plan_d_action.submit_request")), icon=":material/send:"):
                # On déclenche la suite — mais on NE traite rien maintenant
                st.session_state["plan_d_action"] = 1
                st.rerun()  # force la réexécution immédiatement avec état mis à jour

    elif st.session_state["plan_d_action"] == 1:
        st.markdown("")
        # Traitement lourd une seule fois
        if not st.session_state["g_md_done"]:
            write_dot(st.session_state["G_final"], st.session_state["G_dot"])
            config = load_config()
            graph, ref_graph = parse_graphs(st.session_state["G_dot"])
            data = extract_data_from_graph(graph, ref_graph)
            results = calculate_vulnerabilities(data, config)
            report, file_names = generate_report(data, results, config)
            write_report(remplacer_par_badge(report), st.session_state["G_md"])
            st.session_state["g_md_done"] = True  # pour ne pas re-traiter à chaque affichage

        # Affichage de l’interface Streamlit
        initialiser_interface(st.session_state["G_md"])
        if (st.button("Réinitialiser", icon=":material/refresh:")):
            st.session_state["plan_d_action"] = 0
            st.session_state["g_md_done"] = False
            st.session_state.sel_prod = None
            st.session_state.sel_comp = None
            st.session_state.sel_miner = None
            for f in JOBS.glob(f"*{st.session_state["uuid"]}*"):
                if f.is_file():
                    f.unlink()
            st.rerun()