【实操日记】使用 PyQt5 设计下载远程服务器日志文件程序

Program：/Users/macbookpro/workspace/projects/DownloadServerLog/venv/bin/python3.9Arguments：-m PyQt5.uic.pyuic $FileName$ -o $FileNameWithoutExtension$.pyWorking directory：/Users/macbookpro/workspace/projects/DownloadServerLog/ui

DownloadServerLog├── app│   ├── downloadlog.py│   └── downloadlog_qtui.py├── main.py└── ui│   └── downloadlog_qtui.ui├── .env

import sys
from PyQt5 import QtCorefrom PyQt5.QtCore import QObject, pyqtSignalfrom PyQt5.QtWidgets import QApplication, QMainWindowfrom threading import Thread
from app.downloadlog_qtui import Ui_Dialogfrom app.downloadlog import DownloadLog

class CommunicateSignal(QObject):    text_print = pyqtSignal(str)

# MyWindow 是主窗口程序，继承自 PyQt5.QtWidgets.QMainWindow# 和通过 ui 文件生成的 downloadlog_qtui.py 中的 Ui_Dialog 类class MyWindow(QMainWindow, Ui_Dialog):    def __init__(self, parent=None):        super().__init__(parent)        self.setupUi(self)        self.btn_download.clicked.connect(self.click_download)
        # 自定义信号处理函数        self.comm_signal = CommunicateSignal()        self.comm_signal.text_print.connect(self.show_text)
        self.set_window_init_data()
    def set_window_init_data(self):        """设置程序窗体初始值"""        # 从 .env 读取环境变量        result_dict = dict()        with open('.env', 'r', encoding='utf-8') as f:            for line in f.readlines():                key = line.split('=')[0].strip()                value = line.split('=')[-1].strip()                result_dict[key] = value
        # 设置输入框值        _translate = QtCore.QCoreApplication.translate        self.host.setText(_translate("Dialog", result_dict.get("HOST", '')))        self.port.setText(_translate("Dialog", result_dict.get("PORT", '22')))        self.username.setText(_translate("Dialog", result_dict.get("USERNAME", 'root')))        self.password.setText(_translate("Dialog", result_dict.get("PASSWORD", '')))        self.directory.setPlainText(_translate("Dialog", result_dict.get("DIRECTORY", '')))        self.startTime.setDate(QtCore.QDate.currentDate())        self.endTime.setDate(QtCore.QDate.currentDate())

    def get_window_input_value(self):        """获取程序各「输入框」组件值"""        return {            "host": self.host.text(),            "port": self.port.text(),            "username": self.username.text(),            "password": self.password.text(),            "directory": self.directory.toPlainText(),            "start_time": self.startTime.date().toString("yyyy-MM-dd"),            "end_time": self.endTime.date().toString("yyyy-MM-dd"),            "suffix": ".log",        }
    def show_text(self, text):        """将文本内容追加到程序「展示框」"""        self.textBrowser.append(text)
    def click_download(self):        """处理点击「下载」按钮事件"""        params = self.get_window_input_value()
        def run():            res = DownloadLog(conn_type='ssh', comm_signal=self.comm_signal, **params)            res.main()
        t = Thread(target=run)        t.start()

if __name__ == '__main__':    app = QApplication(sys.argv)    myWin = MyWindow()    myWin.show()    sys.exit(app.exec_())

class DownloadLog(object):    def __init__(self, **kwargs):        """初始化一些参数"""        ...
    def main(self):        # 获取 Transport 实例        tran = paramiko.Transport((self.host, int(self.port)))        # 连接 SSH 服务端        tran.connect(username=self.username, password=self.password)        # 创建 SFTP 实例        self.sftp = paramiko.SFTPClient.from_transport(tran)        # 下载文件        # :param str remotepath: the remote file to copy        # :param str localpath: the destination path on the local host        self.sftp.get(remotepath=self.remote_path, localpath=self.local_path)

class CommunicateSignal(QObject):    text_print = pyqtSignal(str)

# 自定义信号处理函数self.comm_signal = CommunicateSignal()self.comm_signal.text_print.connect(self.show_text)

# 通过该信号对象的 emit 方法发出信号，emit 方法的参数传递必要的数据。# 参数类型遵循定义 Signal 时指定的类型。self.comm_signal.text_print.emit(text)

def show_text(self, text):    """将文本内容追加到程序「展示框」"""    self.textBrowser.append(text)