python - PYQT5：桂冻结

Question

我用 PyQt5 为 Mikrotik 制作程序，但是当我开始我的功能时 Gui 冻结了。我添加了 time.sleep() 函数，但它仍然冻结。这是它冻结的功能。我该如何解决？谢谢

def button_run(self):
        try:
            import paramiko
            from paramiko.client import SSHClient
            from paramiko.client import AutoAddPolicy

            import logging
            logging.basicConfig()
            logging.getLogger('paramiko.transport').setLevel(logging.DEBUG)

        except ImportError:
            sys.tracebacklimit = 0
            with open("error.log", "ab") as e:
                e.write(
                    bytes(time.strftime("%Y.%m.%d") + " " + time.strftime(
                        "%H:%M:%S") + "\"Paramiko\" module missing! Please visit http://www.paramiko.org/installing.html "
                                      "for more details." + "\r\n"))
            e.close()
            raise ImportError(
                "\rPlease install \"paramiko\" module! Visit http://www.paramiko.org/installing.html for more "
                "details.\r\n")

        ip = self.ipAddress.text()
        try:
            socket.inet_aton(ip)
            print("Your ip is: " + ip)
        except socket.error:
            print("Check your ip")

        split_ip = ip.split('.')
        ini_range = self.initRange.text()
        end_range = self.fineRange.text()

        file = open("hosts", "w")
        for x in range(int(ini_range), int(end_range)):
            file.write(split_ip[0] + '.' + split_ip[1] + '.' + split_ip[2] + '.' + str(x) + '\n')
        file.close()

        nlines = 0
        f = open("hosts", "r")
        mt_username = self.username.text()
        mt_password = self.password.text()
        timeout = 20

        for line in f:

            if nlines > 0:
                # print("\r\nWaiting 3s before logging on to the next device...\n")
                logging.info(bytes("Waiting 3s before logging on to the next device...", 'utf8'))
                time.sleep(3)

            nlines += 1
            conn_date = time.strftime("%Y.%m.%d")
            conn_time = time.strftime("%H:%M:%S")
            host = line.rstrip("\n")
            client = SSHClient()
            # ssh = paramiko.SSHClient()

            # ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            client.set_missing_host_key_policy(AutoAddPolicy)
            logging.info(bytes("Connecting to " + str(nlines) + ". host: " + host, 'utf8'))
            # print("\r\nConnecting to " + str(nlines) + ". host: " + host + "\r\n")

            try:
                client.connect(host, username=mt_username, password=mt_password, timeout=timeout, allow_agent=False)
            except socket.timeout as e:
                logging.warning(bytes("Connection timeout. Log entry created.", 'utf8'))
                # print("\r\nConnection timeout. Log entry created.")
                with open("error.log", "ab") as e:
                    e.write(
                        bytes(
                            conn_date + " " + conn_time + " " + mt_username + "@" + host + " Timeout connecting to the device." + "\r\n",
                            'utf8'))
                e.close()
                continue

            except paramiko.AuthenticationException:
                logging.warning(bytes("Wrong credentials. Log entry created.", 'utf8'))
                # print("\r\nWrong credentials. Log entry created.")
                with open("error.log", "ab") as e:
                    e.write(bytes(
                        conn_date + " " + conn_time + " " + mt_username + "@" + host + " Wrong credentials." + "\r\n",
                        'utf8'))
                e.close()
                continue

            except:
                logging.warning(bytes("Error connecting to the device. Log entry created.", 'utf8'))
                # print("\r\nError connecting to the device. Log entry created.")
                with open("error.log", "ab") as e:
                    e.write(
                        bytes(
                            conn_date + " " + conn_time + " " + mt_username + "@" + host + " Unknown error while connecting to the device." + "\r\n",
                            'utf8'))
                e.close()
                continue
            logging.info(bytes("Succsessfully connected to the host. Executing commands from command file", 'utf8'))
            # print("\r\nSuccsessfully connected to the host. Executing commands from the external file:\r\n")
            command = self.comandMT.toPlainText()
            k = open("commands", "w")
            k.write(command)
            k.close()

            k = open("commands", "r")
            for line in k:
                mt_command = line.rstrip("\n")
                # channel = client.get_transport().open_session()
                # channel.get_pty()
                # channel.settimeout(5)
                time.sleep(.3)
                # out1 = client.exec_command(command)
                stdin, stdout, stderr = client.exec_command(mt_command)
                stdin.close()
                stdout.close()
                stderr.close()
                logging.info(bytes(mt_command, 'utf8'))
            logging.info(bytes("External commands are executed successfully.", 'utf8'))
            # print("\nExternal commands are executed successfully.\n")
            # k.close()
            # thread1 = threading.Thread(target=self.read_until_EOF, args=(out1,))
            # thread1.start()
            # thread1.join()

            # channel.close()

            client.get_transport().close()
            client.close()

        if nlines == 0:
            logging.info(bytes("List of hosts is empty.", 'utf8'))
            # print("\nList of hosts is empty.\n")
        else:
            logging.info(bytes("End of the program.", 'utf8'))
            # print("\nEnd of the program.\n")
        f.close()

        quit()

完整代码：

from PyQt5 import QtCore, QtGui, QtWidgets
from PyQt5.QtGui import QTextCursor
import ipaddress
import logging
import resource
from script.resource import resource_rc
import socket
import sys
import threading
import time


 class IODirector(object):
    def __init__(self, text_area):
        self.text_area = text_area


 class StdoutDirector(IODirector):
    def __init__(self, text_area):
        super().__init__(text_area)

    def write(self, msg):
        self.text_area.moveCursor(QtGui.QTextCursor.End)
        # cursor = self.text_area.textCursor()
        self.text_area.insertPlainText(msg)
        # self.text_area.setTextCursor(cursor)

    def flush(self):
        pass


class Ui_MainWindow(object):
    def __init__(self):
        self.active = False

    def start(self, MainWindow):
        MainWindow.setObjectName("MainWindow")
        MainWindow.resize(1047, 660)
        icon = QtGui.QIcon()
        icon.addPixmap(QtGui.QPixmap("script\\../../../../.designer/backup/iconapp.png"), QtGui.QIcon.Normal,
                       QtGui.QIcon.Off)
        MainWindow.setWindowIcon(icon)
        MainWindow.setStyleSheet("background-image: url(:/images/images/particle.jpg);")
        self.centralwidget = QtWidgets.QWidget(MainWindow)
        self.centralwidget.setObjectName("centralwidget")
        self.comandMT = QtWidgets.QPlainTextEdit(self.centralwidget)
        self.comandMT.setGeometry(QtCore.QRect(390, 70, 611, 271))
        font = QtGui.QFont()
        font.setFamily("Times New Roman")
        font.setPointSize(12)
        self.comandMT.setFont(font)
        self.comandMT.setStyleSheet("background-color: rgb(0, 0, 0, 0.7);\n"
                                    "color: rgb(255, 255, 255);")
        self.comandMT.setObjectName("comandMT")
        self.startButton = QtWidgets.QPushButton(self.centralwidget)
        self.startButton.setGeometry(QtCore.QRect(40, 460, 121, 41))
        font = QtGui.QFont()
        font.setFamily("Times New Roman")
        font.setPointSize(12)
        font.setBold(True)
        font.setWeight(75)
        self.startButton.setFont(font)
        self.startButton.setStyleSheet("background-color: rgb(0, 0, 0, 0.1);\n"
                                       "color: rgb(255, 255, 255);")
        self.startButton.setObjectName("startButton")
        self.ipAddress = QtWidgets.QLineEdit(self.centralwidget)
        self.ipAddress.setGeometry(QtCore.QRect(60, 140, 221, 41))
        font = QtGui.QFont()
        font.setFamily("Times New Roman")
        font.setPointSize(13)
        font.setBold(True)
        font.setWeight(75)
        self.ipAddress.setFont(font)
        self.ipAddress.setStyleSheet("background-color: rgb(255, 255, 255, 0.1);\n"
                                     "color: rgb(255, 255, 255);")
        self.ipAddress.setAlignment(QtCore.Qt.AlignCenter)
        self.ipAddress.setObjectName("ipAddress")
        self.initRange = QtWidgets.QLineEdit(self.centralwidget)
        self.initRange.setGeometry(QtCore.QRect(60, 200, 131, 41))
        font = QtGui.QFont()
        font.setFamily("Times New Roman")
        font.setPointSize(13)
        font.setBold(True)
        font.setWeight(75)
        self.initRange.setFont(font)
        self.initRange.setStyleSheet("background-color: rgb(0, 0, 0, 0.4);\n"
                                     "color: rgb(255, 255, 255);")
        self.initRange.setAlignment(QtCore.Qt.AlignCenter)
        self.initRange.setObjectName("initRange")
        self.fineRange = QtWidgets.QLineEdit(self.centralwidget)
        self.fineRange.setGeometry(QtCore.QRect(60, 260, 131, 41))
        font = QtGui.QFont()
        font.setFamily("Times New Roman")
        font.setPointSize(13)
        font.setBold(True)
        font.setWeight(75)
        self.fineRange.setFont(font)
        self.fineRange.setStyleSheet("background-color: rgb(0, 0, 0, 0.4);\n"
                                     "color: rgb(255, 255, 255);")
        self.fineRange.setAlignment(QtCore.Qt.AlignCenter)
        self.fineRange.setObjectName("fineRange")
        self.username = QtWidgets.QLineEdit(self.centralwidget)
        self.username.setGeometry(QtCore.QRect(60, 320, 231, 41))
        font = QtGui.QFont()
        font.setFamily("Times New Roman")
        font.setPointSize(13)
        font.setBold(True)
        font.setWeight(75)
        self.username.setFont(font)
        self.username.setStyleSheet("background-color: rgb(0, 0, 0, 0.4);\n"
                                    "color: rgb(255, 255, 255);")
        self.username.setAlignment(QtCore.Qt.AlignCenter)
        self.username.setObjectName("username")
        self.password = QtWidgets.QLineEdit(self.centralwidget)
        self.password.setGeometry(QtCore.QRect(60, 380, 231, 41))
        font = QtGui.QFont()
        font.setFamily("Times New Roman")
        font.setPointSize(13)
        font.setBold(True)
        font.setWeight(75)
        self.password.setFont(font)
        self.password.setStyleSheet("background-color: rgb(0, 0, 0, 0.4);\n"
                                    "color: rgb(255, 255, 255);")
        self.password.setAlignment(QtCore.Qt.AlignCenter)
        self.password.setObjectName("password")
        self.text_area = QtWidgets.QPlainTextEdit(self.centralwidget)
        self.text_area.setGeometry(QtCore.QRect(390, 360, 611, 241))
        font = QtGui.QFont()
        font.setFamily("Times New Roman")
        font.setPointSize(12)
        font.setItalic(False)
        font.setStrikeOut(False)
        self.text_area.setFont(font)
        self.text_area.setStyleSheet("background-color: rgb(0, 0, 0, 0.7);\n"
                                     "color: rgb(255, 255, 255);")
        self.text_area.setFrameShape(QtWidgets.QFrame.Panel)
        self.text_area.setFrameShadow(QtWidgets.QFrame.Plain)
        self.text_area.setVerticalScrollBarPolicy(QtCore.Qt.ScrollBarAsNeeded)
        self.text_area.setSizeAdjustPolicy(QtWidgets.QAbstractScrollArea.AdjustIgnored)
        self.text_area.setReadOnly(True)
        self.text_area.setCenterOnScroll(True)
        self.text_area.setObjectName("text_area")
        self.stopButton = QtWidgets.QPushButton(self.centralwidget)
        self.stopButton.setGeometry(QtCore.QRect(180, 460, 121, 41))
        font = QtGui.QFont()
        font.setFamily("Times New Roman")
        font.setPointSize(12)
        font.setBold(True)
        font.setWeight(75)
        self.stopButton.setFont(font)
        self.stopButton.setStyleSheet("background-color: rgb(0, 0, 0, 0.1);\n"
                                      "color: rgb(255, 255, 255);")
        self.stopButton.setObjectName("stopButton")

        self.label = QtWidgets.QLabel(self.centralwidget)
        self.label.setGeometry(QtCore.QRect(640, 20, 131, 31))
        font = QtGui.QFont()
        font.setFamily("Times New Roman")
        font.setPointSize(12)
        font.setBold(True)
        font.setWeight(75)
        self.label.setFont(font)
        self.label.setStyleSheet("color: rgb(255, 255, 255);\n"
                                 "background-color: rgb(0,0,0,0.1);")
        self.autoButton = QtWidgets.QPushButton(self.centralwidget)

        self.autoButton.setGeometry(QtCore.QRect(110, 520, 121, 41))
        font = QtGui.QFont()
        font.setFamily("Times New Roman")
        font.setPointSize(12)
        font.setBold(True)
        font.setWeight(75)
        self.autoButton.setFont(font)
        self.autoButton.setStyleSheet("background-color: rgb(0, 0, 0, 0.1);\n"
                                      "color: rgb(255, 255, 255);")
        self.autoButton.setObjectName("autoButton")
        self.label.setAlignment(QtCore.Qt.AlignCenter)
        self.label.setObjectName("label")
        self.graphicsView = QtWidgets.QGraphicsView(self.centralwidget)
        self.graphicsView.setGeometry(QtCore.QRect(70, 40, 191, 41))
        self.graphicsView.setStyleSheet("border-image: url(:/images/images/logo-bianco-azzurro.png);")
        self.graphicsView.setObjectName("graphicsView")
        self.startButton.raise_()
        self.ipAddress.raise_()
        self.initRange.raise_()
        self.fineRange.raise_()
        self.username.raise_()
        self.password.raise_()
        self.text_area.raise_()
        self.stopButton.raise_()
        self.label.raise_()
        self.graphicsView.raise_()
        self.comandMT.raise_()
        self.autoButton.raise_()
        MainWindow.setCentralWidget(self.centralwidget)

        self.retranslateUi(MainWindow)
        QtCore.QMetaObject.connectSlotsByName(MainWindow)
        self.threadpool = QtCore.QThreadPool()
        print("Multithreading with maximum %d threads" % self.threadpool.maxThreadCount())
        self.startButton.clicked.connect(self.do_run)
        self.stopButton.clicked.connect(self.stop_thread)

    def retranslateUi(self, MainWindow):
        _translate = QtCore.QCoreApplication.translate
        MainWindow.setWindowTitle(_translate("MainWindow", "MainWindow"))
        self.comandMT.setPlaceholderText(_translate("MainWindow", "Mikrotik Command"))
        self.startButton.setText(_translate("MainWindow", "START"))
        self.ipAddress.setPlaceholderText(_translate("MainWindow", "IP Address"))
        self.initRange.setPlaceholderText(_translate("MainWindow", "Init Range"))
        self.fineRange.setPlaceholderText(_translate("MainWindow", "Fine Range"))
        self.username.setPlaceholderText(_translate("MainWindow", "Username"))
        self.password.setPlaceholderText(_translate("MainWindow", "Password"))
        self.text_area.setPlaceholderText(_translate("MainWindow", "Logger"))
        self.stopButton.setText(_translate("MainWindow", "STOP"))
        self.autoButton.setText(_translate("MainWindow", "AUTO"))
        self.label.setText(_translate("MainWindow", " Mikrotik"))

    def do_run(self):

        t = threading.Thread(target=self.button_run)
        sys.stdout = StdoutDirector(self.text_area)
        # configure the nameless "root" logger to also write           # added
        # to the redirected sys.stdout                                 # added
        logger = logging.getLogger()  # added
        console = logging.StreamHandler(stream=sys.stdout)  # added
        logger.addHandler(console)  # added

        t.start()
        if t.is_alive():
            t.join()
            print('Still running')

        else:
            print('Completed')

    def stop_thread(self):
        self.packetizer.close()
        if self is not threading.current_thread():
            self.join(10)
            if self.is_alive():
                raise Exception("Timed out while trying to kill thread")

    # def read_until_EOF(fileobj):
    #     s = fileobj.readline()
    #     while s:
    #         print(s.strip())
    #         s = fileobj.readline()

    def button_run(self):
        try:
            import paramiko
            from paramiko.client import SSHClient
            from paramiko.client import AutoAddPolicy

            import logging
            logging.basicConfig()
            logging.getLogger('paramiko.transport').setLevel(logging.DEBUG)

        except ImportError:
            sys.tracebacklimit = 0
            with open("error.log", "ab") as e:
                e.write(
                    bytes(time.strftime("%Y.%m.%d") + " " + time.strftime(
                        "%H:%M:%S") + "\"Paramiko\" module missing! Please visit http://www.paramiko.org/installing.html "
                                      "for more details." + "\r\n"))
            e.close()
            raise ImportError(
                "\rPlease install \"paramiko\" module! Visit http://www.paramiko.org/installing.html for more "
                "details.\r\n")

        ip = self.ipAddress.text()
        try:
            socket.inet_aton(ip)
            print("Your ip is: " + ip)
        except socket.error:
            print("Check your ip")

        split_ip = ip.split('.')
        ini_range = self.initRange.text()
        end_range = self.fineRange.text()

        file = open("hosts", "w")
        for x in range(int(ini_range), int(end_range)):
            file.write(split_ip[0] + '.' + split_ip[1] + '.' + split_ip[2] + '.' + str(x) + '\n')
        file.close()

        nlines = 0
        f = open("hosts", "r")
        mt_username = self.username.text()
        mt_password = self.password.text()
        timeout = 20

        for line in f:

            if nlines > 0:
                # print("\r\nWaiting 3s before logging on to the next device...\n")
                logging.info(bytes("Waiting 3s before logging on to the next device...", 'utf8'))
                time.sleep(3)

            nlines += 1
            conn_date = time.strftime("%Y.%m.%d")
            conn_time = time.strftime("%H:%M:%S")
            host = line.rstrip("\n")
            client = SSHClient()
            # ssh = paramiko.SSHClient()

            # ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            client.set_missing_host_key_policy(AutoAddPolicy)
            logging.info(bytes("Connecting to " + str(nlines) + ". host: " + host, 'utf8'))
            # print("\r\nConnecting to " + str(nlines) + ". host: " + host + "\r\n")

            try:
                client.connect(host, username=mt_username, password=mt_password, timeout=timeout, allow_agent=False)
            except socket.timeout as e:
                logging.warning(bytes("Connection timeout. Log entry created.", 'utf8'))
                # print("\r\nConnection timeout. Log entry created.")
                with open("error.log", "ab") as e:
                    e.write(
                        bytes(
                            conn_date + " " + conn_time + " " + mt_username + "@" + host + " Timeout connecting to the device." + "\r\n",
                            'utf8'))
                e.close()
                continue

            except paramiko.AuthenticationException:
                logging.warning(bytes("Wrong credentials. Log entry created.", 'utf8'))
                # print("\r\nWrong credentials. Log entry created.")
                with open("error.log", "ab") as e:
                    e.write(bytes(
                        conn_date + " " + conn_time + " " + mt_username + "@" + host + " Wrong credentials." + "\r\n",
                        'utf8'))
                e.close()
                continue

            except:
                logging.warning(bytes("Error connecting to the device. Log entry created.", 'utf8'))
                # print("\r\nError connecting to the device. Log entry created.")
                with open("error.log", "ab") as e:
                    e.write(
                        bytes(
                            conn_date + " " + conn_time + " " + mt_username + "@" + host + " Unknown error while connecting to the device." + "\r\n",
                            'utf8'))
                e.close()
                continue
            logging.info(bytes("Succsessfully connected to the host. Executing commands from command file", 'utf8'))
            # print("\r\nSuccsessfully connected to the host. Executing commands from the external file:\r\n")
            command = self.comandMT.toPlainText()
            k = open("commands", "w")
            k.write(command)
            k.close()

            k = open("commands", "r")
            for line in k:
                mt_command = line.rstrip("\n")
                # channel = client.get_transport().open_session()
                # channel.get_pty()
                # channel.settimeout(5)
                time.sleep(.3)
                # out1 = client.exec_command(command)
                stdin, stdout, stderr = client.exec_command(mt_command)
                stdin.close()
                stdout.close()
                stderr.close()
                logging.info(bytes(mt_command, 'utf8'))
            logging.info(bytes("External commands are executed successfully.", 'utf8'))
            # print("\nExternal commands are executed successfully.\n")
            # k.close()
            # thread1 = threading.Thread(target=self.read_until_EOF, args=(out1,))
            # thread1.start()
            # thread1.join()

            # channel.close()

            client.get_transport().close()
            client.close()

        if nlines == 0:
            logging.info(bytes("List of hosts is empty.", 'utf8'))
            # print("\nList of hosts is empty.\n")
        else:
            logging.info(bytes("End of the program.", 'utf8'))
            # print("\nEnd of the program.\n")
        f.close()

        quit()


def main():
    import sys
    logging.basicConfig(level=logging.INFO)  # enable logging           # added
    if not QtWidgets.QApplication.instance():
        app = QtWidgets.QApplication(sys.argv)
    else:
        app = QtWidgets.QApplication.instance()

    MainWindow = QtWidgets.QMainWindow()
    ui = Ui_MainWindow()
    ui.start(MainWindow)
    MainWindow.show()
    sys.exit(app.exec_())


if __name__ == "__main__":
    main()

标签： pythonpyqt5