Lizandro GuarnizoandClaude Sonnet 4.6
2053d8a36f
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
418 lines
16 KiB
Python
418 lines
16 KiB
Python
import re as _re
|
|
from collections import defaultdict
|
|
from datetime import datetime, date, timedelta
|
|
from typing import Optional
|
|
# v1.5.0
|
|
|
|
|
|
# ── 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": "E", "PT": "E", "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": (lambda v: v if v and len(v) <= 2 else "99")(str(row.get("CODETNIA") or "").strip()),
|
|
"ciudadExp": cod_ciudad,
|
|
"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 agrupar_por_factura(rows: list) -> dict:
|
|
"""Agrupa filas por NUM_FACTURA — para ventas CMXC donde una factura puede tener varias recepciones."""
|
|
grupos = defaultdict(list)
|
|
for row in rows:
|
|
key = str(row.get("NUM_FACTURA") or "")
|
|
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 = "", contrato_map: dict = None,
|
|
prefijo_override: str = "", sin_contrato_set: set = None) -> 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("CUPS") or row.get("COD_EXAMEN") or "").strip(),
|
|
"codigoBodega": "00",
|
|
"cantidad": 1,
|
|
"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() or "Z017",
|
|
"fechaHoraRealizacion": fecha_real or recepcion_dt,
|
|
})
|
|
|
|
# Egreso = fechaHoraRealizacion del último examen + 5 minutos
|
|
_egreso_raw = max(
|
|
(d["fechaHoraRealizacion"] for d in detalle_pedido if d["fechaHoraRealizacion"]),
|
|
default=recepcion_dt,
|
|
)
|
|
try:
|
|
_egreso_dt = datetime.strptime(_egreso_raw, "%d/%m/%Y %H:%M:%S") + timedelta(minutes=5)
|
|
egreso = _egreso_dt.strftime("%d/%m/%Y %H:%M:%S")
|
|
except Exception:
|
|
egreso = _egreso_raw
|
|
|
|
nit_empresa = str(h.get("NIT_EMPRESA") or "").strip()
|
|
cod_contrato_raw = str(h.get("CODCONTRATO") or "").strip()
|
|
|
|
# Resolver tipo desde contrato_map primero (tiene prioridad sobre heurísticas)
|
|
_TIPOUSU_MAP = {"1": "11", "5": "07"} # EPS→11, Póliza→07
|
|
_nc_s = cod_contrato_raw.lstrip("0") or cod_contrato_raw
|
|
_map_tipo = ""
|
|
if contrato_map and cod_contrato_raw:
|
|
_map_tipo = contrato_map.get(cod_contrato_raw) or contrato_map.get(_nc_s) or ""
|
|
|
|
# es_particular: NIT "PART", o mapa dice "12", o heurística solo si no está en el mapa
|
|
es_particular = (
|
|
nit_empresa.upper() in ("PART", "PARTICULAR", "", "0")
|
|
or (_map_tipo == "12")
|
|
or (not _map_tipo and cod_contrato_raw.lstrip("0") == "12")
|
|
)
|
|
if sin_contrato_set and cod_contrato_raw and cod_contrato_raw in sin_contrato_set:
|
|
cod_contrato = None
|
|
else:
|
|
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: prioridad 1→mapa contratos, 2→particular, 3→TIPOUSUSISPRO/TIPOUSU
|
|
if _map_tipo:
|
|
tipoususispro = _map_tipo
|
|
elif 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
|
|
|
|
_total = float(h.get("VALORTOTAL") or 0)
|
|
_desc = float(h.get("VALORDESC") or 0)
|
|
descuento_pct = round(_desc * 100 / _total, 2) if _total > 0 else 0
|
|
|
|
result = {
|
|
"codigoPrefijo": prefijo_override or 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 ""
|
|
)).zfill(5) or "",
|
|
"fecha": fecha,
|
|
"codTercero": str(h.get("COD_PACIENTE") or "").strip(),
|
|
"codVendedor": "00",
|
|
"codigoCentroCosto": "00",
|
|
"tipoIngreso": "1",
|
|
"fechaHoraIngreso": ingreso,
|
|
"fechaHoraEgreso": egreso,
|
|
"modalidadAtencion": modalidad,
|
|
"numeroContrato": cod_contrato,
|
|
"descuento": descuento_pct,
|
|
"diagnosticoprincipal": str(h.get("DIAG_PPAL") or "").strip() or "Z017",
|
|
"discapacidad": "08",
|
|
"tipousuario": tipoususispro,
|
|
"viaIngreso": via_ingreso,
|
|
"esTerapia": False,
|
|
"esProcedimiento": False,
|
|
"numeroAutorizacion": autorizacion,
|
|
**({
|
|
"fechaAutorizacion": date.today().strftime("%d/%m/%Y"),
|
|
"vigenciaAutorizacion": vigencia,
|
|
} if autorizacion else {}),
|
|
"detallePedido": detalle_pedido,
|
|
}
|
|
return _clean_times(result)
|
|
|
|
|
|
# ── Ventas/Crear ─────────────────────────────────────────────────────────────
|
|
|
|
def generar_factura_venta(rows: list, default_vendedor: str = "00",
|
|
default_prefijo: str = "00", numero_override: str = "") -> dict:
|
|
if not rows:
|
|
return {}
|
|
h = rows[0]
|
|
|
|
num_fac = (numero_override if numero_override else str(h.get("NUM_FACTURA") or "")).zfill(5)
|
|
fecha = _fmt_fecha(h.get("FECHA_RECEPCION"))
|
|
|
|
detalle_pedido = []
|
|
for row in rows:
|
|
precio = float(row.get("PRECIO_TARIFA") or row.get("PRECIO") or 0)
|
|
detalle_pedido.append({
|
|
"codMat": str(row.get("CUPS") or row.get("COD_EXAMEN") or "").strip(),
|
|
"codBodega": "00",
|
|
"codTalla": "",
|
|
"codColor": "",
|
|
"cantidad": 1,
|
|
"tipoUnidad": "M",
|
|
"descuento": 0,
|
|
"descuentoValor": 0,
|
|
"centrosCostos": "00",
|
|
"porcIva": 0,
|
|
"valor": precio,
|
|
"impConsumo": 0,
|
|
"observacion": "",
|
|
"lote": "",
|
|
"fechaVenceLote": "",
|
|
"nroDocumento": "",
|
|
"itemsSerial": [],
|
|
"tipoSerial": "",
|
|
})
|
|
|
|
nit_raw = str(h.get("NIT_EMPRESA") or "").strip()
|
|
cod_tercero = nit_raw if nit_raw else str(h.get("COD_PACIENTE") or "").strip()
|
|
|
|
prefijo_real = str(h.get("PREFIJO") or "").strip() or default_prefijo or "00"
|
|
|
|
return _clean_times({
|
|
"codigoPrefijo": prefijo_real,
|
|
"numero": num_fac,
|
|
"numeroFactura": num_fac,
|
|
"fecha": fecha,
|
|
"kardexId": 0,
|
|
"codigoPedido": "",
|
|
"nombreCliente": "",
|
|
"codTercero": cod_tercero,
|
|
"codVendedor": default_vendedor or "00",
|
|
"codDespachar": "00",
|
|
"codFormaPago": "CO",
|
|
"codBanco": "00",
|
|
"fechaVence": fecha,
|
|
"fechaEntrega": fecha,
|
|
"plazoDias": 30,
|
|
"observacion": "",
|
|
"latitud": "",
|
|
"longitud": "",
|
|
"motivo": "",
|
|
"numeroFacturaDevolucion": "",
|
|
"tipoOperacion": "",
|
|
"codigoCentroCosto": "00",
|
|
"codigoArea": "00",
|
|
"terminal": "00",
|
|
"detallePedido": detalle_pedido,
|
|
"detalleFormaPago": [],
|
|
"asentar": 0,
|
|
"detalleDescuentos": [],
|
|
})
|
|
|
|
|
|
# ── 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 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
|