from __future__ import annotations import re import unicodedata from datetime import datetime from typing import Any from app.rips.application.errors import RipsGenerationError from app.rips.application.models import BuiltRipsPayload, RipsFieldTrace, RipsGenerationOptions from app.rips.domain.models import ( RipsHospitalizacion, RipsMedicamento, RipsOtroServicio, RipsPayload, RipsProcedimiento, RipsServiciosUsuario, RipsUsuario, ) from app.rips.domain.value_objects import ( CodigoCIE10, CodigoCUPS, CodigoPrestador, CodigoTecnologiaSalud, DocumentoIdentificacion, FechaHoraRips, FechaRips, ) _DOCUMENT_PATTERN = re.compile(r"\b(CC|CE|TI|RC|PA|MS|AS|CD|SC|PE|PT|DE|SI|CN|NIT)\b\D*([A-Z0-9\-\.]+)", re.IGNORECASE) class _TraceCollector: def __init__(self) -> None: self._items: list[RipsFieldTrace] = [] def add(self, *, field_path: str, source_path: str, transform: str) -> None: self._items.append( RipsFieldTrace( field_path=field_path, source_path=source_path, transform=transform, ) ) def items(self) -> list[RipsFieldTrace]: return list(self._items) def _clean_text(value: Any) -> str: return str(value or "").strip() def _normalize_key(value: Any) -> str: text = unicodedata.normalize("NFKD", _clean_text(value)).encode("ascii", "ignore").decode("ascii") text = re.sub(r"[^a-z0-9]+", "-", text.lower()) return text.strip("-") def _digits(value: Any) -> str: return re.sub(r"\D", "", _clean_text(value)) def _parse_int(value: Any) -> int | None: text = _clean_text(value) if not text: return None digits = re.sub(r"[^\d\-]", "", text) if not digits: return None try: return int(digits) except ValueError: return None def _parse_float(value: Any) -> float | None: text = _clean_text(value) if not text: return None normalized = text.replace(",", ".") normalized = re.sub(r"[^0-9.\-]", "", normalized) if not normalized: return None try: return float(normalized) except ValueError: return None def _normalize_date(value: Any) -> str | None: text = _clean_text(value) if not text: return None for fmt in ("%Y-%m-%d", "%d/%m/%Y", "%Y/%m/%d", "%d-%m-%Y"): try: return datetime.strptime(text, fmt).strftime("%Y-%m-%d") except ValueError: continue match = re.search(r"(\d{4})-(\d{2})-(\d{2})", text) if match: return f"{match.group(1)}-{match.group(2)}-{match.group(3)}" return None def _normalize_datetime(value: Any) -> str | None: text = _clean_text(value) if not text: return None for fmt in ( "%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S", "%d/%m/%Y %H:%M", "%d/%m/%Y %H:%M:%S", "%Y-%m-%dT%H:%M", "%Y-%m-%dT%H:%M:%S", ): try: return datetime.strptime(text, fmt).strftime("%Y-%m-%d %H:%M") except ValueError: continue date_only = _normalize_date(text) if date_only: return f"{date_only} 00:00" return None def _resolve_total_value(item: dict[str, Any], *, total_key: str = "total") -> int | None: return _parse_int(item.get(total_key)) def _resolve_unit_value(item: dict[str, Any], *, fallback_from_total: bool = False) -> int | None: unit_value = _parse_int(item.get("valor_unitario") or item.get("tarifa")) if unit_value is not None: return unit_value if fallback_from_total: return _resolve_total_value(item) return None def _resolve_quantity_value(item: dict[str, Any], *, default: float | None = None) -> float | None: quantity = _parse_float(item.get("cantidad") or item.get("dias")) if quantity is not None: return quantity return default def _collect_mapping_errors(errors: list[str]) -> None: if errors: raise RipsGenerationError(errors) def _resolve_factura(context: dict[str, Any]) -> dict[str, Any]: factura = context.get("factura") if not isinstance(factura, dict): raise RipsGenerationError(["El contexto de epicrisis no contiene factura estructurada."]) return factura def _resolve_factura_json(context: dict[str, Any]) -> dict[str, Any]: factura = _resolve_factura(context) factura_json = factura.get("factura_json") if not isinstance(factura_json, dict): raise RipsGenerationError(["La factura del caso no contiene factura_json utilizable para RIPS."]) return factura_json def _resolve_documento_identificacion( context: dict[str, Any], options: RipsGenerationOptions, trace: _TraceCollector, ) -> DocumentoIdentificacion: explicit_type = options.user.tipo_documento_identificacion explicit_number = options.user.num_documento_identificacion if explicit_type and explicit_number: trace.add( field_path="usuarios[0].tipoDocumentoIdentificacion", source_path="options.user.tipo_documento_identificacion", transform="mapped to RIPS field after DocumentoIdentificacion validation.", ) trace.add( field_path="usuarios[0].numDocumentoIdentificacion", source_path="options.user.num_documento_identificacion", transform="mapped to RIPS field after DocumentoIdentificacion normalization.", ) return DocumentoIdentificacion(tipo=explicit_type, numero=explicit_number) factura_json = _resolve_factura_json(context) info_paciente = factura_json.get("informacion_paciente") or {} metadatos = context.get("metadatos_hc") or {} candidate_pairs = [ f"{explicit_type or ''} {explicit_number or ''}", metadatos.get("datos_identificacion_paciente"), info_paciente.get("numero_identificacion"), ] for candidate in candidate_pairs: match = _DOCUMENT_PATTERN.search(_clean_text(candidate).upper()) if match: source_path = ( "metadatos_hc.datos_identificacion_paciente" if candidate == metadatos.get("datos_identificacion_paciente") else "factura.factura_json.informacion_paciente.numero_identificacion" ) trace.add( field_path="usuarios[0].tipoDocumentoIdentificacion", source_path=source_path, transform="extracted from source text and mapped to RIPS document type.", ) trace.add( field_path="usuarios[0].numDocumentoIdentificacion", source_path=source_path, transform="extracted from source text and normalized to RIPS document number.", ) return DocumentoIdentificacion(tipo=match.group(1), numero=match.group(2)) if explicit_number: trace.add( field_path="usuarios[0].tipoDocumentoIdentificacion", source_path="options.user.tipo_documento_identificacion", transform="defaulted to CC when only document number was explicitly provided.", ) trace.add( field_path="usuarios[0].numDocumentoIdentificacion", source_path="options.user.num_documento_identificacion", transform="mapped to RIPS field after DocumentoIdentificacion normalization.", ) return DocumentoIdentificacion(tipo=explicit_type or "CC", numero=explicit_number) raise RipsGenerationError( ["No fue posible resolver tipo y número de documento del usuario para el RIPS."] ) def _resolve_provider_code(context: dict[str, Any], trace: _TraceCollector) -> CodigoPrestador: factura_json = _resolve_factura_json(context) proveedor = factura_json.get("proveedor") or {} provider_code = _clean_text(proveedor.get("nit")) if not _clean_text(provider_code): raise RipsGenerationError(["La factura no contiene NIT del proveedor para numDocumentoIdObligado."]) trace.add( field_path="numDocumentoIdObligado", source_path="factura.factura_json.proveedor.nit", transform="normalized digits and mapped to RIPS root field.", ) return CodigoPrestador(value=provider_code) def _resolve_invoice_number(context: dict[str, Any], trace: _TraceCollector) -> str: factura_json = _resolve_factura_json(context) invoice_info = factura_json.get("informacion_factura") or {} prefijo = _clean_text(invoice_info.get("prefijo")) numero = _clean_text(invoice_info.get("numero_factura")) if prefijo and numero and not numero.startswith(prefijo): trace.add( field_path="numFactura", source_path="factura.factura_json.informacion_factura.prefijo + numero_factura", transform="concatenated prefijo and numero_factura to build the RIPS invoice number.", ) return f"{prefijo}{numero}" if numero: trace.add( field_path="numFactura", source_path="factura.factura_json.informacion_factura.numero_factura", transform="mapped invoice number directly to RIPS root field.", ) return numero raise RipsGenerationError(["La factura no contiene número de factura para el RIPS."]) def _resolve_primary_diagnosis(context: dict[str, Any], trace: _TraceCollector) -> CodigoCIE10: diagnosticos = context.get("diagnosticos_consolidados") or [] for item in diagnosticos: if isinstance(item, dict) and _clean_text(item.get("codigo")): trace.add( field_path="usuarios[0].servicios.*.codDiagnosticoPrincipal", source_path="diagnosticos_consolidados[].codigo", transform="selected first consolidated diagnosis and normalized it as CodigoCIE10.", ) return CodigoCIE10(value=item["codigo"]) for doc_key in ("historia", "quirurgico"): doc = context.get(doc_key) or {} for item in doc.get("codigos_cie10") or []: if isinstance(item, dict) and _clean_text(item.get("codigo")): trace.add( field_path="usuarios[0].servicios.*.codDiagnosticoPrincipal", source_path=f"{doc_key}.codigos_cie10[].codigo", transform="selected first available diagnosis and normalized it as CodigoCIE10.", ) return CodigoCIE10(value=item["codigo"]) raise RipsGenerationError(["No se encontró diagnóstico principal CIE-10 para construir el RIPS."]) def _resolve_birth_date(context: dict[str, Any], options: RipsGenerationOptions, trace: _TraceCollector) -> str: metadatos = context.get("metadatos_hc") or {} birth_date = _normalize_date(options.user.fecha_nacimiento or metadatos.get("fecha_nacimiento")) if not birth_date: raise RipsGenerationError(["No se encontró fecha de nacimiento válida para el usuario RIPS."]) trace.add( field_path="usuarios[0].fechaNacimiento", source_path=( "options.user.fecha_nacimiento" if options.user.fecha_nacimiento else "metadatos_hc.fecha_nacimiento" ), transform="normalized source date to the RIPS YYYY-MM-DD format.", ) return FechaRips(value=birth_date).value def _resolve_sex(context: dict[str, Any], options: RipsGenerationOptions, trace: _TraceCollector) -> str: metadatos = context.get("metadatos_hc") or {} value = _clean_text(options.user.cod_sexo or metadatos.get("sexo")).upper() aliases = {"MASCULINO": "M", "FEMENINO": "F"} normalized = aliases.get(value, value) if normalized not in {"M", "F", "I"}: raise RipsGenerationError(["No se encontró codSexo válido para el usuario RIPS."]) trace.add( field_path="usuarios[0].codSexo", source_path="options.user.cod_sexo" if options.user.cod_sexo else "metadatos_hc.sexo", transform="normalized source sex value to the RIPS codSexo domain.", ) return normalized def _resolve_attention_datetime(context: dict[str, Any], trace: _TraceCollector) -> str: factura_json = _resolve_factura_json(context) info_paciente = factura_json.get("informacion_paciente") or {} metadatos = context.get("metadatos_hc") or {} resolved = _normalize_datetime(info_paciente.get("fecha_ingreso") or metadatos.get("fecha_ingreso")) if not resolved: raise RipsGenerationError(["No se encontró fecha de ingreso válida para los servicios del RIPS."]) trace.add( field_path="usuarios[0].servicios.*.fechaInicioAtencion|fechaDispensAdmon", source_path=( "factura.factura_json.informacion_paciente.fecha_ingreso" if info_paciente.get("fecha_ingreso") else "metadatos_hc.fecha_ingreso" ), transform="normalized source datetime to the RIPS YYYY-MM-DD HH:MM format.", ) return FechaHoraRips(value=resolved).value def _resolve_discharge_datetime(context: dict[str, Any], trace: _TraceCollector) -> str | None: factura_json = _resolve_factura_json(context) info_paciente = factura_json.get("informacion_paciente") or {} resolved = _normalize_datetime(info_paciente.get("fecha_egreso")) if not resolved: return None trace.add( field_path="usuarios[0].servicios.hospitalizacion[0].fechaEgreso", source_path="factura.factura_json.informacion_paciente.fecha_egreso", transform="normalized the discharge datetime to the RIPS hospitalizacion format.", ) return FechaHoraRips(value=resolved).value def _resolve_authorization(context: dict[str, Any], configured_value: str | None) -> str | None: if _clean_text(configured_value): return _clean_text(configured_value) factura_json = _resolve_factura_json(context) pagador = factura_json.get("pagador") or {} value = _clean_text(pagador.get("numero_autorizacion")) return value or None def _build_procedimientos( *, context: dict[str, Any], options: RipsGenerationOptions, provider_code: CodigoPrestador, patient_document: DocumentoIdentificacion, primary_diagnosis: CodigoCIE10, attention_datetime: str, trace: _TraceCollector, ) -> list[RipsProcedimiento]: factura_json = _resolve_factura_json(context) servicios = factura_json.get("servicios_procedimientos") or {} procedimientos_raw = servicios.get("procedimientos_quirurgicos") or [] procedimientos: list[RipsProcedimiento] = [] errors: list[str] = [] authorization = _resolve_authorization(context, options.procedure.num_autorizacion) for index, item in enumerate(procedimientos_raw, start=1): if not isinstance(item, dict): continue code = _clean_text(item.get("codigo_cups")) total = _parse_int(item.get("total")) if not code: errors.append(f"Procedimiento #{index}: falta codigo_cups en la factura.") continue if total is None: errors.append(f"Procedimiento #{index}: no fue posible interpretar el valor total.") continue procedimientos.append( RipsProcedimiento( codPrestador=provider_code.value, fechaInicioAtencion=attention_datetime, idMIPRES=options.procedure.id_mipres, numAutorizacion=authorization, codProcedimiento=CodigoCUPS(value=code).value, viaIngresoServicioSalud=options.procedure.via_ingreso_servicio_salud, modalidadGrupoServicioTecSal=options.procedure.modalidad_grupo_servicio_tec_sal, grupoServicios=options.procedure.grupo_servicios, codServicio=options.procedure.cod_servicio, finalidadTecnologiaSalud=options.procedure.finalidad_tecnologia_salud, tipoDocumentoIdentificacion=patient_document.tipo, numDocumentoIdentificacion=patient_document.numero, codDiagnosticoPrincipal=primary_diagnosis.value, codDiagnosticoRelacionado=options.procedure.cod_diagnostico_relacionado or primary_diagnosis.value, codComplicacion=options.procedure.cod_complicacion or primary_diagnosis.value, vrServicio=total, conceptoRecaudo=options.procedure.concepto_recaudo, valorPagoModerador=options.procedure.valor_pago_moderador, numFEVPagoModerador=options.procedure.num_fev_pago_moderador, consecutivo=index, ) ) base_path = f"usuarios[0].servicios.procedimientos[{len(procedimientos) - 1}]" trace.add( field_path=f"{base_path}.codPrestador", source_path="factura.factura_json.proveedor.nit", transform="normalized provider NIT and mapped it to the RIPS procedure provider code.", ) trace.add( field_path=f"{base_path}.fechaInicioAtencion", source_path="factura.factura_json.informacion_paciente.fecha_ingreso|metadatos_hc.fecha_ingreso", transform="normalized the attention date to the RIPS procedure datetime format.", ) trace.add( field_path=f"{base_path}.codProcedimiento", source_path=f"factura.factura_json.servicios_procedimientos.procedimientos_quirurgicos[{index - 1}].codigo_cups", transform="validated and mapped the source code to the RIPS CUPS procedure field.", ) trace.add( field_path=f"{base_path}.vrServicio", source_path=f"factura.factura_json.servicios_procedimientos.procedimientos_quirurgicos[{index - 1}].total", transform="parsed the billed total and mapped it to the RIPS procedure service value.", ) trace.add( field_path=f"{base_path}.codDiagnosticoPrincipal", source_path="diagnosticos_consolidados[].codigo|historia.codigos_cie10[].codigo|quirurgico.codigos_cie10[].codigo", transform="mapped the selected diagnosis to the RIPS procedure diagnosis fields.", ) trace.add( field_path=f"{base_path}.viaIngresoServicioSalud", source_path="options.procedure.via_ingreso_servicio_salud", transform="mapped configured administrative value to the RIPS procedure model.", ) _collect_mapping_errors(errors) return procedimientos def _build_medication_catalog(options: RipsGenerationOptions) -> dict[str, Any]: return {_normalize_key(item.match_key): item for item in options.medication_catalog} def _resolve_medication_name(item: dict[str, Any]) -> str: return _clean_text(item.get("medicamento") or item.get("nombre") or item.get("texto_original")) def _build_other_service_catalog(options: RipsGenerationOptions) -> dict[tuple[str, str], Any]: return { (item.source_section, _normalize_key(item.match_key)): item for item in options.other_service_catalog } def _build_other_service_defaults(options: RipsGenerationOptions) -> dict[str, Any]: return {item.source_section: item for item in options.other_service_defaults} def _resolve_other_service_name(source_section: str, item: dict[str, Any]) -> str: if source_section == "examenes_laboratorio": return _clean_text(item.get("prueba") or item.get("descripcion")) if source_section == "imagenologia": return _clean_text(item.get("estudio") or item.get("descripcion")) if source_section == "hospitalizacion": return _clean_text(item.get("habitacion") or item.get("descripcion")) if source_section == "honorarios_medicos": role = _clean_text(item.get("rol")) professional = _clean_text(item.get("profesional")) return _clean_text(" - ".join(value for value in (role, professional) if value)) if source_section == "otros_servicios": return _clean_text(item.get("descripcion") or item.get("concepto")) return "" def _resolve_other_service_match_key(source_section: str, item: dict[str, Any]) -> str: code_candidates = ( item.get("codigo_facturacion"), item.get("codigo_referencia"), item.get("codigo_cups"), item.get("codigo"), item.get("codigo_servicio"), ) for value in code_candidates: normalized = _normalize_key(value) if normalized: return normalized return _normalize_key(_resolve_other_service_name(source_section, item)) def _resolve_other_service_lookup_keys(source_section: str, item: dict[str, Any]) -> list[str]: keys: list[str] = [] primary_key = _resolve_other_service_match_key(source_section, item) if primary_key: keys.append(primary_key) name_key = _normalize_key(_resolve_other_service_name(source_section, item)) if name_key and name_key not in keys: keys.append(name_key) return keys def _resolve_other_service_source_code(item: dict[str, Any]) -> str | None: code_candidates = ( item.get("cod_tecnologia_salud"), item.get("codigo_tecnologia_salud"), item.get("codigo_facturacion"), item.get("codigo_referencia"), item.get("codigo_cups"), item.get("codigo_servicio"), item.get("codigo"), item.get("codigo_soat"), ) for value in code_candidates: normalized = _clean_text(value).upper() if normalized: return normalized return None def _resolve_other_service_tipo_os(source_section: str, item: dict[str, Any], catalog_entry: Any, defaults_entry: Any) -> str | None: return ( (catalog_entry.tipo_os if catalog_entry else None) or _clean_text(item.get("tipoOS") or item.get("tipo_os")).upper() or (defaults_entry.tipo_os if defaults_entry else None) or None ) def _resolve_other_service_code(item: dict[str, Any], catalog_entry: Any) -> str | None: return (catalog_entry.cod_tecnologia_salud if catalog_entry else None) or _resolve_other_service_source_code(item) def _resolve_other_service_field(item_override: Any, section_default: Any, attribute: str) -> Any: if item_override is not None: return item_override if section_default is not None: return getattr(section_default, attribute) return None def _build_medicamentos( *, context: dict[str, Any], options: RipsGenerationOptions, provider_code: CodigoPrestador, patient_document: DocumentoIdentificacion, primary_diagnosis: CodigoCIE10, attention_datetime: str, trace: _TraceCollector, ) -> list[RipsMedicamento]: factura_json = _resolve_factura_json(context) servicios = factura_json.get("servicios_procedimientos") or {} medicamentos_raw = servicios.get("medicamentos") or [] catalog = _build_medication_catalog(options) medicamentos: list[RipsMedicamento] = [] errors: list[str] = [] for index, item in enumerate(medicamentos_raw, start=1): if not isinstance(item, dict): continue medication_name = _resolve_medication_name(item) if not medication_name: errors.append(f"Medicamento #{index}: no fue posible resolver el nombre para mapearlo al RIPS.") continue catalog_entry = catalog.get(_normalize_key(item.get("codigo") or medication_name)) if catalog_entry is None: errors.append( f"Medicamento '{medication_name}': falta entrada en medication_catalog para completar el RIPS." ) continue quantity = _parse_float(item.get("cantidad")) unit_value = _parse_int(item.get("valor_unitario")) total_value = _parse_int(item.get("total")) if quantity is None or quantity <= 0: errors.append(f"Medicamento '{medication_name}': cantidad inválida.") continue if unit_value is None: errors.append(f"Medicamento '{medication_name}': valor_unitario inválido.") continue if total_value is None: errors.append(f"Medicamento '{medication_name}': total inválido.") continue prescriber_type = catalog_entry.tipo_documento_identificacion or patient_document.tipo prescriber_number = catalog_entry.num_documento_identificacion or patient_document.numero medicamentos.append( RipsMedicamento( codPrestador=provider_code.value, numAutorizacion=_resolve_authorization(context, catalog_entry.num_autorizacion), idMIPRES=catalog_entry.id_mipres, fechaDispensAdmon=attention_datetime, codDiagnosticoPrincipal=primary_diagnosis.value, codDiagnosticoRelacionado=primary_diagnosis.value, tipoMedicamento=catalog_entry.tipo_medicamento, codTecnologiaSalud=CodigoTecnologiaSalud(value=catalog_entry.cod_tecnologia_salud).value, nomTecnologiaSalud=medication_name, concentracionMedicamento=catalog_entry.concentracion_medicamento, unidadMedida=catalog_entry.unidad_medida, formaFarmaceutica=catalog_entry.forma_farmaceutica, unidadMinDispensa=catalog_entry.unidad_min_dispensa, cantidadMedicamento=quantity, diasTratamiento=catalog_entry.dias_tratamiento, tipoDocumentoIdentificacion=DocumentoIdentificacion( tipo=prescriber_type, numero=prescriber_number, ).tipo, numDocumentoIdentificacion=DocumentoIdentificacion( tipo=prescriber_type, numero=prescriber_number, ).numero, vrUnitMedicamento=unit_value, vrServicio=total_value, conceptoRecaudo=catalog_entry.concepto_recaudo, valorPagoModerador=catalog_entry.valor_pago_moderador, numFEVPagoModerador=catalog_entry.num_fev_pago_moderador, consecutivo=index, ) ) base_path = f"usuarios[0].servicios.medicamentos[{len(medicamentos) - 1}]" trace.add( field_path=f"{base_path}.codPrestador", source_path="factura.factura_json.proveedor.nit", transform="normalized provider NIT and mapped it to the RIPS medication provider code.", ) trace.add( field_path=f"{base_path}.fechaDispensAdmon", source_path="factura.factura_json.informacion_paciente.fecha_ingreso|metadatos_hc.fecha_ingreso", transform="normalized the source datetime to the RIPS medication datetime format.", ) trace.add( field_path=f"{base_path}.codTecnologiaSalud", source_path=f"options.medication_catalog[{index - 1}].cod_tecnologia_salud", transform="validated the catalog technology code and mapped it to the RIPS medication model.", ) trace.add( field_path=f"{base_path}.nomTecnologiaSalud", source_path=f"factura.factura_json.servicios_procedimientos.medicamentos[{index - 1}].medicamento|nombre|texto_original", transform="mapped the source medication description to the RIPS medication name.", ) trace.add( field_path=f"{base_path}.cantidadMedicamento", source_path=f"factura.factura_json.servicios_procedimientos.medicamentos[{index - 1}].cantidad", transform="parsed and mapped the billed quantity to the RIPS medication amount.", ) trace.add( field_path=f"{base_path}.vrServicio", source_path=f"factura.factura_json.servicios_procedimientos.medicamentos[{index - 1}].total", transform="parsed and mapped the billed total to the RIPS medication service value.", ) _collect_mapping_errors(errors) return medicamentos def _build_hospitalizacion( *, context: dict[str, Any], options: RipsGenerationOptions, provider_code: CodigoPrestador, primary_diagnosis: CodigoCIE10, attention_datetime: str, trace: _TraceCollector, ) -> list[RipsHospitalizacion]: factura_json = _resolve_factura_json(context) servicios = factura_json.get("servicios_procedimientos") or {} hospitalizacion_raw = [item for item in (servicios.get("hospitalizacion") or []) if isinstance(item, dict)] has_hospitalization_source = bool(hospitalizacion_raw) or _clean_text( (factura_json.get("informacion_paciente") or {}).get("fecha_egreso") ) if not has_hospitalization_source: return [] if options.hospitalization is None: raise RipsGenerationError( [ "El caso tiene señales de hospitalización facturada, pero falta options.hospitalization " "para construir el segmento hospitalizacion del RIPS." ] ) discharge_datetime = _resolve_discharge_datetime(context, trace) if not discharge_datetime: raise RipsGenerationError( ["No se encontró fecha_egreso válida para construir el segmento hospitalizacion del RIPS."] ) hospitalization = RipsHospitalizacion( codPrestador=provider_code.value, viaIngresoServicioSalud=options.hospitalization.via_ingreso_servicio_salud, fechaInicioAtencion=attention_datetime, numAutorizacion=_resolve_authorization(context, options.hospitalization.num_autorizacion), causaMotivoAtencion=options.hospitalization.causa_motivo_atencion, codDiagnosticoPrincipal=primary_diagnosis.value, codDiagnosticoPrincipalE=( CodigoCIE10( value=options.hospitalization.cod_diagnostico_principal_egreso or primary_diagnosis.value ).value ), codDiagnosticoRelacionadoE1=( CodigoCIE10(value=options.hospitalization.cod_diagnostico_relacionado_e1).value if options.hospitalization.cod_diagnostico_relacionado_e1 else None ), codDiagnosticoRelacionadoE2=( CodigoCIE10(value=options.hospitalization.cod_diagnostico_relacionado_e2).value if options.hospitalization.cod_diagnostico_relacionado_e2 else None ), codDiagnosticoRelacionadoE3=( CodigoCIE10(value=options.hospitalization.cod_diagnostico_relacionado_e3).value if options.hospitalization.cod_diagnostico_relacionado_e3 else None ), codComplicacion=( CodigoCIE10(value=options.hospitalization.cod_complicacion).value if options.hospitalization.cod_complicacion else None ), condicionDestinoUsuarioEgreso=options.hospitalization.condicion_destino_usuario_egreso, codDiagnosticoCausaMuerte=( CodigoCIE10(value=options.hospitalization.cod_diagnostico_causa_muerte).value if options.hospitalization.cod_diagnostico_causa_muerte else None ), fechaEgreso=discharge_datetime, consecutivo=1, ) base_path = "usuarios[0].servicios.hospitalizacion[0]" trace.add( field_path=f"{base_path}.codPrestador", source_path="factura.factura_json.proveedor.nit", transform="normalized provider NIT and mapped it to the RIPS hospitalizacion provider code.", ) trace.add( field_path=f"{base_path}.fechaInicioAtencion", source_path="factura.factura_json.informacion_paciente.fecha_ingreso|metadatos_hc.fecha_ingreso", transform="normalized the source admission datetime to the RIPS hospitalizacion model.", ) trace.add( field_path=f"{base_path}.codDiagnosticoPrincipal", source_path="diagnosticos_consolidados[].codigo|historia.codigos_cie10[].codigo|quirurgico.codigos_cie10[].codigo", transform="mapped the selected diagnosis to the RIPS hospitalizacion model.", ) trace.add( field_path=f"{base_path}.viaIngresoServicioSalud", source_path="options.hospitalization.via_ingreso_servicio_salud", transform="mapped configured administrative value to the RIPS hospitalizacion model.", ) trace.add( field_path=f"{base_path}.causaMotivoAtencion", source_path="options.hospitalization.causa_motivo_atencion", transform="mapped configured administrative value to the RIPS hospitalizacion model.", ) trace.add( field_path=f"{base_path}.condicionDestinoUsuarioEgreso", source_path="options.hospitalization.condicion_destino_usuario_egreso", transform="mapped configured administrative value to the RIPS hospitalizacion model.", ) return [hospitalization] def _build_otros_servicios( *, context: dict[str, Any], options: RipsGenerationOptions, provider_code: CodigoPrestador, patient_document: DocumentoIdentificacion, attention_datetime: str, trace: _TraceCollector, ) -> list[RipsOtroServicio]: factura_json = _resolve_factura_json(context) servicios = factura_json.get("servicios_procedimientos") or {} catalog = _build_other_service_catalog(options) defaults = _build_other_service_defaults(options) otros_servicios: list[RipsOtroServicio] = [] errors: list[str] = [] source_sections = ( "examenes_laboratorio", "imagenologia", "hospitalizacion", "honorarios_medicos", "otros_servicios", ) for source_section in source_sections: source_items = servicios.get(source_section) or [] for source_index, item in enumerate(source_items): if not isinstance(item, dict): continue item_name = _resolve_other_service_name(source_section, item) lookup_keys = _resolve_other_service_lookup_keys(source_section, item) match_key = lookup_keys[0] if lookup_keys else "" if not match_key: errors.append( f"{source_section} #{source_index + 1}: no fue posible resolver un match_key para otrosServicios." ) continue catalog_entry = next( (catalog.get((source_section, lookup_key)) for lookup_key in lookup_keys if catalog.get((source_section, lookup_key))), None, ) defaults_entry = defaults.get(source_section) resolved_tipo_os = _resolve_other_service_tipo_os(source_section, item, catalog_entry, defaults_entry) resolved_code = _resolve_other_service_code(item, catalog_entry) if not resolved_tipo_os: errors.append( f"{source_section} '{item_name or match_key}': no se pudo resolver tipoOS desde la fuente ni " "desde other_service_catalog/other_service_defaults." ) continue if not resolved_code: errors.append( f"{source_section} '{item_name or match_key}': no se pudo resolver codTecnologiaSalud desde " "la fuente ni desde other_service_catalog." ) continue quantity = _resolve_other_service_field( catalog_entry.cantidad_os if catalog_entry else None, None, "cantidad_os", ) or _resolve_quantity_value( item, default=1.0 if source_section in {"hospitalizacion", "honorarios_medicos"} else None, ) unit_value = _resolve_other_service_field( catalog_entry.vr_unit_os if catalog_entry else None, None, "vr_unit_os", ) or _resolve_unit_value( item, fallback_from_total=source_section == "hospitalizacion", ) total_value = _resolve_total_value(item) if quantity is None or quantity <= 0: errors.append(f"{source_section} '{item_name or match_key}': cantidadOS inválida.") continue if unit_value is None: errors.append(f"{source_section} '{item_name or match_key}': vrUnitOS inválido.") continue if total_value is None: errors.append(f"{source_section} '{item_name or match_key}': vrServicio inválido.") continue if resolved_tipo_os == "03": document_type = None document_number = None else: document_type = ( _resolve_other_service_field( catalog_entry.tipo_documento_identificacion if catalog_entry else None, defaults_entry, "tipo_documento_identificacion", ) or patient_document.tipo ) document_number = ( _resolve_other_service_field( catalog_entry.num_documento_identificacion if catalog_entry else None, defaults_entry, "num_documento_identificacion", ) or patient_document.numero ) other_service = RipsOtroServicio( codPrestador=provider_code.value, numAutorizacion=_resolve_authorization( context, _resolve_other_service_field( catalog_entry.num_autorizacion if catalog_entry else None, defaults_entry, "num_autorizacion", ), ), idMIPRES=_resolve_other_service_field( catalog_entry.id_mipres if catalog_entry else None, defaults_entry, "id_mipres", ), fechaSuministroTecnologia=attention_datetime, tipoOS=resolved_tipo_os, codTecnologiaSalud=CodigoTecnologiaSalud(value=resolved_code).value, nomTecnologiaSalud=(catalog_entry.nom_tecnologia_salud if catalog_entry else None) or item_name or match_key, cantidadOS=quantity, tipoDocumentoIdentificacion=( DocumentoIdentificacion(tipo=document_type, numero=document_number).tipo if document_type and document_number else None ), numDocumentoIdentificacion=( DocumentoIdentificacion(tipo=document_type, numero=document_number).numero if document_type and document_number else None ), vrUnitOS=unit_value, vrServicio=total_value, conceptoRecaudo=_resolve_other_service_field( catalog_entry.concepto_recaudo if catalog_entry else None, defaults_entry, "concepto_recaudo", ) or "02", valorPagoModerador=_resolve_other_service_field( catalog_entry.valor_pago_moderador if catalog_entry else None, defaults_entry, "valor_pago_moderador", ) or 0, numFEVPagoModerador=_resolve_other_service_field( catalog_entry.num_fev_pago_moderador if catalog_entry else None, defaults_entry, "num_fev_pago_moderador", ), consecutivo=len(otros_servicios) + 1, ) otros_servicios.append(other_service) base_path = f"usuarios[0].servicios.otrosServicios[{len(otros_servicios) - 1}]" trace.add( field_path=f"{base_path}.codPrestador", source_path="factura.factura_json.proveedor.nit", transform="normalized provider NIT and mapped it to the RIPS otrosServicios provider code.", ) trace.add( field_path=f"{base_path}.fechaSuministroTecnologia", source_path="factura.factura_json.informacion_paciente.fecha_ingreso|metadatos_hc.fecha_ingreso", transform="normalized the source datetime to the RIPS otrosServicios model.", ) trace.add( field_path=f"{base_path}.tipoOS", source_path=( f"options.other_service_catalog[{options.other_service_catalog.index(catalog_entry)}].tipo_os" if catalog_entry and catalog_entry in options.other_service_catalog and catalog_entry.tipo_os else ( f"factura.factura_json.servicios_procedimientos.{source_section}[{source_index}].tipoOS|tipo_os" if _clean_text(item.get('tipoOS') or item.get('tipo_os')) else ( f"options.other_service_defaults[{options.other_service_defaults.index(defaults_entry)}].tipo_os" if defaults_entry and defaults_entry in options.other_service_defaults else "derived" ) ) ), transform="resolved tipoOS from auditor override, source data, or section defaults and mapped it to the RIPS otrosServicios model.", ) trace.add( field_path=f"{base_path}.codTecnologiaSalud", source_path=( f"options.other_service_catalog[{options.other_service_catalog.index(catalog_entry)}].cod_tecnologia_salud" if catalog_entry and catalog_entry in options.other_service_catalog and catalog_entry.cod_tecnologia_salud else ( f"factura.factura_json.servicios_procedimientos.{source_section}[{source_index}]" ) ), transform="resolved codTecnologiaSalud from auditor override or source item code and mapped it to the RIPS otrosServicios model.", ) trace.add( field_path=f"{base_path}.nomTecnologiaSalud", source_path=f"factura.factura_json.servicios_procedimientos.{source_section}[{source_index}]", transform="mapped the source service description to the RIPS otrosServicios model.", ) trace.add( field_path=f"{base_path}.vrServicio", source_path=f"factura.factura_json.servicios_procedimientos.{source_section}[{source_index}].total", transform="parsed and mapped the billed total to the RIPS otrosServicios service value.", ) _collect_mapping_errors(errors) return otros_servicios def build_rips_payload( *, context: dict[str, Any], options: RipsGenerationOptions, ) -> BuiltRipsPayload: trace = _TraceCollector() provider_code = _resolve_provider_code(context, trace) patient_document = _resolve_documento_identificacion(context, options, trace) primary_diagnosis = _resolve_primary_diagnosis(context, trace) attention_datetime = _resolve_attention_datetime(context, trace) servicios = RipsServiciosUsuario( procedimientos=_build_procedimientos( context=context, options=options, provider_code=provider_code, patient_document=patient_document, primary_diagnosis=primary_diagnosis, attention_datetime=attention_datetime, trace=trace, ), medicamentos=_build_medicamentos( context=context, options=options, provider_code=provider_code, patient_document=patient_document, primary_diagnosis=primary_diagnosis, attention_datetime=attention_datetime, trace=trace, ), hospitalizacion=_build_hospitalizacion( context=context, options=options, provider_code=provider_code, primary_diagnosis=primary_diagnosis, attention_datetime=attention_datetime, trace=trace, ), otrosServicios=_build_otros_servicios( context=context, options=options, provider_code=provider_code, patient_document=patient_document, attention_datetime=attention_datetime, trace=trace, ), ) usuario = RipsUsuario( tipoDocumentoIdentificacion=patient_document.tipo, numDocumentoIdentificacion=patient_document.numero, tipoUsuario=options.user.tipo_usuario, fechaNacimiento=_resolve_birth_date(context, options, trace), codSexo=_resolve_sex(context, options, trace), codPaisResidencia=options.user.cod_pais_residencia, codMunicipioResidencia=options.user.cod_municipio_residencia, codZonaTerritorialResidencia=options.user.cod_zona_territorial_residencia, incapacidad=options.user.incapacidad, consecutivo=1, codPaisOrigen=options.user.cod_pais_origen, servicios=servicios, ) trace.add( field_path="usuarios[0].tipoUsuario", source_path="options.user.tipo_usuario", transform="mapped configured administrative value to the RIPS user model.", ) trace.add( field_path="usuarios[0].codPaisResidencia", source_path="options.user.cod_pais_residencia", transform="mapped configured administrative value to the RIPS user model.", ) trace.add( field_path="usuarios[0].codMunicipioResidencia", source_path="options.user.cod_municipio_residencia", transform="mapped configured administrative value to the RIPS user model.", ) trace.add( field_path="usuarios[0].codZonaTerritorialResidencia", source_path="options.user.cod_zona_territorial_residencia", transform="mapped configured administrative value to the RIPS user model.", ) trace.add( field_path="usuarios[0].incapacidad", source_path="options.user.incapacidad", transform="mapped configured administrative value to the RIPS user model.", ) payload = RipsPayload( numDocumentoIdObligado=provider_code.value, numFactura=_resolve_invoice_number(context, trace), tipoNota=options.transaction.tipo_nota, numNota=options.transaction.num_nota, usuarios=[usuario], ) if options.transaction.tipo_nota: trace.add( field_path="tipoNota", source_path="options.transaction.tipo_nota", transform="mapped configured note type to the RIPS transaction model.", ) if options.transaction.num_nota: trace.add( field_path="numNota", source_path="options.transaction.num_nota", transform="mapped configured note number to the RIPS transaction model.", ) return BuiltRipsPayload( payload=payload, field_trace_map=trace.items(), )