import streamlit as st import json import os from datetime import date from utils.translations import _ from dotenv import load_dotenv from pathlib import Path load_dotenv(".env") def initialise(): SAVE_SESSIONS_PATH.mkdir(parents=True, exist_ok=True) def get_session_id(): if "session_id" not in st.session_state: session_id = st.context.headers.get("x-session-id", "anonymous") st.session_state["session_id"] = session_id else: session_id = st.session_state["session_id"] return session_id SAVE_STATUT = os.getenv("SAVE_STATUT", "statut_general.json") session_id = get_session_id() SAVE_SESSIONS_PATH = Path(f"tmp/sessions/{session_id}") SAVE_STATUT_PATH = SAVE_SESSIONS_PATH / SAVE_STATUT initialise() def _maj_champ(fichier, cle: str, contenu: str = "") -> bool: def serialize(obj): return obj.isoformat() if isinstance(obj, date) else obj def inserer_cle_json(structure: dict, cle: str, valeur: any) -> dict: """ Insère une clé de type 'a.b.c' dans un dictionnaire JSON imbriqué. Args: structure: Dictionnaire racine à mettre à jour cle: Chaîne de clé séparée par des points valeur: Valeur à insérer Returns: Le dictionnaire mis à jour """ parties = cle.split(".") d = structure for p in parties[:-1]: d = d.setdefault(p, {}) d[parties[-1]] = valeur return structure if fichier.exists(): try: with open(fichier, "r", encoding="utf-8") as f: sauvegarde = json.load(f) sauvegarde = inserer_cle_json(sauvegarde, cle, serialize(contenu)) except Exception as e: st.error(_("persistance.errors.read_file").format(function="_maj_champ", file=fichier, error=e)) return False else: sauvegarde = {} sauvegarde = inserer_cle_json(sauvegarde, cle, serialize(contenu)) try: with open(fichier, "w", encoding="utf-8") as f: json.dump(sauvegarde, f, ensure_ascii=False, indent=2) return True except Exception as e: st.error(_("persistance.errors.write_file").format(function="_maj_champ", file=fichier, error=e)) return False def _get_champ(fichier, cle: str) -> str: def extraire_valeur_par_cle(structure: dict, cle: str): """ Extrait une valeur depuis un dictionnaire imbriqué avec une clé au format 'a.b.c'. Args: structure: Dictionnaire d'origine cle: Chaîne représentant la clé imbriquée Returns: La valeur trouvée, ou `par_defaut` si elle est introuvable """ parties = cle.split(".") d = structure for p in parties: if isinstance(d, dict) and p in d: d = d[p] else: return "" return d import json def charger_json_sain(fichier: str) -> dict: with open(fichier, "r", encoding="utf-8") as f: contenu = json.load(f) if isinstance(contenu, str): try: contenu = json.loads(contenu) # On essaie de parser une 2e fois si nécessaire except json.JSONDecodeError: raise ValueError("Le fichier contient une chaîne JSON invalide.") if not isinstance(contenu, dict): raise ValueError("Le contenu JSON n'est pas un objet/dictionnaire valide.") return contenu if fichier.exists(): try: sauvegarde = charger_json_sain(fichier) return extraire_valeur_par_cle(sauvegarde, cle) except Exception as e: st.error(_("persistance.errors.read_file").format(function="_get_champ", file=fichier, error=e)) return "" def _supprime_champ(fichier: Path, cle: str) -> bool: def supprimer_cle_profonde(d: dict, chemin: str, separateur="."): cles = chemin.split(separateur) sous_dict = d for cle in cles[:-1]: sous_dict = sous_dict.get(cle, {}) if not isinstance(sous_dict, dict): return False # Le chemin est invalide return sous_dict.pop(cles[-1], None) is not None if fichier.exists(): try: with open(fichier, "r", encoding="utf-8") as f: sauvegarde = json.load(f) except Exception as e: st.error(_("persistance.errors.read_file").format(function="_maj_champ", file=fichier, error=e)) return False supprimer_cle_profonde(sauvegarde, cle) try: with open(fichier, "w", encoding="utf-8") as f: json.dump(sauvegarde, f, indent=4) except Exception as e: st.error(_("persistance.errors.write_file").format(function="_supprime_champ", file=fichier, error=e)) return False return True def maj_champ_statut(cle: str, contenu: str = "") -> bool: return _maj_champ(SAVE_STATUT_PATH, cle, contenu) def get_champ_statut(cle: str) -> str: return _get_champ(SAVE_STATUT_PATH, cle) def supprime_champ_statut(cle: str) -> None: _supprime_champ(SAVE_STATUT_PATH, cle) def get_full_structure() -> dict|None: fichier = SAVE_STATUT_PATH if fichier.exists(): try: with open(fichier, "r", encoding="utf-8") as f: sauvegarde = json.load(f) return sauvegarde except Exception as e: st.error(_("persistance.errors.read_file").format(function="_maj_champ", file=fichier, error=e)) return None else: return None