#!/usr/bin/env python3 # -*- coding: utf-8 -*- import os import json import argparse import logging import pymysql from datetime import datetime, timezone from dotenv import load_dotenv # ───────────────────────────────────────────────────────────── # Helpers PlanOK/GCI # ───────────────────────────────────────────────────────────── from interaccion_planok import ( crear_cliente_natural, buscar_cliente_natural, crear_seguimiento, obtener_rut_vendedor_por_proyecto ) # ───────────────────────────────────────────────────────────── # Config & ENV # ───────────────────────────────────────────────────────────── logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s') load_dotenv() DB_USER = os.getenv('DB_USER') DB_PASSWORD = os.getenv('DB_PASSWORD') DB_HOST = os.getenv('DB_HOST') DB_NAME = os.getenv('DB_NAME') TIPO_CONTACTOPOK = int(os.getenv("TIPO_CONTACTOPOK", "1")) ENVIAR_PLANOK = os.getenv("ENVIAR_PLANOK", "0") # "1" para habilitar PROYECTOS_ACTIVO_STR = os.getenv("PROYECTOS_ACTIVO", "") CHATFUEL_BOT_ID = os.getenv("CHATFUEL_BOT_ID", "") PROYECTOS_ACTIVO = set() if PROYECTOS_ACTIVO_STR.strip(): try: PROYECTOS_ACTIVO = {int(x) for x in PROYECTOS_ACTIVO_STR.split(",")} except ValueError: logging.warning(f"PROYECTOS_ACTIVO contiene valores no enteros: {PROYECTOS_ACTIVO_STR}") # ───────────────────────────────────────────────────────────── # DB # ───────────────────────────────────────────────────────────── def _db_conn(): return pymysql.connect( host=DB_HOST, user=DB_USER, password=DB_PASSWORD, database=DB_NAME, cursorclass=pymysql.cursors.DictCursor, charset='utf8mb4', use_unicode=True, ) # ───────────────────────────────────────────────────────────── # Utilidades de fecha # ───────────────────────────────────────────────────────────── def normalize_dt(dt): """Quita microsegundos para coincidir con DATETIME(0).""" if not dt: return None return dt.replace(microsecond=0) def parse_date(date_str): if not date_str: return None try: formats = [ "%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%SZ", "%a, %d %b %Y %H:%M:%S %Z" ] for fmt in formats: try: return datetime.strptime(date_str, fmt) except ValueError: pass raise ValueError(f"Date '{date_str}' no coincide con formatos esperados.") except Exception as e: logging.error(f"Error parseando fecha '{date_str}': {e}") return None def parse_int(value, default=None): try: return int(value) except (ValueError, TypeError): return default def list_json_files_sorted(folder_path): if not os.path.isdir(folder_path): raise FileNotFoundError(f"La carpeta no existe: {folder_path}") files = [] for name in os.listdir(folder_path): if not name.lower().endswith(".json"): continue full = os.path.join(folder_path, name) if os.path.isfile(full): files.append(full) # Antiguo → Nuevo por fecha de modificación files.sort(key=lambda p: os.path.getmtime(p)) return files def file_mtime_as_dt(path): """Devuelve el mtime del archivo como datetime sin microsegundos (hora local del sistema).""" ts = os.path.getmtime(path) # timestamp en segundos (epoch) # Lo mapeamos a datetime local (naive) y truncamos microsegundos dt_local = datetime.fromtimestamp(ts) return normalize_dt(dt_local) # ───────────────────────────────────────────────────────────── # Operaciones BD auxiliares # ───────────────────────────────────────────────────────────── def get_existing_record(whatsapp_user_id, last_seen_dt): """Obtiene (id, enviado_crm, fecha_registro) si existe el par (wa_id, last_seen).""" if not whatsapp_user_id or last_seen_dt is None: return None try: with _db_conn() as connection: with connection.cursor() as cursor: cursor.execute( """ SELECT id, enviado_crm, fecha_registro FROM UserInteraction WHERE whatsapp_user_id = %s AND last_seen = %s LIMIT 1 """, (whatsapp_user_id, last_seen_dt) ) return cursor.fetchone() except pymysql.MySQLError as e: logging.error(f"Error consultando existencia: {str(e)}") return None def save_json_full_to_db(raw_data): """Actualiza Json_full por whatsapp_user_id (si existe la fila).""" try: if isinstance(raw_data, (bytes, bytearray)): json_text = raw_data.decode('utf-8', errors='strict') elif isinstance(raw_data, str): json_text = raw_data else: json_text = json.dumps(raw_data, ensure_ascii=False) data_obj = json.loads(json_text) wa_id = data_obj.get('whatsapp user id') if not wa_id: raise ValueError("El JSON no contiene la llave 'whatsapp user id'.") with _db_conn() as connection: with connection.cursor() as cursor: sql = """ UPDATE UserInteraction SET Json_full = CAST(%s AS JSON) WHERE whatsapp_user_id = %s """ cursor.execute(sql, (json_text, wa_id)) connection.commit() logging.info(f"✅ Json_full actualizado para whatsapp_user_id={wa_id}; filas afectadas: {cursor.rowcount}") return cursor.rowcount except Exception as e: logging.error(f"Error guardando Json_full: {e}") raise def update_as_sent(whatsapp_user_id): """Marca enviado_crm=TRUE para ese usuario (todas sus filas).""" try: with _db_conn() as connection: with connection.cursor() as cursor: sql = """ UPDATE UserInteraction SET enviado_crm = TRUE WHERE whatsapp_user_id = %s """ cursor.execute(sql, (whatsapp_user_id,)) connection.commit() logging.info(f"📬 Marcado enviado_crm=TRUE: {whatsapp_user_id}") except pymysql.MySQLError as e: logging.error(f"Error al actualizar enviado_crm: {str(e)}") raise def upsert_user_interaction(user_data, fecha_registro_dt): """ Upsert por (whatsapp_user_id, last_seen): - Si existe → UPDATE (sobre-escribe campos + fecha_registro con mtime del archivo). - Si no existe → INSERT (fecha_registro = mtime archivo). Retorna dict: {'action': 'updated'|'inserted', 'id': int|None, 'enviado_crm': bool} """ user_data = dict(user_data) # copia defensiva user_data['fecha_registro'] = fecha_registro_dt with _db_conn() as connection: with connection.cursor() as cursor: # UPDATE primero sql_update = """ UPDATE UserInteraction SET whatsapp_phone = %(whatsapp_phone)s, source = %(source)s, signed_up = %(signed_up)s, sessions = %(sessions)s, last_visited_block_name = %(last_visited_block_name)s, last_visited_block_id = %(last_visited_block_id)s, last_clicked_button_name = %(last_clicked_button_name)s, entrada_del_usuario = %(entrada_del_usuario)s, numero_interacciones = %(numero_interacciones)s, motivo_de_consulta = %(motivo_de_consulta)s, topico = %(topico)s, proyecto_id = %(proyecto_id)s, proyecto_consultado = %(proyecto_consultado)s, acuerdo = %(acuerdo)s, medio_de_contacto = %(medio_de_contacto)s, dentro_de_presupuesto = %(dentro_de_presupuesto)s, renta = %(renta)s, situacion_de_vivienda = %(situacion_de_vivienda)s, inicial = %(inicial)s, fechaentrega = %(fechaentrega)s, nombre_cliente = %(nombre_cliente)s, apellido_cliente = %(apellido_cliente)s, numero_ambientes = %(numero_ambientes)s, dni = %(dni)s, fecha_de_visita = %(fecha_de_visita)s, horario = %(horario)s, mail = %(mail)s, crm_email = %(crm_email)s, estado_flujo = %(estado_flujo)s, nivel_de_interes = %(nivel_de_interes)s, subsidio = %(subsidio)s, subsidio_puede_postular = %(subsidio_puede_postular)s, link_de_la_conversacion = %(link_de_la_conversacion)s, observacion_bot = %(observacion_bot)s, contactar_por = %(contactar_por)s, credito = %(credito)s, crm_id_cliente = %(crm_id_cliente)s, crm_id_trato = %(crm_id_trato)s, crm_medio_id = %(crm_medio_id)s, crm_proyecto = %(crm_proyecto)s, digitoverificador = %(digitoverificador)s, modelo = %(modelo)s, motivo_de_problema = %(motivo_de_problema)s, pie = %(pie)s, rentadeclarada = %(rentadeclarada)s, crm_Nombre_vendedor = %(crm_Nombre_vendedor)s, boton_interes = %(boton_interes)s, primer_boton = %(primer_boton)s, boton_seguimiento = %(boton_seguimiento)s, primera_respuesta = %(primera_respuesta)s, crm_Rut_vendedor = %(crm_Rut_vendedor)s, fecha_registro = %(fecha_registro)s WHERE whatsapp_user_id = %(whatsapp_user_id)s AND last_seen = %(last_seen)s """ cursor.execute(sql_update, user_data) if cursor.rowcount > 0: # Obtener enviado_crm + id tras el update cursor.execute( """ SELECT id, enviado_crm FROM UserInteraction WHERE whatsapp_user_id = %s AND last_seen = %s LIMIT 1 """, (user_data['whatsapp_user_id'], user_data['last_seen']) ) row = cursor.fetchone() connection.commit() logging.info(f"✏️ UPDATE por (wa_id, last_seen): {user_data['whatsapp_user_id']} {user_data['last_seen']}") return {'action': 'updated', 'id': row['id'] if row else None, 'enviado_crm': bool(row['enviado_crm']) if row else False} # INSERT si no existía sql_insert = """ INSERT INTO UserInteraction ( whatsapp_user_id, whatsapp_phone, source, last_seen, signed_up, sessions, last_visited_block_name, last_visited_block_id, last_clicked_button_name, entrada_del_usuario, numero_interacciones, motivo_de_consulta, topico, proyecto_id, proyecto_consultado, acuerdo, medio_de_contacto, dentro_de_presupuesto, renta, situacion_de_vivienda, inicial, fechaentrega, nombre_cliente, apellido_cliente, numero_ambientes, dni, fecha_de_visita, horario, mail, crm_email, estado_flujo, nivel_de_interes, subsidio, subsidio_puede_postular, link_de_la_conversacion, observacion_bot, contactar_por, credito, crm_id_cliente, crm_id_trato, crm_medio_id, crm_proyecto, digitoverificador, modelo, motivo_de_problema, pie, rentadeclarada, crm_Nombre_vendedor, boton_interes, primer_boton, boton_seguimiento, primera_respuesta, crm_Rut_vendedor, fecha_registro ) VALUES ( %(whatsapp_user_id)s, %(whatsapp_phone)s, %(source)s, %(last_seen)s, %(signed_up)s, %(sessions)s, %(last_visited_block_name)s, %(last_visited_block_id)s, %(last_clicked_button_name)s, %(entrada_del_usuario)s, %(numero_interacciones)s, %(motivo_de_consulta)s, %(topico)s, %(proyecto_id)s, %(proyecto_consultado)s, %(acuerdo)s, %(medio_de_contacto)s, %(dentro_de_presupuesto)s, %(renta)s, %(situacion_de_vivienda)s, %(inicial)s, %(fechaentrega)s, %(nombre_cliente)s, %(apellido_cliente)s, %(numero_ambientes)s, %(dni)s, %(fecha_de_visita)s, %(horario)s, %(mail)s, %(crm_email)s, %(estado_flujo)s, %(nivel_de_interes)s, %(subsidio)s, %(subsidio_puede_postular)s, %(link_de_la_conversacion)s, %(observacion_bot)s, %(contactar_por)s, %(credito)s, %(crm_id_cliente)s, %(crm_id_trato)s, %(crm_medio_id)s, %(crm_proyecto)s, %(digitoverificador)s, %(modelo)s, %(motivo_de_problema)s, %(pie)s, %(rentadeclarada)s, %(crm_Nombre_vendedor)s, %(boton_interes)s, %(primer_boton)s, %(boton_seguimiento)s, %(primera_respuesta)s, %(crm_Rut_vendedor)s, %(fecha_registro)s ) """ cursor.execute(sql_insert, user_data) last_id = cursor.lastrowid connection.commit() logging.info(f"🟢 INSERT por (wa_id, last_seen) inexistente: {user_data['whatsapp_user_id']} -> {last_id}") return {'action': 'inserted', 'id': last_id, 'enviado_crm': False} # ───────────────────────────────────────────────────────────── # Mapeo idéntico al endpoint # ───────────────────────────────────────────────────────────── def build_user_data(data: dict): proyecto_id_str = data.get('proyecto id') if proyecto_id_str is None or proyecto_id_str == 'null': proyecto_id = 0 else: proyecto_id = int(proyecto_id_str) crm_id_trato_raw = data.get('crm id trato') or data.get('crm_id') or data.get('crm id') crm_id_trato = None if crm_id_trato_raw in [None, "", "null", "NULL"] else crm_id_trato_raw medio_llegada_raw = data.get('medio_llegada_id', '') if medio_llegada_raw and str(medio_llegada_raw).isdigit(): medio_llegada_id = int(medio_llegada_raw) else: medio_llegada_id = None rut_vendedor_raw = data.get('crm_Rut vendedor') if rut_vendedor_raw and str(rut_vendedor_raw).isdigit(): crm_rut_vendedor = int(rut_vendedor_raw) else: crm_rut_vendedor = None user_data = { 'whatsapp_user_id': data.get('whatsapp user id'), 'whatsapp_phone': ''.join(filter(str.isdigit, data.get('whatsapp phone', ''))), 'source': data.get('source'), 'last_seen': parse_date(data.get('last seen')), 'signed_up': parse_date(data.get('signed up')), 'sessions': data.get('sessions'), 'last_visited_block_name':data.get('last visited block name'), 'last_visited_block_id': data.get('last visited block id'), 'last_clicked_button_name':data.get('last clicked button name'), 'entrada_del_usuario': data.get('entrada del usuario'), 'numero_interacciones': data.get('numero interacciones'), 'motivo_de_consulta': data.get('motivo de problema') or data.get('motivo_de_consulta'), 'topico': data.get('topico'), 'proyecto_id': proyecto_id, 'proyecto_consultado': data.get('proyecto_consultado'), 'acuerdo': data.get('acuerdo') or data.get('Acuerdo'), 'medio_de_contacto': data.get('medio de contacto') or data.get('crm_medio_llegada') or data.get('medio_de_contacto'), 'dentro_de_presupuesto': data.get('dentro de presupuesto'), 'renta': data.get('rentadeclarada'), 'situacion_de_vivienda': data.get('situacion_de_vivienda'), 'inicial': data.get('inicial'), 'fechaentrega': data.get('fechaentrega'), 'nombre_cliente': data.get('crm_nombre_cliente') or data.get('nombre_cliente'), 'apellido_cliente': data.get('crm_apellido_cliente') or data.get('apellido_cliente'), 'numero_ambientes': data.get('numero ambientes'), 'dni': data.get('dni') or data.get('dni_sin_verificador') or data.get('DNI'), 'fecha_de_visita': data.get('fecha de visita'), 'horario': data.get('horario'), 'mail': data.get('mail'), 'crm_email': data.get('crm_email') or data.get('crm_email_cliente'), 'estado_flujo': data.get('estado_flujo'), 'nivel_de_interes': parse_int( data.get('nivel_de_interes') or data.get('nivel de interes') or data.get('nivel_de_interes_crm') ), 'subsidio': data.get('subsidio'), 'subsidio_puede_postular':data.get('subsidio_puede_postular'), 'link_de_la_conversacion': f"https://dashboard.chatfuel.com/bot/{CHATFUEL_BOT_ID}/livechat?folder=all&conversationId=wa_{data.get('whatsapp user id')}&platform=whatsapp", 'observacion_bot': data.get('observacion_bot'), 'contactar_por': data.get('contactar_por'), 'credito': data.get('credito'), 'crm_id_cliente': data.get('crm_id_cliente') or data.get('id_cliente'), 'id_cliente': data.get('id_cliente') or data.get('crm_id_cliente'), 'crm_id_trato': crm_id_trato, 'crm_medio_id': data.get('crm_medio_id'), 'crm_proyecto': data.get('crm_proyecto'), 'digitoverificador': data.get('digitoverificador') or data.get('digito_dni'), 'digito_dni': data.get('digito_dni') or data.get('digitoverificador'), 'modelo': data.get('modelo'), 'motivo_de_problema': data.get('motivo_de_problema'), 'pie': data.get('pie'), 'rentadeclarada': data.get('rentadeclarada'), 'crm_Rut_vendedor': (data.get('crm_Rut vendedor') or data.get('Rut vendedor') if (data.get('crm_Rut vendedor') or data.get('Rut vendedor')) and str(data.get('crm_Rut vendedor') or data.get('Rut vendedor')).isdigit() else None), 'crm_Nombre_vendedor': data.get('crm Nombre_vendedor'), 'boton_interes': data.get('boton_interes'), 'primer_boton': data.get('primer_boton'), 'boton_seguimiento': data.get('boton_seguimiento'), 'primera_respuesta': data.get('primera_respuesta'), 'medio_llegada_id': medio_llegada_id } # Normaliza precisión temporal user_data['last_seen'] = normalize_dt(user_data['last_seen']) user_data['signed_up'] = normalize_dt(user_data['signed_up']) return user_data # ───────────────────────────────────────────────────────────── # PlanOK/GCI (no reenvía si ya enviado_crm = TRUE) # ───────────────────────────────────────────────────────────── def maybe_send_to_planok_if_needed(user_data, already_sent: bool): """Envia a GCI solo si: - estado_flujo == 'completo' - ENVIAR_PLANOK == '1' - y aún NO fue enviado (already_sent == False) """ estado = (user_data.get('estado_flujo') or '').strip().lower() if estado != "completo": return if ENVIAR_PLANOK != "1": return if already_sent: logging.info("⛔ Ya marcado enviado_crm anteriormente. No se reenvía a GCI.") return # a) Resolver id_cliente / cliente_id_gci cliente_id_gci = None id_cliente_str = str(user_data.get('id_cliente', '')).strip() if id_cliente_str.isdigit(): posible_id = int(id_cliente_str) if posible_id > 0: logging.info(f"id_cliente válido ya presente: {posible_id}") cliente_id_gci = posible_id else: logging.info(f"id_cliente={posible_id} <= 0; se buscará/creará.") else: logging.info("Sin id_cliente válido; se buscará/creará cliente en GCI.") if not cliente_id_gci: dni_int = None if user_data.get('dni'): try: dni_int = int(user_data['dni']) except ValueError: logging.warning(f"DNI no convertible a int: {user_data['dni']}") if dni_int: cliente_id_gci = buscar_cliente_natural(dni_int) if cliente_id_gci: logging.info(f"Cliente ya existe en GCI con ID={cliente_id_gci}") else: digito = user_data.get('digito_dni') or user_data.get('digitoverificador') or "" nombre = user_data.get('nombre_cliente', '') apellidopat = user_data.get('apellido_cliente', '') or '' celular = user_data['whatsapp_phone'] email_final = user_data['mail'] or user_data['crm_email'] or "" logging.info(f"Creando cliente GCI: RUT={dni_int}-{digito}, nombre={nombre} {apellidopat}") nuevo_id = crear_cliente_natural( identificador_personal=dni_int, digito_verificador=digito, nombre=nombre, apellido_paterno=apellidopat, apellido_materno="", celular=celular, email=email_final ) cliente_id_gci = nuevo_id update_as_sent(user_data['whatsapp_user_id']) logging.info(f"Cliente creado en GCI con ID={cliente_id_gci}") else: logging.info("No se busca/crea cliente: sin DNI válido.") # b) Seguimiento si proyecto activo id_medio_cf = parse_int(user_data.get('medio_llegada_id')) id_espectativa_cf = parse_int(user_data.get('nivel_de_interes')) observacion_bot_cf = user_data.get('observacion_bot') proyecto_id = user_data.get('proyecto_id') rutvendedor_dni = user_data.get('crm_Rut_vendedor') if not rutvendedor_dni or str(rutvendedor_dni).strip().lower() in ('', '0', 'none', 'null'): rutvendedor_dni = obtener_rut_vendedor_por_proyecto(proyecto_id) try: rut_int = int(str(rutvendedor_dni).strip()) if rut_int < 1000000: rutvendedor_dni = None except (TypeError, ValueError): rutvendedor_dni = None if proyecto_id and isinstance(proyecto_id, int) and (proyecto_id in PROYECTOS_ACTIVO): if cliente_id_gci: fecha_recontacto_str = datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ') resp_seguimiento = crear_seguimiento( id_cliente=cliente_id_gci, id_proyecto=proyecto_id, id_medio=id_medio_cf, id_tipo_contacto=TIPO_CONTACTOPOK, id_expectativa=id_espectativa_cf, comentario=observacion_bot_cf, fecha_recontacto_str=fecha_recontacto_str, mantener_ejecutivo=False, rutvendedor=rutvendedor_dni ) if resp_seguimiento is not None: logging.info(f"Seguimiento creado en GCI, ID={resp_seguimiento}") else: logging.warning("No se pudo crear el seguimiento o API no devolvió ID.") update_as_sent(user_data['whatsapp_user_id']) else: logging.info("No se crea seguimiento: cliente_id_gci es None.") else: logging.info("No se crea seguimiento: proyecto_id no activo o inexistente.") # ───────────────────────────────────────────────────────────── # Procesamiento de un archivo (solo lectura) # ───────────────────────────────────────────────────────────── def process_one_file(path: str): try: with open(path, 'rb') as f: raw = f.read() data = json.loads(raw.decode('utf-8')) user_data = build_user_data(data) if not user_data['whatsapp_user_id'] or not user_data['whatsapp_phone']: raise ValueError("Los campos 'whatsapp user id' y 'whatsapp phone' no pueden ser nulos") # fecha_registro = mtime del archivo (cuando llegó) fecha_registro_dt = file_mtime_as_dt(path) # UPSERT por (whatsapp_user_id, last_seen) + set fecha_registro desde archivo upsert_result = upsert_user_interaction(user_data, fecha_registro_dt) # Sincroniza Json_full (última versión vista para ese usuario) save_json_full_to_db(raw) if upsert_result['action'] == 'inserted': logging.info(f"Lead insertado con ID interno: {upsert_result['id']}") else: logging.info("Lead sobre-escrito (par existente).") # No re-enviar a GCI si ya estaba enviado maybe_send_to_planok_if_needed(user_data, already_sent=upsert_result['enviado_crm']) return True, None except Exception as e: return False, str(e) # ───────────────────────────────────────────────────────────── # Batch # ───────────────────────────────────────────────────────────── def run_batch(folder: str, limit: int = None): files = list_json_files_sorted(folder) if limit is not None and limit > 0: files = files[:limit] total = len(files) ok = 0 errors = 0 logging.info(f"📁 Carpeta: {folder}") logging.info(f"🗂️ Archivos a procesar: {total}") for idx, path in enumerate(files, start=1): logging.info(f"[{idx}/{total}] Procesando: {os.path.basename(path)}") success, err = process_one_file(path) if success: ok += 1 logging.info(f"[{idx}/{total}] ✅ OK: {os.path.basename(path)}") else: errors += 1 logging.error(f"[{idx}/{total}] ❌ Error: {os.path.basename(path)} -> {err}") logging.info("────────────────────────────────────────") logging.info(f"✅ OK: {ok} ❌ Errores: {errors} 🧮 Total: {total}") # ───────────────────────────────────────────────────────────── # CLI # ───────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser(description="Procesa en batch archivos JSON tipo /inter (solo lectura).") parser.add_argument("--folder", "-f", default="interacciones", help="Carpeta con los .json") parser.add_argument("--limit", "-n", type=int, default=None, help="Procesar solo los primeros N") args = parser.parse_args() run_batch(args.folder, args.limit) if __name__ == "__main__": main()