import os import glob import re import csv import pdfplumber import unicodedata # --- CONFIGURATION DES DOSSIERS --- BASE_DIR = os.path.dirname(os.path.abspath(__file__)) PDF_SLEEP_DIR = os.path.join(BASE_DIR, "PDF", "SLEEP") PDF_SPIRO_DIR = os.path.join(BASE_DIR, "PDF", "SPIRO") CSV_SLEEP_DIR = os.path.join(BASE_DIR, "CSV", "SLEEP") CSV_SPIRO_DIR = os.path.join(BASE_DIR, "CSV", "SPIRO") # Création des dossiers s'ils n'existent pas for d in [PDF_SLEEP_DIR, PDF_SPIRO_DIR, CSV_SLEEP_DIR, CSV_SPIRO_DIR]: os.makedirs(d, exist_ok=True) def clean_filename_part(part): """Nettoie une chaîne pour l'utiliser dans un nom de fichier (retire accents, majuscules, etc).""" if not part: return "" # Normalisation pour retirer les accents (É -> E) nfkd_form = unicodedata.normalize('NFKD', str(part)) part_ascii = "".join([c for c in nfkd_form if not unicodedata.combining(c)]) # On ne garde que les lettres, chiffres et tirets, puis on met en majuscules return re.sub(r'[^\w\-]', '', part_ascii).strip().upper() PARAMETRES_CLES = ["CVF", "VEMS", "VEMS/", "DEP", "DEM2575", "DEM75%", "DEM50%", "DEM25%", "TempsDE", "VEXT", "VEM6"] def point_vers_virgule(texte): if not isinstance(texte, str): return texte test_texte = texte.lstrip('-') if test_texte.replace('.', '', 1).isdigit(): return texte.replace('.', ',') return texte def extract_spirometry_data(pdf_path): """Extrait les infos patient et les paramètres médicaux d'un PDF de spirométrie MIR.""" infos_patient = {} lignes_donnees = [] nb_mesures_detectees = 0 try: with pdfplumber.open(pdf_path) as pdf: page = pdf.pages[0] texte = page.extract_text() if not texte: print(" ⚠️ Aucun texte extrait de la page.") return None texte_global = " ".join(texte.split()) # --- Infos patient --- m = re.search(r"Nom\s+([\w-]+)", texte_global) infos_patient["Nom"] = m.group(1) if m else "Inconnu" m = re.search(r"Prénom\s+([\w-]+)", texte_global) infos_patient["Prénom"] = m.group(1) if m else "Inconnu" m = re.search(r"(\d+)\s*kg", texte_global) infos_patient["Poids"] = m.group(1) if m else "Non trouvé" m = re.search(r"IMC\s+([\d\.]+)", texte_global) infos_patient["IMC"] = point_vers_virgule(m.group(1)) if m else "Non trouvé" m = re.search(r"VISITE\s+(\d{2}/\d{2}/\d{4})", texte_global) infos_patient["Date Visite"] = m.group(1) if m else "Non trouvée" m = re.search(r"naissance\s+(\d{2}/\d{2}/\d{4})", texte_global) infos_patient["Date de Naissance"] = m.group(1) if m else "Non trouvée" m = re.search(r"Taille\s+(\d+\s?cm)", texte_global) infos_patient["Taille"] = m.group(1) if m else "Non trouvée" # --- Paramètres médicaux --- for ligne in texte.split('\n'): ligne = ligne.strip() for param in PARAMETRES_CLES: if ligne.startswith(param): elements = ligne.split() if len(elements) > 3: parametre = elements[0] unite = elements[1] valeurs = [point_vers_virgule(v) for v in elements[2:]] nb_mesures_actuel = len(valeurs) - 5 if nb_mesures_actuel > nb_mesures_detectees: nb_mesures_detectees = nb_mesures_actuel lignes_donnees.append([parametre, unite] + valeurs) break return infos_patient, lignes_donnees, nb_mesures_detectees except Exception as e: print(f" ❌ Erreur extraction spirométrie : {e}") return None def save_spirometry_csv(infos_patient, lignes_donnees, nb_mesures_detectees, path): try: with open(path, mode='w', newline='', encoding='utf-8-sig') as f: writer = csv.writer(f, delimiter=';') writer.writerow(["--- INFORMATIONS PATIENT ---"]) for cle in ["Nom", "Prénom", "Date de Naissance", "Date Visite", "Taille", "Poids", "IMC"]: writer.writerow([cle, infos_patient.get(cle, "")]) writer.writerow([]) writer.writerow(["--- RÉSULTATS MÉDICAUX ---"]) en_tete = ["Paramètre", "Unité", "Meilleur", "Théorique", "LLN", "%Théor.", "Z-score"] for i in range(1, nb_mesures_detectees + 1): en_tete.append(f"PRE#{i}") writer.writerow(en_tete) writer.writerows(lignes_donnees) print(f" ✅ Succès ! Spirométrie générée : {path}") return True except Exception as e: print(f" ❌ Erreur écriture CSV spirométrie : {e}") return False # --- NOUVELLES FONCTIONS POUR LE SOMMEIL (Sleep Analyzer) --- def extract_all_sleep_data(pdf_path): data = { "infos_patient": { "Nom": "Inconnu", "Prénom": "Inconnu", "Âge": "N/A", "Période": "N/A" }, "resume_global": {}, "routine": {}, "agenda": [] } try: with pdfplumber.open(pdf_path) as pdf: # === PAGE 1 : Infos patient === # Structure: "Vue d'ensemble · 28 Févr.-28 Mars 2026\nThomas Ozifkf\n23 ans" p1_lines = [l.strip() for l in (pdf.pages[0].extract_text() or '').split('\n') if l.strip()] for i, line in enumerate(p1_lines): if "Vue d'ensemble" in line and "·" in line: data["infos_patient"]["Période"] = line.split("·", 1)[1].strip() if i + 1 < len(p1_lines): name_parts = p1_lines[i + 1].split() if len(name_parts) >= 2: data["infos_patient"]["Prénom"] = name_parts[0] data["infos_patient"]["Nom"] = " ".join(name_parts[1:]) elif len(name_parts) == 1: data["infos_patient"]["Nom"] = name_parts[0] if i + 2 < len(p1_lines) and "ans" in p1_lines[i + 2]: data["infos_patient"]["Âge"] = p1_lines[i + 2] break # === PAGE 4 : Données de sommeil (index 3) === p4_text = pdf.pages[3].extract_text() or "" # Durée moyenne ex: "6h46\nDurée moyenne du sommeil" m = re.search(r'(\d+h\d+)\nDurée moyenne du sommeil', p4_text) data["resume_global"]["Durée moyenne"] = m.group(1) if m else "N/A" # Temps d'endormissement ex: "18 min\nTemps d.endormissement" m = re.search(r'(\d+)\s*min\nTemps d.endormissement', p4_text) data["resume_global"]["Temps d'endormissement"] = (m.group(1) + " min") if m else "N/A" # TIB, TST, Efficacité sur une ligne ex: "7h18 6h46 93 %" m = re.search(r'(\d+h\d+)\s+(\d+h\d+)\s+(\d+\s*%)', p4_text) data["resume_global"]["Temps au lit (TIB)"] = m.group(1) if m else "N/A" data["resume_global"]["Temps de sommeil (TST)"] = m.group(2) if m else "N/A" data["resume_global"]["Efficacité"] = m.group(3) if m else "N/A" # Heure de coucher: deux heures sur deux lignes "23:50\n23:32\nHeure de coucher" # La première = week-end, la deuxième = moyenne m = re.search(r'(\d+:\d+)\n(\d+:\d+)\nHeure de coucher', p4_text) data["routine"]["Coucher moyen"] = m.group(2) if m else "N/A" # Heure de lever: "6:50 6:50\nHeure de réveil" m = re.search(r'(\d+:\d+)\s+(\d+:\d+)\nHeure de réveil', p4_text) data["routine"]["Lever moyen"] = m.group(1) if m else "N/A" # IAH: 3 valeurs (min, moy, max) - on prend la moyenne (2ème) m = re.search(r'(\d+)\s*événement/heure\s+(\d+)\s*événement/heure\s+(\d+)\s*événement/heure', p4_text) data["routine"]["IAH Moyen"] = (m.group(2) + " év/h") if m else "0 év/h" # Ronflements: "0 min\nRonflements" m = re.search(r'(\d+)\s*min\nRonflements', p4_text) data["routine"]["Ronflements"] = (m.group(1) + " min") if m else "N/A" # Fréquence cardiaque nuit: "61 bpm ... \nFréquence cardiaque de nuit" m = re.search(r'(\d+)\s*bpm.+\nFréquence cardiaque de nuit', p4_text) data["routine"]["Fréq. Cardiaque Nuit"] = (m.group(1) + " bpm") if m else "N/A" # === PAGE 5 : Agenda (index 4) === # Chaque ligne de donnée ressemble à: "27-28 6h46 7h18 93 % 0 0 0 61" # La ligne suivante contient le mois: "Mar Endormi Au lit Efficacité Min bpm" p5_lines = [l.strip() for l in (pdf.pages[4].extract_text() or '').split('\n') if l.strip()] for i, line in enumerate(p5_lines): row_m = re.match(r'^(\d+-\d+)\s+(\d+h\d+)\s+(\d+h\d+)\s+(\d+\s*%)\s+(\d+)\s+(\d+)\s+(\d+)\s+(\d+)', line) if row_m: month = "" if i + 1 < len(p5_lines): month_m = re.match(r'^([A-Za-zÀ-ÿ]+)', p5_lines[i + 1]) if month_m: month = month_m.group(1) date_str = f"{row_m.group(1)} {month}".strip() # Ordre dans le PDF: date | TST(Endormi) | TIB(Au lit) | Eff | Sorties | IAH | Ronf | FC data["agenda"].append([ date_str, row_m.group(3), # TIB (Au lit) row_m.group(2), # TST (Endormi) row_m.group(4).strip(), # Efficacité row_m.group(5), # Sorties du lit row_m.group(6), # IAH row_m.group(7) + " min", # Ronflements row_m.group(8) + " bpm" # FC ]) return data except Exception as e: print(f"Erreur lors de l'extraction : {e}") return None def save_to_csv_sleep(data, path): try: with open(path, mode='w', newline='', encoding='utf-8-sig') as f: writer = csv.writer(f, delimiter=';') writer.writerow(["INFORMATIONS PATIENT"]) for k, v in data["infos_patient"].items(): writer.writerow([k, v]) writer.writerow([]) writer.writerow(["RÉSUMÉ GLOBAL SOMMEIL"]) for k, v in data["resume_global"].items(): writer.writerow([k, v]) writer.writerow([]) writer.writerow(["ROUTINE ET MESURES"]) for k, v in data["routine"].items(): writer.writerow([k, v]) writer.writerow([]) writer.writerow(["AGENDA DÉTAILLÉ"]) writer.writerow(["Date", "Temps au lit (TIB)", "Temps endormi (TST)", "Efficacité", "Sorties du lit", "IAH", "Ronflements", "FC moy."]) writer.writerows(data["agenda"]) print(f"✅ Succès ! Fichier de sommeil généré ici : {path}") except PermissionError: print("⚠️ Le fichier CSV est verrouillé (souvent à cause de l'aperçu Windows Explorer ou d'un processus en tâche de fond).") import time new_path = path.replace(".csv", f"_{int(time.time())}.csv") print(f"🔁 Tentative de sauvegarde sous un autre nom : {new_path}...") try: with open(new_path, mode='w', newline='', encoding='utf-8-sig') as f: writer = csv.writer(f, delimiter=';') writer.writerow(["INFORMATIONS PATIENT"]) for k, v in data["infos_patient"].items(): writer.writerow([k, v]) writer.writerow([]) writer.writerow(["RÉSUMÉ GLOBAL SOMMEIL"]) for k, v in data["resume_global"].items(): writer.writerow([k, v]) writer.writerow([]) writer.writerow(["ROUTINE ET MESURES"]) for k, v in data["routine"].items(): writer.writerow([k, v]) writer.writerow([]) writer.writerow(["AGENDA DÉTAILLÉ"]) writer.writerow(["Date", "Temps au lit (TIB)", "Temps endormi (TST)", "Efficacité", "Sorties du lit", "IAH", "Ronflements", "FC moy."]) writer.writerows(data["agenda"]) print(f"✅ Succès ! Fichier généré sous le nom : {new_path}") except Exception as e: print(f"❌ Échec de la sauvegarde alternative : {e}") except Exception as e: print(f"❌ Une erreur est survenue : {e}") def main(): any_file = False # ========================================================= # 1. TRAITEMENT DES RAPPORTS SOMMEIL (PDF/SLEEP/) # ========================================================= sleep_files = glob.glob(os.path.join(PDF_SLEEP_DIR, "*.pdf")) for pdf_path in sleep_files: any_file = True base_name = os.path.splitext(os.path.basename(pdf_path))[0] print(f"\n⚙️ [SLEEP] Traitement de : {base_name}.pdf...") resultats = extract_all_sleep_data(pdf_path) if not resultats: print(f"❌ Échec de l'extraction pour {base_name}.pdf") continue nom = clean_filename_part(resultats["infos_patient"].get("Nom", "NOMINCONNU")) prenom = clean_filename_part(resultats["infos_patient"].get("Prénom", "PRENOMINCONNU")) periode = clean_filename_part(resultats["infos_patient"].get("Période", base_name)) final_csv_name = f"{nom}_{prenom}_{periode}.csv" final_csv_path = os.path.join(CSV_SLEEP_DIR, final_csv_name) if os.path.exists(final_csv_path): print(f"⏩ Ignoré (déjà traité) : {final_csv_name}") continue save_to_csv_sleep(resultats, final_csv_path) # ========================================================= # 2. TRAITEMENT DES SPIROMÉTRIES (PDF/SPIRO/) # ========================================================= spiro_files = glob.glob(os.path.join(PDF_SPIRO_DIR, "*.pdf")) for pdf_path in spiro_files: any_file = True base_name = os.path.splitext(os.path.basename(pdf_path))[0] print(f"\n⚙️ [SPIRO] Traitement de : {base_name}.pdf...") result = extract_spirometry_data(pdf_path) if not result: print(f" ❌ Échec de l'extraction pour {base_name}.pdf") continue infos_patient, lignes_donnees, nb_mesures = result nom_sp = clean_filename_part(infos_patient.get("Nom", "NOMINCONNU")) prenom_sp = clean_filename_part(infos_patient.get("Prénom", "PRENOMINCONNU")) date_sp = clean_filename_part(infos_patient.get("Date Visite", "DATEINCONNUE").replace('/', '-')) final_csv_name = f"{nom_sp}_{prenom_sp}_{date_sp}_Spirometry.csv" final_csv_path = os.path.join(CSV_SPIRO_DIR, final_csv_name) if os.path.exists(final_csv_path): print(f" ⏩ Ignoré (déjà traité) : {final_csv_name}") continue save_spirometry_csv(infos_patient, lignes_donnees, nb_mesures, final_csv_path) if not any_file: print("ℹ️ Aucun PDF trouvé dans PDF/SLEEP/ ni dans PDF/SPIRO/") if __name__ == "__main__": main()