333 lines
12 KiB
Python
333 lines
12 KiB
Python
import re, yaml, pandas as pd, textwrap
|
|
import unicodedata
|
|
from jinja2 import Template
|
|
import streamlit as st
|
|
|
|
# -------- 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
|
|
|
|
|
|
# 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 = []
|
|
|
|
def pastille(indice, valeur):
|
|
if not valeur:
|
|
return ""
|
|
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
|
|
|
|
# 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
|