Code/utils/fiche_dynamic.py
Fabrication du Numérique 8710014345 Modifications du jour
2025-05-08 21:35:37 +02:00

553 lines
21 KiB
Python

import re, yaml, pandas as pd, textwrap
import unicodedata
from jinja2 import Template
import streamlit as st
def pastille(indice, valeur):
try:
SEUILS = st.session_state['seuils']
VERT = SEUILS[indice]["vert"]["max"]
ROUGE = SEUILS[indice]["rouge"]["min"]
pastille_verte = ""
pastille_orange = "🔶"
pastille_rouge = "🔴"
if float(valeur) < VERT:
return pastille_verte
elif float(valeur) > ROUGE:
return pastille_rouge
else:
return pastille_orange
except:
return ""
# -------- repère chaque bloc ```yaml … ``` -------------
PAIR_RE = re.compile(r"```yaml[^\n]*\n(.*?)```", re.S | re.I)
def _pairs_dataframe(md: str) -> pd.DataFrame:
rows = []
for raw in PAIR_RE.findall(md):
bloc = yaml.safe_load(raw)
if isinstance(bloc, dict) and "pair" in bloc:
rows.append(bloc["pair"])
return pd.DataFrame(rows)
def _normalize_unicode(text: str) -> str:
return unicodedata.normalize("NFKC", text)
def _fill(segment: str, pair: dict) -> str:
segment = _normalize_unicode(segment)
for k, v in pair.items():
val = f"{v:.2f}" if isinstance(v, (int, float)) else str(v)
segment = re.sub(
rf"{{{{\s*{re.escape(k)}\s*}}}}",
val,
segment,
flags=re.I,
)
segment = re.sub(
r"ICS\s*=\s*[-+]?\d+(?:\.\d+)?",
f"ICS = {pair['ics']:.2f}",
segment,
count=1,
)
return segment
def _segments(md: str):
blocs = list(PAIR_RE.finditer(md))
for i, match in enumerate(blocs):
pair = yaml.safe_load(match.group(1))["pair"]
start = match.end()
end = blocs[i + 1].start() if i + 1 < len(blocs) else len(md)
segment = md[start:end]
yield pair, segment
def _pivot(df: pd.DataFrame) -> str:
out = []
for min_, g in df.groupby("minerai"):
out += [f"## {min_}",
"| Composant | ICS | Faisabilité technique | Délai d'implémentation | Impact économique |",
"| :-- | :--: | :--: | :--: | :--: |"]
for _, r in g.sort_values("ics", ascending=False).iterrows():
out += [f"| {r.composant} | {r.ics:.2f} | {r.f_tech:.2f} | "
f"{r.delai:.2f} | {r.cout:.2f} |"]
out.append("")
return "\n".join(out)
def _synth(df: pd.DataFrame) -> str:
lignes = ["| Composant | Minerai | ICS |", "| :-- | :-- | :--: |"]
for _, r in df.sort_values("ics", ascending=False).iterrows():
lignes.append(f"| {r.composant} | {r.minerai} | {r.ics:.2f} |")
return "\n".join(lignes)
def build_dynamic_sections(md_raw: str) -> str:
md_raw = _normalize_unicode(md_raw)
df = _pairs_dataframe(md_raw)
if df.empty:
return md_raw
couples = ["# Criticité par couple Composant -> Minerai"]
for pair, seg in _segments(md_raw):
if pair:
couples.append(_fill(seg, pair))
couples_md = "\n".join(couples)
pivot_md = _pivot(df)
synth_md = _synth(df)
md = re.sub(r"#\s+Criticité par couple.*", couples_md, md_raw, flags=re.S | re.I)
md = re.sub(r"<!---- AUTO-BEGIN:PIVOT -->.*?<!---- AUTO-END:PIVOT -->",
f"<!---- AUTO-BEGIN:PIVOT -->\n{pivot_md}\n<!---- AUTO-END:PIVOT -->",
md, flags=re.S)
md = re.sub(r"<!---- AUTO-BEGIN:TABLEAU-FINAL -->.*?<!---- AUTO-END:TABLEAU-FINAL -->",
f"<!---- AUTO-BEGIN:TABLEAU-FINAL -->\n{synth_md}\n<!---- AUTO-END:TABLEAU-FINAL -->",
md, flags=re.S)
return textwrap.dedent(md)
IVC_RE = re.compile(r"```yaml\s+minerai:(.*?)```", re.S | re.I)
def _synth_ivc(minerais: list[dict]) -> str:
"""Crée un tableau de synthèse pour les IVC des minerais."""
lignes = [
"| Minerai | IVC | Vulnérabilité |",
"| :-- | :-- | :-- |"
]
for minerai in minerais:
lignes.append(
f"| {minerai['nom']} | {minerai['ivc']} | {minerai['vulnerabilite']} |"
)
return "\n".join(lignes)
def _ivc_segments(md: str):
"""Yield (dict, segment) pour chaque bloc IVC yaml."""
pos = 0
for m in IVC_RE.finditer(md):
bloc = yaml.safe_load("minerai:" + m.group(1))
start = m.end()
next_match = IVC_RE.search(md, start)
end = next_match.start() if next_match else len(md)
yield bloc["minerai"], md[start:end].strip()
pos = end
yield None, md[pos:] # reste éventuel
def build_ivc_sections(md: str) -> str:
"""Remplace les blocs YAML minerai + segment avec rendu Jinja2, conserve l'intro."""
segments = []
minerais = [] # Pour collecter les données de chaque minerai
intro = None
matches = list(IVC_RE.finditer(md))
if matches:
first = matches[0]
intro = md[:first.start()].strip()
else:
return md # pas de blocs à traiter
for m in matches:
bloc = yaml.safe_load("minerai:" + m.group(1))
minerais.append(bloc["minerai"]) # Collecte les données
start = m.end()
next_match = IVC_RE.search(md, start)
end = next_match.start() if next_match else len(md)
rendered = Template(md[start:end].strip()).render(**bloc["minerai"])
segments.append(rendered)
if intro:
segments.insert(0, intro)
# Créer et insérer le tableau de synthèse
synth_table = _synth_ivc(minerais)
md_final = "\n\n".join(segments)
# Remplacer la section du tableau final
md_final = re.sub(
r"## Tableau de synthèse\s*\n<!---- AUTO-BEGIN:TABLEAU-FINAL -->.*?<!---- AUTO-END:TABLEAU-FINAL -->",
f"## Tableau de synthèse\n<!---- AUTO-BEGIN:TABLEAU-FINAL -->\n{synth_table}\n<!---- AUTO-END:TABLEAU-FINAL -->",
md_final,
flags=re.S
)
return md_final
def build_assemblage_sections(md: str) -> str:
"""Traite les fiches d'assemblage/fabrication et génère le tableau des assembleurs/fabricants à partir du bloc YAML."""
# Extraire le schéma depuis l'en-tête YAML de la fiche
schema = None
front_match = re.match(r"(?s)^---\n(.*?)\n---\n", md)
if front_match:
try:
front_matter = yaml.safe_load(front_match.group(1))
schema = front_matter.get("schema")
type_fiche = front_matter.get("type_fiche")
# Vérifier si c'est bien une fiche d'assemblage ou de fabrication
if type_fiche not in ["assemblage", "fabrication"] or not schema:
return md
except Exception as e:
st.error(f"Erreur lors du chargement du front matter: {e}")
return md
# Rechercher le bloc YAML dans la fiche (entre ```yaml et ```)
yaml_block = re.search(r"```yaml\n(.+?)\n```", md, re.DOTALL)
if not yaml_block:
return md # Pas de bloc YAML trouvé, retourner la fiche inchangée
# Charger les données YAML
try:
yaml_data = yaml.safe_load(yaml_block.group(1))
except Exception as e:
st.error(f"Erreur lors du chargement du YAML: {e}")
return md
# Vérifier la structure du YAML (clé principale suivie des pays)
if not isinstance(yaml_data, dict) or len(yaml_data) == 0:
return md
# Récupérer la clé principale (nom du produit assemblé)
produit_key = list(yaml_data.keys())[0]
produit_data = yaml_data[produit_key]
# Structure pour stocker les données triées
pays_data = []
# Pour chaque pays, extraire ses données et celles de ses acteurs
for pays_key, pays_info in produit_data.items():
nom_pays = pays_info.get('nom_du_pays', '')
part_marche_pays = pays_info.get('part_de_marche', '0%')
# Convertir la part de marché en nombre pour le tri
part_marche_num = float(part_marche_pays.strip('%'))
acteurs = []
for acteur_key, acteur_info in pays_info.get('acteurs', {}).items():
nom_acteur = acteur_info.get('nom_de_l_acteur', '')
part_marche_acteur = acteur_info.get('part_de_marche', '0%')
pays_origine = acteur_info.get('pays_d_origine', '')
# Convertir la part de marché en nombre pour le tri
part_marche_acteur_num = float(part_marche_acteur.strip('%'))
acteurs.append({
'nom': nom_acteur,
'part_marche': part_marche_acteur,
'part_marche_num': part_marche_acteur_num,
'pays_origine': pays_origine
})
# Trier les acteurs par part de marché décroissante
acteurs_tries = sorted(acteurs, key=lambda x: x['part_marche_num'], reverse=True)
pays_data.append({
'nom': nom_pays,
'part_marche': part_marche_pays,
'part_marche_num': part_marche_num,
'acteurs': acteurs_tries
})
# Trier les pays par part de marché décroissante
pays_tries = sorted(pays_data, key=lambda x: x['part_marche_num'], reverse=True)
# Générer le tableau des assembleurs
lignes_tableau = [
"| **Pays d'implantation** | **Entreprise** | **Pays d'origine** | **Part de marché** |",
"| :-- | :-- | :-- | :-- |"
]
for pays in pays_tries:
for acteur in pays['acteurs']:
# Formater la part de marché (retirer le % du texte et ajouter un espace avant %)
part_marche_formattee = acteur['part_marche'].strip('%') + ' %'
lignes_tableau.append(
f"| {pays['nom']} | {acteur['nom']} | {acteur['pays_origine']} | {part_marche_formattee} |"
)
# Ajouter la ligne de total pour le pays (en gras)
part_marche_pays_formattee = pays['part_marche'].strip('%') + ' %'
lignes_tableau.append(
f"| **{pays['nom']}** | **Total** | **{pays['nom']}** | **{part_marche_pays_formattee}** |"
)
# Construire le tableau final
tableau_final = "\n".join(lignes_tableau)
# Remplacer la section du tableau dans la fiche (assembleurs ou fabricants selon le type)
if type_fiche == "fabrication":
md_modifie = re.sub(
r"<!---- AUTO-BEGIN:TABLEAU-FABRICANTS -->.*?<!---- AUTO-END:TABLEAU-FABRICANTS -->",
f"<!---- AUTO-BEGIN:TABLEAU-FABRICANTS -->\n{tableau_final}\n<!---- AUTO-END:TABLEAU-FABRICANTS -->",
md,
flags=re.DOTALL
)
else: # type_fiche == "assemblage"
md_modifie = re.sub(
r"<!---- AUTO-BEGIN:TABLEAU-ASSEMBLEURS -->.*?<!---- AUTO-END:TABLEAU-ASSEMBLEURS -->",
f"<!---- AUTO-BEGIN:TABLEAU-ASSEMBLEURS -->\n{tableau_final}\n<!---- AUTO-END:TABLEAU-ASSEMBLEURS -->",
md,
flags=re.DOTALL
)
# Chercher et remplacer la section IHH si un schéma a été identifié
if schema:
# Charger le contenu de la fiche technique IHH
try:
# Essayer de lire le fichier depuis le système de fichiers
ihh_path = "Fiches/Criticités/Fiche technique IHH.md"
with open(ihh_path, "r", encoding="utf-8") as f:
ihh_content = f.read()
# Chercher la section IHH correspondant au schéma et au type de fiche
# Format de la section : ## Assemblage/Fabrication - [Schema]
if type_fiche == "fabrication":
ihh_section_pattern = rf"## Fabrication - {schema}\s*\n### Indice de Herfindahl-Hirschmann[\s\S]*?(?=\n## |$)"
else: # type_fiche == "assemblage"
ihh_section_pattern = rf"## Assemblage - {schema}\s*\n### Indice de Herfindahl-Hirschmann[\s\S]*?(?=\n## |$)"
ihh_section_match = re.search(ihh_section_pattern, ihh_content)
if ihh_section_match:
# Extraire la section complète sans le titre principal
ihh_section = ihh_section_match.group(0).split("\n", 2)[2].strip()
# Remplacer la section IHH dans la fiche d'assemblage
md_modifie = re.sub(
r"<!---- AUTO-BEGIN:SECTION-IHH -->.*?<!---- AUTO-END:SECTION-IHH -->",
f"<!---- AUTO-BEGIN:SECTION-IHH -->\n{ihh_section}\n<!---- AUTO-END:SECTION-IHH -->",
md_modifie,
flags=re.DOTALL
)
else:
# Si aucune section IHH n'est trouvée pour ce schéma, laisser la section existante
st.warning(f"Aucune section IHH trouvée pour le schéma {schema} dans la fiche technique IHH.")
except Exception as e:
st.error(f"Erreur lors de la lecture/traitement de la fiche IHH: {e}")
return md_modifie
# Regex pour capturer les blocs YAML d'opération dans les fiches IHH
IHH_RE = re.compile(r"```yaml\s+opération:(.*?)```", re.S | re.I)
def _synth_ihh(operations: list[dict]) -> str:
"""Crée un tableau de synthèse pour les IHH."""
# Créer un dictionnaire pour regrouper les données par minerai/produit/composant
data_by_item = {}
for op in operations:
nom = op.get('nom', '')
item_id = op.get('minerai', op.get('produit', op.get('composant', '')))
if not item_id:
continue
# Initialiser l'entrée si elle n'existe pas encore
if item_id not in data_by_item:
data_by_item[item_id] = {
'type': 'minerai' if 'extraction' in op or 'reserves' in op or 'traitement' in op else
'produit' if 'assemblage' in op else 'composant',
'extraction_ihh_pays': '-',
'extraction_ihh_acteurs': '-',
'reserves_ihh_pays': '-',
'traitement_ihh_pays': '-',
'traitement_ihh_acteurs': '-',
'assemblage_ihh_pays': '-',
'assemblage_ihh_acteurs': '-',
'fabrication_ihh_pays': '-',
'fabrication_ihh_acteurs': '-'
}
# Mettre à jour les valeurs selon le type d'opération
if 'extraction' in op:
data_by_item[item_id]['extraction_ihh_pays'] = op['extraction'].get('ihh_pays', '-')
data_by_item[item_id]['extraction_ihh_acteurs'] = op['extraction'].get('ihh_acteurs', '-')
data_by_item[item_id]['reserves_ihh_pays'] = op['reserves'].get('ihh_pays', '-')
data_by_item[item_id]['traitement_ihh_pays'] = op['traitement'].get('ihh_pays', '-')
data_by_item[item_id]['traitement_ihh_acteurs'] = op['traitement'].get('ihh_acteurs', '-')
elif 'assemblage' in op:
data_by_item[item_id]['assemblage_ihh_pays'] = op['assemblage'].get('ihh_pays', '-')
data_by_item[item_id]['assemblage_ihh_acteurs'] = op['assemblage'].get('ihh_acteurs', '-')
elif 'fabrication' in op:
data_by_item[item_id]['fabrication_ihh_pays'] = op['fabrication'].get('ihh_pays', '-')
data_by_item[item_id]['fabrication_ihh_acteurs'] = op['fabrication'].get('ihh_acteurs', '-')
# Compléter avec les autres types si présents
result = []
# Tableau des produits
produits = {k: v for k, v in data_by_item.items() if v['type'] == 'produit'}
if produits:
result.append("\n\n## Assemblage des produits\n")
produit_lignes = [
"| Produit | Assemblage IHH Pays | Assemblage IHH Acteurs |",
"| :-- | :--: | :--: |"
]
for produit, data in sorted(produits.items()):
pastille_1 = pastille("IHH", data['assemblage_ihh_pays'])
pastille_2 = pastille("IHH", data['assemblage_ihh_acteurs'])
produit_lignes.append(
f"| {produit} | {pastille_1} {data['assemblage_ihh_pays']} | {pastille_2} {data['assemblage_ihh_acteurs']} |"
)
result.append("\n".join(produit_lignes))
# Tableau des composants
composants = {k: v for k, v in data_by_item.items() if v['type'] == 'composant'}
if composants:
result.append("\n\n## Fabrication des composants\n")
composant_lignes = [
"| Composant | Fabrication IHH Pays | Fabrication IHH Acteurs |",
"| :-- | :--: | :--: |"
]
for composant, data in sorted(composants.items()):
pastille_1 = pastille("IHH", data['fabrication_ihh_pays'])
pastille_2 = pastille("IHH", data['fabrication_ihh_acteurs'])
composant_lignes.append(
f"| {composant} | {pastille_1} {data['fabrication_ihh_pays']} | {pastille_2} {data['fabrication_ihh_acteurs']} |"
)
result.append("\n".join(composant_lignes))
# Trier et créer le tableau de minerais (celui demandé)
minerais = {k: v for k, v in data_by_item.items() if v['type'] == 'minerai'}
if minerais:
result.append("\n\n## Opérations sur les minerais\n")
minerai_lignes = [
"| Minerai | Extraction IHH Pays | Extraction IHH Acteurs | Réserves IHH Pays | Traitement IHH Pays | Traitement IHH Acteurs |",
"| :-- | :--: | :--: | :--: | :--: | :--: |"
]
for minerai, data in sorted(minerais.items()):
pastille_1 = pastille("IHH", data['extraction_ihh_pays'])
pastille_2 = pastille("IHH", data['extraction_ihh_acteurs'])
pastille_3 = pastille("IHH", data['reserves_ihh_pays'])
pastille_4 = pastille("IHH", data['traitement_ihh_pays'])
pastille_5 = pastille("IHH", data['traitement_ihh_acteurs'])
minerai_lignes.append(
f"| {minerai} | {pastille_1} {data['extraction_ihh_pays']} | {pastille_2} {data['extraction_ihh_acteurs']} | "
f"{pastille_3} {data['reserves_ihh_pays']} | {pastille_4} {data['traitement_ihh_pays']} | {pastille_5} {data['traitement_ihh_acteurs']} |"
)
result.append("\n".join(minerai_lignes))
return "\n".join(result)
def build_ihh_sections(md: str) -> str:
"""Traite les fiches IHH pour les opérations, produits, composants et minerais."""
segments = []
operations = [] # Pour collecter les données de chaque opération
intro = None
matches = list(IHH_RE.finditer(md))
if matches:
first = matches[0]
intro = md[:first.start()].strip()
else:
return md # pas de blocs à traiter
# Traiter chaque bloc YAML et sa section correspondante
for m in matches:
bloc_text = m.group(1)
bloc = yaml.safe_load("opération:" + bloc_text)
operations.append(bloc["opération"]) # Collecte les données
start = m.end()
next_match = IHH_RE.search(md, start)
end = next_match.start() if next_match else len(md)
# Utiliser Jinja2 pour le rendu de la section
section_template = md[start:end].strip()
rendered = Template(section_template).render(**bloc["opération"])
segments.append(rendered)
if intro:
segments.insert(0, intro)
# Créer et insérer le tableau de synthèse si nécessaire
if "# Tableaux de synthèse" in md:
synth_table = _synth_ihh(operations)
md_final = "\n\n".join(segments)
# Remplacer la section du tableau final
md_final = re.sub(
r"(?:##?|#) Tableaux de synthèse\s*\n<!---- AUTO-BEGIN:TABLEAU-FINAL -->.*?<!---- AUTO-END:TABLEAU-FINAL -->",
f"# Tableaux de synthèse\n<!---- AUTO-BEGIN:TABLEAU-FINAL -->\n{synth_table}\n<!---- AUTO-END:TABLEAU-FINAL -->",
md_final,
flags=re.S
)
else:
md_final = "\n\n".join(segments)
return md_final
def _synth_isg(md: str) -> str:
"""Crée un tableau de synthèse pour les ISG à partir du bloc YAML des pays."""
# Regex pour trouver le bloc YAML des pays
yaml_block = re.search(r"```yaml\n(.+?)\n```", md, re.DOTALL)
if not yaml_block:
return "*(aucune donnée de pays trouvée)*"
# Charger les données YAML
yaml_data = yaml.safe_load(yaml_block.group(1))
# Préparer les lignes du tableau
lignes = ["| Pays | WGI | FSI | NDGAIN | ISG |", "| :-- | :-- | :-- | :-- | :-- |"]
# Trier les pays par nom
sorted_pays = sorted(yaml_data.items(), key=lambda x: x[1]['pays'].lower())
# Ajouter chaque pays au tableau
for identifiant, data in sorted_pays:
pays = data['pays']
wgi_ps = data['wgi_ps']
fsi = data['fsi']
ndgain = data['ndgain']
isg = data['isg']
# Ajouter pastilles en fonction des seuils
pastille_isg = pastille("ISG", isg) if 'ISG' in st.session_state.get('seuils', {}) else ""
lignes.append(f"| {pays} | {wgi_ps} | {fsi} | {ndgain} | {pastille_isg} {isg} |")
return "\n".join(lignes)
def build_isg_sections(md: str) -> str:
"""Traite les fiches ISG pour générer le tableau de synthèse des pays."""
# Vérifier si c'est une fiche ISG (via indice_court dans les frontmatter)
front_match = re.match(r"(?s)^---\n(.*?)\n---\n", md)
if front_match:
front_matter = yaml.safe_load(front_match.group(1))
if front_matter.get("indice_court") != "ISG":
return md # Ce n'est pas une fiche ISG
# Générer le tableau de synthèse
synth_table = _synth_isg(md)
# Remplacer la section du tableau final
md_final = re.sub(
r"## Tableau de synthèse\s*\n<!---- AUTO-BEGIN:TABLEAU-FINAL -->.*?<!---- AUTO-END:TABLEAU-FINAL -->",
f"## Tableau de synthèse\n<!---- AUTO-BEGIN:TABLEAU-FINAL -->\n{synth_table}\n<!---- AUTO-END:TABLEAU-FINAL -->",
md,
flags=re.S
)
# Supprimer le bloc YAML des pays (entre # Criticité par pays et le prochain titre ##)
md_final = re.sub(
r"# Criticité par pays\s*\n```yaml[\s\S]*?```\s*",
"# Criticité par pays\n\n",
md_final,
flags=re.S
)
return md_final