import os import uuid import httpx import logging import datetime import io import matplotlib.pyplot as plt from fpdf import FPDF from config.settings import settings logger = logging.getLogger("pdf-generator") class ExecutiveReportPDF(FPDF): def __init__( self, requester_name="Jhoana Castillo", requester_role="Jefe de TI", **kwargs ): super().__init__(**kwargs) self.requester_name = requester_name self.requester_role = requester_role def header(self): # 1. Logos Corporativos try: # Sincronizamos con los nombres reales en la carpeta assets (Case Sensitive) logo_agrota = os.path.join("assets", "LogoAgrota.jpg") logo_terra = os.path.join("assets", "LogoGrupoTerra.webp") if os.path.exists(logo_agrota): self.image(logo_agrota, 10, 8, 30) if os.path.exists(logo_terra): self.image(logo_terra, 160, 8, 30) except Exception as e: logger.warning(f"No se pudieron cargar los logos: {e}") # 2. Título y Información del Solicitante self.set_font("Arial", "B", 15) self.set_y(15) self.cell(0, 10, "Resumen Ejecutivo Felizia - Agrota", 0, 1, "C") # Datos del Solicitante y Fecha (Hora Ecuador) import pytz ecuador_tz = pytz.timezone("America/Guayaquil") now_ec = datetime.datetime.now(ecuador_tz) now_str = now_ec.strftime("%Y-%m-%d %H:%M") self.set_font("Arial", "I", 8) self.cell(0, 5, f"Generado el: {now_str} (Ecuador)", 0, 1, "C") self.ln(5) def footer(self): # Posición a 1.5 cm desde el fondo self.set_y(-15) self.set_font("Arial", "I", 8) self.cell(0, 10, f"Página {self.page_no()}", 0, 0, "C") def create_visualizations(metrics: dict) -> dict: """Genera múltiples gráficos para el reporte y retorna un dict de bytes.""" charts = {} # 1. Gráfico de Barras: Tendencia de Ventas (3 meses) try: trim_data = {k: v for k, v in metrics.items() if k.startswith("TRIM_") and k.endswith("_V")} if trim_data: fig, ax = plt.subplots(figsize=(6, 3.5)) labels = [v["label"].replace("Venta ", "") for v in trim_data.values()] values = [v.get("valor_raw", 0) for v in trim_data.values()] # Si no hay raw, intentar extraer de texto if all(v == 0 for v in values): import re values = [float(re.sub(r'[^\d.]', '', v["valor_texto"])) for v in trim_data.values()] ax.bar(labels, values, color="#4A90E2", alpha=0.8) ax.set_title("Tendencia de Ventas (Últimos 3 Meses)", fontsize=10, pad=10) ax.set_ylabel("Monto ($)", fontsize=9) plt.tight_layout() buf = io.BytesIO() plt.savefig(buf, format="png", dpi=100) plt.close() buf.seek(0) charts["ventas_trend"] = buf.read() except Exception as e: logger.error(f"Error en gráfico de tendencia: {e}") # 2. Gráfico de Pastel: Distribución UB por División try: ub_maq = metrics.get("VENTAS_UTILIDAD_BRUTA_MAQUINARIA", {}).get("valor_raw", 0) ub_agro = metrics.get("VENTAS_UTILIDAD_BRUTA_AGROQUIMICOS", {}).get("valor_raw", 0) if ub_maq > 0 or ub_agro > 0: fig, ax = plt.subplots(figsize=(5, 4)) labels = ['Maquinaria', 'Agroquímicos'] sizes = [ub_maq, ub_agro] colors = ['#FF9800', '#4CAF50'] ax.pie(sizes, labels=labels, autopct='%1.1f%%', startangle=140, colors=colors, textprops={'fontsize': 8}) ax.set_title("Distribución UB por División", fontsize=10) plt.tight_layout() buf = io.BytesIO() plt.savefig(buf, format="png", dpi=100) plt.close() buf.seek(0) charts["division_pie"] = buf.read() except Exception as e: logger.error(f"Error en gráfico de pastel: {e}") return charts def draw_kpi_card(pdf, x, y, w, h, title, value, variance=None): """Dibuja una tarjeta (card) de estilo ejecutivo para un KPI.""" # Sombra/Borde sutil pdf.set_fill_color(255, 255, 255) pdf.rect(x, y, w, h, 'F') pdf.set_draw_color(220, 220, 220) pdf.rect(x, y, w, h, 'D') # Título pdf.set_xy(x + 2, y + 3) pdf.set_font("Arial", "B", 8) pdf.set_text_color(100, 100, 100) pdf.cell(w - 4, 4, title.upper(), 0, 1, "C") # Valor Principal pdf.set_xy(x + 2, y + 10) pdf.set_font("Arial", "B", 14) pdf.set_text_color(44, 62, 80) pdf.cell(w - 4, 10, value, 0, 1, "C") # Variación (si existe) if variance and variance != "-": pdf.set_xy(x + 2, y + 20) pdf.set_font("Arial", "B", 8) is_negative = "-" in variance if is_negative: pdf.set_text_color(231, 76, 60) # Rojo icon = "v " else: pdf.set_text_color(46, 204, 113) # Verde icon = "^ " pdf.cell(w - 4, 5, f"{icon}{variance}", 0, 1, "C") pdf.set_text_color(0, 0, 0) async def generate_and_upload_executive_summary( metrics: dict, text_conclusion: str, requester_name: str, requester_role: str ) -> str: """Genera un PDF con diseño de Tablero (Cards + Gráficos).""" pdf = ExecutiveReportPDF(requester_name=requester_name, requester_role=requester_role) pdf.add_page() # --- SECCIÓN: CARDS EJECUTIVAS (Fila 1) --- pdf.ln(2) pdf.set_font("Arial", "B", 11) pdf.set_text_color(52, 152, 219) pdf.cell(0, 8, "VISTAZO GENERAL DEL PERIODO", ln=1) pdf.set_text_color(0, 0, 0) curr_y = pdf.get_y() # Card 1: Venta Total v_total = metrics.get("VENTAS_TOTALES_MES", {}).get("valor_texto", "N/D") v_var = metrics.get("VENTAS_CRECIMIENTO_MOM", {}).get("valor_texto", "-") draw_kpi_card(pdf, 10, curr_y, 45, 28, "Venta Total", v_total, v_var) # Card 2: Utilidad Bruta ub_total = metrics.get("VENTAS_UTILIDAD_BRUTA_MES", {}).get("valor_texto", "N/D") draw_kpi_card(pdf, 58, curr_y, 45, 28, "Utilidad Bruta", ub_total) # Card 3: Margen margen = metrics.get("VENTAS_MARGEN_ACTUAL", {}).get("valor_texto", "N/D") draw_kpi_card(pdf, 106, curr_y, 45, 28, "Margen", margen) # Card 4: Peso Rol peso_rol = metrics.get("AUTO_DAME_EL_PESO_DEL_ROL_SOBRE_UB", {}).get("valor_texto", "N/D") draw_kpi_card(pdf, 154, curr_y, 45, 28, "Peso Rol/UB", peso_rol) pdf.set_y(curr_y + 35) # --- SECCIÓN: GRÁFICOS --- charts = create_visualizations(metrics) import tempfile y_graficos = pdf.get_y() if "ventas_trend" in charts: tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".png") tmp.write(charts["ventas_trend"]) tmp.close() pdf.image(tmp.name, x=10, y=y_graficos, w=90) os.unlink(tmp.name) if "division_pie" in charts: tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".png") tmp.write(charts["division_pie"]) tmp.close() pdf.image(tmp.name, x=105, y=y_graficos, w=90) os.unlink(tmp.name) pdf.set_y(y_graficos + 65) # --- SECCIÓN: CARTERA Y RIESGO (Cards Fila 2) --- pdf.set_font("Arial", "B", 11) pdf.set_text_color(231, 76, 60) pdf.cell(0, 8, "EXPOSICIÓN FINANCIERA (CARTERA)", ln=1) pdf.set_text_color(0, 0, 0) y_cartera = pdf.get_y() # Card Cartera Total c_total = metrics.get("CARTERA_SALDO_TOTAL_ACTUAL", {}).get("valor_texto", "N/D") draw_kpi_card(pdf, 10, y_cartera, 60, 25, "Cartera Total", c_total) # Card Cartera Vencida (FIX: Sincronización de clave) c_vencida = metrics.get("CARTERA_VALOR_VENCIDO_ACTUAL", {}).get("valor_texto", "N/D") draw_kpi_card(pdf, 75, y_cartera, 60, 25, "Cartera Vencida", c_vencida) # Card RCC rcc = metrics.get("AUTO_DAME_EL_RCC", {}).get("valor_texto", "N/D") draw_kpi_card(pdf, 140, y_cartera, 60, 25, "RCC (Días)", rcc) pdf.set_y(y_cartera + 32) # --- SECCIÓN 4: CONCLUSIONES --- pdf.ln(2) pdf.set_font("Arial", "B", 11) pdf.set_fill_color(52, 152, 219) pdf.set_text_color(255, 255, 255) pdf.cell(0, 8, " ANÁLISIS ESTRATÉGICO", ln=1, fill=True) pdf.set_text_color(0, 0, 0) pdf.ln(2) pdf.set_font("Arial", "", 10) clean_text = text_conclusion.encode("latin1", "replace").decode("latin1") pdf.multi_cell(0, 5, clean_text) # Finalización y subida (Restaurando el flujo de Keycloak completo) pdf_bytes = bytes(pdf.output()) import base64 pdf_base64 = base64.b64encode(pdf_bytes).decode("utf-8") upload_url = settings.UPLOAD_DOCUMENT_URL file_name = f"Dashboard_Ejecutivo_{uuid.uuid4().hex[:8]}.pdf" payload_json = { "base64": pdf_base64, "contentType": "application/pdf", "folder": "executive-reports", "fileName": file_name, } try: async with httpx.AsyncClient() as client: # 1. Obtener token de Keycloak auth_data = { "username": settings.KEYCLOAK_USERNAME, "password": settings.KEYCLOAK_PASSWORD, "client_id": settings.KEYCLOAK_CLIENT_ID, "grant_type": "password", } auth_resp = await client.post(settings.KEYCLOAK_TOKEN_URL, data=auth_data, timeout=15.0) token = auth_resp.json().get("access_token") headers = {"Content-Type": "application/json"} if token: headers["Authorization"] = f"Bearer {token}" response = await client.post(upload_url, json=payload_json, headers=headers, timeout=60.0) response.raise_for_status() return response.json().get("payload", {}).get("path") except Exception as e: logger.error(f"❌ Error al subir PDF: {e}") raise e