""" Lee los archivos JSONL de output del batch, parsea las respuestas de la IA y guarda los datos en la DB. Mueve cada archivo a la carpeta de procesados. Uso: python procesar_output.py Variables de entorno requeridas: PATH_BATCH_OUTPUT, BATCH_PATH_OUTPUT_PROCESADO, USER_ID, SERVER_D, DATABASE_D, USERNAME_D, PASSWORD_D """ import sys import os import pathlib import socket sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) from dotenv import load_dotenv import logging import shutil import json import helpers.db as db from helpers.logging_setup import setup_logging, setup_logging_db load_dotenv() _run_id = setup_logging("procesar_output") USER_ID = os.getenv('USER_ID', '0') ID_PROCESO = int(os.getenv("ID_PROCESO_PROCESAR_OUTPUT", 0)) def safe_extract(value, max_len=None): if value is None: return None if isinstance(value, list): result = ";".join(str(v).strip() for v in value if v) else: result = str(value).strip() if not result or result.lower() == 'none': return None if max_len and len(result) > max_len: logging.warning(f"TRUNCADO | valor de {len(result)} chars truncado a {max_len}") return result[:max_len] return result def _cuit_valido(cuit): return cuit and str(cuit).isdigit() and len(str(cuit)) >= 11 def _levenshtein(s1, s2): s1, s2 = s1.lower().strip(), s2.lower().strip() if len(s1) < len(s2): s1, s2 = s2, s1 if not s2: return len(s1) prev = list(range(len(s2) + 1)) for c1 in s1: curr = [prev[0] + 1] for j, c2 in enumerate(s2): curr.append(min(curr[j] + 1, prev[j + 1] + 1, prev[j] + (c1 != c2))) prev = curr return prev[-1] def _normalizar_nombre(nombre): return ' '.join(sorted(nombre.lower().strip().split())) def _normalizar_razon_social(nombre): """Quita puntos/comas y colapsa espacios para comparar razones sociales. Ej: 'S. A.' -> 'SA', 'S.R.L.' -> 'SRL'""" return ' '.join(nombre.replace('.', '').replace(',', '').split()) def _palabras_contenidas(nombre_ai, nombre_padron): """True si todas las palabras del nombre de la IA están en el nombre del padrón.""" palabras_ai = set(nombre_ai.lower().strip().split()) palabras_padron = set(nombre_padron.lower().strip().split()) return palabras_ai and palabras_ai.issubset(palabras_padron) def _validar_persona_padron(persona, conn): """ Busca la persona en el padrón usando DNI o CUIT que trajo la IA. Si el nombre coincide (Levenshtein ≤ 3), corrige persona_cuit y persona_DNI en el dict con los valores del padrón. No bloquea el guardado si no hay match. """ nombre = persona.get("apellido_nombre") cuit_ai = persona.get("persona_cuit") dni_ai = persona.get("persona_DNI") if not nombre: return if cuit_ai: cuit_limpio_raw = ''.join(filter(str.isdigit, str(cuit_ai))) if cuit_limpio_raw != str(cuit_ai): logging.info(f"PADRON_CUIT_LIMPIEZA | nombre='{nombre}' cuit_ai='{cuit_ai}' -> '{cuit_limpio_raw}'") persona["persona_cuit"] = cuit_limpio_raw cuit_ai = cuit_limpio_raw if cuit_ai and not _cuit_valido(cuit_ai): logging.warning(f"PADRON_CUIT_FORMATO | nombre='{nombre}' cuit_ai='{cuit_ai}' motivo=formato_invalido") dni_limpio = ''.join(filter(str.isdigit, str(dni_ai))) if dni_ai else '' cuit_limpio = ''.join(filter(str.isdigit, str(cuit_ai))) if cuit_ai else '' # Rama 1: hay DNI (≤ 8 dígitos) — sp_buscar_cuit_dni_nombre resuelve CUIT+DNI if dni_limpio and len(dni_limpio) <= 8: try: cursor = conn.cursor() cursor.execute("{CALL sp_buscar_cuit_dni_nombre (?,?)}", (nombre, dni_limpio)) columns = [col[0] for col in cursor.description] rows = cursor.fetchall() cursor.close() except Exception as e: logging.error(f"PADRON_ERROR_BUSQUEDA | nombre='{nombre}' dni='{dni_limpio}' error={e}") rows = [] cuit_padron = dict(zip(columns, rows[0])).get('cuit') if rows else None if cuit_padron: row_dict = dict(zip(columns, rows[0])) nombre_padron = row_dict.get('nombre') or '' dist = _levenshtein(_normalizar_nombre(nombre), _normalizar_nombre(nombre_padron)) if dist <= 3 or _palabras_contenidas(nombre, nombre_padron): cuit_padron = str(cuit_padron) if cuit_limpio != cuit_padron: logging.warning( f"PADRON_CORRECCION_CUIT | nombre='{nombre}' " f"cuit_ai='{cuit_ai}' -> cuit_padron='{cuit_padron}' dist_nombre={dist}" ) persona["persona_cuit"] = cuit_padron else: logging.info(f"PADRON_OK | nombre='{nombre}' cuit='{cuit_padron}' dist_nombre={dist}") dni_padron = row_dict.get('dni') if dni_padron and str(dni_ai) != str(dni_padron): logging.warning( f"PADRON_CORRECCION_DNI | nombre='{nombre}' " f"dni_ai='{dni_ai}' -> dni_padron='{dni_padron}' dist_nombre={dist}" ) persona["persona_DNI"] = str(dni_padron) persona["id_padron_integrante"] = row_dict.get('id_padron') return # resuelto por DNI else: logging.warning( f"PADRON_NOMBRE_NO_MATCH | nombre='{nombre}' nombre_padron='{nombre_padron}' " f"cuit_padron='{cuit_padron}' dist={dist}" ) # nombre no coincide → no corregimos, pero tampoco fallback por CUIT return else: logging.warning(f"PADRON_SIN_RESULTADO | nombre='{nombre}' dni='{dni_limpio}'") # DNI no encontró nada → intentar por CUIT si lo hay # Rama 2: validar por CUIT (11 dígitos) — buscador_paises_padron if not cuit_limpio: logging.debug(f"PADRON_SIN_IDENTIFICACION | nombre='{nombre}' cuit_ai='{cuit_ai}' dni_ai='{dni_ai}'") return try: cursor = conn.cursor() cursor.execute( 'EXEC [dbo].[buscador_paises_padron] ?,?,?,?,?,?,?,?,?,?,?,?', ('buscador', None, cuit_limpio, 'AR', None, None, None, None, None, None, None, None) ) rows = cursor.fetchall() cursor.close() except Exception as e: logging.error(f"PADRON_ERROR_BUSQUEDA_CUIT | nombre='{nombre}' cuit='{cuit_limpio}' error={e}") return if not rows: logging.warning(f"PADRON_SIN_RESULTADO | nombre='{nombre}' cuit='{cuit_limpio}'") return if len(rows) > 1: logging.info(f"PADRON_MULTIPLE | nombre='{nombre}' cuit='{cuit_limpio}' resultados={len(rows)}") return # r[2]=id_padron, r[3]=CUIT, r[4]=nombre_persona, r[12]=cuerpo_clave(DNI) r = rows[0] mejor_id = r[2] mejor_cuit = str(r[3]) if r[3] is not None else None mejor_nombre = r[4] mejor_dni = r[12] if len(r) > 12 else None if not mejor_cuit or not mejor_nombre: persona["id_padron_integrante"] = None return mejor_dist = _levenshtein(_normalizar_nombre(nombre), _normalizar_nombre(mejor_nombre)) if mejor_dist > 3 and not _palabras_contenidas(nombre, mejor_nombre): logging.warning( f"PADRON_NOMBRE_NO_MATCH | nombre='{nombre}' nombre_padron='{mejor_nombre}' " f"cuit='{cuit_limpio}' dist={mejor_dist}" ) persona["id_padron_integrante"] = None return if mejor_cuit != cuit_limpio: logging.warning( f"PADRON_CORRECCION_CUIT | nombre='{nombre}' " f"cuit_ai='{cuit_ai}' -> cuit_padron='{mejor_cuit}' dist_nombre={mejor_dist}" ) persona["persona_cuit"] = mejor_cuit else: logging.info(f"PADRON_OK | nombre='{nombre}' cuit='{mejor_cuit}' dist_nombre={mejor_dist}") if mejor_dni and str(dni_ai) != str(mejor_dni): logging.warning( f"PADRON_CORRECCION_DNI | nombre='{nombre}' " f"dni_ai='{dni_ai}' -> dni_padron='{mejor_dni}' dist_nombre={mejor_dist}" ) persona["persona_DNI"] = str(mejor_dni) persona["id_padron_integrante"] = mejor_id def _parsear_json(texto): texto = texto.strip() if texto.startswith("```"): lineas = texto.splitlines() texto = "\n".join(lineas[1:-1]).strip() try: return json.loads(texto) except json.JSONDecodeError as e: logging.warning(f"JSON inválido: {e}") return None def _extraer_cargos(persona): if isinstance(persona.get("Persona_Cargo"), list): return persona["Persona_Cargo"] if "Persona_Cargo" in persona: return [persona["Persona_Cargo"]] if isinstance(persona.get("p_rol"), list): return persona["p_rol"] if "p_rol" in persona: return [persona["p_rol"]] return [] def _validar_empresa_padron(nombre, cuit, conn): """ Busca la empresa en el padrón por CUIT (primario) o por nombre (fallback cuando no hay CUIT). Si el nombre coincide (Levenshtein ≤ 5) retorna (id_padron, nombre_padron, cuit_padron), sino (None, None, None). """ cuit_limpio = ''.join(filter(str.isdigit, str(cuit))) if cuit else '' if not cuit_limpio and not nombre: logging.info("PADRON_EMPRESA_SIN_CUIT_NI_NOMBRE") return None, None, None nombre_busqueda = _normalizar_razon_social(nombre) if nombre else None try: cursor = conn.cursor() if cuit_limpio: cursor.execute( 'EXEC [dbo].[buscador_paises_padron] ?,?,?,?,?,?,?,?,?,?,?,?', ('buscador', 'J', cuit_limpio, 'AR', None, None, None, None, None, None, None, None) ) else: logging.info(f"PADRON_EMPRESA_SIN_CUIT | nombre='{nombre}' nombre_busqueda='{nombre_busqueda}' -> buscando por razon social") cursor.execute( 'EXEC [dbo].[buscador_paises_padron] ?,?,?,?,?,?,?,?,?,?,?,?', ('buscador', 'J', nombre_busqueda, 'AR', None, None, None, None, None, None, None, None) ) rows = cursor.fetchall() cursor.close() except Exception as e: label = f"cuit='{cuit_limpio}'" if cuit_limpio else f"nombre='{nombre}'" logging.error(f"PADRON_EMPRESA_ERROR | {label} error={e}") return None, None label = f"cuit='{cuit_limpio}'" if cuit_limpio else f"nombre='{nombre}'" if not rows: logging.warning(f"PADRON_EMPRESA_SIN_RESULTADO | {label}") return None, None, None # r[2]=id_padron, r[3]=identificacion(CUIT), r[4]=nombre_persona def _score_row(r): nombre_r = r[4] or '' return _levenshtein( _normalizar_nombre(_normalizar_razon_social(nombre or '')), _normalizar_nombre(_normalizar_razon_social(nombre_r)) ) if len(rows) > 1: if not nombre: logging.info(f"PADRON_EMPRESA_MULTIPLE | {label} resultados={len(rows)} sin nombre para desambiguar") return None, None, None rows = sorted(rows, key=_score_row) logging.info(f"PADRON_EMPRESA_MULTIPLE | {label} resultados={len(rows)} -> eligiendo más cercano") r = rows[0] id_padron_r, cuit_padron_r, nombre_r = r[2], str(r[3]) if r[3] is not None else None, r[4] if id_padron_r is None or nombre_r is None: return None, None, None # Si la IA no trajo nombre, tomamos el del padrón directamente if not nombre: logging.info(f"PADRON_EMPRESA_NOMBRE_COMPLETADO | {label} nombre_padron='{nombre_r}'") return id_padron_r, nombre_r, cuit_padron_r dist = _score_row(r) if dist > 5: logging.warning( f"PADRON_EMPRESA_NOMBRE_NO_MATCH | nombre='{nombre}' {label} nombre_padron='{nombre_r}' dist={dist}" ) return None, None, None logging.info(f"PADRON_EMPRESA_OK | nombre='{nombre}' {label} nombre_padron='{nombre_r}' id_padron={id_padron_r} dist={dist}") return id_padron_r, nombre_r, cuit_padron_r def _enriquecer_con_padron(datos): """ Realiza todas las búsquedas en el padrón ANTES de abrir la transacción principal. Muta los dicts de empresa y personas con los valores corregidos y los id_padron. Retorna id_padron_sociedad. """ id_padron_sociedad = None conn_padron = db.conectar() try: empresa = datos.get('Datos_Empresa') or {} nombre_actual = safe_extract(empresa.get("nombre_actual")) cuit_empresa = safe_extract(empresa.get("cuit_empresa")) logging.info(f"PADRON_ENRIQUECIMIENTO | nombre='{nombre_actual}' cuit='{cuit_empresa}'") id_padron_sociedad, nombre_padron, cuit_padron = _validar_empresa_padron(nombre_actual, cuit_empresa, conn_padron) logging.info(f"PADRON_ENRIQUECIMIENTO_RESULTADO | id_padron_sociedad={id_padron_sociedad} nombre_padron='{nombre_padron}' cuit_padron='{cuit_padron}'") if not nombre_actual and nombre_padron: empresa["nombre_actual"] = nombre_padron if not cuit_empresa and cuit_padron: logging.info(f"PADRON_EMPRESA_CUIT_COMPLETADO | cuit_padron='{cuit_padron}'") empresa["cuit_empresa"] = cuit_padron personas = datos.get('Relacion_Personas', []) if not isinstance(personas, list): personas = [personas] for persona in personas: if isinstance(persona, dict): _validar_persona_padron(persona, conn_padron) empresas = datos.get('Relacion_Empresas', []) if not isinstance(empresas, list): empresas = [empresas] for empresa_v in empresas: if not isinstance(empresa_v, dict): continue nombre_v = safe_extract(empresa_v.get("Vinculada_nombre"), 500) cuit_v = safe_extract(empresa_v.get("Vinculada_cuit"), 20) id_padron_v, nombre_padron_v, cuit_padron_v = _validar_empresa_padron(nombre_v, cuit_v, conn_padron) empresa_v["_id_padron_vinculada"] = id_padron_v if not nombre_v and nombre_padron_v: empresa_v["Vinculada_nombre"] = nombre_padron_v if not cuit_v and cuit_padron_v: empresa_v["Vinculada_cuit"] = cuit_padron_v except Exception as e: logging.error(f"PADRON_ENRIQUECIMIENTO_ERROR | error={e}") id_padron_sociedad = None finally: conn_padron.close() return id_padron_sociedad def _guardar_empresa(id_fide, datos, file_path, fuente, id_padron_sociedad, conn): empresa = datos.get('Datos_Empresa') or {} generales = datos.get('Datos_Generales') or {} db.ejecutar_sp('openai_insert_sociedades_constituciones', [ id_fide, 'update', USER_ID, safe_extract(empresa.get("nombre_actual")), safe_extract(empresa.get("nombre_anterior")), safe_extract(empresa.get("cuit_empresa")), safe_extract(empresa.get("domicilio_actual")), safe_extract(empresa.get("domicilio_anterior")), safe_extract(empresa.get("empresa_tipo")), safe_extract(generales.get("tipo_edicto")), file_path, fuente, None, # @fechas_asambleas id_padron_sociedad, # @id_padron ], conn) def _guardar_json_texto(id_fide, tipo_edicto, file_path, custom_id, txt_json, conn): result = db.ejecutar_sp('openai_insert_json', [ 'insert', id_fide, tipo_edicto or None, file_path or None, custom_id or None, txt_json, USER_ID, ], conn) return result[0].get('id', '0') if result else '0' def _guardar_personas(id_fide, relacion_personas, id_padron_sociedad, conn): for persona in relacion_personas: if not isinstance(persona, dict): continue cargos = _extraer_cargos(persona) id_padron_integrante = persona.get("id_padron_integrante") apellido_nombre = safe_extract(persona.get("apellido_nombre"), 200) dni = safe_extract(persona.get("persona_DNI"), 20) cuit = safe_extract(persona.get("persona_cuit"), 50) domicilio = safe_extract(persona.get("persona_domicilio"), 500) estado_civil = safe_extract(persona.get("persona_estado_civil"), 50) conyuge = safe_extract(persona.get("persona_conyuge"), 200) if not cargos: db.ejecutar_sp('openai_insert_personas', [ 'insert', id_fide, apellido_nombre, dni, cuit, domicilio, estado_civil, conyuge, None, persona.get("permanece_en_cargo"), USER_ID, id_padron_integrante, id_padron_sociedad, ], conn) else: for cargo in cargos: p_rol = safe_extract(cargo.get("p_rol") if isinstance(cargo, dict) else cargo, 200) permanece = cargo.get("permanece_en_cargo", persona.get("permanece_en_cargo")) if isinstance(cargo, dict) else persona.get("permanece_en_cargo") db.ejecutar_sp('openai_insert_personas', [ 'insert', id_fide, apellido_nombre, dni, cuit, domicilio, estado_civil, conyuge, p_rol, permanece, USER_ID, id_padron_integrante, id_padron_sociedad, ], conn) def _guardar_vinculadas(id_fide, relacion_empresas, id_padron_sociedad, conn): for empresa in relacion_empresas: if not isinstance(empresa, dict): continue vinculos = empresa.get("Vinculada_vinculo") if isinstance(vinculos, list): pass elif vinculos is not None: vinculos = [vinculos] else: vinculos = [] nombre_vinculada = safe_extract(empresa.get("Vinculada_nombre"), 500) cuit_vinculada = safe_extract(empresa.get("Vinculada_cuit"), 20) domicilio_vinculada = safe_extract(empresa.get("Vinculada_domicilio"), 500) id_padron_vinculada = empresa.get("_id_padron_vinculada") roles = vinculos if vinculos else [None] for vinculo in roles: rol = safe_extract(vinculo.get("e_rol") if isinstance(vinculo, dict) else vinculo, 100) db.ejecutar_sp('openai_insert_vinculadas', [ 'insert', id_fide, nombre_vinculada, cuit_vinculada, safe_extract(empresa.get("Vinculada_Pais"), 100), domicilio_vinculada, rol, USER_ID, id_padron_sociedad, id_padron_vinculada, ], conn) def _procesar_archivo(file_path): with open(file_path, 'r', encoding='utf-8') as f: lineas = f.readlines() for numero, linea in enumerate(lineas): linea = linea.replace('permanece_en_cargo?', 'permanece_en_cargo') try: registro = json.loads(linea) except json.JSONDecodeError as e: logging.error(f"Error decodificando línea {numero}: {e}") continue custom_id = registro.get("custom_id", "") partes = custom_id.split("_") id_fide = partes[2] if len(partes) > 2 else 0 control = db.ejecutar_sp('openai_insert_sociedades_constituciones', [id_fide, 'control_ya_procesado']) if control and control[0].get('procesado') == 1: logging.info(f"id_fide {id_fide} ya procesado, saltando") continue choices = registro.get("response", {}).get("body", {}).get("choices", []) for choice in choices: txt_json = choice.get("message", {}).get("content", "") txt_json = txt_json.replace("```json", "").replace("```", "").strip() datos = _parsear_json(txt_json) if not datos: db.ejecutar_sp('openai_insert_sociedades_constituciones', [id_fide, 'error_2', USER_ID]) continue generales = datos.get('Datos_Generales') or {} tipo_edicto = safe_extract(generales.get("tipo_edicto")) personas = datos.get('Relacion_Personas', []) empresas = datos.get('Relacion_Empresas', []) # Si la IA no extrajo CUIT, intentar recuperarlo de la DB empresa = datos.setdefault('Datos_Empresa', {}) if not safe_extract(empresa.get('cuit_empresa')): result = db.ejecutar_sp('openai_insert_sociedades_constituciones', [id_fide, 'get_cuit']) cuit_db = result[0].get('cuit') if result else None if cuit_db: logging.info(f"CUIT_FALLBACK_DB | id_fide={id_fide} cuit='{cuit_db}'") empresa['cuit_empresa'] = cuit_db # Búsquedas en padrón ANTES de abrir la transacción principal id_padron_sociedad = _enriquecer_con_padron(datos) conn = db.conectar() try: _guardar_empresa(id_fide, datos, file_path, None, id_padron_sociedad, conn) _guardar_json_texto(id_fide, tipo_edicto, file_path, custom_id, txt_json, conn) _guardar_personas(id_fide, personas if isinstance(personas, list) else [personas], id_padron_sociedad, conn) _guardar_vinculadas(id_fide, empresas if isinstance(empresas, list) else [empresas], id_padron_sociedad, conn) conn.commit() logging.info(f"id_fide {id_fide} procesado correctamente") except Exception as e: conn.rollback() db.ejecutar_sp('openai_insert_sociedades_constituciones', [id_fide, 'error_2', USER_ID]) logging.error(f"Error procesando id_fide {id_fide}: {e}") finally: conn.close() def procesar_output(): path_input = os.getenv('PATH_BATCH_OUTPUT') path_procesado = os.getenv('BATCH_PATH_OUTPUT_PROCESADO') if not path_input or not os.path.isdir(path_input): logging.error(f"Directorio no encontrado: {path_input}") return 0, 0, 0 archivos = [f for f in os.listdir(path_input) if os.path.isfile(os.path.join(path_input, f))] if not archivos: logging.info("No hay archivos para procesar") return 0, 0, 0 logging.info(f"Archivos a procesar: {len(archivos)}") os.makedirs(path_procesado, exist_ok=True) ok_count = 0 error_count = 0 for archivo in archivos: full_path = os.path.join(path_input, archivo) logging.info(f"Procesando: {archivo}") try: _procesar_archivo(full_path) shutil.move(full_path, os.path.join(path_procesado, archivo)) logging.info(f"Movido a procesado: {archivo}") ok_count += 1 except Exception as e: logging.error(f"Error procesando/moviendo {archivo}: {e}") error_count += 1 return len(archivos), ok_count, error_count if __name__ == "__main__": logging.info("Inicio del proceso de guardado en la DB") id_log = None try: result = db.ejecutar_sp('PRS_iniciar', [ ID_PROCESO, socket.gethostbyname(socket.gethostname()), os.getpid(), socket.gethostname(), 12, _run_id, ]) if result: id_log = result[0].get('id_log') setup_logging_db(id_log) except Exception as e: logging.warning(f"PRS_iniciar falló: {e}") total = ok_count = error_count = 0 try: total, ok_count, error_count = procesar_output() except Exception as e: error_count = 1 logging.error(f"Error en procesar_output: {e}") finally: if id_log is not None: try: db.ejecutar_sp('PRS_finalizar', [id_log, None, total, ok_count, error_count]) except Exception as e: logging.warning(f"PRS_finalizar falló: {e}")