import os import uuid import tempfile import pandas as pd from openai import OpenAI from utils.utils import limpiar_nombre_archivo, preparar_bloques_json, preparar_bloques_texto_redes, markdown_a_word_docx, extraer_texto_de_docx, validar_y_leer_excel_encuesta, validar_y_leer_excel_redes, combinar_archivos_docx, clean_text, sintetizar_resumenes_por_pregunta, cargar_prompt from utils.prompts import obtener_prompts_por_informe from analysis.global_summary import generar_resumen_global from analysis.global_summary_final import generar_resumen_global_final from datetime import datetime import zipfile # Inicializar cliente OpenAI openai = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) PALABRAS_SOCIODEMOGRAFICAS = [ "edad", "género", "genero", "sexo", "nivel educativo", "educación", "educacion", "escolaridad", "ocupación", "profesión", "profesion", "ingresos", "ingreso", "trabajo", "zona", "lugar de residencia", "distrito", "estrato", "estado civil" ] def es_pregunta_sociodemografica(nombre_pregunta: str) -> bool: texto = nombre_pregunta.lower() return any(palabra in texto for palabra in PALABRAS_SOCIODEMOGRAFICAS) def analizar_excel_encuesta_y_generar_pdf(archivo_excel, informe): """ Analiza respuestas abiertas de encuestas cargadas desde un archivo Excel (.xlsx) y genera resúmenes por pregunta utilizando modelos de lenguaje (GPT). La función identifica cada columna como una pregunta, aplica un prompt específico por pregunta, y genera resúmenes individuales por bloques de respuestas. Luego, sintetiza un resumen global por pregunta (exceptuando preguntas sociodemográficas) y produce un informe general en formato Word. Parámetros: - archivo_excel (str): Ruta al archivo .xlsx o CSV cargado por el usuario. - informe (str): Tipo de informe a generar (actualmente soporta "Análisis Focalizado por Pregunta"). Yields: - Tuple: - [ruta_zip]: Lista con la ruta al archivo ZIP que contiene: - resúmenes por pregunta en formato Word (.docx) - un resumen global unificado (excluyendo preguntas sociodemográficas) - mensaje (str): Mensaje de estado o error. - progreso (str): Descripción textual del progreso del análisis. """ try: # Validar que el archivo fue cargado correctamente if archivo_excel is None: yield None, "❌ Por favor, sube un archivo Excel o csv válido antes de generar el análisis.", None return # Leer el archivo y validar formato df, error_msg = validar_y_leer_excel_encuesta(archivo_excel) if error_msg: yield None, error_msg, None return nombre_excel = archivo_excel.name base_name = os.path.splitext(os.path.basename(nombre_excel))[0] if informe == "Análisis Focalizado por Pregunta": # Preparar prompts para cada pregunta usar_contexto_entre_claves = False plantilla_prompt = cargar_prompt("prompts/analisis_pregunta.txt") prompts = {} for i, pregunta in enumerate([col for col in df.columns if str(col).strip()]): pregunta_limpia = str(pregunta).strip() clave = f"pregunta_{i+1:02d}" # ejemplo: pregunta_01, pregunta_02, etc. prompts[clave] = { "nombre": pregunta_limpia, "system": "Eres un analista especializado en interpretación de respuestas abiertas en encuestas. Tu tarea es identificar argumentos, preocupaciones, emociones, patrones comunes y ejemplos ilustrativos expresados por los participantes en relación con la pregunta planteada.", "user_prompt": plantilla_prompt.replace("{{PREGUNTA}}", pregunta_limpia) } if not prompts: yield None, "⚠️ No se encontraron preguntas válidas en el archivo.", None return # Preparar bloques de texto para cada pregunta bloques_json = preparar_bloques_json(nombre_excel) # Crear carpeta temporal para almacenar resúmenes session_id = str(uuid.uuid4())[:8] temp_dir = os.path.join(tempfile.gettempdir(), f"resumenes_{session_id}") os.makedirs(temp_dir, exist_ok=True) resumen_dir_global = os.path.join(temp_dir, "resumenes_globales") os.makedirs(resumen_dir_global, exist_ok=True) # Crear directorios para resúmenes por clave for clave in prompts: os.makedirs(os.path.join(temp_dir, f"resumenes_{clave}"), exist_ok=True) # Iterar por cada bloque de respuestas for i, bloque_json in enumerate(bloques_json, start=1): message_history = [] # Iterar por cada clave de pregunta y generar resumen for j, (clave, prompt_info) in enumerate(prompts.items()): if usar_contexto_entre_claves: if j == 0: # Primer bloque inicia con contexto limpio message_history = [ {"role": "system", "content": prompt_info["system"]}, {"role": "user", "content": prompt_info["user_prompt"] + "\n\n```json\n" + str(bloque_json) + "\n```"} ] else: # Consecutivos bloques usan el contexto acumulado # (es decir, el resumen anterior puede influir en la próxima respuesta) message_history.append({ "role": "user", "content": prompt_info["user_prompt"] + "\n\n```json\n" + str(bloque_json) + "\n```" }) else: # Si no se usa contexto entre claves, reiniciar mensaje de sistema message_history = [ {"role": "system", "content": prompt_info["system"]}, {"role": "user", "content": prompt_info["user_prompt"] + "\n\n```json\n" + str(bloque_json) + "\n```"} ] # Generar resumen con OpenAI try: response = openai.chat.completions.create( model="gpt-4o-mini", messages=message_history, temperature=0.4 ) resumen = response.choices[0].message.content if usar_contexto_entre_claves: # Si se usa contexto entre claves, actualizar el mensaje de assistant message_history.append({"role": "assistant", "content": resumen}) except Exception as e: resumen = f"⚠️ Error en bloque {i} ({clave}): {str(e)}" # Guardar resumen en archivo .md por clave resumen_dir_clave = os.path.join(temp_dir, f"resumenes_{clave}") path = os.path.join(resumen_dir_clave, f"resumen_bloque_{i}.md") with open(path, "w", encoding="utf-8") as f: f.write(resumen) yield None, None, f"\n\n🧩 **Resumen del bloque {i}/{len(bloques_json)} para clave ({clave})**\n{'-'*80}" # Resumen global por clave resumenes_docx_por_clave = [] total_claves = len(prompts) for idx, (clave, prompt_info) in enumerate(prompts.items(), start=1): yield None, None, f"📝 Generando resumen global {idx}/{total_claves} para: **{prompt_info['nombre']}**..." resumenes_dir_clave = os.path.join(temp_dir, f"resumenes_{clave}") resumen_md = os.path.join(resumen_dir_global, f"{clave}_resumen.md") resumen_docx = os.path.join(resumen_dir_global, f"{clave}_resumen.docx") generar_resumen_global(resumenes_dir_clave, resumen_md) markdown_a_word_docx(resumen_md, resumen_docx) resumenes_docx_por_clave.append(resumen_docx) # Concatenar todos los resúmenes globales yield None, None, "📚 Generando resumen global unificado del informe completo..." resumen_final_md = os.path.join(temp_dir, "resumen_final.md") # OPCION 1: unir todos los _resumen.md en un solo archivo with open(resumen_final_md, "w", encoding="utf-8") as f_final: f_final.write(f"# 🧾 Informe: {informe}\n\n") for clave, prompt_info in prompts.items(): resumen_md = os.path.join(resumen_dir_global, f"{clave}_resumen.md") if os.path.exists(resumen_md): with open(resumen_md, "r", encoding="utf-8") as f_in: nombre_pregunta = prompt_info["nombre"] if not es_pregunta_sociodemografica(nombre_pregunta): contenido = f_in.read() f_final.write(f"\n\n## 📌 {prompt_info['nombre']}\n\n") f_final.write(contenido) else: yield None, f"⏭️ Omitiendo pregunta sociodemográfica: **{nombre_pregunta}** del informe final...", None resumen_final_docx = os.path.join(temp_dir, f"{base_name}_resumen_global.docx") markdown_a_word_docx(resumen_final_md, resumen_final_docx) # Crear archivo ZIP con todos los .docx (por clave + resumen global) zip_path = os.path.join(temp_dir, f"{base_name}_resumenes.zip") with zipfile.ZipFile(zip_path, "w") as zipf: # Incluir resumenes por clave for docx_file in resumenes_docx_por_clave: zipf.write(docx_file, arcname=os.path.basename(docx_file)) # Incluir resumen global zipf.write(resumen_final_docx, arcname=os.path.basename(resumen_final_docx)) if os.path.exists(zip_path): yield [zip_path], "✅ Análisis completado correctamente.", "✅ ZIP con todos los resúmenes generados" else: yield None, "⚠️ No se pudo generar el archivo ZIP.", None except Exception as e: yield None, f"⚠️ Error inesperado: {str(e)}", None def analizar_excel_redes_y_generar_pdf(archivo_excel, informe, nombre_cliente=None, nombre_competidor=None): """ Analiza comentarios de redes sociales a partir de un archivo Excel, CSV o un archivo .pkl con un DataFrame, y genera informes automáticos por red social utilizando modelos de lenguaje (GPT). Según el tipo de informe solicitado, aplica distintos prompts para generar análisis como desempeño, contenido temático, comparaciones o diagnóstico estratégico. Los resultados se consolidan y devuelven en un archivo .zip con documentos Word (.docx) por red y un resumen global. Parámetros: - archivo_excel (str): Ruta al archivo a procesar. Puede ser: • un archivo Excel (.xlsx) o CSV (.csv) subido por el usuario • un archivo .pkl generado desde la base de datos con comentarios ya cargados - informe (str): Tipo de informe solicitado por el usuario. - nombre_cliente (str, opcional): Nombre del cliente (si aplica según el informe). - nombre_competidor (str, opcional): Nombre del competidor (si aplica según el informe). Yields: - Tuple: - [ruta_zip]: Lista con la ruta al archivo ZIP generado (o None si hubo error). - mensaje (str): Estado final o mensaje de error. - progreso (str): Texto informativo con el detalle del avance. """ try: # Validación del archivo de entrada if archivo_excel is None: yield None, "❌ Por favor, sube un archivo Excel o csv válido antes de generar el análisis.", None return df, error_msg = validar_y_leer_excel_redes(archivo_excel) if error_msg: yield None, error_msg, None return # Limpieza básica del texto y detección de redes sociales df["texto"] = df["texto"].apply(clean_text) redes_presentes = sorted(df["red_social"].dropna().unique().tolist()) lista_redes = ", ".join(redes_presentes) # Obtención de los prompts según el tipo de informe seleccionado prompts, error_prompt = obtener_prompts_por_informe(informe, nombre_cliente, nombre_competidor) if error_prompt: yield None, error_prompt, None return # Agrupación de comentarios por red social grouped = df.groupby("red_social")["texto"].apply(list) # Carpeta temporal donde se guardarán los resúmenes y archivos session_id = str(uuid.uuid4())[:8] temp_dir = os.path.join(tempfile.gettempdir(), f"resumenes_{session_id}") os.makedirs(temp_dir, exist_ok=True) archivos_finales = [] # Iterar por cada red social (Twitter, Facebook, etc.) for red, textos in grouped.items(): yield None, None, f"📊 Generando análisis para {red}..." # 🚫 Saltear redes sin texto válido if not textos or all(t.strip() == "" for t in textos): yield None, f"⚠️ La red {red} no contiene texto válido para analizar.", None continue # Crear directorio temporal para resúmenes de esta red resumen_dir_red = os.path.join(temp_dir, red.lower()) os.makedirs(resumen_dir_red, exist_ok=True) # Preparar bloques de texto (para evitar límite de tokens) bloques = preparar_bloques_texto_redes(textos) if not bloques: yield None, f"⚠️ No se pudieron generar bloques de texto para {red}. Verifica el contenido.", None continue # Crear directorios para resúmenes por clave for clave in prompts: resumen_dir_clave = os.path.join(resumen_dir_red, f"resumenes_{clave}") os.makedirs(resumen_dir_clave, exist_ok=True) # Iterar por cada bloque de comentarios for i, bloque in enumerate(bloques, start=1): comentarios = [c.strip() for c in bloque.split("\n") if c.strip()] if not comentarios: yield None, f"⚠️ El bloque {i} de {red} no contiene comentarios válidos.", None continue message_history = [] # Generar resumen por cada clave de prompt, manteniendo el contexto for clave in prompts: prompt_info = prompts[clave] user_msg = ( prompt_info["user_prompt"].replace("{{REDES_PRESENTES}}", red) + f"\n\n⚠️ Este bloque contiene solo {len(comentarios)} comentarios.\n" + "Limitá el análisis a lo observable y no infieras más allá.\n\n" + "Comentarios:\n\n" + bloque ) # Si es la primera clave, iniciamos el contexto con system y user_msg de la prinera clave if len(message_history) == 0: message_history = [ {"role": "system", "content": prompt_info["system"]}, {"role": "user", "content": user_msg} ] else: # Agregamos nuevo mensaje del usuario para el siguiente análisis encadenado (siguiente clave) message_history.append({ "role": "user", "content": user_msg }) # Intentar generar el resumen con OpenAI try: response = openai.chat.completions.create( model="gpt-4o-mini", messages=message_history, temperature=0.4 ) resumen = response.choices[0].message.content message_history.append({"role": "assistant", "content": resumen}) # Agrega respuesta al contexto except Exception as e: resumen = f"⚠️ Error al generar el resumen del bloque {i} ({clave}): {str(e)}" # Guardar resumen en directorio por clave resumen_dir_clave = os.path.join(resumen_dir_red, f"resumenes_{clave}") os.makedirs(resumen_dir_clave, exist_ok=True) path_resumen = os.path.join(resumen_dir_clave, f"resumen_bloque_{i}.md") with open(path_resumen, "w", encoding="utf-8") as f: f.write(resumen) yield None, None, f"\n\n🧩 **Resumen del bloque {i}/{len(bloques)} de la red {red} generado ({clave})**\n{'-'*80}" # Después de los bloques: resumen global por clave for clave in prompts: yield None, None, f"\n\n🧩 **Resumen global de {clave}**\n{'-'*80}" resumen_dir_clave = os.path.join(resumen_dir_red, f"resumenes_{clave}") resumen_md = os.path.join(resumen_dir_red, f"{clave}_resumen.md") generar_resumen_global(resumen_dir_clave, resumen_md) # ✅ Concatenar todos los .md de esta red en resumen_final_red yield None, None, f"📘 Generando informe final de la red **{red}**..." resumen_final_red = os.path.join(resumen_dir_red, f"resumen_final_{red.lower()}.md") with open(resumen_final_red, "w", encoding="utf-8") as f_final: f_final.write(f"# 🧾 Análisis de {red.capitalize()}\n\n") for clave in prompts.keys(): resumen_md = os.path.join(resumen_dir_red, f"{clave}_resumen.md") if os.path.exists(resumen_md): with open(resumen_md, "r", encoding="utf-8") as f_in: contenido = f_in.read() f_final.write(f"\n\n## 📌 Resumen ({clave.upper()})\n\n") f_final.write(contenido) # ✨ Generar documento Word del resumen final de la red nombre_informe_limpio = limpiar_nombre_archivo(informe.replace(" ", "_")) docx_file = os.path.join(resumen_dir_red, f"{nombre_cliente.upper()}_{nombre_informe_limpio}_{red.upper()}_resumen.docx") markdown_a_word_docx(resumen_final_red, docx_file) archivos_finales.append(docx_file) # ✅ Combinar los Words individuales por red en uno solo fecha = datetime.today().strftime("%Y%m%d") nombre_archivo = f"{nombre_cliente.upper()}_Analisis_x_Red_Social_{nombre_informe_limpio}_{fecha}.docx" archivo_docx_unico = os.path.join(temp_dir, nombre_archivo) combinar_archivos_docx(archivos_finales, archivo_docx_unico) archivos_finales.append(archivo_docx_unico) # 🧾 Leer todos los resúmenes por red texto_completo = "" for archivo in archivos_finales: if archivo.endswith("_resumen.docx"): texto = extraer_texto_de_docx(archivo) red = os.path.basename(archivo).replace("_resumen.docx", "") texto_completo += f"\n\n---\n\n📘 **Resumen de {red.capitalize()}**\n\n{texto}" # ✨ Llamada al modelo para generar resumen global resumen_global = generar_resumen_global_final(texto_completo) # 💾 Guardar en .md y .docx el resumen global resumen_md_final = os.path.join(temp_dir, f"resumen_global_{fecha}.md") with open(resumen_md_final, "w", encoding="utf-8") as f: f.write(resumen_global) resumen_docx_final = os.path.join(temp_dir, f"Resumen_Global_{nombre_cliente.upper()}_{nombre_informe_limpio}_{fecha}.docx") markdown_a_word_docx(resumen_md_final, resumen_docx_final) archivos_finales.append(resumen_docx_final) # Crear archivo ZIP con todos los .docx (por clave + resumen global) zip_path = os.path.join(temp_dir, f"{nombre_cliente.upper()}_Analisis_{nombre_informe_limpio}_{fecha}.zip") with zipfile.ZipFile(zip_path, 'w') as zipf: for file_path in archivos_finales: if os.path.exists(file_path): arcname = os.path.relpath(file_path, temp_dir) # Nombre dentro del ZIP zipf.write(file_path, arcname=arcname) # ✅ Devolver solo el ZIP para descarga if os.path.exists(zip_path): yield [zip_path], "✅ Análisis completado correctamente.", "✅ ZIP con todos los resúmenes generados" else: yield None, "⚠️ No se pudo generar el archivo ZIP.", None except Exception as e: yield None, f"⚠️ Error inesperado: {str(e)}", None