import os import sys import openai import mysql.connector from dotenv import load_dotenv from datetime import datetime from typing import List from pydantic import BaseModel import json from db_utils import get_db, save_message load_dotenv('../.env') #openai.api_key = os.getenv("OPENAI_API_KEY") MESSAGE_LIMIT = int(os.getenv("MESSAGE_ANALYSIS_LIMIT", 100)) client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY")) class LeadAnalysis(BaseModel): is_lead: bool funnel_stage: str topics_detected: List[str] possible_project: str search_criteria: str suggested_next_action: str def obtener_conversaciones_por_contacto(db): cursor = db.cursor(dictionary=True) cursor.execute("SELECT id FROM contact where source != 'email'") contactos = cursor.fetchall() conversaciones = [] for contacto in contactos: # Obtener última fecha de análisis cursor.execute("SELECT MAX(analyzed_at) AS last_analysis FROM message_analysis WHERE contact_id = %s", (contacto["id"],)) last_analysis = cursor.fetchone()["last_analysis"] # Verificar si hay mensajes nuevos desde la última fecha de análisis if last_analysis: cursor.execute(""" SELECT COUNT(*) AS nuevos FROM message WHERE contact_id = %s and channel != 'crm' AND timestamp > %s """, (contacto["id"], last_analysis)) if cursor.fetchone()["nuevos"] == 0: continue # No hay mensajes nuevos cursor.execute(f""" SELECT m.*, u.name AS user_name FROM message m JOIN user_account u ON m.user_id = u.id WHERE m.contact_id = %s and m.channel != 'crm' ORDER BY m.timestamp ASC LIMIT {MESSAGE_LIMIT} """, (contacto["id"],)) mensajes = cursor.fetchall() if mensajes: conversaciones.append((contacto["id"], mensajes)) return conversaciones def construir_mensaje_texto(mensajes): texto = "" total_tokens = 0 for msg in reversed(mensajes): autor = "VENDEDOR" if msg["direction"] == "outgoing" else "LEAD" # Formatear la fecha y hora del mensaje timestamp = msg["timestamp"] if isinstance(timestamp, str): fecha_hora = timestamp else: fecha_hora = timestamp.strftime("%Y-%m-%d %H:%M:%S") linea = f"[{fecha_hora}] {msg['subject']} {autor}: {msg['content']}\n" #tokens = len(encoder.encode(linea)) #if total_tokens + tokens > MAX_PROMPT_TOKENS: # break texto = linea + texto #total_tokens += tokens return texto def analizar_conversacion(conversacion): try: response = client.responses.create( model="gpt-4o", input=[ { "role": "system", "content": """ Sos un asistente especializado en ventas inmobiliarias para la desarrolladora Quartier (marca de Argencons S.A.). Vas a analizar conversaciones de WhatsApp entre leads y vendedores. Tu tarea es: 1. Leer el historial de mensajes. 2. Determinar si es un lead válido. 3. Clasificar al lead en una etapa del embudo de ventas. 4. Sugerir la próxima acción que debería realizar el asesor comercial (si corresponde). IMPORTANTE: Si en la conversación figura algun mensaje de envío de formulario de Meta (ej. "Completó formulario Meta: ..." ), **no lo consideres como una conversación real**. Ese tipo de mensajes pueden ser enviados por error, de forma automática o por clics accidentales. ### Etapas del embudo (`funnel_stage`): - **new**: El lead aún no respondió ningún mensaje o solo figura un mensaje de ingreso desde formulario automático. - **warm**: Mostró interés general. Pidió precios, ubicación o información general, pero sin intención clara de avanzar. Ejemplos: "¿Cuánto están?", "¿Dónde queda?", "Pasame más info", "Me interesa". - **appt**: Está evaluando seriamente avanzar. Mencionó que quiere hablar, coordinar una reunión o visitar. También si dice estar analizando con alguien más. Ejemplos: "Podemos agendar?", "Quiero ver el lugar", "Estoy decidiendo con mi pareja". - **cold**: Había conversación (warm o appt) pero **dejó de responder durante más de 10 días**. Se considera inactivo. - **won**: Confirmó compra, reserva o seña. - **lost**: Dijo explícitamente que no está interesado o fue marcado como perdido luego de 2 follow-ups sin respuesta en 30 días. - **wrong**: Mencionó algo que no ofrecemos: alquiler, zonas no disponibles, otros productos que no son real estate. - **no_lead**: No se logró establecer contacto durante 30 días, o el mensaje recibido es completamente irrelevante o automatizado (sin señales de interés humano). ### Reglas adicionales: - No asumas `appt` si solo pidió precios o info. Eso es `warm`. - Si estaba en conversación (`warm` o `appt`) y dejó de responder por más de 10 días, pasalo a `cold`. - Si vuelve a responder luego de estar en `cold` o `lost`, la acción sugerida debe ser `reactivar`. ### Acciones posibles (`suggested_next_action`): - **enviar_info**: Si pidió datos, precios o brochures. - **followup_1**: Si hace 3 días no responde después de enviar info o propuesta. - **followup_2**: Si hace 7 días no responde después del followup_1. - **agendar_meet**: Si quiere hablar, entender más, o necesita definición. - **agendar_visita**: Si quiere conocer el lugar, showroom o la obra. - **reactivar**: Si estaba marcado como perdido o frío y vuelve a hablar. - **cerrar**: Confirmó compra o seña. - **esperar**: La conversación está activa y reciente (<48hs). - **archivar**: No respondió en más de 2 semanas o no hay interés real. Actuá como un analista de ventas inmobiliarias con criterio y sentido común. """ }, {"role": "user", "content": conversacion} ], text={ "format": { "type": "json_schema", "name": "lead_analysis", "schema": { "type": "object", "properties": { "is_lead": { "type": "boolean", "description": "es un lead" }, "funnel_stage": { "type": "string", "description": "etapa del embudo de ventas según tabla crestados", "enum": [ "new", "warm", "appt", "cold", "won", "lost", "wrong", "no_lead" ] }, "topics_detected": { "type": "array", "items": { "type": "string" } }, "possible_project": { "type": "string", "description": "proyecto posible de interés", "enum": [ "Quartier Juan B. Justo", "Quartier Brava 30", "Quartier Bajo Belgrano", "Quartier +Colonia", "Quartier Del Bajo", "Distrito Quartier", "Edificio Plaza", "Quartier Lacroze", "Quartier Dorrego", "Quartier Nordelta", "Otros Quartier" ] }, "search_criteria": { "type": "string", "description": "criterios de búsqueda" }, "suggested_next_action": { "type": "string", "description": "siguiente acción sugerida según tabla cracciones", "enum": [ "followup_1", "followup_2", "enviar_info", "reactivar", "cerrar", "agendar_meet", "agendar_visita", "esperar", "archivar" ] } }, "required": [ "is_lead", "funnel_stage", "topics_detected", "possible_project", "search_criteria", "suggested_next_action" ], "additionalProperties": False }, "strict": True } } ) return LeadAnalysis.model_validate(json.loads(response.output_text)) except Exception as e: print(f"❌ Error al llamar a OpenAI: {e}") return None def guardar_analisis(db, contact_id, datos: LeadAnalysis): cursor = db.cursor() resumen = f"Lead: {'sí' if datos.is_lead else 'no'}, Proyecto: {datos.possible_project}, Estado: {datos.funnel_stage}, Acción: {datos.suggested_next_action}" # Obtener el estado y acción actual del contacto cursor.execute("SELECT status, accion FROM contact WHERE id = %s", (contact_id,)) contacto_actual = cursor.fetchone() estado_anterior = contacto_actual[0] if contacto_actual else None accion_anterior = contacto_actual[1] if contacto_actual else None # Insertar el análisis cursor.execute(""" INSERT INTO message_analysis ( contact_id, is_lead, funnel_stage, topics_detected, suggested_next_action, possible_project, search_criteria, analyzed_at, resumen ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( contact_id, datos.is_lead, datos.funnel_stage, ', '.join(datos.topics_detected), datos.suggested_next_action, datos.possible_project, datos.search_criteria, datetime.now(), resumen )) # Actualizar estado y acción del contacto cursor.execute(""" UPDATE contact SET status = %s, accion = %s WHERE id = %s """, (datos.funnel_stage, datos.suggested_next_action, contact_id)) # Verificar si hubo cambios en estado o acción cambio_estado = estado_anterior != datos.funnel_stage cambio_accion = accion_anterior != datos.suggested_next_action hubo_cambios = cambio_estado or cambio_accion if hubo_cambios: # Determinar qué cambió cambios = [] if cambio_estado: cambios.append(f"estado: {estado_anterior or 'sin estado'} → {datos.funnel_stage}") if cambio_accion: cambios.append(f"acción: {accion_anterior or 'sin acción'} → {datos.suggested_next_action}") cambios_texto = " y ".join(cambios) # Insertar actividad en message sobre el análisis realizado contenido_actividad = f"Analizado por IA - Cambios: {cambios_texto} - Proyecto: *{datos.possible_project}*" cursor.execute(""" INSERT INTO message (user_id, contact_id, direction, content, timestamp, channel, automated) VALUES (%s, %s, %s, %s, %s, %s, %s) """, (-1, contact_id, '', contenido_actividad, datetime.now(), 'crm', True)) print(f"📝 Actividad registrada: {contenido_actividad}") db.commit() return resumen def obtener_conversacion_contacto_especifico(db, contact_id): """Obtiene la conversación de un contacto específico""" cursor = db.cursor(dictionary=True) # Verificar que el contacto existe cursor.execute("SELECT id FROM contact WHERE id = %s", (contact_id,)) contacto = cursor.fetchone() if not contacto: print(f"❌ Contacto {contact_id} no encontrado") return None cursor.execute(f""" SELECT m.*, u.name AS user_name FROM message m JOIN user_account u ON m.user_id = u.id WHERE m.contact_id = %s and m.channel != 'crm' ORDER BY m.timestamp ASC LIMIT {MESSAGE_LIMIT} """, (contact_id,)) mensajes = cursor.fetchall() if not mensajes: print(f"❌ No se encontraron mensajes para el contacto {contact_id}") return None return (contact_id, mensajes) def main(): db = get_db() # Verificar si se proporcionó un ID de contacto como argumento if len(sys.argv) > 1: try: contact_id = int(sys.argv[1]) print(f"🎯 Analizando contacto específico: {contact_id}") conversacion = obtener_conversacion_contacto_especifico(db, contact_id) if not conversacion: print("❌ No se pudo obtener la conversación del contacto especificado") return contact_id, mensajes = conversacion print(f"🔍 Analizando mensajes del contacto {contact_id}: {len(mensajes)} mensajes...") texto = construir_mensaje_texto(mensajes) print(texto) respuesta = analizar_conversacion(texto) if respuesta: try: resume = guardar_analisis(db, contact_id, respuesta) print(f"✅ Análisis guardado para contacto {contact_id}: {resume}") except Exception as e: print(f"❌ Error al parsear/guardar respuesta: {e}\nRespuesta:\n{respuesta}") else: print(f"❌ No se pudo analizar la conversación del contacto {contact_id}") except ValueError: print("❌ Error: El ID del contacto debe ser un número entero") print("Uso: python analyze_conversations.py [contact_id]") return else: # Análisis de todas las conversaciones (comportamiento original) conversaciones = obtener_conversaciones_por_contacto(db) print(f"🧠 Analizando {len(conversaciones)} conversaciones...") for contact_id, mensajes in conversaciones: print(f"🔍 Analizando mensajes del contacto {contact_id} : {len(mensajes)} mensajes...") texto = construir_mensaje_texto(mensajes) respuesta = analizar_conversacion(texto) if respuesta: try: resume = guardar_analisis(db, contact_id, respuesta) print(f"✅ Análisis guardado para contacto {contact_id}: {resume}") except Exception as e: print(f"❌ Error al parsear/guardar respuesta: {e}\nRespuesta:\n{respuesta}") if __name__ == '__main__': main() # Luego tendra que hacer # insert into message(user_id, contact_id, direction, content, timestamp, channel, automated) select -1, contact_id, '', concat('Analizado por IA , estado *', funnel_stage, '* , accion sugerida *',suggested_next_action, '* en proyecto *',possible_project, '*') as content, analyzed_at as timestamp, 'crm', true from message_analysis where status is null; # update contact set status = (select funnel_stage from message_analysis where contact_id = contact.id order by analyzed_at desc limit 1), accion = (select suggested_next_action from message_analysis where contact_id = contact.id order by analyzed_at desc limit 1); """ Uso del script: - Sin argumentos: Analiza todas las conversaciones que necesitan análisis python analyze_conversations.py - Con ID de contacto: Analiza solo el contacto especificado python analyze_conversations.py 123 """