from collections import defaultdict from datetime import datetime from typing import Optional # ── helpers de formato de fecha ────────────────────────────────────────────── def _fmt_fecha(val) -> str: if not val: return "" if hasattr(val, "strftime"): return val.strftime("%d/%m/%Y") s = str(val)[:10] parts = s.split("-") if len(parts) == 3: return f"{parts[2]}/{parts[1]}/{parts[0]}" return s def _fmt_datetime(val) -> str: if not val: return "" if hasattr(val, "strftime"): return val.strftime("%d/%m/%Y %H:%M:%S") # Eliminar espacios dentro de la hora ANTES de truncar ("06: 59: 28" → "06:59:28") s = str(val).replace(": ", ":").replace(" :", ":") s = s[:19].replace("T", " ") parts = s.split(" ") if len(parts) == 2: dp = parts[0].split("-") if len(dp) == 3: return f"{dp[2]}/{dp[1]}/{dp[0]} {parts[1]}" return s # ── Tercero/Crear ───────────────────────────────────────────────────────────── _TIPO_DOC_MAP = { "CC": "C", "TI": "T", "RC": "R", "CE": "E", "PA": "P", "AS": "A", "MS": "M", "NU": "U", "SC": "S", "PE": "PE", "PT": "PT", "SI": "A", "CN": "C", "DE": "E", "CD": "P", } _ZONA_MAP = {"U": "01", "R": "02"} def generar_tercero_api(row: dict) -> dict: nombres = (row.get("NOMBRES") or "").strip().upper().split() apellidos = (row.get("APELLIDOS") or "").strip().upper().split() nombre_completo = f"{row.get('NOMBRES', '')} {row.get('APELLIDOS', '')}".strip().upper() tipo_doc = _TIPO_DOC_MAP.get(str(row.get("TIPOIDENT") or "CC").strip(), "C") zona = _ZONA_MAP.get(str(row.get("TIPORES") or "U").strip(), "01") return { "codigo": str(row.get("CODIGO") or "").strip(), "natJuridica": "N", "tipoDocumento": tipo_doc, "nit": str(row.get("DOCIDENT") or "").strip(), "nombre": nombre_completo, "nombre1": nombres[0] if nombres else "", "nombre2": nombres[1] if len(nombres) > 1 else "", "apellido1": apellidos[0] if apellidos else "", "apellido2": apellidos[1] if len(apellidos) > 1 else "", "direccion": (row.get("DIRECCION") or "").strip(), "codigoCiudad": str(row.get("COD_CIUDAD") or "00").strip(), "nombreCiudad": (row.get("NOM_CIUDAD") or "SIN CIUDAD").strip().upper(), "zona1": "00", "clasificacion": "00", "nomRegTri": nombre_completo, "telefono": str(row.get("TELEFONOS") or "").strip() or "0000000", "email": (row.get("EMAIL") or "").strip() or f"{str(row.get('DOCIDENT') or 'paciente').strip()}@sinregistro.co", "inactivo": False, "privada": "N", "mixta": "N", "codigoBarrio": "00", "comision": 0, "fechaNacimiento": _fmt_fecha(row.get("F_NACIMIENTO")), "sexo": str(row.get("SEXO") or "M").strip(), "zona": zona, "etnia": str(row.get("CODETNIA") or "99").strip(), "cliente": "S", "enfermedadCronica": False, "noEnviarMinSalud": False, } # ── RdaPaciente/Insertar ────────────────────────────────────────────────────── def agrupar_por_recepcion(rows: list) -> dict: grupos = defaultdict(list) for row in rows: key = row.get("IDRECEPCION") grupos[key].append(dict(row)) return grupos def generar_rda_paciente(rows: list, default_profesional: str = "", default_especialidad: str = "") -> dict: if not rows: return {} h = rows[0] fecha = _fmt_fecha(h.get("FECHA_RECEPCION")) ingreso = _fmt_datetime(h.get("HORAINICIORECEPCION") or h.get("FECHA_RECEPCION")) # Egreso = última fecha de reporte válida entre los exámenes, o fecha de recepción ultima_fecha = max( (r.get("FECHA_REPORTADO") for r in rows if str(r.get("FECHA_REPORTADO") or "")[:4] != "1900"), default=None, ) egreso = _fmt_datetime(ultima_fecha or h.get("FECHA_RECEPCION")) detalle_pedido = [] for row in rows: fecha_real = _fmt_datetime( row.get("FECHA_REPORTADO") if str(row.get("FECHA_REPORTADO") or "")[:4] != "1900" else h.get("FECHA_RECEPCION") ) detalle_pedido.append({ "codigoMaterial": str(row.get("COD_EXAMEN") or "").strip(), "codigoBodega": "00", "cantidad": 1, "tipoUnidad": "D", "valor": float(row.get("PRECIO") or 0), "descuento": 0, "porcentajeIva": 0, "impConsumo": 0, "observacion": "", "profesional": default_profesional or None, "especialidad": default_especialidad or None, "diagnosticoprincipal": str(h.get("DIAG_PPAL") or "").strip(), "fechaHoraRealizacion": fecha_real or egreso, }) autorizacion = str(h.get("AUTORIZACION") or "").strip() or None nit_empresa = str(h.get("NIT_EMPRESA") or "").strip() es_particular = nit_empresa.upper() in ("PART", "PARTICULAR", "", "0") cod_contrato = str(h.get("CODCONTRATO") or "").strip() or None # Códigos configurados en TNS para este laboratorio (GET /v2/tablas/FormaPago/ObtenerFormasDePago) cod_forma_pago = "CLIP" if es_particular else "INST" # tipousuario: usar TIPOUSUSISPRO; si es NULL derivar de TIPOUSU + NIT_EMPRESA _TIPOUSU_MAP = {"1": "11", "5": "07"} # EPS→11, Póliza→07 tipoususispro = str(h.get("TIPOUSUSISPRO") or "").strip() if not tipoususispro: tipousu = str(h.get("TIPOUSU") or "").strip() tipoususispro = _TIPOUSU_MAP.get(tipousu, "12" if es_particular else "11") # viaIngreso y modalidadAtencion no existen en RECEPCION → valores fijos de laboratorio via_ingreso = "01" # Demanda espontánea (pacientes llegan directo al lab) modalidad = "01" # Intramural (laboratorio en sede fija) return { "codigoPrefijo": str(h.get("PREFIJO") or "00").strip(), "numero": str(h.get("NUM_FACTURA") or "").strip(), "fecha": fecha, "codTercero": str(h.get("COD_PACIENTE") or "").strip(), "codVendedor": "00", "codFormaPago": cod_forma_pago, "codBanco": "00", "codigoCentroCosto": "00", "tipoIngreso": str(h.get("CLASEPROC") or "1").strip() or "1", "fechaHoraIngreso": ingreso, "fechaHoraEgreso": egreso, "modalidadAtencion": modalidad, "numeroContrato": cod_contrato, "diagnosticoprincipal": str(h.get("DIAG_PPAL") or "").strip(), "tipousuario": tipoususispro, "viaIngreso": via_ingreso, "esTerapia": False, "esProcedimiento": False, "numeroAutorizacion": autorizacion, "detallePedido": detalle_pedido, "detalleFormaPago": [{ "codigoFormaPago": cod_forma_pago, "plazoDias": "0" if es_particular else "30", "fechaVencimiento": fecha, "valor": str(int(float(h.get("VALORTOTAL") or 0))), }], } # ── funciones legacy RIPS 2.0 (se mantienen) ───────────────────────────────── def generar_terceros(row: dict) -> dict: return { "tipoDocumentoIdentificacion": row.get("tipo_documento", "CC"), "numDocumentoIdentificacion": row.get("numero_documento", ""), "primerNombre": (row.get("primer_nombre") or "").upper(), "segundoNombre": (row.get("segundo_nombre") or "").upper(), "primerApellido": (row.get("primer_apellido") or "").upper(), "segundoApellido": (row.get("segundo_apellido") or "").upper(), "fechaNacimiento": str(row.get("fecha_nacimiento", ""))[:10], "codSexo": row.get("cod_sexo", "M"), "codEntidadAdministradora": row.get("cod_entidad", ""), "tipoUsuario": row.get("tipo_usuario", "01"), "codPaisResidencia": row.get("cod_pais", "170"), "codMunicipioResidencia": row.get("cod_municipio", ""), "codZonaTerritorialResidencia": row.get("cod_zona", "01"), "incapacidad": row.get("incapacidad", "NO"), "codPaisOrigen": row.get("cod_pais_origen", "170"), "direccionResidencia": row.get("direccion", ""), "codZonaResidencia": row.get("cod_zona", "01"), } def generar_procedimiento(row: dict, consecutivo: int) -> dict: return { "consecutivo": consecutivo, "codProcedimiento": row.get("cod_procedimiento", ""), "fechaInicioAtencion": str(row.get("fecha_atencion", datetime.now().strftime("%Y-%m-%d %H:%M")))[:16], "codDiagnosticoPrincipal": row.get("cod_diagnostico", "R790"), "codDiagnosticoRelacionado": row.get("cod_diagnostico_rel", None), "finalidadTecnologiaSalud": row.get("finalidad", "23"), "viaIngresoServicioSalud": row.get("via_ingreso", "02"), "modalidadGrupoServicioTecSal": row.get("modalidad", "01"), "grupoServicios": row.get("grupo_servicio", "02"), "codServicio": int(row.get("cod_servicio", 706)), "codPrestador": row.get("cod_prestador", ""), "tipoDocumentoIdentificacion": row.get("tipo_doc_profesional", "CC"), "numDocumentoIdentificacion": row.get("num_doc_profesional", ""), "vrServicio": int(row.get("vr_servicio", 0)), "valorPagoModerador": int(row.get("valor_pago_moderador", 0)), "conceptoRecaudo": row.get("concepto_recaudo", "05"), "numAutorizacion": row.get("num_autorizacion", None), "idMIPRES": row.get("id_mipres", None), "codComplicacion": row.get("cod_complicacion", None), "numFEVPagoModerador": row.get("num_fev_pago_moderador", None), } def agrupar_por_factura(rows: list, factura_default: str = "") -> dict: grupos = defaultdict(lambda: {"factura": "", "procedimientos": []}) for row in rows: doc_key = (row.get("tipo_doc_paciente", "CC"), row.get("num_doc_paciente", "")) fact = row.get("num_factura", factura_default) grupos[(fact, doc_key)]["factura"] = fact grupos[(fact, doc_key)]["procedimientos"].append(dict(row)) return grupos def generar_transaccion( factura: str, num_doc_obligado: str, rows_tercero: Optional[dict], rows_procedimientos: list, ) -> dict: transaccion = { "numDocumentoIdObligado": num_doc_obligado, "numFactura": factura, "tipoNota": None, "numNota": None, "usuarios": [], } if rows_tercero and rows_procedimientos: usuario = { "tipoDocumentoIdentificacion": rows_tercero.get("tipoDocumentoIdentificacion", "CC"), "numDocumentoIdentificacion": rows_tercero.get("numDocumentoIdentificacion", ""), "codEntidadAdministradora": rows_tercero.get("codEntidadAdministradora", ""), "tipoUsuario": rows_tercero.get("tipoUsuario", "01"), "fechaNacimiento": rows_tercero.get("fechaNacimiento", ""), "codSexo": rows_tercero.get("codSexo", "M"), "codPaisResidencia": rows_tercero.get("codPaisResidencia", "170"), "codMunicipioResidencia": rows_tercero.get("codMunicipioResidencia", ""), "codZonaTerritorialResidencia": rows_tercero.get("codZonaTerritorialResidencia", "01"), "incapacidad": rows_tercero.get("incapacidad", "NO"), "consecutivo": 1, "codPaisOrigen": rows_tercero.get("codPaisOrigen", "170"), "servicios": { "procedimientos": [ generar_procedimiento(p, i + 1) for i, p in enumerate(rows_procedimientos) ] }, } transaccion["usuarios"].append(usuario) return transaccion