Files
rips_manager/app/services/json_generator.py
T
Lizandro GuarnizoandClaude Sonnet 4.6 774ce7d728 feat: queries reales Firebird y generadores JSON para API ERP
- json_generator: nuevas funciones generar_tercero_api y generar_rda_paciente
  mapeando PACIENTE/RECEPCION/RELACION al formato de los endpoints
  Tercero/Crear y RdaPaciente/Insertar
- queries: reemplaza placeholders por consultas reales sobre PACIENTE,
  RECEPCION, RELACION, MEDICO y CIUDAD
- config: agrega api_version y api_sucursal; actualiza url y credenciales Firebird
- terceros/transaccion routes: usan nuevos generadores y endpoints correctos

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-25 17:27:00 -05:00

256 lines
10 KiB
Python

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")
s = str(val)[: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",
"Telefono": str(row.get("TELEFONOS") or "").strip(),
"Email": (row.get("EMAIL") or "").strip(),
"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) -> 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 = _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")
)
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": str(h.get("profesional") or "").strip(),
"especialidad": str(row.get("COD_ESPECIALIDAD") or "").strip(),
"diagnosticoprincipal": str(h.get("DIAG_PPAL") or "").strip(),
"fechaHoraRealizacion": fecha_real or egreso,
})
autorizacion = str(h.get("AUTORIZACION") or "").strip() or None
return {
"CodigoPrefijo": str(h.get("PREFIJO") or "00").strip(),
"Numero": "",
"Fecha": fecha,
"CodTercero": str(h.get("COD_PACIENTE") or "").strip(),
"CodVendedor": "00",
"CodFormaPago": "CO",
"CodBanco": "00",
"CodigoCentroCosto": "00",
"tipoIngreso": str(h.get("CLASEPROC") or "1").strip() or "1",
"fechaHoraIngreso": ingreso,
"fechaHoraEgreso": egreso,
"modalidadAtencion": "01",
"numeroContrato": str(h.get("NIT_EMPRESA") or "").strip(),
"diagnosticoprincipal": str(h.get("DIAG_PPAL") or "").strip(),
"tipousuario": str(h.get("TIPOUSUSISPRO") or "12").strip(),
"viaIngreso": "01",
"esTerapia": False,
"esProcedimiento": False,
"numeroAutorizacion": autorizacion,
"DetallePedido": detalle_pedido,
"DetalleFormaPago": [{
"CodigoFormaPago": "CO",
"PlazoDias": "0",
"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