"""Motor de matching por evidencia para soportes del cargue individual.""" from __future__ import annotations import re import unicodedata from dataclasses import dataclass, field from difflib import SequenceMatcher from typing import Any, Literal from app.services.patient_name_extraction import sanitize_patient_name_candidate IdentitySignalType = Literal[ "case_number", "patient_id", "patient_name", "payer", "provider", "admin_entity", ] IdentitySource = Literal["signal", "label", "inline", "neighbor_line", "filename"] IdentityPosition = Literal["first_page", "header_zone", "body_zone", "table_zone", "unknown"] IdentityQuality = Literal["clean", "ocr_noisy", "truncated", "suffix_contaminated"] AttributeStatus = Literal["match", "contradiction", "low_confidence"] _PATIENT_LABEL_PATTERN = re.compile( r"(?:nombre(?:\s+del)?\s+paciente|paciente|usuario)\s*[:#-]?\s*(.*)$", re.IGNORECASE, ) _CASE_LABEL_PATTERN = re.compile( r"(?:caso\s*(?:no\.?)?|no\.?\s*de\s*caso|n[°o]\s*caso)\s*[:#-]?\s*([A-Z0-9-]{4,20})", re.IGNORECASE, ) _DOCUMENT_NUMBER_PATTERN = re.compile(r"\b(\d{5,16}[A-Z]{0,3})\b", re.IGNORECASE) _PATIENT_ID_LABEL_PATTERN = re.compile( r"(?:identificaci[oó]n|documento|c[ée]dula|\bcc\b|\bti\b|\bce\b|\bdni\b)\s*[:#-]?\s*([A-Z0-9.-]{5,20})", re.IGNORECASE, ) _PAYER_LABEL_PATTERN = re.compile( r"(?:señores|aseguradora|eps|ips|convenio|prestador|entidad)\s*[:#-]?\s*(.+)$", re.IGNORECASE, ) _STOP_LABEL_PATTERN = re.compile( r"(?:fecha|egreso|ingreso|direcci[oó]n|convenio|autorizaci[oó]n|tipo de servicio|servicio|factura)\s*[:#-]?", re.IGNORECASE, ) _ADMIN_ENTITY_TOKENS = { "aseguradora", "cooperativa", "colombia", "convenio", "entidad", "eps", "fundacion", "fundación", "hospital", "ips", "ltda", "medicas", "médicas", "prestador", "salud", "sas", "seguros", "servicios", "solidaria", } _JOINER_TOKENS = {"de", "del", "la", "las", "los", "y"} _FILENAME_IGNORE_TOKENS = { "archivo", "caso", "cirugia", "cirugía", "clinica", "clínica", "control", "doc", "documento", "egreso", "factura", "fecha", "generico", "genérico", "hc", "historia", "ingreso", "laboratorio", "orden", "paciente", "prefactura", "quirurgica", "quirúrgica", "quirurgico", "quirúrgico", "radiografia", "radiografía", "radiologia", "radiología", "soporte", "tipo", "usuario", } @dataclass(frozen=True) class IdentityCandidate: original_value: str normalized_value: str signal_type: IdentitySignalType source: IdentitySource position: IdentityPosition quality: IdentityQuality score: int = 0 @dataclass(frozen=True) class AttributeMatchResult: attribute: Literal["case_number", "patient_id", "patient_name"] status: AttributeStatus extracted_value: str = "" preferred_value: str = "" matched_by_normalization: bool = False message: str = "" candidates: list[IdentityCandidate] = field(default_factory=list) @dataclass(frozen=True) class SupportIdentityDecision: patient_id_result: AttributeMatchResult case_number_result: AttributeMatchResult patient_name_result: AttributeMatchResult warnings: list[str] = field(default_factory=list) contradictions: list[str] = field(default_factory=list) @property def extracted_patient_id(self) -> str: return self.patient_id_result.extracted_value @property def extracted_case_number(self) -> str: return self.case_number_result.extracted_value @property def extracted_patient_name(self) -> str: return self.patient_name_result.extracted_value @dataclass(frozen=True) class _CanonicalIdentifier: original: str compact: str digits: str digits_only: bool suffix_contaminated: bool class SupportIdentityMatchEngine: """Evalúa coincidencias entre señales extraídas y el caso destino.""" def evaluate( self, *, filename: str, raw_text: str, detected_type: str, preferred_case_context: dict[str, Any], extracted_signals: Any, ) -> SupportIdentityDecision: candidates = self._mine_candidates( filename=filename, raw_text=raw_text, detected_type=detected_type, extracted_signals=extracted_signals, ) patient_id_result = self._resolve_numeric_attribute( attribute="patient_id", label="Identificación", preferred_value=preferred_case_context.get("patient_id"), candidates=[item for item in candidates if item.signal_type == "patient_id"], ) case_number_result = self._resolve_numeric_attribute( attribute="case_number", label="Número de caso", preferred_value=preferred_case_context.get("case_number"), candidates=[item for item in candidates if item.signal_type == "case_number"], ) patient_name_result = self._resolve_patient_name( preferred_value=preferred_case_context.get("patient_name"), patient_candidates=[item for item in candidates if item.signal_type == "patient_name"], admin_candidates=[item for item in candidates if item.signal_type == "admin_entity"], ) warnings = self._unique_messages( [ message for message in ( patient_id_result.message if patient_id_result.status == "low_confidence" else "", case_number_result.message if case_number_result.status == "low_confidence" else "", patient_name_result.message if patient_name_result.status == "low_confidence" else "", ) if message ] ) contradictions = self._unique_messages( [ message for message in ( patient_id_result.message if patient_id_result.status == "contradiction" else "", case_number_result.message if case_number_result.status == "contradiction" else "", patient_name_result.message if patient_name_result.status == "contradiction" else "", ) if message ] ) if self._needs_missing_identity_warning( detected_type=detected_type, patient_id_result=patient_id_result, case_number_result=case_number_result, patient_name_result=patient_name_result, ): warnings.append( "No se pudo validar con confianza la identidad extraída de la factura contra el caso seleccionado." ) return SupportIdentityDecision( patient_id_result=patient_id_result, case_number_result=case_number_result, patient_name_result=patient_name_result, warnings=self._unique_messages(warnings), contradictions=contradictions, ) def _mine_candidates( self, *, filename: str, raw_text: str, detected_type: str, extracted_signals: Any, ) -> list[IdentityCandidate]: candidates: list[IdentityCandidate] = [] candidates.extend(self._build_signal_candidates(extracted_signals)) candidates.extend(self._build_numeric_candidates(raw_text)) candidates.extend(self._build_name_candidates(raw_text, detected_type=detected_type)) if filename: filename_candidate = sanitize_patient_name_candidate(self._name_candidate_from_filename(filename)) if filename_candidate: candidates.append( self._build_candidate( value=filename_candidate, signal_type="patient_name", source="filename", position="unknown", ) ) return self._dedupe_candidates(candidates) def _build_signal_candidates(self, extracted_signals: Any) -> list[IdentityCandidate]: candidates: list[IdentityCandidate] = [] patient_id = str(getattr(extracted_signals, "patient_id", "") or "").strip() if patient_id: candidates.append( self._build_candidate( value=patient_id, signal_type="patient_id", source="signal", position="first_page", ) ) case_number = str(getattr(extracted_signals, "case_number", "") or "").strip() if case_number: candidates.append( self._build_candidate( value=case_number, signal_type="case_number", source="signal", position="first_page", ) ) patient_name = str(getattr(extracted_signals, "patient_name", "") or "").strip() if patient_name: signal_type: IdentitySignalType = ( "admin_entity" if self._looks_like_admin_entity(patient_name) else "patient_name" ) candidates.append( self._build_candidate( value=patient_name, signal_type=signal_type, source="signal", position="first_page", ) ) return candidates def _build_numeric_candidates(self, raw_text: str) -> list[IdentityCandidate]: candidates: list[IdentityCandidate] = [] lines = self._clean_lines(raw_text) for index, line in enumerate(lines[:40]): position = self._position_for_index(index) case_match = _CASE_LABEL_PATTERN.search(line) if case_match: candidates.append( self._build_candidate( value=case_match.group(1), signal_type="case_number", source="label", position=position, ) ) patient_id_match = _PATIENT_ID_LABEL_PATTERN.search(line) if patient_id_match: candidates.append( self._build_candidate( value=patient_id_match.group(1), signal_type="patient_id", source="label", position=position, ) ) if re.search(r"^cc\s*:", line, re.IGNORECASE): cc_match = _DOCUMENT_NUMBER_PATTERN.search(line) if cc_match: candidates.append( self._build_candidate( value=cc_match.group(1), signal_type="patient_id", source="inline", position=position, ) ) return candidates def _build_name_candidates(self, raw_text: str, *, detected_type: str) -> list[IdentityCandidate]: candidates: list[IdentityCandidate] = [] lines = self._clean_lines(raw_text) for index, line in enumerate(lines[:40]): position = self._position_for_index(index) payer_match = _PAYER_LABEL_PATTERN.search(line) if payer_match: value = self._extract_name_text(payer_match.group(1)) if value: candidates.append( self._build_candidate( value=value, signal_type="admin_entity", source="label", position=position, ) ) patient_label_match = _PATIENT_LABEL_PATTERN.search(line) if not patient_label_match: continue inline_value = self._extract_name_text(patient_label_match.group(1)) if inline_value: signal_type: IdentitySignalType = ( "admin_entity" if self._looks_like_admin_entity(inline_value) else "patient_name" ) candidates.append( self._build_candidate( value=inline_value, signal_type=signal_type, source="inline", position=position, ) ) for probe in range(index + 1, min(index + 5, len(lines))): candidate_line = lines[probe] if _STOP_LABEL_PATTERN.search(candidate_line): break candidate_text = self._extract_name_text(candidate_line) if not candidate_text: continue signal_type = ( "admin_entity" if self._looks_like_admin_entity(candidate_text) else "patient_name" ) candidates.append( self._build_candidate( value=candidate_text, signal_type=signal_type, source="neighbor_line", position=position if detected_type == "factura" else self._position_for_index(probe), ) ) return candidates def _resolve_numeric_attribute( self, *, attribute: Literal["case_number", "patient_id"], label: str, preferred_value: Any, candidates: list[IdentityCandidate], ) -> AttributeMatchResult: preferred_text = str(preferred_value or "").strip() preferred_identifier = self._canonical_identifier(preferred_text) if not candidates: return AttributeMatchResult( attribute=attribute, status="low_confidence", preferred_value=preferred_text, message="", candidates=[], ) ranked = sorted( candidates, key=lambda item: self._numeric_rank(item, preferred_identifier), reverse=True, ) best_candidate = ranked[0] best_identifier = self._canonical_identifier(best_candidate.original_value) is_match, matched_by_normalization = self._identifiers_match(preferred_identifier, best_identifier) if preferred_text and is_match: return AttributeMatchResult( attribute=attribute, status="match", extracted_value=best_candidate.original_value, preferred_value=preferred_text, matched_by_normalization=matched_by_normalization, candidates=ranked, ) if not preferred_text: return AttributeMatchResult( attribute=attribute, status="low_confidence", extracted_value=best_candidate.original_value, preferred_value=preferred_text, candidates=ranked, ) return AttributeMatchResult( attribute=attribute, status="contradiction", extracted_value=best_candidate.original_value, preferred_value=preferred_text, message=f"{label} extraída ({best_candidate.original_value}) no coincide con el caso seleccionado ({preferred_text}).", candidates=ranked, ) def _resolve_patient_name( self, *, preferred_value: Any, patient_candidates: list[IdentityCandidate], admin_candidates: list[IdentityCandidate], ) -> AttributeMatchResult: preferred_text = str(preferred_value or "").strip() if not patient_candidates: message = "" if admin_candidates: message = "Se detectó una entidad administrativa donde se esperaba el nombre del paciente." return AttributeMatchResult( attribute="patient_name", status="low_confidence", preferred_value=preferred_text, message=message, candidates=[*patient_candidates, *admin_candidates], ) ranked = sorted( patient_candidates, key=lambda item: self._name_rank(item, preferred_text), reverse=True, ) best_candidate = ranked[0] similarity = self._name_similarity(best_candidate.original_value, preferred_text) if preferred_text and similarity >= 0.72: return AttributeMatchResult( attribute="patient_name", status="match", extracted_value=best_candidate.original_value, preferred_value=preferred_text, candidates=[*ranked, *admin_candidates], ) if not preferred_text: return AttributeMatchResult( attribute="patient_name", status="low_confidence", extracted_value=best_candidate.original_value, preferred_value=preferred_text, candidates=[*ranked, *admin_candidates], ) return AttributeMatchResult( attribute="patient_name", status="contradiction", extracted_value=best_candidate.original_value, preferred_value=preferred_text, message=( f"Nombre de paciente extraído ({best_candidate.original_value}) no coincide con el caso seleccionado " f"({preferred_text})." ), candidates=[*ranked, *admin_candidates], ) def _build_candidate( self, *, value: str, signal_type: IdentitySignalType, source: IdentitySource, position: IdentityPosition, ) -> IdentityCandidate: normalized_value = self._normalize_spaces(value) quality = self._classify_quality(normalized_value) score = self._base_score(source=source, position=position, quality=quality, signal_type=signal_type) return IdentityCandidate( original_value=normalized_value, normalized_value=self._normalize_candidate_value(normalized_value, signal_type), signal_type=signal_type, source=source, position=position, quality=quality, score=score, ) def _dedupe_candidates(self, candidates: list[IdentityCandidate]) -> list[IdentityCandidate]: deduped: dict[tuple[str, str, str], IdentityCandidate] = {} for candidate in candidates: key = (candidate.signal_type, candidate.normalized_value, candidate.source) current = deduped.get(key) if current is None or candidate.score > current.score: deduped[key] = candidate return list(deduped.values()) def _base_score( self, *, source: IdentitySource, position: IdentityPosition, quality: IdentityQuality, signal_type: IdentitySignalType, ) -> int: source_score = { "signal": 50, "label": 42, "inline": 40, "neighbor_line": 34, "filename": 16, }[source] position_score = { "first_page": 8, "header_zone": 6, "body_zone": 3, "table_zone": -2, "unknown": 0, }[position] quality_score = { "clean": 14, "ocr_noisy": 0, "truncated": 4, "suffix_contaminated": 9, }[quality] signal_bonus = 4 if signal_type in {"patient_id", "case_number"} else 0 return source_score + position_score + quality_score + signal_bonus def _numeric_rank( self, candidate: IdentityCandidate, preferred_identifier: _CanonicalIdentifier ) -> tuple[int, int]: candidate_identifier = self._canonical_identifier(candidate.original_value) is_match, matched_by_normalization = self._identifiers_match( preferred_identifier, candidate_identifier ) match_score = 40 if is_match and not matched_by_normalization else 25 if is_match else 0 return match_score, candidate.score def _name_rank(self, candidate: IdentityCandidate, preferred_value: str) -> tuple[int, int]: similarity_bonus = int(self._name_similarity(candidate.original_value, preferred_value) * 30) return similarity_bonus, candidate.score def _canonical_identifier(self, value: str) -> _CanonicalIdentifier: compact = re.sub(r"[^A-Z0-9]+", "", str(value or "").upper()) digits = "".join(char for char in compact if char.isdigit()) digits_only = bool(compact) and compact.isdigit() suffix_contaminated = bool(re.fullmatch(r"\d{4,16}[A-Z]{1,3}", compact)) return _CanonicalIdentifier( original=str(value or "").strip(), compact=compact, digits=digits, digits_only=digits_only, suffix_contaminated=suffix_contaminated, ) def _identifiers_match( self, preferred_identifier: _CanonicalIdentifier, candidate_identifier: _CanonicalIdentifier, ) -> tuple[bool, bool]: if not preferred_identifier.compact or not candidate_identifier.compact: return False, False if preferred_identifier.compact == candidate_identifier.compact: return True, False if ( preferred_identifier.digits and preferred_identifier.digits == candidate_identifier.digits and ( preferred_identifier.digits_only or candidate_identifier.digits_only or preferred_identifier.suffix_contaminated or candidate_identifier.suffix_contaminated ) ): return True, True return False, False def _needs_missing_identity_warning( self, *, detected_type: str, patient_id_result: AttributeMatchResult, case_number_result: AttributeMatchResult, patient_name_result: AttributeMatchResult, ) -> bool: if self._normalize_spaces(detected_type).lower() != "factura": return False return not any( [ patient_id_result.extracted_value.strip(), case_number_result.extracted_value.strip(), patient_name_result.extracted_value.strip(), ] ) def _name_similarity(self, left: str, right: str) -> float: normalized_left = self._normalize_name(left) normalized_right = self._normalize_name(right) if not normalized_left or not normalized_right: return 0.0 if normalized_left == normalized_right: return 1.0 left_tokens = self._meaningful_tokens(normalized_left) right_tokens = self._meaningful_tokens(normalized_right) if not left_tokens or not right_tokens: return 0.0 overlap = len(left_tokens.intersection(right_tokens)) / max(len(left_tokens), len(right_tokens)) sequence = SequenceMatcher(a=normalized_left, b=normalized_right).ratio() return (overlap * 0.6) + (sequence * 0.4) def _meaningful_tokens(self, value: str) -> set[str]: return { token for token in re.findall(r"[a-z0-9áéíóúñ]+", value) if token and token not in _JOINER_TOKENS } def _normalize_candidate_value(self, value: str, signal_type: IdentitySignalType) -> str: if signal_type in {"case_number", "patient_id"}: return self._canonical_identifier(value).compact return self._normalize_name(value) def _extract_name_text(self, value: str) -> str: cleaned = self._normalize_spaces(value) if not cleaned or _STOP_LABEL_PATTERN.search(cleaned): return "" return sanitize_patient_name_candidate(cleaned) def _looks_like_admin_entity(self, value: str) -> bool: tokens = re.findall(r"[a-záéíóúñ]+", self._normalize_name(value)) return sum(1 for token in tokens if token in _ADMIN_ENTITY_TOKENS) >= 2 def _classify_quality(self, value: str) -> IdentityQuality: if not value: return "ocr_noisy" compact = re.sub(r"\s+", "", value) if re.fullmatch(r"\d{4,16}[A-Z]{1,3}", compact, re.IGNORECASE): return "suffix_contaminated" if len(value.split()) == 1 and len(value) <= 4: return "truncated" letters = sum(1 for char in compact if char.isalpha()) digits = sum(1 for char in compact if char.isdigit()) if letters and digits and letters / max(len(compact), 1) > 0.5: return "ocr_noisy" return "clean" def _position_for_index(self, index: int) -> IdentityPosition: if index <= 6: return "header_zone" if index <= 20: return "body_zone" return "table_zone" def _name_candidate_from_filename(self, filename: str) -> str: basename = re.sub(r"\.pdf$", "", str(filename or ""), flags=re.IGNORECASE) tokens = re.findall(r"[A-Za-zÁÉÍÓÚÑáéíóúñ]+", basename) filtered = [] for token in tokens: normalized = self._normalize_name(token) if len(normalized) < 2 or normalized in _FILENAME_IGNORE_TOKENS: continue filtered.append(token) if len(filtered) < 2: return "" return self._normalize_spaces(" ".join(filtered[:5])) def _clean_lines(self, raw_text: str) -> list[str]: return [ self._normalize_spaces(line) for line in str(raw_text or "").splitlines() if self._normalize_spaces(line) ] def _normalize_spaces(self, value: str) -> str: return re.sub(r"\s+", " ", str(value or "")).strip() def _normalize_name(self, value: str) -> str: normalized = unicodedata.normalize("NFKD", str(value or "")) normalized = "".join(char for char in normalized if not unicodedata.combining(char)) normalized = normalized.lower() normalized = re.sub(r"[^a-z0-9 ]+", " ", normalized) return re.sub(r"\s+", " ", normalized).strip() def _unique_messages(self, messages: list[str]) -> list[str]: seen: set[str] = set() ordered: list[str] = [] for message in messages: normalized = str(message or "").strip() if not normalized or normalized in seen: continue seen.add(normalized) ordered.append(normalized) return ordered