使用 PyQt5 設計下載遠程服務器日志文件程序的思路

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

import sys

from PyQt5 import QtCore
from PyQt5.QtCore import QObject, pyqtSignal
from PyQt5.QtWidgets import QApplication, QMainWindow
from threading import Thread

from app.downloadlog_qtui import Ui_Dialog
from 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)