import os import sys import clr from datetime import datetime, timedelta, timezone from pathlib import Path import logger_utilities # Importar configuración y utilidades personalizadas from outlook import get_access_token, send_email from config_loader import CONFIG # --- CRYSTAL REPORTS RUNTIME INITIALIZATION --- ######################################################## def init_crystal_runtime(): """ Inicializa el runtime de Crystal Reports cargando las DLLs necesarias desde el GAC. Se basa en la ruta configurada en config.json para determinar la versión. """ # gac_path = r"C:\Windows\Microsoft.NET\assembly\GAC_MSIL" gac_path = CONFIG.crystal.path # La versión del runtime de Crystal Reports instalada en este sistema version_folder = CONFIG.crystal.version dlls = [ os.path.join(gac_path, "CrystalDecisions.CrystalReports.Engine", version_folder, "CrystalDecisions.CrystalReports.Engine.dll"), os.path.join(gac_path, "CrystalDecisions.Shared", version_folder, "CrystalDecisions.Shared.dll"), os.path.join(gac_path, "CrystalDecisions.ReportSource", version_folder, "CrystalDecisions.ReportSource.dll") ] for dll in dlls: if not os.path.exists(dll): print(f"CRITICAL ERROR: No se encontró la DLL necesaria: {dll}") print("Verifique que el SAP Crystal Reports Runtime esté instalado correctamente.") sys.exit(1) clr.AddReference(dll) ######################################################## # Inicializar el runtime antes de importar los tipos de .NET init_crystal_runtime() # Importar tipos de .NET después de cargar las referencias # pyrefly: ignore [missing-import] from System import DateTime # pyrefly: ignore [missing-import] from CrystalDecisions.CrystalReports.Engine import ReportDocument # pyrefly: ignore [missing-import] from CrystalDecisions.Shared import ExportFormatType # --- HELPERS --- ######################################################## def configure_database(report): """ Configura las credenciales de la base de datos HANA para el reporte y todas sus sub-tablas. """ hc = CONFIG.hana # Login general para el reporte report.SetDatabaseLogon(hc.user, hc.password, hc.server, hc.database) # Asegurar conexión en cada tabla individual for table in report.Database.Tables: conn_info = table.LogOnInfo.ConnectionInfo conn_info.ServerName = hc.server conn_info.DatabaseName = hc.database conn_info.UserID = hc.user conn_info.Password = hc.password table.ApplyLogOnInfo(table.LogOnInfo) ######################################################## def ejecutar_reporte(rpt_filename: str, pdf_path: Path, params: dict) -> bool: """ Carga un reporte .rpt, aplica parámetros y lo exporta a PDF. :param rpt_filename: Nombre o ruta del archivo .rpt :param pdf_path: Ruta de salida del archivo PDF :param params: Diccionario de parámetros {Nombre: Valor} :return: True si tuvo éxito, False en caso contrario """ print(f"--- Generando: {pdf_path.name} ---") report = ReportDocument() try: report.Load(rpt_filename) configure_database(report) # Asignar parámetros dinámicamente for key, value in params.items(): report.SetParameterValue(key, value) # Exportar a disco report.ExportToDisk(ExportFormatType.PortableDocFormat, str(pdf_path)) print(f"SUCCESS: Reporte guardado en {pdf_path}") return True except Exception as e: print(f"ERROR: Falló la ejecución del reporte '{rpt_filename}': {e}") return False finally: # Siempre liberar recursos de Crystal Reports report.Close() report.Dispose() ######################################################## def get_report_dates(): """ Calcula las fechas de reporte: Ayer y el primer día del mes actual. Retorna objetos System.DateTime compatibles con Crystal Reports. """ now = datetime.now(timezone.utc) # Si hoy es lunes (weekday() == 0), retroceder al viernes anterior (3 días atrás) days_back = 3 if now.weekday() == 0 else 1 previous_business_day = now - timedelta(days=days_back) # Objetos .NET DateTime dt_yesterday = DateTime(previous_business_day.year, previous_business_day.month, previous_business_day.day) dt_month_start = DateTime(previous_business_day.year, previous_business_day.month, 1) return dt_month_start, dt_yesterday # --- PROCESO PRINCIPAL --- ######################################################## def main(): print("====================================================") print("INICIANDO AUTOMATIZACIÓN DE CRYSTAL REPORTS") print(f"Fecha/Hora: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print("====================================================") # 1. Autenticación con Microsoft Graph access_token = get_access_token() if not access_token: print("ABORT: Error crítico de autenticación con Outlook.") return # 2. Preparación del Entorno base_path = Path(__file__).parent output_dir = base_path / CONFIG.outlook.attachments_folder output_dir.mkdir(parents=True, exist_ok=True) # 3. Cálculo de Fechas dt_month_start, dt_yesterday = get_report_dates() str_ayer = dt_yesterday.ToString("dd/MM/yyyy") str_inicio = dt_month_start.ToString("dd/MM/yyyy") # 4. Generación de Reportes rpt_file = "venta_presupuesto_linea.rpt" adjuntos = [] # REPORTE 1: Venta del día de ayer pdf_yesterday = output_dir / f"venta_presupuesto_linea_{dt_yesterday.ToString('dd_MM_yyyy')}.pdf" if ejecutar_reporte(rpt_file, pdf_yesterday, {"FecIni": dt_yesterday, "FecFin": dt_yesterday}): adjuntos.append(str(pdf_yesterday)) # REPORTE 2: Acumulado del mes actual (MTD) pdf_mtd = output_dir / f"venta_presupuesto_linea_{dt_month_start.ToString('dd_MM_yyyy')}_al_{dt_yesterday.ToString('dd_MM_yyyy')}.pdf" if ejecutar_reporte(rpt_file, pdf_mtd, {"FecIni": dt_month_start, "FecFin": dt_yesterday}): adjuntos.append(str(pdf_mtd)) # 5. Envío de Correo Electrónico if adjuntos: subject = f"Informe de Venta de Presupuestos por Línea - {str_ayer}" body = f"""

Informe Automático de Ventas por Línea

Buenos días

Me complace compartir con ustedes el reporte de avance de ventas por línea correspondiente al {str_ayer}.

La información se encuentra disponible en el siguiente dashboard:

Ver reporte Power BI

Quedamos atentos a cualquier consulta o comentario.


Saludos cordiales,

Equipo de Inteligencia de Negocios

Este es un mensaje generado automáticamente por el sistema de reportes.

""" success = send_email( access_token=access_token, from_email=CONFIG.outlook.from_email, to_recipients=CONFIG.outlook.to_email, subject=subject, html_body=body, attachment_paths=adjuntos ) if success: print("PROCESO FINALIZADO: Correo enviado correctamente.") else: print("ERROR: No se generaron archivos PDF. No hay nada que enviar.") ######################################################## if __name__ == "__main__": try: # Configurar Logger (Redirige stdout a archivo y consola) log_path = logger_utilities.setup_logger() logger_utilities.limpiar_logs_antiguos(dias=30) inicio_ejecucion = datetime.now() print("="*60) print(f"🚀 INICIO DE EJECUCIÓN: {inicio_ejecucion.strftime('%Y-%m-%d %H:%M:%S')}") print(f"📄 Log guardado en: {log_path}") print("="*60) main() fin_ejecucion = datetime.now() duracion = fin_ejecucion - inicio_ejecucion print("\n" + "="*60) print(f"🏁 FIN DE EJECUCIÓN: {fin_ejecucion.strftime('%Y-%m-%d %H:%M:%S')}") print(f"⏱️ Duración total: {str(duracion).split('.')[0]}") # Sin microsegundos print("="*60) except Exception as e: print(f"FATAL MAIN ERROR: {e}")