import difflib import logging import sys import os BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')) sys.path.insert(0, BASE_DIR) from helpers.db import conectar, conectar_readonly, ejecutar_sp, ejecutar_sp_raw from helpers.provincias import PROVINCIAS_CON_NUMERO from helpers.cuit_homonimia import _normalizar_razon_social, seleccionar_mejor_cuit from helpers.logging_setup import setup_logging _TABLAS_PERMITIDAS = { 'edictos_ia', 'sociedades_constituciones_relaciones_vinculadas', 'sociedades_constituciones_relaciones_personas', } def _exec(conn, sql, *params): c = conn.cursor() try: c.execute(sql, *params) return c.fetchall() finally: c.close() def _limpiar_criterio(texto): return ( _normalizar_razon_social(texto) .replace('"', '').replace('"', '').replace('"', '') .replace("'", '').replace(''', '').replace(''', '') ) def _buscar_en_padron(conn_read, criterio, cod_prov=None): return _exec( conn_read, 'EXEC [dbo].[buscador_paises_padron] ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?', ('buscador', None, criterio, 'AR', cod_prov, None, None, None, None, None, None, None) ) def _preparar_busqueda(cuit, nombre, provincia): criterio = _limpiar_criterio(nombre) if not cuit or str(cuit).lower() == 'null' else cuit cod_prov = PROVINCIAS_CON_NUMERO.get(str(provincia).strip().upper()) if provincia and isinstance(provincia, str) else None return criterio, cod_prov def _id_padron_para_cuit(candidatos, cuit): return next((r[2] for r in candidatos if r[3] == cuit), None) def resolver_multiples(candidatos, criterio, cod_prov, conn_read): source = candidatos pares = [(r[3], r[4]) for r in source] if cod_prov is not None: filtrados = _buscar_en_padron(conn_read, criterio, cod_prov) or [] if filtrados: pares_filtrados = [(r[3], r[4]) for r in filtrados] if len(pares_filtrados) == 1: return pares_filtrados[0][0], filtrados[0][2] if pares_filtrados: pares = pares_filtrados source = filtrados cuit = seleccionar_mejor_cuit(criterio, pares) return cuit, _id_padron_para_cuit(source, cuit) if cuit else None def resolver_via_dni(conn_read, nombre, dni): """Devuelve (cuit, id_padron) buscando por DNI+nombre, o (None, None) si no resuelve.""" if not dni: return None, None dni_limpio = ''.join(filter(str.isdigit, str(dni))) if not dni_limpio or len(dni_limpio) > 8: return None, None try: c = conn_read.cursor() c.execute("{CALL sp_buscar_cuit_dni_nombre (?,?)}", (nombre, dni_limpio)) columns = [col[0] for col in c.description] rows = c.fetchall() c.close() except Exception as e: logging.error(f" DNI_ERROR | nombre='{nombre}' dni='{dni_limpio}' error={e}") return None, None if not rows: return None, None row = dict(zip(columns, rows[0])) cuit_padron = row.get('cuit') if not cuit_padron: return None, None nombre_padron = row.get('nombre') or '' nombre_norm = _normalizar_razon_social(nombre) nombre_p_norm = _normalizar_razon_social(nombre_padron) similitud = difflib.SequenceMatcher(None, nombre_norm, nombre_p_norm).ratio() if similitud < 0.6: logging.info(f" DNI_NOMBRE_NO_MATCH | nombre='{nombre}' padron='{nombre_padron}' sim={similitud:.2f}") return None, None return str(cuit_padron), row.get('id_padron') def buscar_y_resolver(conn_read, cuit, nombre, provincia): """Devuelve (cuit_encontrado, id_padron, motivo, candidatos).""" # Si viene con CUIT, buscamos primero por CUIT en modo masivos para obtener id_padron cuit_limpio = cuit.replace('-', '') if cuit and str(cuit).lower() != 'null' else None if cuit_limpio: from helpers.cuit_homonimia import ejecutar_busqueda_cuit resultados_cuit = ejecutar_busqueda_cuit(conn_read, cuit=cuit_limpio) or [] if len(resultados_cuit) == 1: nombre_padron = resultados_cuit[0][4] if nombre and _normalizar_razon_social(nombre) != _normalizar_razon_social(nombre_padron): logging.info(f" CUIT_NOMBRE_MISMATCH | fuente='{nombre}' padron='{nombre_padron}' — se asigna igual por CUIT único") return resultados_cuit[0][3], resultados_cuit[0][2], 'único', resultados_cuit if len(resultados_cuit) > 1: id_padron = _id_padron_para_cuit(resultados_cuit, cuit_limpio) if id_padron: return cuit_limpio, id_padron, 'cuit_directo', resultados_cuit criterio, cod_prov = _preparar_busqueda(cuit, nombre, provincia) if not criterio: return None, None, 'sin criterio', [] resultados = _buscar_en_padron(conn_read, criterio) n = len(resultados) if n == 1: return resultados[0][3], resultados[0][2], 'único', resultados elif n > 1: cuit_encontrado, id_padron = resolver_multiples(resultados, criterio, cod_prov, conn_read) if cuit_encontrado: return cuit_encontrado, id_padron, 'fuzzy', resultados return None, None, f'múltiple sin ganador ({n} candidatos)', resultados else: return None, None, 'sin resultados', [] def obtener_edicto_padre(conn_read, id_fide): rows = ejecutar_sp_raw('EDT_homonimia_obtener_sociedad', (id_fide,), conn_read) if rows: return rows[0][0], rows[0][1], rows[0][2] return None, None, None def cuits_en_tablas_historicas(conn_read, cuit_soc): """Devuelve set de CUITs que aparecen en socios o accionistas para esa sociedad.""" cuits = set() rows = ejecutar_sp_raw('EDT_homonimia_buscar_en_socios', (cuit_soc,), conn_read) or [] for r in rows: if r[0]: cuits.add(r[0]) rows = ejecutar_sp_raw('EDT_homonimia_buscar_en_accionistas', (cuit_soc,), conn_read) or [] for r in rows: if r[0]: cuits.add(r[0]) return cuits def resolver_via_historicos(conn_read, candidatos, nombre, cuit_soc): """Devuelve (cuit, id_padron, motivo).""" historicos = cuits_en_tablas_historicas(conn_read, cuit_soc) if not historicos: return None, None, 'sin históricos' matches = [r for r in candidatos if r[3] in historicos] if not matches: return None, None, 'sin intersección con históricos' if len(matches) == 1: return matches[0][3], matches[0][2], 'histórico único' pares = [(r[3], r[4]) for r in matches] cuit_ganador = seleccionar_mejor_cuit(nombre, pares) if cuit_ganador: return cuit_ganador, _id_padron_para_cuit(matches, cuit_ganador), 'histórico fuzzy' return None, None, f'histórico múltiple sin ganador ({len(matches)})' def asignar(conn, cuit, id_padron, id_edicto, tabla): ejecutar_sp('EDT_homonimia_asignar_cuit', (tabla, id_edicto, cuit, id_padron), conn) conn.commit() def marcar_negativo(conn, id_edicto, tabla): ejecutar_sp('EDT_homonimia_marcar_negativo', (tabla, id_edicto), conn) conn.commit() def marcar_multiple(conn, id_edicto, tabla): ejecutar_sp('EDT_homonimia_marcar_multiple', (tabla, id_edicto), conn) conn.commit() if __name__ == '__main__': tabla = sys.argv[1] if len(sys.argv) > 1 else 'edictos_ia' if tabla not in _TABLAS_PERMITIDAS: print(f"Tabla no permitida: {tabla!r}. Opciones: {', '.join(_TABLAS_PERMITIDAS)}") sys.exit(1) setup_logging("homonimia") logging.info(f"PROCESO_INICIADO | tabla={tabla}") conn_read = conectar_readonly() conn_write = conectar() hola = ejecutar_sp_raw('EDT_homonimia_obtener', (tabla, '2016-01-01', '2020-01-01'), conn_read) stats = { 'asignado_directo': 0, 'asignado_fuzzy': 0, 'sociedad_resuelta': 0, 'asignado_historico': 0, 'multiple_sin_ganador': 0, 'sin_resultados': 0, 'error': 0, } for i, edicto in enumerate(hola, start=1): id_edicto = edicto[0] cuit = edicto[1] nombre = edicto[2] provincia = edicto[3] if len(edicto) > 3 else None dni = edicto[4] if len(edicto) > 4 else None id_fide = edicto[5] if len(edicto) > 5 else None logging.info(f"[{i}] id={id_edicto} cuit={cuit} nombre='{nombre}'") try: cuit_encontrado, id_padron_encontrado, motivo, candidatos = buscar_y_resolver(conn_read, cuit, nombre, provincia) if cuit_encontrado: asignar(conn_write, cuit_encontrado, id_padron_encontrado, id_edicto, tabla) if motivo == 'único': stats['asignado_directo'] += 1 else: stats['asignado_fuzzy'] += 1 logging.info(f" ASIGNADO ({motivo}): {cuit_encontrado} id_padron={id_padron_encontrado}") continue logging.info(f" {motivo}") if dni: cuit_dni, id_padron_dni = resolver_via_dni(conn_read, nombre, dni) if cuit_dni: asignar(conn_write, cuit_dni, id_padron_dni, id_edicto, tabla) stats['asignado_fuzzy'] += 1 logging.info(f" ASIGNADO (dni): {cuit_dni} id_padron={id_padron_dni}") continue resuelto_historico = False if tabla != 'edictos_ia' and id_fide: cuit_soc, rs_soc, prov_soc = obtener_edicto_padre(conn_read, id_fide) if not cuit_soc: logging.info(f" Sociedad sin CUIT (id_fide={id_fide}), intentando resolverla...") cuit_soc, id_padron_soc, motivo_soc, _ = buscar_y_resolver(conn_read, None, rs_soc, prov_soc) if cuit_soc: asignar(conn_write, cuit_soc, id_padron_soc, id_fide, 'edictos_ia') stats['sociedad_resuelta'] += 1 logging.info(f" SOCIEDAD ASIGNADA ({motivo_soc}): {cuit_soc}") else: logging.info(f" Sociedad también irresoluble: {motivo_soc}") if cuit_soc and candidatos: cuit_historico, id_padron_historico, motivo_hist = resolver_via_historicos(conn_read, candidatos, nombre, cuit_soc) if cuit_historico: asignar(conn_write, cuit_historico, id_padron_historico, id_edicto, tabla) stats['asignado_historico'] += 1 resuelto_historico = True logging.info(f" ASIGNADO ({motivo_hist}): {cuit_historico} id_padron={id_padron_historico}") else: logging.info(f" Irresoluble: {motivo_hist}") elif cuit_soc and not candidatos: logging.info(f" Sin candidatos del padrón para cruzar con históricos") if not resuelto_historico: if 'múltiple' in motivo: stats['multiple_sin_ganador'] += 1 marcar_multiple(conn_write, id_edicto, tabla) else: stats['sin_resultados'] += 1 marcar_negativo(conn_write, id_edicto, tabla) except Exception as e: stats['error'] += 1 logging.error(f" ERROR: {e}") try: conn_write.rollback() except Exception: pass total = len(hola) asignados = stats['asignado_directo'] + stats['asignado_fuzzy'] + stats['asignado_historico'] logging.info( f"PROCESO_FINALIZADO | tabla={tabla} total={total} asignados={asignados} " f"directo={stats['asignado_directo']} fuzzy={stats['asignado_fuzzy']} " f"historico={stats['asignado_historico']} sociedad_resuelta={stats['sociedad_resuelta']} " f"multiple={stats['multiple_sin_ganador']} sin_resultado={stats['sin_resultados']} errores={stats['error']}" )