PyQt6异步任务管理：QThreadPool与QThread的选择与应用

# 原始 TaskRunner (QRunnable)
class TaskRunner(QRunnable):
    def __init__(self, parent: Loading | None, task: Callable):
        super().__init__()
        self.parent = parent
        self.task = task

    def run(self):
        self.task(self.parent)
        if self.parent:
            self.parent.task_done(None)

# 原始 Loading 类片段
class Loading(Ui_Form, QWidget):
    # ...
    def __init__(self, # ...):
        # ...
        self.thread_pool = QThreadPool(self)
        self.thread_pool.destroyed.connect(self.quit) # 此信号可能不会在预期时机触发
        self.run_tasks()
        # ...

    def closeEvent(self, a0):
        self.timer.stop()
        self.thread_pool.waitForDone(-1) # 阻塞等待，但线程池可能永不“完成”
        self.close() # 再次调用close，可能导致递归或无用

    def run_tasks(self): self.thread_pool.start(self.task)

from src.gui.loading import Ui_Form
from PyQt6.QtWidgets import QWidget, QApplication
from PyQt6.QtCore import QThreadPool, QRunnable, QTimer, QThread, pyqtSignal
from typing import Callable
import time # 用于模拟耗时操作

# 1. TaskRunner 继承自 QThread
class TaskRunner(QThread):
    # 定义一个信号，用于在任务完成时通知主线程
    task_finished = pyqtSignal()

    def __init__(self, parent: QWidget | None, task: Callable):
        super().__init__(parent) # 将父对象传递给QThread
        self.task = task
        self.loading_page = parent # 保持对Loading实例的引用，如果需要从任务中更新UI

    def run(self):
        """
        线程的执行入口。
        在这里执行耗时操作。
        """
        try:
            # 模拟耗时操作，并将Loading实例传递给任务函数
            self.task(self.loading_page)
        finally:
            # 任务完成后发出信号
            self.task_finished.emit()

# 2. Loading 类调整
class Loading(Ui_Form, QWidget):
    def __init__(self,
                 parent: QWidget | None,
                 next_widget: QWidget | None,
                 action: str,
                 time: int,
                 task: Callable,
                 task_len: int,
                 initial_task: str):
        super().__init__()
        self.setupUi(self)
        self.setParent(parent)
        self.parent = parent
        self.next_widget = next_widget
        self.time = time

        # 不再使用 QThreadPool
        # self.thread_pool = QThreadPool(self)
        # self.thread_pool.destroyed.connect(self.quit)

        # 实例化 TaskRunner 作为 QThread
        self.task_thread = TaskRunner(self, task)
        # 连接任务完成信号到关闭窗口的槽函数
        self.task_thread.task_finished.connect(self.on_task_finished)

        self.current_time = 0
        self.tasks_done = 0
        self.all_tasks = task_len

        self.Task.setText(action)
        self.Estimation.setText(f"estimated time: {self.int_to_time(time)}")
        self.progressBar.setValue(0)
        self.TimeLeft.setText("")
        self.Current.setText("")
        self.Task.setText("")

        self.run_tasks() # 启动任务
        self.task_done(initial_task)

        self.timer = QTimer(self)
        self.timer.timeout.connect(self.update_time)
        self.timer.start(1000)

    @staticmethod
    def int_to_time(time: int) -> str:
        if time >= 3600:
            return f"{time / 3600} hours"
        elif time >= 60:
            return f"{time / 60} minutes"
        else:
            return f"{time} seconds"

    def update_time(self):
        self.current_time += 1
        self.TimeLeft.setText(self.int_to_time(self.current_time))

    def task_done(self, next_task: str = None):
        # 这里的逻辑需要根据实际任务数量调整，
        # 如果只有一个主任务，则在on_task_finished中处理关闭逻辑
        self.tasks_done += 1
        if not next_task: # 假设这是最后一个任务完成的标志
            self.Current.setText("finished all tasks, closing window")
            self.Tasks.setText(f"{self.tasks_done} out of {self.all_tasks}")
            # 如果是多个任务，可以在这里检查是否所有任务都完成
            # self.try_close_window() # 如果有多个任务，可以在这里调用尝试关闭
        elif self.tasks_done <= self.all_tasks: # 修正条件，避免超出总任务数
            self.Current.setText(f"currently: {next_task}")
            self.Tasks.setText(f"{self.tasks_done} out of {self.all_tasks}")

    def on_task_finished(self):
        """
        当后台任务完成时，此槽函数会被调用。
        """
        self.task_done(None) # 更新UI显示任务完成
        self.try_close_window() # 尝试关闭窗口

    def try_close_window(self):
        """
        检查是否可以安全关闭窗口。
        """
        # 确保所有任务都已完成，并且线程已安全退出
        if self.tasks_done >= self.all_tasks: # 确保所有任务都计数完成
            self.timer.stop() # 停止计时器
            self.task_thread.quit() # 请求线程退出
            self.task_thread.wait() # 等待线程真正退出，避免资源泄露
            self.close() # 关闭窗口
            # 如果有next_widget，可以在这里显示它
            if self.next_widget:
                self.next_widget.show()
                self.parent.close() # 如果有父窗口，也关闭父窗口

    def closeEvent(self, event):
        """
        重写 closeEvent 以确保在用户关闭窗口时，线程也能被正确终止。
        """
        self.timer.stop()
        if self.task_thread.isRunning():
            self.task_thread.quit()
            self.task_thread.wait() # 确保线程在窗口关闭前退出
        super().closeEvent(event) # 调用父类的closeEvent

    def run_tasks(self):
        """
        启动后台任务线程。
        """
        self.task_thread.start()

# 示例测试代码
if __name__ == '__main__':
    class TestLoading:
        def test_task(self):
            def foo(loading_page: Loading):
                print("Task started...")
                for i in range(5):
                    time.sleep(1) # 模拟耗时操作
                    print(f"Task progress: {i+1}/5")
                    # 从子线程更新UI需要使用信号槽机制
                    # loading_page.progressBar.setValue(int((i+1)/5 * 100)) # 错误：直接访问UI
                print("Task finished.")

            app = QApplication([])
            # 假设有一个主窗口或父窗口
            main_window = QWidget()
            main_window.setWindowTitle("Main Application")
            main_window.resize(300, 200)

            # 假设有一个后续要显示的窗口
            next_window = QWidget()
            next_window.setWindowTitle("Next Window")
            next_window.resize(400, 300)

            # 传递 main_window 作为 parent，next_window 作为 next_widget
            self.loader = Loading(main_window, next_window, "doing something", 5, foo, 1, "testing")
            self.loader.show()

            # 如果 loader 是顶层窗口，可以直接执行app.exec()
            # 如果 loader 是子窗口，则应该在某个地方调用它的show()
            # 这里为了测试，直接显示loader，并在任务完成后关闭它
            app.exec()