""" History Widget for Python Scheduler Display execution history """ from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QTableWidget, QTableWidgetItem, QPushButton, QHeaderView, QLabel, QSpinBox, QMessageBox ) from PyQt6.QtCore import Qt from PyQt6.QtGui import QColor from loguru import logger import socket from core.database import SchedulerDatabase from core.config import SchedulerConfig from gui.log_monitor_window import LogMonitorWindow class HistoryWidget(QWidget): """Execution history widget""" def __init__(self, db: SchedulerDatabase, config: SchedulerConfig): super().__init__() self.db = db self.config = config self.limit = 50 self._setup_ui() self.refresh() def _setup_ui(self): """Setup UI""" layout = QVBoxLayout(self) # Header header_layout = QHBoxLayout() header_label = QLabel("Execution History") header_label.setStyleSheet("font-size: 14px; font-weight: bold;") header_layout.addWidget(header_label) header_layout.addStretch() # Limit selector limit_label = QLabel("Show last:") header_layout.addWidget(limit_label) self.limit_spin = QSpinBox() self.limit_spin.setMinimum(10) self.limit_spin.setMaximum(500) self.limit_spin.setValue(50) self.limit_spin.setSingleStep(10) self.limit_spin.valueChanged.connect(self._on_limit_changed) header_layout.addWidget(self.limit_spin) # Monitor Log button self.monitor_btn = QPushButton("📺 Monitor Log") self.monitor_btn.clicked.connect(self._open_monitor) self.monitor_btn.setToolTip("Open log monitor for selected execution") header_layout.addWidget(self.monitor_btn) layout.addLayout(header_layout) # History table self.table = QTableWidget() self.table.setColumnCount(7) self.table.setHorizontalHeaderLabels([ "ID", "Job Name", "Status", "Start Time", "Duration", "Exit Code", "Triggered By" ]) self.table.horizontalHeader().setSectionResizeMode(1, QHeaderView.ResizeMode.Stretch) self.table.setSelectionBehavior(QTableWidget.SelectionBehavior.SelectRows) self.table.setAlternatingRowColors(True) self.table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers) layout.addWidget(self.table) def _on_limit_changed(self, value): """Handle limit change""" self.limit = value self.refresh() def refresh(self): """Refresh execution history""" try: executions = self.db.get_recent_executions(limit=self.limit, hostname=socket.gethostname()) self.table.setRowCount(len(executions)) for row, execution in enumerate(executions): # ID self.table.setItem(row, 0, QTableWidgetItem(str(execution['id']))) # Job name name_item = QTableWidgetItem(execution['job_name']) self.table.setItem(row, 1, name_item) # Status with color status = execution['status'] status_item = QTableWidgetItem(status.upper()) if status == 'completed': status_item.setForeground(Qt.GlobalColor.darkGreen) status_item.setText("✓ COMPLETED") elif status == 'failed': status_item.setForeground(Qt.GlobalColor.red) status_item.setText("✗ FAILED") elif status == 'running': status_item.setForeground(Qt.GlobalColor.blue) status_item.setText("▶ RUNNING") elif status == 'timeout': status_item.setForeground(Qt.GlobalColor.darkYellow) status_item.setText("⏱ TIMEOUT") self.table.setItem(row, 2, status_item) # Start time start_time = execution['start_time'][:19] if execution['start_time'] else "N/A" self.table.setItem(row, 3, QTableWidgetItem(start_time)) # Duration if execution['duration_seconds']: duration_str = f"{execution['duration_seconds']:.1f}s" else: duration_str = "N/A" self.table.setItem(row, 4, QTableWidgetItem(duration_str)) # Exit code exit_code = str(execution['exit_code']) if execution['exit_code'] is not None else "N/A" exit_item = QTableWidgetItem(exit_code) if execution['exit_code'] == 0: exit_item.setForeground(Qt.GlobalColor.darkGreen) elif execution['exit_code'] is not None and execution['exit_code'] != 0: exit_item.setForeground(Qt.GlobalColor.red) self.table.setItem(row, 5, exit_item) # Triggered by self.table.setItem(row, 6, QTableWidgetItem(execution['triggered_by'])) logger.debug(f"Refreshed history table: {len(executions)} executions") except Exception as e: logger.error(f"Error refreshing history: {e}") def _open_monitor(self): """Open log monitor window for selected execution""" try: # Get selected row selected_rows = self.table.selectedIndexes() if not selected_rows: QMessageBox.warning(self, "No Selection", "Please select an execution to monitor") return # Get execution ID from first column of selected row row = selected_rows[0].row() execution_id_item = self.table.item(row, 0) if not execution_id_item: return execution_id = int(execution_id_item.text()) # Open monitor window monitor_window = LogMonitorWindow(execution_id, self.db, self) monitor_window.show() logger.info(f"Opened log monitor for execution {execution_id}") except Exception as e: logger.error(f"Error opening monitor: {e}") QMessageBox.critical(self, "Error", f"Failed to open monitor: {e}")