Lizandro GuarnizoandClaude Sonnet 4.6
8732db9c1f
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
331 lines
13 KiB
Python
331 lines
13 KiB
Python
import re as _re
|
|
from collections import defaultdict
|
|
from datetime import datetime, date
|
|
from typing import Optional
|
|
|
|
|
|
# ── helpers de formato de fecha ──────────────────────────────────────────────
|
|
|
|
def _clean_dt(s: str) -> str:
|
|
return _re.sub(r'\s*:\s*', ':', s)
|
|
|
|
|
|
def _clean_times(obj):
|
|
"""Limpia recursivamente espacios alrededor de ':' en todos los strings del dict."""
|
|
if isinstance(obj, str):
|
|
return _re.sub(r'\s*:\s*', ':', obj) if ':' in obj else obj
|
|
if isinstance(obj, dict):
|
|
return {k: _clean_times(v) for k, v in obj.items()}
|
|
if isinstance(obj, list):
|
|
return [_clean_times(i) for i in obj]
|
|
return obj
|
|
|
|
|
|
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 ""
|
|
s = _clean_dt(str(val))
|
|
s = s[:19].replace("T", " ")
|
|
parts = s.strip().split()
|
|
if len(parts) == 2:
|
|
dp = parts[0].split("-")
|
|
if len(dp) == 3:
|
|
return f"{dp[2]}/{dp[1]}/{dp[0]} {parts[1]}"
|
|
return f"{parts[0]} {parts[1]}"
|
|
if len(parts) == 1:
|
|
dp = parts[0].split("-")
|
|
if len(dp) == 3:
|
|
# ponytail: solo fecha ISO → agregar hora 00:00:00
|
|
return f"{dp[2]}/{dp[1]}/{dp[0]} 00:00:00"
|
|
return f"{parts[0]} 00:00:00"
|
|
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")
|
|
|
|
sexo = str(row.get("SEXO") or "M").strip().upper()
|
|
identidad_genero = "01" if sexo == "M" else "02"
|
|
cod_ciudad = str(row.get("COD_CIUDAD") or "54001").strip() or "54001"
|
|
|
|
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": cod_ciudad,
|
|
"nombreCiudad": (row.get("NOM_CIUDAD") or "CUCUTA").strip().upper(),
|
|
"divisionPoliticaAdministrativa": cod_ciudad,
|
|
"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": sexo,
|
|
"identidadGenero": identidad_genero,
|
|
"zona": zona,
|
|
"etnia": str(row.get("CODETNIA") or "99").strip(),
|
|
"antecedentes": "",
|
|
"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 = "",
|
|
default_remisionante: str = "00", default_prefijo: str = "00",
|
|
numero_override: str = "") -> dict:
|
|
if not rows:
|
|
return {}
|
|
h = rows[0]
|
|
|
|
fecha = _fmt_fecha(h.get("FECHA_RECEPCION"))
|
|
_fecha_iso = str(h.get("FECHA_RECEPCION") or "")[:10]
|
|
_hora_raw = _clean_dt(str(h.get("HORAINICIORECEPCION") or "")).strip()
|
|
# ponytail: HORAINICIORECEPCION es campo TIME (sin fecha) → combinar con la fecha de FECHA_RECEPCION
|
|
if _hora_raw and len(_hora_raw) <= 8 and ":" in _hora_raw and _fecha_iso:
|
|
ingreso = _fmt_datetime(f"{_fecha_iso}T{_hora_raw}")
|
|
else:
|
|
ingreso = _fmt_datetime(_hora_raw or _fecha_iso)
|
|
recepcion_dt = _fmt_datetime(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")
|
|
)
|
|
_doc = str(row.get("DOCIDMEDICO") or "").strip()
|
|
cedula_medico = _re.sub(r"[^0-9]", "", _doc)
|
|
detalle_pedido.append({
|
|
"codigoMaterial": str(row.get("COD_EXAMEN") or "").strip(),
|
|
"codigoBodega": "00",
|
|
"cantidad": 1,
|
|
"tipoUnidad": "D",
|
|
"descuento": float(row.get("DESCUENTO") or 0),
|
|
"porcentajeIva": 0,
|
|
"impConsumo": 0,
|
|
"observacion": "",
|
|
"profesional": default_profesional or None,
|
|
"especialidad": default_especialidad or None,
|
|
"profesionalRemisionante": cedula_medico or default_remisionante or None,
|
|
"diagnosticoprincipal": str(h.get("DIAG_PPAL") or "").strip(),
|
|
"fechaHoraRealizacion": fecha_real or recepcion_dt,
|
|
})
|
|
|
|
# Egreso = fechaHoraRealizacion del último examen en el detalle
|
|
egreso = max(
|
|
(d["fechaHoraRealizacion"] for d in detalle_pedido if d["fechaHoraRealizacion"]),
|
|
default=recepcion_dt,
|
|
)
|
|
|
|
nit_empresa = str(h.get("NIT_EMPRESA") or "").strip()
|
|
cod_contrato_raw = str(h.get("CODCONTRATO") or "").strip()
|
|
# Particular: por NIT_EMPRESA "PART" o por contrato "12"/"012"
|
|
es_particular = (
|
|
nit_empresa.upper() in ("PART", "PARTICULAR", "", "0")
|
|
or cod_contrato_raw.lstrip("0") == "12"
|
|
)
|
|
cod_contrato = cod_contrato_raw or ("012" if es_particular else None)
|
|
cod_forma_pago = "CLIP" if es_particular else "INST"
|
|
autorizacion = None if es_particular else (str(h.get("AUTORIZACION") or "").strip() or None)
|
|
|
|
# tipousuario: particular siempre "12"; para convenios usar TIPOUSUSISPRO o derivar
|
|
_TIPOUSU_MAP = {"1": "11", "5": "07"} # EPS→11, Póliza→07
|
|
if es_particular:
|
|
tipoususispro = "12"
|
|
else:
|
|
tipoususispro = str(h.get("TIPOUSUSISPRO") or "").strip()
|
|
if not tipoususispro:
|
|
tipousu = str(h.get("TIPOUSU") or "").strip()
|
|
tipoususispro = _TIPOUSU_MAP.get(tipousu, "11")
|
|
|
|
via_ingreso = "01"
|
|
modalidad = "01"
|
|
|
|
vigencia = 30
|
|
|
|
result = {
|
|
"codigoPrefijo": str(h.get("PREFIJO") or "").strip() or default_prefijo or "00",
|
|
"numero": numero_override if numero_override else (
|
|
str(h.get("NUM_FACTURA") or "").strip() if (h.get("NUM_FACTURA") or 0) != 0 else ""
|
|
),
|
|
"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,
|
|
"fechaAutorizacion": date.today().strftime("%d/%m/%Y"),
|
|
"vigenciaAutorizacion": vigencia,
|
|
"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))),
|
|
}],
|
|
}
|
|
return _clean_times(result)
|
|
|
|
|
|
# ── 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
|