"""Clasificador documental heurístico para V1 del procesamiento masivo.""" from __future__ import annotations import re import unicodedata from app.clinical_pipeline.domain.models import DocumentClassificationDecision class HeuristicDocumentClassifier: """Detecta el tipo documental por nombre de archivo y contenido extraído.""" _REAL_FACTURA_MARKERS = ( "factura electronica de venta", "autorizacion de facturacion electronica", "numero de factura", "fecha factura", "prefijo no factura", "valor total de la factura", ) _PATIENT_ID_MARKERS = ( "identificacion", "documento", "cedula", " cc ", ) _CASE_PATTERN = re.compile( r"(?:caso\s*(?:no\.?)?|no\.?\s*de\s*caso|n[°o]\s*caso|caso)\s*[:#-]?\s*(?:cm\s*-\s*)?\d{4,12}", re.IGNORECASE, ) _PREFACTURA_MARKERS = ( "prefactura", "pre factura", "pre factura de servicios", "extracto de cuenta", "cuenta preliminar", "prefacturacion", ) _HEADER_RULES: tuple[tuple[str, str, tuple[str, ...]], ...] = ( ("prefactura", "PreFactura de Servicios", ("prefactura de servicios", "extracto de cuenta")), ( "historia_clinica", "Historia Clínica de Urgencias", ("historia clinica de urgencias",), ), ("historia_clinica", "Nota Observación", ("nota observacion",)), ("historia_clinica", "Notas Sala Procedimientos", ("notas sala procedimientos",)), ("historia_clinica", "Evolución Médica", ("evolucion medica",)), ("historia_clinica", "Notas de Curaciones", ("notas de curaciones",)), ("prescripcion", "Hoja de Drogas", ("hoja de drogas",)), ( "radiologia", "Órdenes de Paraclínicos Generadas en Historias Clínicas", ("ordenes de paraclinicos generadas en historias clinicas", "ordenes de paraclinicos"), ), ( "prescripcion", "Órdenes de Medicamentos Generadas en Historias Clínicas", ("ordenes de medicamentos generadas en historias clinicas", "ordenes de medicamentos"), ), ( "prescripcion", "Órdenes Médicas Generadas en Historias Clínicas", ("ordenes medicas generadas en historias clinicas", "ordenes medicas"), ), ( "prescripcion", "Órdenes Generadas en Historias Clínicas", ("ordenes generadas en historias clinicas",), ), ) _REPORT_CODE_RULES: dict[str, tuple[str, str]] = { "rephistoriaurgencias": ("historia_clinica", "Historia Clínica de Urgencias"), "rephistoriaevoluciones": ("historia_clinica", "Historia clínica"), "rephistorianotas": ("historia_clinica", "Historia clínica"), } _REPORT_CODE_PATTERN = re.compile(r"\b(rep[a-z]+)\b", re.IGNORECASE) _GENERIC_TITLE_PATTERNS = ( "inversiones medicas valle salud", "nit 900631361 6", "codigo", "medicamento", "descripcion", "edad", "sexo", "identificacion", "nombre del paciente", "no de caso", "servicio", "consecutivo", "orden no", "page ", "fecha", "hora", "tipo y n", ) _RULES = ( ( "factura", ( "factura electronica de venta", "numero de factura", "valor total de la factura", "autorizacion de facturacion electronica", ), ), ( "quirurgico", ( "quirurg", "cirugia", "quirofano", "procedimiento quirurgico", "qx", ), ), ( "laboratorio", ( "laboratorio", "hemograma", "bioquimica", "quimica sanguinea", "resultado de laboratorio", ), ), ( "radiologia", ( "radiologia", "tomografia", "tac", "resonancia", "ecografia", "rayos x", "radiografia", "rx", ), ), ( "prescripcion", ( "formula medica", "prescripcion", "orden medica", "medicamento", "tratamiento", ), ), ( "historia_clinica", ( "historia clinica", "epicrisis", "evolucion", "anamnesis", ), ), ) _HISTORIA_FILENAME_PATTERN = re.compile( r"(?:^|[^a-z0-9])hc\s*[-_]?\s*\d{4,}|historia\s+clinica", re.IGNORECASE, ) _HISTORIA_COMPOSITE_MARKERS = ( "historia clinica", "motivo de consulta", "enfermedad actual", "antecedentes", "evolucion medica", "examen fisico", ) def describe(self, filename: str, text: str) -> tuple[str, str]: decision = self.inspect(filename, text) return decision.document_type, decision.title def inspect(self, filename: str, text: str) -> DocumentClassificationDecision: haystack = self._normalize_haystack(filename, text) if self._looks_like_prefactura(haystack): return DocumentClassificationDecision( document_type="prefactura", title="PreFactura de Servicios", confidence=0.98, reasons=["prefactura_markers"], scores={"prefactura": 9.0}, ) simplified_filename = self._simplify_text(filename) composite_hits = [marker for marker in self._HISTORIA_COMPOSITE_MARKERS if marker in haystack] strong_historia_filename = bool(self._HISTORIA_FILENAME_PATTERN.search(simplified_filename)) if strong_historia_filename or len(composite_hits) >= 3: reasons = [] if strong_historia_filename: reasons.append("historia_filename") if composite_hits: reasons.append("historia_composite_content") historia_score = 8.0 + min(len(composite_hits), 6) quirurgico_score = float(sum(haystack.count(keyword) for keyword in self._RULES[1][1])) return DocumentClassificationDecision( document_type="historia_clinica", title=self._resolve_title("historia_clinica", "Historia clínica", text), confidence=0.98 if strong_historia_filename else 0.9, reasons=reasons, scores={"historia_clinica": historia_score, "quirurgico": quirurgico_score}, ) header_match = self._find_header_match(text) if header_match: document_type, document_title = header_match return DocumentClassificationDecision( document_type=document_type, title=self._resolve_title(document_type, document_title, text), confidence=0.99, reasons=["recognized_document_header"], scores={document_type: 10.0}, ) report_code = self._extract_report_code(text) if report_code: report_match = self._REPORT_CODE_RULES.get(report_code) if report_match: return DocumentClassificationDecision( document_type=report_match[0], title=report_match[1], confidence=0.99, reasons=[f"report_code:{report_code}"], scores={report_match[0]: 10.0}, ) for detected_type, keywords in self._RULES: if any(keyword in haystack for keyword in keywords): hits = [keyword for keyword in keywords if keyword in haystack] return DocumentClassificationDecision( document_type=detected_type, title=self._resolve_title( detected_type, self._humanize_type(detected_type), text, ), confidence=min(0.95, 0.72 + (0.05 * len(hits))), reasons=[f"keyword:{item}" for item in hits[:5]], scores={detected_type: float(len(hits))}, ) return DocumentClassificationDecision( document_type="generico", title="Documento general", confidence=0.35, reasons=["no_specific_markers"], scores={"generico": 1.0}, ) def classify(self, filename: str, text: str) -> str: return self.inspect(filename, text).document_type def _find_header_match(self, text: str) -> tuple[str, str] | None: normalized_lines = [ self._simplify_text(self._normalize_line(line)) for line in str(text or "").splitlines() if self._normalize_line(line) ] candidate_lines = normalized_lines[:24] for line in candidate_lines: for document_type, document_title, markers in self._HEADER_RULES: if any(marker in line for marker in markers): return document_type, document_title joined = " ".join(candidate_lines) for document_type, document_title, markers in self._HEADER_RULES: if any(marker in joined for marker in markers): return document_type, document_title return None def _looks_like_prefactura(self, haystack: str) -> bool: has_prefactura_marker = any(marker in haystack for marker in self._PREFACTURA_MARKERS) has_case_number = bool(self._CASE_PATTERN.search(haystack)) has_real_factura = any(marker in haystack for marker in self._REAL_FACTURA_MARKERS) has_patient_id = any(marker in f" {haystack} " for marker in self._PATIENT_ID_MARKERS) if has_prefactura_marker and has_case_number and not has_real_factura: return True return has_case_number and "factura" in haystack and not has_real_factura and not has_patient_id def _normalize_haystack(self, filename: str, text: str) -> str: haystack = self._simplify_text(f"{filename} {text[:4000]}") haystack = re.sub(r"[_\-]+", " ", haystack) haystack = re.sub(r"\s+", " ", haystack) return haystack.strip() def _humanize_type(self, detected_type: str) -> str: labels = { "factura": "Factura", "quirurgico": "Documento quirúrgico", "laboratorio": "Laboratorio", "radiologia": "Radiología", "prescripcion": "Prescripción", "historia_clinica": "Historia clínica", } return labels.get(detected_type, "Documento general") def _resolve_title(self, document_type: str, base_title: str, text: str) -> str: if document_type == "radiologia": radiology_title = self._extract_radiology_title(text) if radiology_title: return radiology_title if document_type == "prescripcion" and base_title in {"Prescripción", "Documento general"}: support_title = self._extract_support_title(text) if support_title: return support_title return base_title def _extract_support_title(self, text: str) -> str: lines = [self._normalize_line(line) for line in str(text or "").splitlines()] for line in lines[:40]: if not line: continue normalized = self._simplify_text(line) if any(pattern in normalized for pattern in self._GENERIC_TITLE_PATTERNS): continue if re.fullmatch(r"[0-9 .:/-]+", line): continue if len(re.findall(r"[a-z]", normalized)) < 3: continue return line return "" def _extract_radiology_title(self, text: str) -> str: lines = [self._normalize_line(line) for line in str(text or "").splitlines()] for line in lines[:40]: normalized = self._simplify_text(line) if any( token in normalized for token in ("radiografia", "tomografia", "ecografia", "resonancia", "rayos x", "rx") ): return line return "" def _extract_report_code(self, text: str) -> str: match = self._REPORT_CODE_PATTERN.search(str(text or "")) if not match: return "" return self._simplify_text(match.group(1)) def _normalize_line(self, value: str) -> str: return re.sub(r"\s+", " ", str(value or "")).strip(" :#-\t") def _simplify_text(self, value: str) -> str: normalized = unicodedata.normalize("NFKD", str(value or "")) normalized = normalized.encode("ascii", "ignore").decode("ascii") normalized = normalized.lower() normalized = re.sub(r"[_\-]+", " ", normalized) normalized = re.sub(r"\s+", " ", normalized) return normalized.strip()