# -*- coding: utf-8 -*- """日志查看器""" import re from datetime import datetime from PySide6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QPlainTextEdit, QPushButton, QLineEdit, QLabel, QComboBox, QCheckBox, QFileDialog, QMessageBox ) from PySide6.QtCore import Qt, Signal, Slot from PySide6.QtGui import QTextCharFormat, QColor, QFont, QTextCursor class LogViewer(QWidget): """日志查看器""" # 信号 log_cleared = Signal() def __init__(self, parent=None): super().__init__(parent) self.max_lines = 10000 self.auto_scroll = True self.filter_text = "" self.filter_level = "ALL" self._all_logs = [] # 存储所有日志 self._init_ui() self._connect_signals() def _init_ui(self): """初始化界面""" layout = QVBoxLayout(self) layout.setContentsMargins(16, 16, 16, 16) layout.setSpacing(8) # 标题和工具栏 header_layout = QHBoxLayout() title = QLabel("执行日志") title.setProperty("heading", True) header_layout.addWidget(title) header_layout.addStretch() # 日志级别过滤 header_layout.addWidget(QLabel("级别:")) self.level_combo = QComboBox() self.level_combo.addItems(["ALL", "INFO", "WARNING", "ERROR", "DEBUG"]) self.level_combo.setFixedWidth(100) header_layout.addWidget(self.level_combo) # 搜索框 header_layout.addWidget(QLabel("搜索:")) self.search_edit = QLineEdit() self.search_edit.setPlaceholderText("输入关键字...") self.search_edit.setFixedWidth(200) header_layout.addWidget(self.search_edit) # 自动滚动 self.auto_scroll_check = QCheckBox("自动滚动") self.auto_scroll_check.setChecked(True) header_layout.addWidget(self.auto_scroll_check) layout.addLayout(header_layout) # 日志文本区域 self.log_text = QPlainTextEdit() self.log_text.setObjectName("logViewer") self.log_text.setReadOnly(True) self.log_text.setFont(QFont("Consolas", 10)) self.log_text.setLineWrapMode(QPlainTextEdit.NoWrap) layout.addWidget(self.log_text, 1) # 底部工具栏 footer_layout = QHBoxLayout() self.line_count_label = QLabel("0 行") self.line_count_label.setProperty("subheading", True) footer_layout.addWidget(self.line_count_label) footer_layout.addStretch() self.copy_btn = QPushButton("复制全部") self.copy_btn.setProperty("secondary", True) footer_layout.addWidget(self.copy_btn) self.export_btn = QPushButton("导出") self.export_btn.setProperty("secondary", True) footer_layout.addWidget(self.export_btn) self.clear_btn = QPushButton("清空") self.clear_btn.setProperty("secondary", True) footer_layout.addWidget(self.clear_btn) layout.addLayout(footer_layout) def _connect_signals(self): """连接信号""" self.level_combo.currentTextChanged.connect(self._apply_filter) self.search_edit.textChanged.connect(self._apply_filter) self.auto_scroll_check.stateChanged.connect(self._toggle_auto_scroll) self.copy_btn.clicked.connect(self._copy_all) self.export_btn.clicked.connect(self._export_log) self.clear_btn.clicked.connect(self._clear_log) @Slot(str) def append_log(self, text: str): """追加日志""" # 添加时间戳（如果没有） if not re.match(r'^\d{4}-\d{2}-\d{2}', text) and not text.startswith('['): timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") text = f"[{timestamp}] {text}" # 存储到全部日志 self._all_logs.append(text) # 限制日志行数 if len(self._all_logs) > self.max_lines: self._all_logs = self._all_logs[-self.max_lines:] # 检查是否通过过滤器 if self._matches_filter(text): self._append_formatted_line(text) # 更新行数 self._update_line_count() def _append_formatted_line(self, text: str): """追加格式化的行""" cursor = self.log_text.textCursor() cursor.movePosition(QTextCursor.End) # 设置格式 fmt = QTextCharFormat() text_lower = text.lower() if "[error]" in text_lower or "错误" in text or "失败" in text: fmt.setForeground(QColor("#d93025")) fmt.setFontWeight(QFont.Bold) elif "[warning]" in text_lower or "警告" in text or "warn" in text_lower: fmt.setForeground(QColor("#f9ab00")) elif "[info]" in text_lower: fmt.setForeground(QColor("#1a73e8")) elif "[debug]" in text_lower: fmt.setForeground(QColor("#9aa0a6")) elif "[gui]" in text_lower: fmt.setForeground(QColor("#1e8e3e")) else: fmt.setForeground(QColor("#333333")) cursor.insertText(text + "\n", fmt) # 自动滚动 if self.auto_scroll: self.log_text.verticalScrollBar().setValue( self.log_text.verticalScrollBar().maximum() ) def _matches_filter(self, text: str) -> bool: """检查是否匹配过滤器""" # 级别过滤 if self.filter_level != "ALL": level_marker = f"[{self.filter_level}]" if level_marker.lower() not in text.lower(): return False # 文本过滤 if self.filter_text: if self.filter_text.lower() not in text.lower(): return False return True def _apply_filter(self): """应用过滤器""" self.filter_level = self.level_combo.currentText() self.filter_text = self.search_edit.text().strip() # 重新显示日志 self.log_text.clear() for line in self._all_logs: if self._matches_filter(line): self._append_formatted_line(line) self._update_line_count() def _toggle_auto_scroll(self, state: int): """切换自动滚动""" self.auto_scroll = state == Qt.Checked def _copy_all(self): """复制全部日志""" from PySide6.QtWidgets import QApplication text = self.log_text.toPlainText() QApplication.clipboard().setText(text) QMessageBox.information(self, "提示", "日志已复制到剪贴板") def _export_log(self): """导出日志""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") default_name = f"etl_log_{timestamp}.txt" file_path, _ = QFileDialog.getSaveFileName( self, "导出日志", default_name, "文本文件 (*.txt);;日志文件 (*.log);;所有文件 (*.*)" ) if not file_path: return try: with open(file_path, "w", encoding="utf-8") as f: f.write(self.log_text.toPlainText()) QMessageBox.information(self, "成功", f"日志已导出到:\n{file_path}") except Exception as e: QMessageBox.critical(self, "错误", f"导出失败: {e}") def _clear_log(self): """清空日志""" reply = QMessageBox.question( self, "确认", "确定要清空所有日志吗?", QMessageBox.Yes | QMessageBox.No, QMessageBox.No ) if reply == QMessageBox.Yes: self._all_logs.clear() self.log_text.clear() self._update_line_count() self.log_cleared.emit() def _update_line_count(self): """更新行数显示""" visible_count = self.log_text.document().blockCount() - 1 total_count = len(self._all_logs) if visible_count < total_count: self.line_count_label.setText(f"{visible_count} / {total_count} 行") else: self.line_count_label.setText(f"{total_count} 行")