""" Real-time Log Monitor Window - PyQt6 Shows job execution log in real-time with auto-scroll """ from PyQt6.QtWidgets import (QDialog, QVBoxLayout, QTextEdit, QLabel, QPushButton, QHBoxLayout, QMessageBox) from PyQt6.QtCore import QTimer, Qt from PyQt6.QtGui import QFont, QTextCursor from pathlib import Path from typing import Optional from core.database import SchedulerDatabase class LogMonitorWindow(QDialog): """Window that displays job execution log in real-time""" def __init__(self, execution_id: int, db: SchedulerDatabase, parent=None): super().__init__(parent) self.execution_id = execution_id self.db = db self.log_file_path: Optional[Path] = None self.log_file = None self.last_position = 0 self.is_monitoring = True self._setup_ui() self._load_execution_info() self._start_monitoring() def _setup_ui(self): """Setup the UI""" self.setWindowTitle(f"Job Monitor - Execution #{self.execution_id}") self.setMinimumSize(900, 600) layout = QVBoxLayout(self) # Header with job info self.header_label = QLabel("Loading...") self.header_label.setStyleSheet(""" QLabel { background-color: #2c3e50; color: white; padding: 10px; font-weight: bold; font-size: 12pt; } """) layout.addWidget(self.header_label) # Status bar self.status_label = QLabel("Status: Initializing...") self.status_label.setStyleSheet(""" QLabel { background-color: #34495e; color: #ecf0f1; padding: 5px; } """) layout.addWidget(self.status_label) # Log text area self.log_text = QTextEdit() self.log_text.setReadOnly(True) self.log_text.setLineWrapMode(QTextEdit.LineWrapMode.NoWrap) # Use monospace font font = QFont("Consolas", 9) if not font.exactMatch(): font = QFont("Courier New", 9) self.log_text.setFont(font) self.log_text.setStyleSheet(""" QTextEdit { background-color: #1e1e1e; color: #d4d4d4; border: 1px solid #3e3e3e; } """) layout.addWidget(self.log_text) # Bottom buttons button_layout = QHBoxLayout() self.auto_scroll_btn = QPushButton("Auto-scroll: ON") self.auto_scroll_btn.setCheckable(True) self.auto_scroll_btn.setChecked(True) self.auto_scroll_btn.clicked.connect(self._toggle_auto_scroll) button_layout.addWidget(self.auto_scroll_btn) self.refresh_btn = QPushButton("Refresh Now") self.refresh_btn.clicked.connect(self._manual_refresh) button_layout.addWidget(self.refresh_btn) button_layout.addStretch() self.close_btn = QPushButton("Close") self.close_btn.clicked.connect(self.close) button_layout.addWidget(self.close_btn) layout.addLayout(button_layout) # Timer for updating log self.update_timer = QTimer(self) self.update_timer.timeout.connect(self._update_log) self.update_timer.setInterval(100) # Update every 100ms def _load_execution_info(self): """Load execution information from database""" execution = self.db.get_execution(self.execution_id) if not execution: QMessageBox.critical(self, "Error", f"Execution ID {self.execution_id} not found") self.close() return # Get job name job = self.db.get_job(execution['job_id']) job_name = job['name'] if job else f"Job ID {execution['job_id']}" # Update window title and header self.setWindowTitle(f"Job Monitor - {job_name}") self.header_label.setText(f"Job: {job_name} | Execution ID: {self.execution_id}") # Get log file path self.log_file_path = Path(execution['log_file']) # Update status status = execution['status'] status_colors = { 'running': '#3498db', 'completed': '#27ae60', 'failed': '#e74c3c', 'timeout': '#e67e22', 'pending': '#95a5a6' } color = status_colors.get(status, '#95a5a6') self.status_label.setText( f"Status: {status.upper()} | " f"Started: {execution['start_time']} | " f"Log: {self.log_file_path.name}" ) self.status_label.setStyleSheet(f""" QLabel {{ background-color: {color}; color: white; padding: 5px; font-weight: bold; }} """) def _start_monitoring(self): """Start monitoring the log file""" if not self.log_file_path or not self.log_file_path.exists(): self.log_text.append("Waiting for log file to be created...\n") # Check periodically if file exists QTimer.singleShot(1000, self._check_log_file_exists) return # Open log file try: self.log_file = open(self.log_file_path, 'r', encoding='utf-8', errors='replace') # Read existing content content = self.log_file.read() if content: self.log_text.setPlainText(content) self._scroll_to_bottom() self.last_position = self.log_file.tell() # Start update timer self.update_timer.start() except Exception as e: self.log_text.append(f"\nERROR opening log file: {e}\n") def _check_log_file_exists(self): """Check if log file exists (called periodically)""" if self.log_file_path and self.log_file_path.exists(): self._start_monitoring() else: # Check again in 1 second QTimer.singleShot(1000, self._check_log_file_exists) def _update_log(self): """Update log content from file""" if not self.log_file or not self.is_monitoring: return try: # Check execution status execution = self.db.get_execution(self.execution_id) if execution['status'] in ['completed', 'failed', 'timeout']: # Job finished - read remaining content and stop monitoring self.log_file.seek(self.last_position) remaining = self.log_file.read() if remaining: self.log_text.append(remaining) self._scroll_to_bottom() self.update_timer.stop() self.is_monitoring = False # Update status self._load_execution_info() # Show completion message status = execution['status'] exit_code = execution['exit_code'] duration = execution['duration_seconds'] completion_msg = f"\n{'='*80}\n" completion_msg += f"JOB FINISHED - Status: {status.upper()}\n" completion_msg += f"Exit Code: {exit_code} | Duration: {duration:.1f}s\n" completion_msg += f"{'='*80}\n" self.log_text.append(completion_msg) self._scroll_to_bottom() return # Read new content self.log_file.seek(self.last_position) new_content = self.log_file.read() if new_content: self.log_text.append(new_content) self.last_position = self.log_file.tell() self._scroll_to_bottom() except Exception as e: self.log_text.append(f"\nERROR reading log: {e}\n") self.update_timer.stop() def _scroll_to_bottom(self): """Scroll to bottom if auto-scroll is enabled""" if self.auto_scroll_btn.isChecked(): cursor = self.log_text.textCursor() cursor.movePosition(QTextCursor.MoveOperation.End) self.log_text.setTextCursor(cursor) def _toggle_auto_scroll(self, checked: bool): """Toggle auto-scroll""" self.auto_scroll_btn.setText(f"Auto-scroll: {'ON' if checked else 'OFF'}") if checked: self._scroll_to_bottom() def _manual_refresh(self): """Manually refresh log content""" self._update_log() def closeEvent(self, event): """Clean up when window closes""" self.update_timer.stop() if self.is_monitoring: reply = QMessageBox.question( self, "Close Monitor", "Job is still running. Close monitor window?\n(Job will continue in background)", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No ) if reply == QMessageBox.StandardButton.No: self.update_timer.start() event.ignore() return if self.log_file: self.log_file.close() event.accept()