""" Notification Manager for Python Scheduler Handles popup notifications and email notifications via Outlook """ import threading from datetime import datetime from typing import Optional from loguru import logger from core.database import SchedulerDatabase from core.config import SchedulerConfig try: from win10toast import ToastNotifier TOAST_AVAILABLE = True except ImportError: logger.warning("win10toast not available, popup notifications disabled") TOAST_AVAILABLE = False class NotificationManager: """Manages notifications for job executions""" def __init__(self, db: SchedulerDatabase, config: SchedulerConfig): self.db = db self.config = config self.toaster = ToastNotifier() if TOAST_AVAILABLE else None self.enabled = config.enable_popup_notifications and TOAST_AVAILABLE logger.info(f"NotificationManager initialized (enabled: {self.enabled})") def notify_job_result(self, execution_id: int, job_name: str, status: str, duration: Optional[float] = None, exit_code: Optional[int] = None): """ Send notification for job execution result Args: execution_id: Execution ID job_name: Job name status: Execution status ('completed', 'failed', 'timeout', etc.) duration: Duration in seconds exit_code: Process exit code """ if not self.enabled: return # Check config for when to notify should_notify = False if status == 'completed' and self.config.get('notifications.notify_on_completion'): should_notify = True elif status == 'failed' and self.config.get('notifications.notify_on_failure'): should_notify = True elif status == 'timeout' and self.config.get('notifications.notify_on_timeout'): should_notify = True if not should_notify: return # Build notification title, message, icon_path = self._build_notification_content( job_name, status, duration, exit_code ) # Send notification in background thread thread = threading.Thread( target=self._send_toast_notification, args=(execution_id, title, message, icon_path), daemon=True ) thread.start() def _build_notification_content(self, job_name: str, status: str, duration: Optional[float], exit_code: Optional[int]) -> tuple: """Build notification title and message""" if status == 'completed': title = f"Job Completed - {job_name}" if duration: message = f"Completed successfully in {duration:.1f}s" else: message = "Completed successfully" icon_path = None elif status == 'failed': title = f"Job Failed - {job_name}" if exit_code is not None: message = f"Failed with exit code {exit_code}. Check logs for details." else: message = "Failed. Check logs for details." icon_path = None elif status == 'timeout': title = f"Job Timeout - {job_name}" message = "Job exceeded timeout limit. Process was terminated." icon_path = None else: title = f"Job Status - {job_name}" message = f"Status: {status}" icon_path = None return title, message, icon_path def _send_toast_notification(self, execution_id: int, title: str, message: str, icon_path: Optional[str]): """ Send Windows toast notification Args: execution_id: Execution ID for logging title: Notification title message: Notification message icon_path: Optional icon path """ if not self.toaster: return try: duration = self.config.get('notifications.popup_duration_seconds', 10) # Show notification self.toaster.show_toast( title=title, msg=message, icon_path=icon_path, duration=duration, threaded=True ) # Log to database self.db.create_notification( execution_id=execution_id, notification_type='popup', status='sent', message=f"{title}: {message}", sent_at=datetime.now() ) logger.info(f"Sent popup notification for execution {execution_id}") except Exception as e: logger.error(f"Failed to send popup notification: {e}") # Log failure to database try: self.db.create_notification( execution_id=execution_id, notification_type='popup', status='failed', message=f"{title}: {message}", error=str(e) ) except: pass def send_custom_notification(self, title: str, message: str): """ Send a custom notification (not tied to a job execution) Args: title: Notification title message: Notification message """ if not self.enabled or not self.toaster: logger.debug(f"Custom notification not sent (disabled): {title}") return try: duration = self.config.get('notifications.popup_duration_seconds', 10) self.toaster.show_toast( title=title, msg=message, duration=duration, threaded=True ) logger.info(f"Sent custom notification: {title}") except Exception as e: logger.error(f"Failed to send custom notification: {e}") def notify_final_failure(self, job_name: str, attempt_count: int): """ Notifica finale quando tutti i retry sono esauriti. Invia popup Windows + email via Outlook (se abilitata). Args: job_name: nome del job attempt_count: numero totale di tentativi falliti """ title = f"Job FALLITO (tutti i retry esauriti) — {job_name}" message = ( f"Il job '{job_name}' ha fallito {attempt_count} volte consecutive.\n" f"Tutti i retry sono esauriti. Verificare i log per i dettagli." ) # Popup Windows if self.enabled: threading.Thread( target=self._send_toast_notification, args=(0, title, message, None), daemon=True ).start() # Email via Outlook if self.config.get('notifications.enable_email', False): email_to = self.config.get('notifications.email_to', '') if email_to: threading.Thread( target=self._send_email_notification, args=(email_to, title, message), daemon=True ).start() else: logger.warning("notify_final_failure: email_to non configurato in notifications") def _send_email_notification(self, to: str, subject: str, body: str): """Invia email via OutlookConnector.""" try: from modules.outlook.outlook_connector import OutlookConnector connector = OutlookConnector() if connector.connect(): connector.send_email(to=to, subject=subject, body=body) else: logger.error("_send_email_notification: connessione Outlook fallita") except Exception as e: logger.error(f"_send_email_notification: errore — {e}") def test_notification(self): """Send a test notification""" self.send_custom_notification( title="Python Scheduler - Test Notification", message="This is a test notification from Python Scheduler. Notifications are working!" )