Évolution du schéma et améliorations mineures

This commit is contained in:
Fabrication du Numérique 2025-05-01 18:37:06 +02:00
parent acaa79d2a6
commit 5436ccff5e
2 changed files with 68 additions and 53 deletions

1
.gitignore vendored
View File

@ -20,7 +20,6 @@ __pycache__/
venv
# Ignorer données Fiches (adapté à ton projet)
schema.txt
Instructions.md
Fiches/

118
fabnum.py
View File

@ -6,7 +6,6 @@ import networkx as nx
import logging
import altair as alt
import numpy as np
from collections import OrderedDict
from dotenv import load_dotenv
import os
import requests
@ -20,13 +19,6 @@ from connexion import connexion, bouton_deconnexion
import tempfile
import json
st.set_page_config(
page_title="Fabnum Analyse de chaîne",
page_icon="assets/weakness.png"
)
session_id = st.context.headers.get("x-session-id")
# Configuration Gitea
load_dotenv()
@ -38,6 +30,14 @@ DEPOT_CODE = os.getenv("DEPOT_CODE", "code")
ENV = os.getenv("ENV")
ENV_CODE = os.getenv("ENV_CODE")
DOT_FILE = os.getenv("DOT_FILE")
INSTRUCTIONS = os.getenv("INSTRUCTIONS", "Instructions.md")
st.set_page_config(
page_title="Fabnum Analyse de chaîne",
page_icon="assets/weakness.png"
)
session_id = st.context.headers.get("x-session-id")
niveau_labels = {
0: "Produit final",
@ -85,6 +85,19 @@ header+="""
st.markdown(header, unsafe_allow_html=True)
def charger_instructions_depuis_gitea(nom_fichier="Instructions.md"):
headers = {"Authorization": f"token {GITEA_TOKEN}"}
url = f"{GITEA_URL}/repos/{ORGANISATION}/{DEPOT_FICHES}/contents/{nom_fichier}?ref={ENV}"
try:
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
contenu_md = base64.b64decode(data["content"]).decode("utf-8")
return contenu_md
except Exception as e:
st.error(f"Erreur lors du chargement des instructions depuis Gitea : {e}")
return None
def afficher_menu():
with st.sidebar:
st.markdown("""
@ -166,11 +179,9 @@ def charger_schema_depuis_gitea(fichier_local="schema_temp.txt"):
st.error(f"Erreur lors du chargement de {DOT_FILE} depuis Gitea : {e}")
return None
@st.cache_data(ttl=600)
def charger_arborescence_fiches():
headers = {"Authorization": f"token {GITEA_TOKEN}"}
branche = "dev" if ENV == "dev" else "public"
url_base = f"{GITEA_URL}/repos/{ORGANISATION}/{DEPOT_FICHES}/contents/Documents?ref={branche}"
url_base = f"{GITEA_URL}/repos/{ORGANISATION}/{DEPOT_FICHES}/contents/Documents?ref={ENV}"
try:
response = requests.get(url_base, headers=headers)
@ -208,46 +219,40 @@ def couleur_noeud(n, niveaux, G):
# Niveau 99 : pays géographique avec isg
if niveau == 99:
isg = int(attrs.get("isg", -1))
if isg >= 60:
return "darkred"
elif isg >= 31:
return "orange"
elif isg >= 0:
return "darkgreen"
else:
return "gray"
return (
"darkred" if isg >= 60 else
"orange" if isg >= 31 else
"darkgreen" if isg >= 0 else
"gray"
)
# Niveau 11 ou 12 connecté à un pays géographique
if niveau in (11, 12):
if niveau in (11, 12, 1011, 1012):
for succ in G.successors(n):
if niveaux.get(succ) == 99:
isg = int(G.nodes[succ].get("isg", -1))
if isg >= 60:
return "darkred"
elif isg >= 31:
return "orange"
elif isg >= 0:
return "darkgreen"
else:
return "gray"
return (
"darkred" if isg >= 60 else
"orange" if isg >= 31 else
"darkgreen" if isg >= 0 else
"gray"
)
# Logique existante pour IHH / IVC
if niveau == 10 and attrs.get("ihh_pays"):
if niveau in (10, 1010) and attrs.get("ihh_pays"):
ihh = int(attrs["ihh_pays"])
if ihh <= 15:
return "darkgreen"
elif ihh <= 25:
return "orange"
else:
return "darkred"
return (
"darkgreen" if ihh <= 15 else
"orange" if ihh <= 25 else
"darkred"
)
elif niveau == 2 and attrs.get("ivc"):
ivc = int(attrs["ivc"])
if ivc <= 15:
return "darkgreen"
elif ivc <= 30:
return "orange"
else:
return "darkred"
return (
"darkgreen" if ivc <= 15 else
"orange" if ivc <= 30 else
"darkred"
)
return "lightblue"
@ -466,7 +471,7 @@ def afficher_sankey(
liens_chemins = set()
chemins_filtres = set()
niveaux_speciaux = [1001]
niveaux_speciaux = [1000, 1001, 1002, 1010, 1011, 1012]
for chemin in chemins:
has_ihh = has_ivc = has_criticite = has_isg_critique = False
@ -484,21 +489,21 @@ def afficher_sankey(
if filtrer_ihh and ihh_type:
ihh_field = "ihh_pays" if ihh_type == "Pays" else "ihh_acteurs"
if niveau_u == 10 and int(G.nodes[u].get(ihh_field, 0)) > 25:
if niveau_u in (10, 1010) and int(G.nodes[u].get(ihh_field, 0)) > 25:
has_ihh = True
if niveau_v == 10 and int(G.nodes[v].get(ihh_field, 0)) > 25:
if niveau_v in (10, 1010) and int(G.nodes[v].get(ihh_field, 0)) > 25:
has_ihh = True
if filtrer_ivc and niveau_u == 2 and int(G.nodes[u].get("ivc", 0)) > 30:
if filtrer_ivc and niveau_u in (2, 1002) and int(G.nodes[u].get("ivc", 0)) > 30:
has_ivc = True
if filtrer_criticite and niveau_u == 1 and niveau_v == 2 and extraire_criticite(u, v) > 0.66:
if filtrer_criticite and ((niveau_u == 1 and niveau_v == 2) or (niveau_u == 1001 and niveau_v == 1002) or (niveau_u == 10 and niveau_v in (1000, 1001))) and extraire_criticite(u, v) > 0.66:
has_criticite = True
for n in (u, v):
if niveaux.get(n) == 99 and int(G.nodes[n].get("isg", 0)) >= 60:
has_isg_critique = True
elif niveaux.get(n) in (11, 12):
elif niveaux.get(n) in (11, 12, 1011, 1012):
for succ in G.successors(n):
if niveaux.get(succ) == 99 and int(G.nodes[succ].get("isg", 0)) >= 60:
has_isg_critique = True
@ -546,8 +551,19 @@ def afficher_sankey(
df_liens["value"] = 0.1
# Ne garder que les nœuds effectivement connectés
niveaux_speciaux = [1000, 1001, 1002, 1010, 1011, 1012]
# Inclure les nœuds connectés + tous les nœuds 10xx traversés dans les chemins
noeuds_utilises = set(df_liens["source"]) | set(df_liens["target"])
sorted_nodes = [n for n in sorted(G.nodes(), key=lambda x: niveaux.get(x, 99), reverse=True) if n in noeuds_utilises]
for chemin in chemins:
for n in chemin:
if niveaux.get(n) in niveaux_speciaux:
noeuds_utilises.add(n)
sorted_nodes = [
n for n in sorted(G.nodes(), key=lambda x: niveaux.get(x, 99), reverse=True)
if n in noeuds_utilises
]
def couleur_criticite(p):
if p <= 0.33:
@ -577,7 +593,7 @@ def afficher_sankey(
niveau = niveaux.get(n, 99)
# Ajout dun ISG hérité si applicable
if niveau in (11, 12):
if niveau in (11, 12, 1011, 1012):
for succ in G.successors(n):
if niveaux.get(succ) == 99 and "isg" in G.nodes[succ]:
isg_val = G.nodes[succ]["isg"]
@ -1036,8 +1052,8 @@ les enregistrer dans un fichier que vous pourrez recharger ultérieurement.
dot_file_path = None
if st.session_state.onglet == "Instructions":
with open("Instructions.md", "r", encoding="utf-8") as f:
markdown_content = f.read()
markdown_content = charger_instructions_depuis_gitea(INSTRUCTIONS)
if markdown_content:
st.markdown(markdown_content)
elif st.session_state.onglet == "Fiches":