# -*- coding:utf-8 -*- import sys import logging from file import fileDataCreating from common import global_var from pathlib import Path from PySide6.QtUiTools import QUiLoader from PySide6.QtCore import Signal,QThread,QObject,QEventLoop,QTimer from PySide6.QtWidgets import QApplication from PySide6.QtGui import QTextCursor # 声明全局变量 global_var._init() # 控制台重定向到GUI class EmittingStr(QObject): # 定义信号 textWritten = Signal(str) # 将控制台的内容输出到QTextBrowser def write(self, text): # 调用 emit方法 发信号时,传入参数 必须是这里指定的 参数类型 self.textWritten.emit(str(text)) loop = QEventLoop() QTimer.singleShot(100, loop.quit) loop.exec_() QApplication.processEvents() # logging重定向到print class PrintHandler(logging.Handler): def emit(self, record): pass # 处理文件的线程 class FileThread(QThread): def __init__(self, parent=None): super().__init__(parent=parent) # 覆盖原来run去调用文件造数据脚本 def run(self): fileDataCreating.main() # GUI的主程序 class Window: def __init__(self): #载入ui path = Path(__file__) ui_file_patch = path.parent.parent / "ui" / "ui_main.ui" self.ui = QUiLoader().load(str(ui_file_patch)) #指定按钮和槽函数 self.ui.startButton.clicked.connect(self.start) self.ui.stopButton.clicked.connect(self.stop) self.ui.conn_Button.clicked.connect(self.conntest) self.setup_thread() # QTextBrowser输出设置,限制条数为100 self.ui.Log_Output.document().setMaximumBlockCount(100) self.ui.Log_Output.ensureCursorVisible() # 将控制台输出重定向到QTextBrowser sys.stdout = EmittingStr() sys.stdout.textWritten.connect(self.outputWritten) # 创建并配置logger self.logger = logging.getLogger() self.logger.setLevel(logging.INFO) # 创建并配置处理程序 handler = PrintHandler() handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')) self.logger.addHandler(handler) # 输出到GUI,信号处理函数 def outputWritten(self, text): cursor = self.ui.Log_Output.textCursor() cursor.movePosition(QTextCursor.End) cursor.insertText(text) self.ui.Log_Output.setTextCursor(cursor) self.ui.Log_Output.ensureCursorVisible() def start(self): self.logger.info("点击了开始按钮") # 输入框参数校验 num = self.ui.tabWidget.currentIndex() if num == 0: self.check_file_input() else: self.conntest() # 开始运行 if num == 0: self.start_thread() else: self.start_run_db() # 按钮置灰 self.ui.startButton.setEnabled(False) def stop(self): self.logger.info("点击了停止按钮") num = self.ui.tabWidget.currentIndex() if num == 0: self.stop_run_file() else: self.stop_run_db() # 按钮恢复 # self.startButton.isEnabled() 获取按钮的状态 self.ui.startButton.setEnabled(True) def conntest(self): self.logger.info("点击了测试连接按钮") self.check_db_input() # TODO(MH):校验连接信息 self.logger.info("数据库连接成功") def setup_thread(self): self.thread = FileThread() global_var.set_value('start_flag',True) self.thread_running = global_var.get_value('start_flag') def start_thread(self): if self.thread_running: self.thread.start() else: self.setup_thread() self.thread.start() def check_file_input(self): self.logger.info("文件参数校验通过") def check_db_input(self): self.logger.info("数据库参数校验通过,开始连接数据库测试。。。") def start_run_file(self): # 调用文件处理脚本 self.logger.info("开始生成文件。。。") def start_run_db(self): pass def stop_run_file(self): # 终止线程的事件循环 global_var.set_value('start_flag',False) self.thread.quit() # 标记线程停止 self.thread_running = global_var.get_value('start_flag') def stop_run_db(self): pass