gitea/Projects
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

完成日志界面的制作

Browse Source
This commit is contained in:
gitea
2025-09-28 18:21:48 +08:00
parent 1c47497fc2
commit 943130b875
12 changed files with 295 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@@ -3,4 +3,5 @@
rokae/testbench.py
**/__pycache__/
NOTRACK/
gui/assets/database/
toolbox/assets/
toolbox/output/

2
toolbox/codes/common/clibs.py
View File

@@ -15,7 +15,7 @@ bg = f"{base_path}/assets/media/bg.jpg"
win_width, win_height = 1100, 500
conn, cursor = None, None
listW_items = {"实用工具": "w10_practical", "效率提升": "w20_efficiency", "财务分析": "w30_financial"}
icon = f"{base_path}/assets/media/icon.ico"
def delete_files_in_directory(directory):
path = Path(directory)

61
toolbox/codes/common/db_operation.py
View File

@@ -1,6 +1,7 @@
import sqlite3
import time
from inspect import currentframe
from functools import singledispatch
from codes.common import clibs
@@ -9,15 +10,15 @@ def db_init():
if clibs.db_file.exists():
return
clibs.conn = sqlite3.connect(clibs.db_file, isolation_level=None, check_same_thread=False, cached_statements=2048, timeout=10.0)
clibs.cursor = clibs.conn.cursor()
clibs.cursor.execute("PRAGMA journal_mode=wal")
clibs.cursor.execute("PRAGMA wal_checkpoint=TRUNCATE")
clibs.cursor.execute("PRAGMA synchronous=normal")
clibs.cursor.execute("PRAGMA temp_store=memory")
clibs.cursor.execute("PRAGMA mmap_size=30000000000")
clibs.cursor.execute("PRAGMA cache_size=200000")
clibs.cursor.execute(
conn = sqlite3.connect(clibs.db_file, isolation_level=None, check_same_thread=False, cached_statements=2048, timeout=10.0)
cursor = conn.cursor()
cursor.execute("PRAGMA journal_mode=wal")
cursor.execute("PRAGMA wal_checkpoint=TRUNCATE")
cursor.execute("PRAGMA synchronous=normal")
cursor.execute("PRAGMA temp_store=memory")
cursor.execute("PRAGMA mmap_size=30000000000")
cursor.execute("PRAGMA cache_size=200000")
cursor.execute(
"""
create table if not exists logs(
id integer primary key autoincrement,
@@ -28,7 +29,7 @@ def db_init():
)
"""
)
clibs.cursor.execute(
cursor.execute(
"""
create table if not exists users(
id integer primary key autoincrement,
@@ -39,14 +40,18 @@ def db_init():
)
"""
)
db_write_logs("数据库初始化成功！", "login_ui")
db_close()
cursor.execute(f"INSERT INTO logs (level, module, content) VALUES (?, ?, ?)", ("info", "login_ui", "数据库初始化成功！"))
cursor.close()
conn.close()
def db_lock(func):
def wrapper(*args, **kwargs):
try:
clibs.lock.acquire(True)
ret = func(*args, **kwargs)
except Exception as e:
print(f"db operation error: {e}")
ret = None
finally:
clibs.lock.release()
return ret
@@ -62,6 +67,13 @@ def db_backup():
db.unlink()
def db_conn():
# import traceback, inspect
# print("[Conn] 被调用", traceback.format_stack()[-2])
# print("[Conn] conn=", clibs.conn, "cursor=", clibs.cursor)
if clibs.conn is not None:
return
clibs.conn = sqlite3.connect(clibs.db_file, isolation_level=None, check_same_thread=False, cached_statements=2048, timeout=3.0)
clibs.cursor = clibs.conn.cursor()
clibs.cursor.execute("PRAGMA journal_mode=wal")
@@ -92,9 +104,30 @@ def db_write_logs(content, module="", level="info"):
clibs.cursor.execute(f"INSERT INTO logs (level, module, content) VALUES (?, ?, ?)", (level, module, content))
@singledispatch
@db_lock
def db_query_logs():
...
def db_query_logs(dummy: bool = True):
clibs.cursor.execute(f"SELECT * FROM logs")
records = clibs.cursor.fetchall()
len_records = len(records)
return records, len_records
@db_query_logs.register
def _(levels: list):
placeholders = ",".join("?" * len(levels))
clibs.cursor.execute(f"SELECT * FROM logs WHERE level IN ({placeholders})", (*levels, ))
records = clibs.cursor.fetchall()
len_records = len(records)
return records, len_records
@db_query_logs.register
def _(search_text: str, records: list):
ids = [_[0] for _ in records]
placeholder = ",".join("?" * len(ids))
clibs.cursor.execute(f"SELECT * FROM logs WHERE id IN ({placeholder}) and content like ?", (ids + [f"%{search_text}%", ]))
records = clibs.cursor.fetchall()
len_records = len(records)
return records, len_records
@db_lock
def db_write_users(username, password_encrypted, salt):

6
toolbox/codes/common/signal_bus.py
View File

@@ -14,7 +14,11 @@ class SignalBus(QObject):
current_stacked_page = Signal(str) # 获取当前页面的page_id
init_stacked_page = Signal(str) # 设置打开侧边栏后的初始页面
qa_stacked_page_switch = Signal(str) # 切换stacked widget页面
stacked_page_switch = Signal(str) # 切换stacked widget的页面
stacked_page_switch_setting = Signal() # 切换stacked widget的设置页面后的触发信号
stacked_page_switch_log = Signal() # 切换stacked widget的日志页面后的触发信号
stacked_page_switch_about = Signal() # 切换stacked widget的关于页面后的触发信号
qa_switch_change = Signal(bool) # 切换折叠侧边栏的状态
home_overlay_trigger = Signal() # 触发软件锁屏
home_overlay_auth = Signal() # 触发密码框的显示与隐藏
home_overlay_close = Signal() # 退出锁屏后的收尾信号

11
toolbox/codes/ui/components/list_widget_ui.py
View File

@@ -1,6 +1,6 @@
from PySide6.QtGui import QFocusEvent
from PySide6.QtWidgets import QListWidget, QListWidgetItem
from PySide6.QtCore import Qt, QEvent
from PySide6.QtCore import Qt
from codes.common import clibs
from codes.common.signal_bus import signal_bus
@@ -18,6 +18,7 @@ class SListWidget(QListWidget):
...
def init_ui(self):
self.setMinimumWidth(150)
for item in clibs.listW_items:
_ = QListWidgetItem(item)
_.setTextAlignment(Qt.AlignmentFlag.AlignCenter)
@@ -25,7 +26,9 @@ class SListWidget(QListWidget):
def setup_slot(self):
self.itemClicked.connect(self.item_clicked)
signal_bus.qa_stacked_page_switch.connect(self.hide)
signal_bus.stacked_page_switch_setting.connect(self.qa_hide)
signal_bus.stacked_page_switch_log.connect(self.qa_hide)
signal_bus.stacked_page_switch_about.connect(self.qa_hide)
signal_bus.list_widget_on_off.connect(self.lw_show_hide)
def item_clicked(self, item):
@@ -40,6 +43,10 @@ class SListWidget(QListWidget):
else:
self.hide()
def qa_hide(self):
self.hide()
signal_bus.qa_switch_change.emit(False)
def focusOutEvent(self, event: QFocusEvent):
self.clearSelection()
super().focusOutEvent(event)

48
toolbox/codes/ui/components/stacked_widget_ui.py
View File

@@ -1,11 +1,12 @@
from pathlib import Path
import importlib.util
from PySide6.QtWidgets import QStackedWidget
from PySide6.QtWidgets import QStackedWidget, QWidget, QLabel
from PySide6.QtCore import Qt
from codes.common import clibs
from codes.common.signal_bus import signal_bus
from codes.ui.stacked_pages.w01_setting import W01Setting
from codes.ui.stacked_pages.w08_log import W08Log
from codes.ui.stacked_pages.w09_about import W09About
from codes.ui.stacked_pages.w10_practical import W10Practical
from codes.ui.stacked_pages.w20_efficiency import W20Efficiency
from codes.ui.stacked_pages.w30_financial import W30Financial
class SStackedWidget(QStackedWidget):
@@ -17,46 +18,23 @@ class SStackedWidget(QStackedWidget):
self.setup_slot()
def predos(self):
self.page_list = {}
self.page_list = {"w01_setting": W01Setting(), "w08_log": W08Log(), "w09_about": W09About(), "w10_practical": W10Practical(), "w20_efficiency": W20Efficiency(), "w30_financial": W30Financial()}
def init_ui(self):
# stacked widget 1x: 10为一级按钮页，其余为二级按钮页，2-9同理 | 0x. 日志/设置/关于等页面
self.load_pages()
for page_id, widget in self.page_list.items():
widget.setObjectName(page_id)
self.addWidget(widget)
w = self.page_list.get("w01_setting")
self.setCurrentWidget(w)
def setup_slot(self):
signal_bus.init_stacked_page.connect(self.set_current_page)
signal_bus.qa_stacked_page_switch.connect(self.set_current_page)
signal_bus.stacked_page_switch.connect(self.set_current_page)
signal_bus.list_widget_click.connect(self.set_current_page)
def set_current_page(self, page_id: str):
w = self.page_list.get(page_id)
self.setCurrentWidget(w)
signal_bus.current_stacked_page.emit(page_id)
def load_pages(self):
def to_camel(snake: str) -> str:
# w01_setting -> W01Setting
return "".join(word.capitalize() for word in snake.split('_'))
def instantiate(pyFile: Path, className: str) -> QWidget | None:
try:
spec = importlib.util.spec_from_file_location(pyFile.stem, pyFile)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
cls = getattr(module, className)
if issubclass(cls, QWidget):
return cls()
except Exception as e:
print(f"加载 {pyFile} 失败: {e}")
pages_dir = clibs.base_path / "codes/ui/stacked_pages/"
for py_file in pages_dir.glob("w*.py"):
page_id = py_file.stem # w01_setting
class_name = to_camel(page_id) # W01Setting
widget = instantiate(py_file, class_name)
if widget:
widget.setObjectName(page_id) # 用于 findChild / 切换
self.addWidget(widget)
self.page_list[page_id] = widget

13
toolbox/codes/ui/components/toolbar_ui.py
View File

@@ -6,7 +6,6 @@ from codes.common import clibs
from codes.common.signal_bus import signal_bus
class SToolBar(QToolBar):
def __init__(self, parent=None):
super().__init__(parent)
@@ -85,24 +84,26 @@ class SToolBar(QToolBar):
self.ac_setting.triggered.connect(self.ac_sp)
self.ac_log.triggered.connect(self.ac_lp)
self.ac_about.triggered.connect(self.ac_ap)
signal_bus.qa_stacked_page_switch.connect(self.change2hide)
signal_bus.qa_switch_change.connect(self.change2hide)
def ac_sw(self, checked: bool):
self.ac_switch.setIcon(self.on_icon if checked else self.off_icon)
print(f"checked: {checked}")
signal_bus.list_widget_on_off.emit(checked)
def ac_hp(self):
signal_bus.home_overlay_trigger.emit()
def ac_sp(self):
signal_bus.qa_stacked_page_switch.emit("w01_setting")
signal_bus.stacked_page_switch.emit("w01_setting")
signal_bus.stacked_page_switch_setting.emit()
def ac_lp(self):
signal_bus.qa_stacked_page_switch.emit("w08_log")
signal_bus.stacked_page_switch.emit("w08_log")
signal_bus.stacked_page_switch_log.emit()
def ac_ap(self):
signal_bus.qa_stacked_page_switch.emit("w09_about")
signal_bus.stacked_page_switch.emit("w09_about")
signal_bus.stacked_page_switch_about.emit()
def change2hide(self):
self.ac_switch.setChecked(False)

6
toolbox/codes/ui/login_ui.py
View File

@@ -126,6 +126,8 @@ class LoginWindow(QWidget):
self.le_password.returnPressed.connect(self.login_check)
self.le_password_reg_confirm.returnPressed.connect(self.register_check)
QShortcut("Esc", self).activated.connect(self.close)
QShortcut("Alt+1", self).activated.connect(lambda: self.tabW_login.setCurrentIndex(0))
QShortcut("Alt+2", self).activated.connect(lambda: self.tabW_login.setCurrentIndex(1))
def onChange_tabW(self):
text = self.tabW_login.tabText(self.tabW_login.currentIndex())
@@ -142,7 +144,7 @@ class LoginWindow(QWidget):
@handle_exception()
def login_check(self):
def login_failed(flag: int = 0):
reason = {-1: "用户名或密码为空", 0: "用户名未注册", 1: "解迷成功，密码错误", 2: "解迷失败，密码错误", 3: "数据库中有重复的用户名"}
reason = {-1: "用户名或密码为空", 0: "用户名未注册", 1: "解密成功，密码错误", 2: "解密失败，密码错误", 3: "数据库中有重复的用户名"}
self.le_username.clear()
self.le_password.clear()
self.le_username.setFocus()
@@ -172,7 +174,6 @@ class LoginWindow(QWidget):
clibs.username = username
clibs.password = password
db_operation.db_write_logs(f"username:{username} 登录成功！")
db_operation.db_close()
self.deleteLater()
except ValueError:
login_failed(flag=2)
@@ -224,7 +225,6 @@ class LoginWindow(QWidget):
validate_register()
def closeEvent(self, event):
db_operation.db_close()
event.accept()

2
toolbox/codes/ui/main_ui.py
View File

@@ -5,7 +5,7 @@ import requests
import json
import sys
from PySide6.QtWidgets import QApplication, QWidget, QHBoxLayout, QListWidget, QStackedWidget, QMessageBox, QToolBar, QMainWindow, QStatusBar
from PySide6.QtWidgets import QApplication, QWidget, QHBoxLayout, QMessageBox, QMainWindow, QStatusBar
from PySide6.QtGui import QFont, QIcon, QResizeEvent, QShortcut, QKeySequence, QAction
from PySide6.QtCore import Qt

2
toolbox/codes/ui/overlay_page/overlay_ui.py
View File

@@ -1,7 +1,7 @@
import sys
from PySide6.QtWidgets import QWidget, QApplication, QSizePolicy, QVBoxLayout, QLabel, QFrame, QHBoxLayout, QLineEdit, QMessageBox
from PySide6.QtGui import QPixmap, QPainter, QFontDatabase, QFont, QBrush, QShortcut, QKeySequence, QColor
from PySide6.QtGui import QPixmap, QPainter, QFontDatabase, QFont, QBrush, QColor
from PySide6.QtCore import Qt, QPoint, QDateTime, Signal, QTimer
from zhdate import ZhDate

209
toolbox/codes/ui/stacked_pages/w08_log.py
View File

@@ -1,25 +1,224 @@
from PySide6.QtWidgets import QWidget, QLabel
from PySide6.QtWidgets import QWidget, QLabel, QMessageBox, QVBoxLayout, QTreeWidget, QHBoxLayout, QPushButton, QFrame, QLineEdit, QCheckBox, QTreeWidgetItem, QDialog
from PySide6.QtCore import Qt, Signal
from PySide6.QtGui import QColor, QIcon, QFont, QKeySequence, QIntValidator, QShortcut
from codes.common.signal_bus import signal_bus
from codes.common import db_operation
from codes.common import clibs
class PageNumberInput(QDialog):
def __init__(self, parent=None):
super().__init__(parent)
self.init_ui()
self.setup_slot()
def setup_slot(self):
QShortcut(QKeySequence("Esc"), self).activated.connect(self.reject)
self.le_page_number.returnPressed.connect(self.accept)
self.le_page_number.returnPressed.connect(self.get_page_number)
def init_ui(self):
self.setMinimumSize(300,100)
self.setMaximumSize(400,120)
self.resize(300, 100)
self.setWindowIcon(QIcon(clibs.icon))
self.setWindowTitle("输入页码")
layout_h = QHBoxLayout()
layout_h.addStretch(1)
self.le_page_number = QLineEdit(self)
self.le_page_number.setText("1")
self.le_page_number.selectAll()
self.le_page_number.setFont(QFont('Consolas', 14))
self.le_page_number.setValidator(QIntValidator(0, 9999999, self))
layout_h.addWidget(self.le_page_number, stretch=4)
layout_h.addStretch(1)
self.setLayout(layout_h)
def get_page_number(self):
text = self.le_page_number.text()
return 1 if text == 0 else int(text)
class ClickableLabel(QLabel):
clicked = Signal()
def mousePressEvent(self, event):
if event.button() == Qt.MouseButton.LeftButton:
self.clicked.emit()
super().mousePressEvent(event)
class W08Log(QWidget):
def __init__(self, parent=None):
super().__init__(parent)
self.predos()
self.ui_init()
self.setup_slot()
self.setup_sc()
def predos(self):
...
self.records, self.len_records = "", ""
self.is_searching = False
self.max_item_number = clibs.config["log_number_per_page"]
def ui_init(self):
self.lb_test = QLabel(f"testing text on widget: \n{__file__}", parent=self)
layout_v = QVBoxLayout(self)
self.treeW = QTreeWidget()
self.treeW.setHeaderLabels(["ID", "时间戳", "告警级别", "模块信息", "告警内容"])
layout_v.addWidget(self.treeW, stretch=9)
layout_h = QHBoxLayout()
self.pb_previous = QPushButton("上一页")
layout_h.addWidget(self.pb_previous, stretch=1)
self.lb_page = ClickableLabel("999999/999999")
self.lb_page.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.lb_page.setMinimumWidth(144)
layout_h.addWidget(self.lb_page, stretch=2)
self.pb_next = QPushButton("下一页")
layout_h.addWidget(self.pb_next, stretch=1)
layout_h.addStretch(9)
self.frame_checkbox = QFrame()
layout_h_checkbox = QHBoxLayout()
self.box_info = QCheckBox("通知", parent=self.frame_checkbox)
self.box_info.setChecked(True)
layout_h_checkbox.addWidget(self.box_info, stretch=1)
self.box_warning = QCheckBox("告警", parent=self.frame_checkbox)
self.box_warning.setChecked(True)
layout_h_checkbox.addWidget(self.box_warning, stretch=1)
self.box_error = QCheckBox("错误", parent=self.frame_checkbox)
self.box_error.setChecked(True)
layout_h_checkbox.addWidget(self.box_error, stretch=1)
self.box_exception = QCheckBox("异常", parent=self.frame_checkbox)
self.box_exception.setChecked(True)
layout_h_checkbox.addWidget(self.box_exception, stretch=1)
self.box_unknown = QCheckBox("未知", parent=self.frame_checkbox)
self.box_unknown.setChecked(True)
layout_h_checkbox.addWidget(self.box_unknown, stretch=1)
layout_h.addLayout(layout_h_checkbox, stretch=4)
self.le_search = QLineEdit()
self.le_search.setPlaceholderText("告警内容")
self.le_search.setMinimumWidth(300)
layout_h.addWidget(self.le_search, stretch=5)
self.pb_search = QPushButton("查找")
layout_h.addWidget(self.pb_search, stretch=1)
layout_v.addLayout(layout_h, stretch=1)
self.setLayout(layout_v)
def setup_slot(self):
...
self.pb_previous.clicked.connect(self.previous_page)
self.pb_next.clicked.connect(self.next_page)
self.pb_search.clicked.connect(self.search_page)
self.le_search.returnPressed.connect(self.search_page)
self.lb_page.clicked.connect(self.goto_page)
signal_bus.stacked_page_switch_log.connect(self.show_latest_page)
def setup_sc(self):
...
def previous_page(self):
if not self.is_searching:
self.records, self.len_records = db_operation.db_query_logs(True)
current, total = self.lb_page.text().split("/")
page_number = int(current) - 1
self.show_page(self.records, self.len_records, page_number=page_number)
def next_page(self):
if not self.is_searching:
self.records, self.len_records = db_operation.db_query_logs(True)
current, total = self.lb_page.text().split("/")
page_number = int(current) + 1
self.show_page(self.records, self.len_records, page_number=page_number)
def search_page(self):
filters = {"info": self.box_info.isChecked(), "warning": self.box_warning.isChecked(), "error": self.box_error.isChecked(), "exception": self.box_exception.isChecked(), "unknown": self.box_unknown.isChecked()}
search_text = self.le_search.text().strip()
flag, levels = False, []
for level, enable in filters.items():
if not enable:
continue
flag = True
levels.append(level)
if not flag:
QMessageBox().warning(None, "警告", "至少选择一个过滤器！")
return
self.records, self.len_records = db_operation.db_query_logs(levels=levels)
if search_text:
# ids = [_[0] for _ in self.records]
# placeholder = ",".join(ids)
# clibs.cursor.execute(f"SELECT * FROM logs WHERE id IN ({placeholder}) and content like ?", (ids + [search_text, ]))
self.records, self.len_records = db_operation.db_query_logs(search_text, self.records)
self.is_searching = True
self.show_page(self.records, self.len_records, page_number=None)
def goto_page(self):
dlg = PageNumberInput()
if dlg.exec() != 1:
return
page_number = dlg.get_page_number()
if not self.is_searching:
self.records, self.len_records = db_operation.db_query_logs(True)
self.show_page(self.records, self.len_records, page_number=page_number)
def show_latest_page(self):
self.records, self.len_records = db_operation.db_query_logs(True)
self.is_searching = False
self.box_info.setChecked(True)
self.box_warning.setChecked(True)
self.box_error.setChecked(True)
self.box_exception.setChecked(True)
self.box_unknown.setChecked(True)
self.le_search.clear()
self.show_page(self.records, self.len_records, page_number=None)
def show_page(self, records, len_records, page_number: int | None):
if len_records == 0:
self.treeW.clear()
return
remainder = len_records % self.max_item_number
total = len_records // self.max_item_number + 1 if remainder else len_records // self.max_item_number
if page_number is None:
current = total
elif page_number <= 0:
current = 1
elif page_number < total:
current = page_number
else:
current = total
self.lb_page.setText(f"{current}/{total}")
if current == 1:
idx_start = 0
idx_end = self.max_item_number if len_records >= self.max_item_number else len_records
elif current == total:
remainder = len_records % self.max_item_number
idx_start = len_records - remainder if remainder else len_records - self.max_item_number
idx_end = len_records
else:
idx_start = self.max_item_number * (current-1)
idx_end = self.max_item_number * current
self.treeW.clear()
for record in records[idx_start:idx_end]:
record = [str(_) for _ in record]
item = QTreeWidgetItem(self.treeW, record)
self.treeW.addTopLevelItem(item)
colors = {"info": QColor(144, 238, 144), "warning": QColor(255, 240, 210), "error": QColor(255, 220, 220), "exception": QColor(255, 220, 220), "unknown": QColor(255, 220, 220)}
color = colors[record[2]]
for col in range(5):
item.setBackground(col, color)
self.treeW.scrollToBottom()

4
toolbox/main.py
View File

@@ -13,7 +13,7 @@ if __name__ == '__main__':
clibs.config = json.load(f)
app = QApplication(sys.argv)
# window = LoginWindow()
window = MainWindow()
window = LoginWindow()
# window = MainWindow()
window.show()
sys.exit(app.exec())