1. 解耦设计

在使用 pyqt5 开发 gui 时，建议遵从 ”UI 设计“ 与”功能逻辑“解耦的原则。

这不仅是软件工程的通用最佳实践，在 PyQt5 这种基于事件驱动和信号槽机制的框架中尤为重要。将界面代码（View）与业务逻辑（Model/Controller）分离，能带来以下核心优势：

1. 为什么要解耦？

• 可维护性 (Maintainability)
  • 如果 UI 代码和业务逻辑混在一起（例如在按钮点击事件中直接写数据库查询或复杂计算），代码会变得极其臃肿且难以阅读。
  • 解耦后，修改界面布局（如调整按钮位置、更换控件）不会影响核心业务逻辑；反之，修改算法或数据处理流程也不需要动界面代码。
• 可测试性 (Testability)
  • 这是最关键的一点。PyQt5 的 UI 类通常依赖 QApplication 实例和图形环境，难以进行自动化单元测试。
  • 如果逻辑解耦为独立的普通 Python 类或函数，你可以直接使用 pytest 或 unittest 对其进行测试，而无需启动 GUI 窗口，极大提高了测试效率和覆盖率。
• 协作开发 (Collaboration)
  • UI 设计师可以使用 Qt Designer (.ui 文件) 专注于界面布局和样式，而开发人员专注于后端逻辑。两者通过明确的接口（信号 / 槽或方法调用）交互，互不干扰。
• 复用性 (Reusability)
  • 解耦后的业务逻辑可以被其他非 GUI 程序（如命令行工具、Web 后端）复用。
  • 同一套逻辑也可以轻松应用到不同的 UI 框架中（虽然需要重写 View 层，但 Model/Controller 层可以保留）。

2. 如何实现解耦

在 PyQt5 中，最常用的解耦模式是 MVC (Model-View-Controller) 或其变体 MVVM(Model-View-ViewModel)。

进化版：Model + Worker + View + Controller来实现全解耦

逻辑分层架构

将代码分为四个主要部分：

1. View (界面层)
   • 只负责显示数据和接收用户输入。
   • 不包含复杂的业务判断。
   • 通过 Signal (信号) 发出用户意图（如 login_requested, data_saved）。
2. Model (数据 / 逻辑层)
   • 纯 Python 类，不引用任何 QtWidgets 或 QtGui 模块（除了可能用到 QObject 做信号发射，但最好完全独立）。
   • 负责数据处理、网络请求、数据库操作、算法计算。
   • 处理完后，通过回调或信号通知 View 更新。
3. Controller / Presenter (控制层)
   • 作为 View 和 Model 的桥梁。
   • 连接 View 的信号到 Model 的方法。
   • 连接 Model 的结果信号到 View 的更新槽函数。
4. Worker 层
   • 线程容器 + 调度器 (有线程，有队列，调用 Model，发射信号)

2. 代码结构示例

2.1 view 界面层

这部分只写基础 ui 布局，构建用户界面

class BaseWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.initUI()
        self.setStyleSheet(qss1)

    def initUI():
        self.setWindowTitle("GUI 示例 ")
        self.resize(500, 400)

        #创建主窗口部件
        main_widget = QWidget()
        self.setCentralWidget(main_widget)
        self.main_layout = QVBoxLayout()

        #创建菜单和主体内容
        self._initMenubar()
        self._initMainContent()

        main_widget.setLayout(self.main_layout)

    def _initMenubar():
        #创建菜单栏

    def _initMainContent():
        grid_layout = QGridLayout()
        grid_layout.addWidget(self.layout1_group, 0, 0)
        grid_layout.addWidget(self.layout2_group, 0, 1)

        self.main_layout.addLayout(grid_layout)

    def layout1_group(self):
        group = QGroupBox("test")

        form_layout = QFormLayout()
        layout1, slef.btn_table1, self.edit_product = self.create_input_field()
        form_layout.addRow("table", layout1)     

注意：在对输入框，按钮等重要元素命名时，建议遵循一定的规律，比如按钮命名 btn_开头，输入框命名 edit_开头

2.2 Controller 核心调度逻辑

作为组装与调度核心，这部分代码主要是信号连接工作。常常我们也将初始化界面层，实例化逻辑层，执行任务等工作也放在这里。

将所有东西在这里组装。

这部分内容可以自由发挥，有时候我们会创建一个中间类或者直接在正式的 ui 中做这些工作，前面的 view 只作为一个基础 base 类。

class MainWindow(BaseWindow)：
    def __init__(self):
        super().__init__()

        self.active_threads = []
        self.run_timer = QTimer(self)
        self.cache_timer = QTimer(self)

        self.btn_table1.click.connect(self.update_table1)
        self.edit_product.click.connect(self.show_result)

    def _start_task(self, task_func, *args):
        ithread = WorkThread(task_func, *args)
        ithread.log_signal.connect(self.log)
        ithread.finished.signal.connect(lambda: self._remove_thread(ithread)
        )
        self.active_threads.append(ithread)
        return ithread

    def _remove_thread(self, thread):
        if thread in self.active_threads:
            self.active_threads.remove(thread)

    def log(self, log:str):
        if "failed" in log:
            QMessageBox.critical(self, "error", log)
        else:
            print(log)

    def get_calc_status(self, result_folder):
        ithread  = self._start_task(parse_log, result_folder)
        ithread.result_signal.connect(lambda res: self.update_opc_status(res, "ok")
        )
        ithread.start()

2.3 Worker 线程任务层

为了避免阻塞主界面，我们需要将计算等耗时任务放到子线程执行。然后将计算的结果通过信号的方式传回给主界面显示。

from PyQt5.QtCore import QThread, pyqtSignal
from collections.abc import Iterator 

class WorkThread(Qthread):

    #定义信号
    log_signal = pyqtSignal(str)
    finish_signal = pyqtSignal()
    result_signal = pyqtSignal(object)

    def __init__(self, task_func, *args, **kwargs):
        supper().__init__()
        self.task_func = task_func
        self.args = args
        self.kwargs = kwargs

    def run(self):
        try:
            result = None
            ret = self.task_func(*args, **kwargs)  #我们的函数是通过 yield 回传 log 的
            if isinstance(ret, Iterator):
                try:
                    while True:
                        log_msg = next(ret)
                        self.log_signal.emit(str(log_msg))
                except StopIteration as e:
                    result = e.value
                    self.result_signal.emit(result)
            else:
               result = e.value
               self.result_signal.emit(result)
        except Exception as e:
            self.log_signal.emit(f"task failed: {str(e)}")
        finally:
            self.finish_signal.emit()
            self.quit()

上述示例中，我们是通过 yield 回传 log。

另一种方法是在 WorkThread 中定义回调函数，并将回调函数传到 model 逻辑层处理。

def run(self):
    try:
        #定义内部回调函数，将 Model 的日志转发为 Qt 信号
        def log_handler(msg):
            self.log_signal.emit(msg)
        ret = self.task_func(log_handler, *args, **kwargs)
        self.result_signal.emit(ret)

    except Exception as e:
        self.log_signal.emit(f"task failed: {str(e)}")
    finally:
        self.finish_signal.emit()
        self.quit()

• 在当前的 PyQt5 + QThread 架构下：回调函数（Callback）更好、更稳健、更符合 Qt 生态。
• 性能最优：没有生成器挂起 / 恢复的开销，直接在子线程函数栈中执行。

2.4 Model 数据 / 逻辑层

这里只写纯业务逻辑层，只有纯 python 计算函数，没有任何 PyQt 依赖，没有信号，没有线程。

# yield 的例子
def parse_log(folder):
    if not os.path.exists(folder): return 0
    yield "search all logs"
    logs = [os.path.join(folder, log) for log in os.listdir(folder) if log.endswith(".log")]
    if logs:
        log_infos = []
        for logfile in logs:
            yield f"grep log: {logfile}"
            progress = calcuate_progress(logfile)
            return progress
#使用回调函数
def parse_log(log_cb, folder):
    if not os.path.exists(folder): return 0
    log_cb("search all logs")
    logs = [os.path.join(folder, log) for log in os.listdir(folder) if log.endswith(".log")]
    if logs:
        log_infos = []
        for logfile in logs:
            log_cb(f"grep log: {logfile}")
            progress = calcuate_progress(logfile)
            return progress

如果你觉得传一个函数 log_cb 不够优雅，可以传一个 轻量级的日志对象 给 Model，这个对象内部封装了信号发射。

class ThreadLogger:
    def __init__(self, signal_emit_func):
        self._emit = signal_emit_func

    def info(self, msg): self._emit(msg, "INFO")
    def debug(self, msg): self._emit(msg, "DEBUG")
    def error(self, msg): self._emit(msg, "ERROR")

def parse_log(logger: ThreadLogger, folder):
    if not os.path.exists(folder): return 0
    logger.info("search all logs")
    logs = [os.path.join(folder, log) for log in os.listdir(folder) if log.endswith(".log")]
    if logs:
        log_infos = []
        for logfile in logs:
            logger.info(f"grep log: {logfile}")
            progress = calcuate_progress(logfile)
            return progress