feat: pipeline T2A - anonymisation, extraction CIM-10 et intégration edsnlp

Pipeline complet de traitement de documents médicaux PDF :
- Extraction texte (pdfplumber) et classification (Trackare/CRH)
- Anonymisation multi-couche (regex + NER CamemBERT + sweep)
- Extraction médicale CIM-10 hybride : edsnlp (AP-HP) enrichit les
  diagnostics, médicaments (codes ATC via Romedi) et négation,
  avec fallback regex pour les patterns spécifiques
- Fix sentencepiece pinné à <0.2.0 pour compatibilité CamemBERT

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
dom
2026-02-10 15:24:12 +01:00
commit 4a12cd2676
25 changed files with 7592 additions and 0 deletions

View File

View File

@@ -0,0 +1,129 @@
"""Parsing des Comptes Rendus d'Hospitalisation (CRH)."""
from __future__ import annotations
import re
def parse_crh(text: str) -> dict:
"""Parse un CRH et retourne les sections structurées."""
result: dict = {
"type": "crh",
"patient": {},
"sejour": {},
"medecins": [],
"contenu_medical": "",
"sections": {},
}
_extract_patient_info(text, result)
_extract_sejour_info(text, result)
_extract_medecins(text, result)
_extract_medical_content(text, result)
return result
def _extract_patient_info(text: str, result: dict) -> None:
"""Extrait les informations patient du CRH."""
# "MME NARBAIS AUDREY" ou "M. NOM PRENOM"
m = re.search(
r"(?:MME|M\.|MR)\s+([A-ZÉÈÊËÀÂÄÙÛÜÔÖÎÏÇ][A-ZÉÈÊËÀÂÄÙÛÜÔÖÎÏÇa-zéèêëàâäùûüôöîïç\- ]+)",
text[:2000],
)
if m:
result["patient"]["nom_complet"] = m.group(1).strip()
# Adresse sous le nom patient — capturer les lignes entre le nom et le CP+Ville
addr_match = re.search(
r"(?:MME|M\.|MR|Madame|Monsieur)\s+[A-ZÉÈÊËÀÂ][A-ZÉÈÊËÀÂa-zéèêëàâ\s\-]+\n((?:.*\n){1,4}?\d{5}\s+[A-Z][A-Z\s\-]+)",
text[:3000],
)
if addr_match:
result["patient"]["adresse"] = addr_match.group(1).strip()
# "née le DD/MM/YYYY" ou "né le DD/MM/YYYY"
m = re.search(r"n[ée]+\s+le\s+(\d{2}/\d{2}/\d{4})", text)
if m:
result["patient"]["date_naissance"] = m.group(1)
# Sexe depuis le titre
if re.search(r"\bMME\b", text[:2000]):
result["patient"]["sexe"] = "F"
elif re.search(r"\b(?:M\.|MR)\b", text[:2000]):
result["patient"]["sexe"] = "M"
# "Votre patiente" / "Votre patient"
if "patiente" in text[:3000].lower():
result["patient"]["sexe"] = "F"
elif "patient" in text[:3000].lower():
result["patient"].setdefault("sexe", "M")
def _extract_sejour_info(text: str, result: dict) -> None:
"""Extrait les dates et motif de séjour."""
# "du DD/MM/YYYY au DD/MM/YYYY"
m = re.search(
r"du\s+(\d{2}/\d{2}/\d{4})\s+au\s+(\d{2}/\d{2}/\d{4})", text
)
if m:
result["sejour"]["date_entree"] = m.group(1)
result["sejour"]["date_sortie"] = m.group(2)
# "pour le motif suivant:" ou "pour le motif suivant :\n..."
m = re.search(
r"pour\s+le\s+motif\s+suivant\s*[:\s]*\n?(.*?)(?:\n\n|\.\s+[A-Z])",
text,
re.DOTALL,
)
if m:
result["sejour"]["motif"] = m.group(1).strip()
def _extract_medecins(text: str, result: dict) -> None:
"""Extrait les noms de médecins mentionnés."""
# "Dr NOM" ou "DR NOM" ou "Dr. NOM" ou "Docteur NOM" ou "Dr F. NOM"
for m in re.finditer(
r"(?:Dr\.?|DR\.?|Docteur)\s+(?:[A-Z]\.\s+)?([A-ZÉÈÊËÀÂ][A-ZÉÈÊËÀÂa-zéèêëàâ\-]+(?:\s+[A-ZÉÈÊËÀÂ][A-ZÉÈÊËÀÂa-zéèêëàâ\-]+)?)",
text,
):
name = m.group(1).strip()
if name not in result["medecins"] and len(name) > 2:
result["medecins"].append(name)
def _extract_medical_content(text: str, result: dict) -> None:
"""Extrait le contenu médical principal."""
# Chercher après "Mon cher confrère," et les infos d'hospitalisation
m = re.search(
r"(?:motif\s+suivant\s*[:\s]*\n?)(.*?)(?:Rédigé par|Cordialement|Confraternellement|Dr\s+\w+\s*$)",
text,
re.DOTALL,
)
if m:
result["contenu_medical"] = m.group(1).strip()
else:
# Fallback : prendre tout après "Mon cher confrère"
m = re.search(
r"Mon cher confrère,?\s*\n(.*?)(?:Rédigé par|$)",
text,
re.DOTALL,
)
if m:
result["contenu_medical"] = m.group(1).strip()
# Sections spécifiques
section_patterns = [
("motif_hospitalisation", r"(?:motif\s+(?:d'hospitalisation|suivant))\s*[:\s]*\n?(.*?)(?=\n\s*(?:Antécédents|Histoire|Examen|Au total|Devenir|TTT)|$)"),
("antecedents", r"(?:Antécédents?)\s*[:\s]*\n?(.*?)(?=\n\s*(?:Histoire|Examen|Traitement|Au total|Devenir)|$)"),
("histoire_maladie", r"(?:Histoire de la maladie)\s*[:\s]*\n?(.*?)(?=\n\s*(?:Examen|Biologie|Au total|Devenir)|$)"),
("examen_clinique", r"(?:Examen clinique)\s*[:\s]*\n?(.*?)(?=\n\s*(?:Biologie|Imagerie|Au total|Devenir)|$)"),
("conclusion", r"(?:Au total|Conclusion)\s*[:\s]*\n?(.*?)(?=\n\s*(?:Devenir|TTT|Traitement)|$)"),
("traitement_sortie", r"(?:TTT de sortie|Traitement de sortie)\s*[:\s]*\n?(.*?)(?=\n\s*(?:Devenir|Rédigé|Cordialement)|$)"),
("devenir", r"(?:Devenir)\s*[:\s]*\n?(.*?)(?=\n\s*(?:TTT|Traitement|Rédigé|Cordialement)|$)"),
]
for key, pattern in section_patterns:
m = re.search(pattern, text, re.DOTALL | re.IGNORECASE)
if m:
result["sections"][key] = m.group(1).strip()

View File

@@ -0,0 +1,45 @@
"""Détection du type de document : CRH vs Trackare."""
from __future__ import annotations
def classify(text: str) -> str:
"""Classifie un document extrait en CRH ou Trackare.
Retourne "crh" ou "trackare".
"""
text_lower = text[:3000].lower()
trackare_markers = [
"dossier patient",
"détails des patients",
"détails épisode",
"liste des contacts",
"notes paramédicales",
"signes vitaux",
"traitements médicamenteux",
"observations médicales",
]
trackare_score = sum(1 for m in trackare_markers if m in text_lower)
crh_markers = [
"mon cher confrère",
"cher confrère",
"compte rendu d'hospitalisation",
"compte-rendu",
"service de gastro",
"pôle spécialités",
"votre patient",
]
crh_score = sum(1 for m in crh_markers if m in text_lower)
if trackare_score >= 2:
return "trackare"
if crh_score >= 2:
return "crh"
# Heuristique : Trackare contient des tableaux avec IPP
if "ipp:" in text_lower or "episode no:" in text_lower:
return "trackare"
return "crh"

View File

@@ -0,0 +1,36 @@
"""Extraction de texte et tableaux depuis les PDF via pdfplumber."""
from __future__ import annotations
from pathlib import Path
import pdfplumber
def extract_text(pdf_path: str | Path) -> str:
"""Extrait le texte de toutes les pages d'un PDF."""
pages_text: list[str] = []
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
text = page.extract_text() or ""
pages_text.append(text)
return "\n\n".join(pages_text)
def extract_pages(pdf_path: str | Path) -> list[str]:
"""Extrait le texte page par page."""
pages: list[str] = []
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
pages.append(page.extract_text() or "")
return pages
def extract_tables(pdf_path: str | Path) -> list[list[list[str | None]]]:
"""Extrait tous les tableaux détectés dans le PDF."""
all_tables: list[list[list[str | None]]] = []
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
tables = page.extract_tables() or []
all_tables.extend(tables)
return all_tables

View File

@@ -0,0 +1,419 @@
"""Parsing des exports Trackare (dossier patient complet)."""
from __future__ import annotations
import re
def parse_trackare(text: str) -> dict:
"""Parse un export Trackare et retourne les sections structurées."""
result: dict = {
"type": "trackare",
"patient": {},
"sejour": {},
"contacts": [],
"medecins": [],
"urgences": {},
"observations_medicales": [],
"notes_paramedicales": [],
"signes_vitaux": {},
"diagnostics": [],
"traitements": [],
"contenu_medical": "",
}
_extract_patient_info(text, result)
_extract_sejour_info(text, result)
_extract_contacts(text, result)
_extract_medecins(text, result)
_extract_urgences(text, result)
_extract_observations(text, result)
_extract_notes_param(text, result)
_extract_diagnostics(text, result)
_extract_traitements(text, result)
_extract_vitals(text, result)
_build_medical_content(result)
return result
def _extract_patient_info(text: str, result: dict) -> None:
"""Extrait les informations du bloc 'Détails des patients'."""
# Nom de naissance
m = re.search(r"Nom de naissance:\s*(\S+)", text)
if m:
result["patient"]["nom_naissance"] = m.group(1).strip()
# Nom et Prénom
m = re.search(r"Nom et Prénom:\s*(.+?)(?:\s+Date de naissance|\n)", text)
if m:
result["patient"]["nom_prenom"] = m.group(1).strip()
# IPP
m = re.search(r"IPP:\s*(\d+)", text)
if m:
result["patient"]["ipp"] = m.group(1)
# Date de naissance
m = re.search(r"Date de naissance:\s*(\d{2}/\d{2}/\d{4})", text)
if m:
result["patient"]["date_naissance"] = m.group(1)
# Sexe
m = re.search(r"Sexe:\s*(\S+)", text)
if m:
sexe_raw = m.group(1).strip().lower()
result["patient"]["sexe"] = "F" if "fém" in sexe_raw else "M"
# Lieu de naissance
m = re.search(r"Lieu de naissance:\s*(.+?)(?:\n|$)", text)
if m:
result["patient"]["lieu_naissance"] = m.group(1).strip()
# Adresse
m = re.search(r"Adresse:\s*(.+?)(?:\s+Ville de résidence|\n)", text)
if m:
result["patient"]["adresse"] = m.group(1).strip()
# Code postal et ville
m = re.search(r"Code Postal:\s*(\d{5})", text)
if m:
result["patient"]["code_postal"] = m.group(1)
m = re.search(r"Ville de résidence:\s*(.+?)(?:\n|$)", text)
if m:
result["patient"]["ville"] = m.group(1).strip()
# Taille, Poids, IMC (footer)
m = re.search(r"Taille:\s*(\d+)\s*cm", text)
if m:
result["patient"]["taille_cm"] = int(m.group(1))
m = re.search(r"Poids:\s*([\d.]+)\s*kg", text)
if m:
result["patient"]["poids_kg"] = float(m.group(1))
m = re.search(r"IMC:\s*([\d.]+)", text)
if m:
result["patient"]["imc"] = float(m.group(1))
def _extract_sejour_info(text: str, result: dict) -> None:
"""Extrait les détails de l'épisode."""
m = re.search(r"Episode No:\s*(\d+)", text)
if m:
result["sejour"]["episode"] = m.group(1)
m = re.search(r"Date d'admission:\s*(\d{2}/\d{2}/\d{4})", text)
if m:
result["sejour"]["date_entree"] = m.group(1)
m = re.search(r"Heure d'admission:\s*(\d{2}:\d{2})", text)
if m:
result["sejour"]["heure_entree"] = m.group(1)
m = re.search(r"Date de sortie:\s*(\d{2}/\d{2}/\d{4})", text)
if m:
result["sejour"]["date_sortie"] = m.group(1)
m = re.search(r"Heure de sortie:\s*(\d{2}:\d{2})", text)
if m:
result["sejour"]["heure_sortie"] = m.group(1)
m = re.search(r"Localisation:\s*(.+?)(?:\s+Médecin courant|\n)", text)
if m:
result["sejour"]["service"] = m.group(1).strip()
m = re.search(r"Médecin courant:\s*(.+?)(?:\n|$)", text)
if m:
result["sejour"]["medecin_courant"] = m.group(1).strip()
def _extract_contacts(text: str, result: dict) -> None:
"""Extrait la liste des contacts."""
# Bloc "Liste des contacts"
contact_block = re.search(
r"Liste des contacts\n(.*?)(?=Passage aux Urgences|Signes Vitaux|Observations médicales)",
text,
re.DOTALL,
)
if not contact_block:
return
block = contact_block.group(1)
# Chaque ligne de contact contient relation, nom, prénom, tél
for line in block.split("\n"):
line = line.strip()
if not line or line.startswith("Type de contact") or line.startswith("Tél"):
continue
# Chercher les noms et téléphones
tel_match = re.search(r"(\d{2}[.\-\s]\d{2}[.\-\s]\d{2}[.\-\s]\d{2}[.\-\s]\d{2})", line)
if tel_match or re.search(r"(?:Epoux|Époux|Épouse|Conjoint|Père|Mère|Fils|Fille|Frère|Soeur)", line, re.IGNORECASE):
result["contacts"].append(line)
def _extract_medecins(text: str, result: dict) -> None:
"""Extrait les noms de médecins/soignants."""
seen: set[str] = set()
def _add(name: str) -> None:
name = _clean_person_name(name)
if name and len(name) > 2 and name.lower() not in seen:
seen.add(name.lower())
result["medecins"].append(name)
# "DR. Prénom NOM" ou "Dr NOM" ou "Docteur NOM Prénom"
for m in re.finditer(
r"(?:DR\.?|Dr\.?|Docteur)\s+([A-ZÉÈÊËÀÂa-zéèêëàâ\.\-]+(?:\s+[A-ZÉÈÊËÀÂ][A-ZÉÈÊËÀÂa-zéèêëàâ\-]+){0,2})",
text,
):
_add(m.group(1))
# Auteurs d'observations : "Note d'évolution NOM Prénom DD/MM/YYYY"
# ou multi-ligne "Note IDE Prénom\nNOM DD/MM/YYYY"
for m in re.finditer(
r"(?:Note d'évolution|Note IDE|Histoire de la maladie|Conclusion Obs\.?\s*médicales?)\s+"
r"(?:DR\.?\s+)?"
r"([A-ZÉÈÊËÀÂa-zéèêëàâ\.\-]+(?:[\s\n]+[A-ZÉÈÊËÀÂa-zéèêëàâ\.\-]+)*?)"
r"\s+\d{2}/\d{2}/\d{4}",
text,
):
_add(m.group(1))
# Médecin de prise en charge / décision médicale
for m in re.finditer(
r"(?:Médecin de (?:la )?(?:prise en charge|décision)\s+médicale)\s+"
r"([A-ZÉÈÊËÀÂ][A-ZÉÈÊËÀÂa-zéèêëàâ\.\-]+(?:\s+[A-ZÉÈÊËÀÂa-zéèêëàâ\.\-]+){0,2})",
text,
):
_add(m.group(1))
# IAO NOM Prénom
for m in re.finditer(
r"IAO\s+([A-ZÉÈÊËÀÂ][A-ZÉÈÊËÀÂa-zéèêëàâ\.\-]+(?:\s+[A-ZÉÈÊËÀÂa-zéèêëàâ\.\-]+){0,2})",
text,
):
_add(m.group(1))
# Prénom seul sur la ligne avant "DD/MM/YYYY...Note IDE...\nNOM HH:MM"
# Ex: "Argitxu 02/03/2023\nNote IDE ...\nHIRIGOYEN 14:05"
# ou "Stephanie 27/02/2023 TDM fait et à voir\nNote IDE\nCONSTANTIN 08:54"
for m in re.finditer(
r"([A-ZÉÈÊËÀÂ][a-zéèêëàâäùûüôöîïç]+)\s+\d{2}/\d{2}/\d{4}[^\n]*\n\s*Note IDE[^\n]*\n\s*([A-ZÉÈÊËÀÂ][A-ZÉÈÊËÀÂa-zéèêëàâ\-]+)\s+\d{2}:\d{2}",
text,
):
prenom = m.group(1)
nom = m.group(2)
_add(f"{prenom} {nom}")
# Mots qui ne sont pas des noms de personnes
_NOT_NAMES = {
"non", "pas", "une", "des", "les", "par", "sur", "pour", "dans",
"avec", "sans", "qui", "que", "est", "sont", "date", "heure",
"cholecystectomie", "cholécystectomie", "cholangiographie",
"complication", "vasculaire", "nécessaire", "donc", "note",
"douleurs", "absence", "douleur", "lotissement", "priorité",
"prescriptions", "technique", "alimentaire", "signé", "réalisé",
"selles", "covid", "devenir", "algique", "normal", "regime",
"reprise", "biprofenid", "orale", "gelule", "comprime",
"glyc", "inj", "lipase", "protéines", "ionogramme",
"créatinine", "glucose", "num", "crp", "ta", "bilirubine",
"tp", "tca", "bh", "bs", "sortie", "transfert",
}
def _clean_person_name(raw: str) -> str:
"""Nettoie un nom extrait en supprimant le texte parasite."""
name = re.sub(r"\n+", " ", raw).strip()
parts = name.split()
clean: list[str] = []
for part in parts:
p = part.strip(".-")
if not p:
continue
if p.lower() in _NOT_NAMES:
break
# Un mot-nom : commence par une majuscule
if re.match(r"^[A-ZÉÈÊËÀÂÄÙÛÜÔÖÎÏÇ]", p):
clean.append(p)
else:
break
result = " ".join(clean).strip()
# Rejeter si un seul mot de 1-2 lettres (initiale)
if len(result) <= 2:
return ""
return result
def _extract_urgences(text: str, result: dict) -> None:
"""Extrait les données du passage aux urgences."""
urg_block = re.search(
r"Passage aux Urgences\n(.*?)(?=Signes Vitaux|Observations médicales|Antécédents)",
text,
re.DOTALL,
)
if not urg_block:
return
block = urg_block.group(1)
m = re.search(r"Mode de transport.*?:\s*(.+)", block)
if m:
result["urgences"]["mode_transport"] = m.group(1).strip()
m = re.search(r"Mode d'entrée\s+(.+)", block)
if m:
result["urgences"]["mode_entree"] = m.group(1).strip()
m = re.search(r"Priorité\s+(Priorité \d)", block)
if m:
result["urgences"]["priorite"] = m.group(1)
# Motifs de prise en charge
motifs = re.findall(
r"Motif de prise en charge\s+(.+?)(?=\n(?:Observ\.|Médecin|Date|IAO))",
block,
re.DOTALL,
)
if motifs:
result["urgences"]["motifs"] = [
line.strip()
for motif in motifs
for line in motif.split("\n")
if line.strip()
]
def _extract_observations(text: str, result: dict) -> None:
"""Extrait les observations médicales."""
obs_block = re.search(
r"Observations médicales\n(.*?)(?=Notes paramédicales|Surveillance Psychiatrie|Traitements médicamenteux|$)",
text,
re.DOTALL,
)
if not obs_block:
return
block = obs_block.group(1)
# Découper par type d'observation
entries = re.split(
r"(Note d'évolution|Conclusion Obs\.\s*médicales|Histoire de la maladie)",
block,
)
i = 1
while i < len(entries) - 1:
obs_type = entries[i].strip()
content = entries[i + 1].strip()
# Extraire auteur et date
m = re.match(
r"(?:DR\.?\s+)?([A-ZÉÈÊËÀÂa-zéèêëàâ\.\-]+(?:\s+[A-ZÉÈÊËÀÂa-zéèêëàâ\.\-]+)*)\s+(\d{2}/\d{2}/\d{4})\s+(\d{2}:\d{2})\s*(.*)",
content,
re.DOTALL,
)
if m:
result["observations_medicales"].append({
"type": obs_type,
"auteur": m.group(1).strip(),
"date": m.group(2),
"heure": m.group(3),
"contenu": m.group(4).strip(),
})
else:
result["observations_medicales"].append({
"type": obs_type,
"contenu": content,
})
i += 2
def _extract_notes_param(text: str, result: dict) -> None:
"""Extrait les notes paramédicales."""
notes_block = re.search(
r"Notes paramédicales\n(.*?)(?=Traitements médicamenteux|Surveillance|$)",
text,
re.DOTALL,
)
if not notes_block:
return
block = notes_block.group(1)
for m in re.finditer(
r"Note IDE\s+([A-Za-zéèêëàâäùûüôöîïçÉÈÊËÀÂÄÙÛÜÔÖÎÏÇ\.\-\s]+?)\s+(\d{2}/\d{2}/\d{4})\s+(\d{2}:\d{2})\s+(.*?)(?=Note IDE|$)",
block,
re.DOTALL,
):
result["notes_paramedicales"].append({
"auteur": m.group(1).strip(),
"date": m.group(2),
"heure": m.group(3),
"contenu": m.group(4).strip(),
})
def _extract_diagnostics(text: str, result: dict) -> None:
"""Extrait les diagnostics codés."""
# "Principal actif CODE DESCRIPTION"
for m in re.finditer(
r"(Principal|Associé|Significatif)\s+(actif|inactif)\s+([A-Z]\d{2}(?:\.\d{1,2})?)\s+(.+?)(?:\s+\[.*?\])?\s+\d{2}/\d{2}/\d{4}",
text,
):
result["diagnostics"].append({
"type": m.group(1),
"statut": m.group(2),
"code_cim10": m.group(3),
"libelle": m.group(4).strip(),
})
def _extract_traitements(text: str, result: dict) -> None:
"""Extrait les traitements médicamenteux."""
ttt_block = re.search(
r"Traitements médicamenteux\n(.*?)$",
text,
re.DOTALL,
)
if not ttt_block:
return
block = ttt_block.group(1)
# Chercher les noms de médicaments (en majuscules)
for m in re.finditer(
r"([A-ZÉÈÊËÀÂ][A-ZÉÈÊËÀÂ0-9\s\-/%.,'`]+(?:MG|ML|SOL|INJ|CPR|GEL|AMP|POCHE)[A-ZÉÈÊËÀÂ0-9\s\-/%.,'`\(\)\[\]]*)\s+([\d\s]+\s*(?:mg|G|GEL|CPR|AMP|ML)?)\s*[-]\s*(.+?)(?=\n[A-Z]|\Z)",
block,
re.DOTALL,
):
result["traitements"].append({
"medicament": m.group(1).strip(),
"dose": m.group(2).strip(),
"frequence": m.group(3).strip().split("\n")[0],
})
def _extract_vitals(text: str, result: dict) -> None:
"""Extrait les données anthropométriques clés."""
m = re.search(r"Taille \[cm\]\s+([\d.]+)", text)
if m:
result["signes_vitaux"]["taille_cm"] = float(m.group(1))
m = re.search(r"Poids \[kg\]\s+([\d.]+)", text)
if m:
result["signes_vitaux"]["poids_kg"] = float(m.group(1))
m = re.search(r"Indice\s*\n?\s*de masse\s+([\d.]+)", text)
if m:
result["signes_vitaux"]["imc"] = float(m.group(1))
def _build_medical_content(result: dict) -> None:
"""Construit le texte médical complet à partir des observations."""
parts: list[str] = []
if result["urgences"].get("motifs"):
parts.append("Motifs: " + ", ".join(result["urgences"]["motifs"]))
for obs in result["observations_medicales"]:
parts.append(obs.get("contenu", ""))
for note in result["notes_paramedicales"]:
parts.append(note.get("contenu", ""))
result["contenu_medical"] = "\n\n".join(parts)