from __future__ import annotations import hashlib import re import unicodedata from collections.abc import Iterable from decimal import Decimal, InvalidOperation from typing import Any from app.clinical_pipeline.domain.medications import ( MedicamentoCaso, MedicationCaseStatus, MedicationEvidenceRole, MedicationInvoiceAssociationStatus, ) _ORDER_TITLE = "ordenes generadas en historias clinicas" _DRUG_SHEET_TITLE = "hoja de drogas" _ROUTE_PATTERN = re.compile( r"\b(INTRAVENOSA|INTRAARTICULAR|INTRAMUSCULAR|SUBCUTANEA|SUBCUTÁNEA|ORAL|TOPIC[AO]|" r"INHALADA|OFTALMICA|OFTÁLMICA|RECTAL|SUBLINGUAL)\b", re.IGNORECASE, ) _ADMIN_DOSE_PATTERN = re.compile( r"(?P\d+(?:[.,]\d+)?\s*(?:mcg|mg|gr|g|ml|cc|ui))\s+" r"(?PINTRAVENOSA|INTRAARTICULAR|INTRAMUSCULAR|SUBCUTANEA|SUBCUTÁNEA|ORAL|TOPIC[AO]|" r"INHALADA|OFTALMICA|OFTÁLMICA|RECTAL|SUBLINGUAL)\b", re.IGNORECASE, ) _ORDER_ROW_PATTERN = re.compile( r"^\s*(?P\d{4})\s+(?P.*?)\s{2,}(?P\d+(?:[.,]\d+)?)\s*$" ) _DATE_PATTERN = re.compile(r"\b\d{1,2}/\d{1,2}/\d{4}\b") _STOP_LINE_PATTERN = re.compile( r"^(?:dr\.|cc\s*-|reg\.m\.|no\. de caso|codigo\s+descripcion|paciente:|tipo y n|page\s+\d+)", re.IGNORECASE, ) _NON_MEDICATION_PATTERN = re.compile( r"\b(?:AGUJA|JERINGA|EQUIPO|GUANTE|BISTURI|CANULA|CATETER|TAPON|ELECTRODO|VENOCLISIS|" r"SUTURA|VICRYL|PROLENE|FIBERWIRE|PLACA|CURACION|CABESTRILLO|FLUOROSCOPIA|CONSULTA|" r"HEMOGRAMA|CREATININA|TROMBOPLASTINA|NITROGENO UREICO)\b", re.IGNORECASE, ) _FORMULATION_PATTERN = re.compile( r"\b(?:SOLUCI[OÓ]N\s+INYECTABLE|INYECTABLE|AMPOLLAS?|TABLETAS?|C[ÁA]PSULAS?|BOLSA\s+X\s+\d+\s*(?:ML|CC)|" r"FRASCO\s+VIAL)\b.*$", re.IGNORECASE, ) _STATUS_PRIORITY = { MedicationCaseStatus.ADMINISTRADO: 0, MedicationCaseStatus.FORMULADO: 1, MedicationCaseStatus.ORDENADO: 2, MedicationCaseStatus.FACTURADO: 3, MedicationCaseStatus.ANTECEDENTE: 4, MedicationCaseStatus.OTRO: 5, } def normalize_medication_key(value: Any) -> str: text = unicodedata.normalize("NFKD", str(value or "")) ascii_text = "".join(char for char in text if not unicodedata.combining(char)).casefold() return re.sub(r"[^a-z0-9]+", " ", ascii_text).strip() def parse_numeric_quantity(value: Any) -> int | float | None: raw = str(value or "").strip().replace(" ", "") if not raw: return None if re.fullmatch(r"-?\d+[.,]00", raw): raw = raw[:-3] elif "," in raw and "." not in raw: raw = raw.replace(",", ".") try: parsed = Decimal(raw) except InvalidOperation: return None return int(parsed) if parsed == parsed.to_integral_value() else float(parsed) def _clean(value: Any) -> str: return re.sub(r"\s+", " ", str(value or "")).strip(" -") def _status(value: Any, default: MedicationCaseStatus = MedicationCaseStatus.OTRO) -> MedicationCaseStatus: normalized = normalize_medication_key(value).replace(" ", "_") aliases = { "administrado": MedicationCaseStatus.ADMINISTRADO, "formulado": MedicationCaseStatus.FORMULADO, "prescrito": MedicationCaseStatus.FORMULADO, "ordenado": MedicationCaseStatus.ORDENADO, "facturado": MedicationCaseStatus.FACTURADO, "antecedente": MedicationCaseStatus.ANTECEDENTE, } return aliases.get(normalized, default) def _split_name_presentation(raw_name: str) -> tuple[str, str]: cleaned = _clean(raw_name) generic_match = re.search(r"\(([A-Za-zÁÉÍÓÚÑáéíóúñ ]{4,})\)", cleaned) presentation_match = _FORMULATION_PATTERN.search(cleaned) presentation = _clean(presentation_match.group(0)) if presentation_match else "" base = _clean(cleaned[: presentation_match.start()] if presentation_match else cleaned) base = re.sub(r"\s+\d+(?:[.,]\d+)?\s*(?:MCG|MG|G|GR|ML|CC)(?:\s*/\s*\d+\s*(?:ML|CC))?.*$", "", base, flags=re.I) if generic_match: generic = _clean(generic_match.group(1)) brand = _clean(re.sub(r"\([^()]+\)", "", base)) return (f"{generic.title()} ({brand.title()})" if brand else generic.title(), presentation) return base.title(), presentation def _build_key(nombre: str, presentacion: str, dosis: str, posologia: str) -> str: material = "|".join( normalize_medication_key(value) for value in (nombre, presentacion, dosis, posologia) ) return hashlib.sha256(material.encode("utf-8")).hexdigest()[:20] def _evidence_role(status: MedicationCaseStatus) -> MedicationEvidenceRole: return { MedicationCaseStatus.ORDENADO: MedicationEvidenceRole.ORDEN, MedicationCaseStatus.FORMULADO: MedicationEvidenceRole.ORDEN, MedicationCaseStatus.ADMINISTRADO: MedicationEvidenceRole.ADMINISTRACION, MedicationCaseStatus.FACTURADO: MedicationEvidenceRole.FACTURACION, }.get(status, MedicationEvidenceRole.CONTEXTO) def _evidence_id( *, nombre: str, fuente: str, estado: MedicationCaseStatus, evidencia: str, cantidad: int | float | None, ) -> str: material = "|".join( ( normalize_medication_key(nombre), normalize_medication_key(fuente), estado.value, normalize_medication_key(evidencia), str(cantidad if cantidad is not None else ""), ) ) return f"medev-{hashlib.sha256(material.encode('utf-8')).hexdigest()[:16]}" def _source_metadata(source: str) -> tuple[str, str]: document_type, separator, document_name = _clean(source).partition(":") return normalize_medication_key(document_type).replace(" ", "_"), _clean( document_name if separator else "" ) def _make_medication( *, nombre: str, presentacion: str = "", dosis: str = "", posologia: str = "", cantidad: Any = None, estado: MedicationCaseStatus = MedicationCaseStatus.OTRO, codigo_facturacion: str = "", codigo_referencia: str = "", fuente: str = "historia_clinica", evidencia: str = "", unidad_cantidad: str = "", ) -> dict[str, Any]: normalized_name, inferred_presentation = _split_name_presentation(nombre) final_presentation = _clean(presentacion) or inferred_presentation parsed_quantity = parse_numeric_quantity(cantidad) document_type, document_name = _source_metadata(fuente) item = MedicamentoCaso( key=_build_key(normalized_name, final_presentation, dosis, posologia), nombre=normalized_name, presentacion=final_presentation, posologia=_clean(posologia), dosis=_clean(dosis), cantidad=parsed_quantity, unidad_cantidad=_clean(unidad_cantidad), estados=[estado], codigo_facturacion=_clean(codigo_facturacion), codigo_referencia=_clean(codigo_referencia), codigo_estado="identificado" if codigo_facturacion or codigo_referencia else "no_identificado", fuentes=[fuente], evidencias=[ { "evidence_id": _evidence_id( nombre=normalized_name, fuente=fuente, estado=estado, evidencia=evidencia, cantidad=parsed_quantity, ), "fuente": fuente, "estado": estado, "document_type": document_type, "document_name": document_name, "role": _evidence_role(estado), "cantidad": parsed_quantity, "unidad": _clean(unidad_cantidad), "evidencia": _clean(evidencia)[:240], } ], ) return _with_legacy_projection(item.model_dump(mode="json")) def _with_legacy_projection(item: dict[str, Any]) -> dict[str, Any]: projected = dict(item) projected["medicamento"] = str(projected.get("nombre") or "") statuses = [str(value) for value in projected.get("estados", []) if str(value)] projected["tipo_uso"] = statuses[0] if statuses else "otro" projected["fuente"] = str((projected.get("fuentes") or [""])[0] or "") projected["codigo"] = str(projected.get("codigo_referencia") or projected.get("codigo_facturacion") or "") projected["texto_original"] = str(projected.get("evidencia") or projected.get("nombre") or "") return projected def _normalize_evidence( evidence: dict[str, Any], *, nombre: str, default_source: str, default_status: MedicationCaseStatus, default_unit: str, ) -> dict[str, Any]: source = _clean(evidence.get("fuente")) or default_source status = _status(evidence.get("estado"), default_status) quantity = parse_numeric_quantity(evidence.get("cantidad")) snippet = _clean(evidence.get("evidencia"))[:240] document_type, document_name = _source_metadata(source) role = str(evidence.get("role") or _evidence_role(status).value) return { "evidence_id": _clean(evidence.get("evidence_id")) or _evidence_id( nombre=nombre, fuente=source, estado=status, evidencia=snippet, cantidad=quantity, ), "fuente": source, "estado": status.value, "document_type": _clean(evidence.get("document_type")) or document_type, "document_name": _clean(evidence.get("document_name")) or document_name, "pagina": _clean(evidence.get("pagina")), "role": role, "fecha": _clean(evidence.get("fecha")), "cantidad": quantity, "unidad": _clean(evidence.get("unidad")) or default_unit, "evidencia": snippet, } def _posology_from_text(text: str, route: str) -> str: normalized = _clean(text) frequency_match = re.search( r"\b(?:AHORA|CADA\s+\d+\s+HORAS?(?:\s+POR\s+\d+\s+D[IÍ]A(?:\(S\)|S)?)?)", normalized, re.IGNORECASE, ) return _clean(" ".join(part for part in (route, frequency_match.group(0) if frequency_match else "") if part)) def _parse_order_block(code: str, body: str, quantity: str, continuations: list[str]) -> dict[str, Any] | None: combined = _clean(" ".join([body, *continuations])) if _NON_MEDICATION_PATTERN.search(combined) or not _ROUTE_PATTERN.search(combined): return None dose_matches = list(_ADMIN_DOSE_PATTERN.finditer(combined)) if not dose_matches: return None administration = dose_matches[-1] raw_name = _clean(combined[: administration.start()]) raw_name = re.sub(r"^\(C\)_?\s*", "", raw_name, flags=re.I) dose = administration.group("dose") route = administration.group("route") status = MedicationCaseStatus.FORMULADO if re.search(r"\bFORMUL", combined, re.I) else MedicationCaseStatus.ORDENADO return _make_medication( nombre=raw_name, dosis=dose, posologia=_posology_from_text(combined[administration.end() :], route), cantidad=quantity, estado=status, codigo_facturacion=code, evidencia=combined, ) def extract_valle_salud_medications(raw_text: str) -> list[dict[str, Any]]: """Interpreta órdenes y administraciones del formato tabular de Valle Salud.""" text = str(raw_text or "") if _ORDER_TITLE not in normalize_medication_key(text) and _DRUG_SHEET_TITLE not in normalize_medication_key(text): return [] lines = text.splitlines() ordered: list[dict[str, Any]] = [] index = 0 while index < len(lines): match = _ORDER_ROW_PATTERN.match(lines[index]) if not match: index += 1 continue continuations: list[str] = [] probe = index + 1 while probe < len(lines) and len(continuations) < 4: candidate = _clean(lines[probe]) if _ORDER_ROW_PATTERN.match(lines[probe]) or _STOP_LINE_PATTERN.match(candidate): break if candidate and not _DATE_PATTERN.search(candidate): continuations.append(candidate) probe += 1 medication = _parse_order_block( match.group("code"), match.group("body"), match.group("quantity"), continuations ) if medication: ordered.append(medication) index = max(index + 1, probe) administered_signatures: list[tuple[str, str, str, str]] = [] in_drug_sheet = False for line in lines: normalized = normalize_medication_key(line) if _DRUG_SHEET_TITLE in normalized: in_drug_sheet = True continue if in_drug_sheet and _ORDER_TITLE in normalized: in_drug_sheet = False if not in_drug_sheet or not _DATE_PATTERN.search(line): continue administration_matches = list(_ADMIN_DOSE_PATTERN.finditer(line)) if not administration_matches: continue first_administration = administration_matches[0] final_dose_match = re.search( r"(?P\d+(?:[.,]\d+)?\s*(?:mcg|mg|gr|g|ml|cc|ui))\s+\d{1,2}/\d{1,2}/\d{4}", line, re.IGNORECASE, ) dose = final_dose_match.group("dose") if final_dose_match else first_administration.group("dose") route = first_administration.group("route") name_key = normalize_medication_key(line[: first_administration.start()]) frequency_fragment = normalize_medication_key( line[first_administration.end() : final_dose_match.start() if final_dose_match else len(line)] ) frequency_kind = "ahora" if "ahora" in frequency_fragment else "cada" if "cada" in frequency_fragment else "" administered_signatures.append( (name_key, normalize_medication_key(dose), normalize_medication_key(route), frequency_kind) ) for item in ordered: item_name = normalize_medication_key(item.get("nombre")) item_dose = normalize_medication_key(item.get("dosis")) item_route = normalize_medication_key(item.get("posologia")) if any( signature_name.split(" ")[0] in item_name and (not signature_dose or signature_dose == item_dose) and (not signature_route or signature_route in item_route) and (not signature_frequency or signature_frequency in item_route) for signature_name, signature_dose, signature_route, signature_frequency in administered_signatures if signature_name ): item["estados"] = [MedicationCaseStatus.ADMINISTRADO.value, *item.get("estados", [])] return consolidate_medication_items(ordered) def medication_from_legacy(item: Any, *, fuente: str = "historia_clinica") -> dict[str, Any] | None: if isinstance(item, dict): name = item.get("nombre") or item.get("medicamento") or item.get("detalle") if not _clean(name) or normalize_medication_key(name) in {"no especificado", "no disponible"}: return None via = _clean(item.get("via")) frecuencia = _clean(item.get("frecuencia")) duracion = _clean(item.get("duracion")) posologia = _clean(item.get("posologia")) or _clean(" ".join(filter(None, (via, frecuencia, duracion)))) raw_statuses = item.get("estados") or [item.get("estado_caso") or item.get("tipo_uso")] statuses = [_status(value) for value in (raw_statuses if isinstance(raw_statuses, list) else [raw_statuses])] result = _make_medication( nombre=str(name), presentacion=str(item.get("presentacion") or ""), dosis=str(item.get("dosis") or ""), posologia=posologia, cantidad=item.get("cantidad"), estado=statuses[0] if statuses else MedicationCaseStatus.OTRO, codigo_facturacion=str(item.get("codigo_facturacion") or item.get("codigo") or ""), codigo_referencia=str(item.get("codigo_referencia") or item.get("codigo_medicamento") or ""), fuente=str(item.get("fuente") or fuente), evidencia=str(item.get("texto_original") or ""), unidad_cantidad=str(item.get("unidad_cantidad") or ""), ) result["estados"] = list(dict.fromkeys(status.value for status in statuses)) if isinstance(item.get("fuentes"), list): result["fuentes"] = list( dict.fromkeys(str(value).strip() for value in item["fuentes"] if str(value).strip()) ) if isinstance(item.get("evidencias"), list): result["evidencias"] = [ _normalize_evidence( value, nombre=str(result.get("nombre") or ""), default_source=str(result.get("fuente") or fuente), default_status=statuses[0] if statuses else MedicationCaseStatus.OTRO, default_unit=str(result.get("unidad_cantidad") or ""), ) for value in item["evidencias"] if isinstance(value, dict) ] result["discrepancia_cantidad"] = bool(item.get("discrepancia_cantidad")) result["cantidad_clinica"] = parse_numeric_quantity(item.get("cantidad_clinica")) result["cantidad_facturada"] = parse_numeric_quantity(item.get("cantidad_facturada")) result["unidad_cantidad"] = _clean(item.get("unidad_cantidad")) result["asociacion_factura"] = _clean(item.get("asociacion_factura")) or "no_evaluable" result["candidatos_factura"] = [ str(value) for value in item.get("candidatos_factura") or [] if str(value).strip() ] if isinstance(item.get("pertinencia"), dict): result["pertinencia"] = dict(item["pertinencia"]) return result text = _clean(re.sub(r"<.*?>", "", str(item or ""))) if not text: return None parts = [_clean(part) for part in re.split(r"\s*[|–—-]\s*", text) if _clean(part)] misplaced = {"administrado", "formulado", "ordenado", "facturado"} code = "" quantity: int | float | None = None posology = "" if "|" in text and len(parts) >= 5: name = parts[0] posology = parts[1] dose = parts[2] quantity = parse_numeric_quantity(parts[3]) elif len(parts) >= 4 and re.fullmatch(r"[A-Z0-9]{2,12}", parts[0], re.I): code = parts[0] quantity_match = re.search(r"\d+(?:[.,]\d+)?", parts[1]) quantity = parse_numeric_quantity(quantity_match.group(0)) if quantity_match else None name = parts[2] dose = parts[3] else: name = next( ( part for part in parts if normalize_medication_key(part) not in misplaced and not re.fullmatch(r"\d+(?:[.,]\d+)?\s*\w*", part) and re.search(r"[A-Za-zÁÉÍÓÚÑáéíóúñ]", part) ), "", ) dose_match = re.search( r"\b\d+(?:[.,]\d+)?\s*(?:mcg|mg|gr|g|ml|cc|ui)\b", text, re.IGNORECASE, ) dose = dose_match.group(0) if dose_match else "" if len(parts) == 1 and dose_match: name = _clean(text[: dose_match.start()]) route_match = _ROUTE_PATTERN.search(text[dose_match.end() :]) route = route_match.group(0) if route_match else "" posology = _posology_from_text(text[dose_match.end() :], route) if not name: return None status_part = next((part for part in parts if normalize_medication_key(part) in misplaced), "") return _make_medication( nombre=name, dosis=dose, posologia=posology, cantidad=quantity, estado=_status(status_part), codigo_referencia=code, fuente=fuente, evidencia=text, ) def _identity(item: dict[str, Any]) -> tuple[str, str, str, str]: return ( normalize_medication_key(item.get("nombre")), normalize_medication_key(item.get("presentacion")), normalize_medication_key(item.get("dosis")), normalize_medication_key(item.get("posologia")), ) def consolidate_medication_items(items: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: merged: dict[tuple[str, str, str, str], dict[str, Any]] = {} for raw in items: item = medication_from_legacy(raw, fuente=str(raw.get("fuente") or "historia_clinica")) if not item or _NON_MEDICATION_PATTERN.search(str(item.get("nombre") or "")): continue identity = _identity(item) if not identity[0]: continue current = merged.get(identity) if current is None: merged[identity] = item continue current["estados"] = list(dict.fromkeys([*current.get("estados", []), *item.get("estados", [])])) current["fuentes"] = list(dict.fromkeys([*current.get("fuentes", []), *item.get("fuentes", [])])) current["evidencias"] = [*current.get("evidencias", []), *item.get("evidencias", [])] current["discrepancia_cantidad"] = bool( current.get("discrepancia_cantidad") or item.get("discrepancia_cantidad") ) if not current.get("pertinencia") and item.get("pertinencia"): current["pertinencia"] = item["pertinencia"] for field in ( "codigo_facturacion", "codigo_referencia", "presentacion", "dosis", "posologia", "cantidad_clinica", "cantidad_facturada", ): if not current.get(field) and item.get(field): current[field] = item[field] if current.get("cantidad") is None and item.get("cantidad") is not None: current["cantidad"] = item["cantidad"] def sort_key(item: dict[str, Any]) -> tuple[int, str, str, str]: states = [_status(value) for value in item.get("estados", [])] priority = min((_STATUS_PRIORITY.get(state, 99) for state in states), default=99) return (priority, normalize_medication_key(item.get("nombre")), normalize_medication_key(item.get("dosis")), normalize_medication_key(item.get("posologia"))) return [_with_legacy_projection(item) for item in sorted(merged.values(), key=sort_key)] _MATCH_NOISE_TOKENS = { "ampolla", "ampollas", "bolsa", "caja", "capsula", "capsulas", "frasco", "inyectable", "solucion", "tableta", "tabletas", "vial", } def _ingredient_signature(item: dict[str, Any]) -> str: material = normalize_medication_key(item.get("nombre")) tokens = [ token for token in material.split() if token not in _MATCH_NOISE_TOKENS and not re.fullmatch(r"\d+(?:\.\d+)?", token) and token not in {"cc", "g", "gr", "mcg", "mg", "ml"} ] return " ".join(tokens) def _concentration_signature(item: dict[str, Any]) -> str: material = " ".join( str(item.get(field) or "") for field in ("nombre", "presentacion", "dosis") ) matches = re.findall( r"\b\d+(?:[.,]\d+)?\s*(?:%|mcg|mg|g|gr|ml|cc)(?:\s*/\s*\d+(?:[.,]\d+)?\s*(?:ml|cc))?", material, re.IGNORECASE, ) return "|".join(normalize_medication_key(value) for value in matches) def _presentation_compatible(left: dict[str, Any], right: dict[str, Any]) -> bool: left_concentration = _concentration_signature(left) right_concentration = _concentration_signature(right) if left_concentration and right_concentration and left_concentration != right_concentration: return False left_presentation = normalize_medication_key(left.get("presentacion")) right_presentation = normalize_medication_key(right.get("presentacion")) return not left_presentation or not right_presentation or left_presentation == right_presentation def _invoice_candidates( invoice: dict[str, Any], clinical: list[dict[str, Any]], ) -> tuple[list[dict[str, Any]], MedicationInvoiceAssociationStatus]: reference = normalize_medication_key(invoice.get("codigo_referencia")) if reference: matches = [ item for item in clinical if normalize_medication_key(item.get("codigo_referencia")) == reference ] if matches: return matches, MedicationInvoiceAssociationStatus.EXACTA internal_code = normalize_medication_key(invoice.get("codigo_facturacion")) if internal_code: matches = [ item for item in clinical if normalize_medication_key(item.get("codigo_facturacion")) == internal_code ] if matches: return matches, MedicationInvoiceAssociationStatus.EXACTA invoice_name = normalize_medication_key(invoice.get("nombre")) exact_name = [ item for item in clinical if normalize_medication_key(item.get("nombre")) == invoice_name and _presentation_compatible(item, invoice) ] if exact_name: return exact_name, MedicationInvoiceAssociationStatus.INEQUIVOCA signature = _ingredient_signature(invoice) if not signature: return [], MedicationInvoiceAssociationStatus.SIN_COINCIDENCIA compatible = [ item for item in clinical if _ingredient_signature(item) == signature and _presentation_compatible(item, invoice) ] return compatible, ( MedicationInvoiceAssociationStatus.INEQUIVOCA if compatible else MedicationInvoiceAssociationStatus.SIN_COINCIDENCIA ) def _administered_quantity(item: dict[str, Any]) -> tuple[int | float | None, str]: quantities: list[int | float] = [] units: set[str] = set() for evidence in item.get("evidencias") or []: if not isinstance(evidence, dict) or _status(evidence.get("estado")) != MedicationCaseStatus.ADMINISTRADO: continue quantity = parse_numeric_quantity(evidence.get("cantidad")) if quantity is None: continue quantities.append(quantity) unit = normalize_medication_key(evidence.get("unidad")) if unit: units.add(unit) if not quantities or len(units) > 1: return None, "" return sum(quantities), next(iter(units), "") def _invoice_group_key(invoice: dict[str, Any]) -> tuple[str, str]: reference = normalize_medication_key(invoice.get("codigo_referencia")) internal_code = normalize_medication_key(invoice.get("codigo_facturacion")) return ( reference or internal_code, normalize_medication_key(invoice.get("nombre")), ) def reconcile_case_medications( clinical_items: Iterable[dict[str, Any]], invoice_items: Iterable[dict[str, Any]] ) -> list[dict[str, Any]]: clinical = consolidate_medication_items(clinical_items) invoices = [ medication_from_legacy( { **item, "nombre": item.get("medicamento") or item.get("descripcion"), "dosis": "", "tipo_uso": "facturado", "fuente": "factura", }, fuente="factura", ) for item in invoice_items if isinstance(item, dict) ] invoice_groups: dict[tuple[str, str], list[dict[str, Any]]] = {} for invoice in (item for item in invoices if item): invoice_groups.setdefault(_invoice_group_key(invoice), []).append(invoice) unmatched_invoices: list[dict[str, Any]] = [] for group in invoice_groups.values(): sample = group[0] candidates, association = _invoice_candidates(sample, clinical) total = sum((parse_numeric_quantity(item.get("cantidad")) or 0) for item in group) if len(candidates) == 1: target = candidates[0] clinical_quantity, clinical_unit = _administered_quantity(target) target["cantidad_clinica"] = clinical_quantity target["cantidad_facturada"] = total or None target["cantidad"] = total or target.get("cantidad") target["unidad_cantidad"] = target.get("unidad_cantidad") or clinical_unit target["discrepancia_cantidad"] = False target["asociacion_factura"] = association.value target["estados"] = list(dict.fromkeys([*target.get("estados", []), "facturado"])) target["fuentes"] = list(dict.fromkeys([*target.get("fuentes", []), "factura"])) target["codigo_referencia"] = target.get("codigo_referencia") or sample.get("codigo_referencia", "") target["codigo_facturacion"] = target.get("codigo_facturacion") or sample.get("codigo_facturacion", "") target["codigo_estado"] = "identificado" target["evidencias"] = [*target.get("evidencias", []), *(evidence for item in group for evidence in item.get("evidencias", []))] elif candidates: candidate_keys = [str(item.get("key") or "") for item in candidates] for target in candidates: target["fuentes"] = list(dict.fromkeys([*target.get("fuentes", []), "factura"])) target["asociacion_factura"] = MedicationInvoiceAssociationStatus.AMBIGUA.value target["candidatos_factura"] = candidate_keys target["evidencias"] = [ *target.get("evidencias", []), *(evidence for item in group for evidence in item.get("evidencias", [])), ] else: sample["cantidad"] = total or sample.get("cantidad") sample["cantidad_facturada"] = total or None sample["asociacion_factura"] = MedicationInvoiceAssociationStatus.SIN_COINCIDENCIA.value unmatched_invoices.append(sample) return consolidate_medication_items([*clinical, *unmatched_invoices])