pyqt5开发linux远程下载文件工具

importsysfrommethodimporthome_uifromPyQt5.QtWidgetsimportQApplication,QMainWindowif__name__=='__main__':
app=QApplication(sys.argv)
# 新建一个主窗口mainWindow=QMainWindow()
# 将创建的py绘制窗口中的UI_MainWindow类赋给uiui=home_ui.Ui_MainWindow()
# 调用类中的setupUi方法，参数为新建的主窗口ui.setupUi(mainWindow)
# 显示该主窗口mainWindow.show()
# 保持停留sys.exit(app.exec_())

# -*- coding: utf-8 -*-# Form implementation generated from reading ui file 'home_ui.ui'## Created by: PyQt5 UI code generator 5.15.4## WARNING: Any manual changes made to this file will be lost when pyuic5 is# run again.  Do not edit this file unless you know what you are doing.frommethod.ssh_methodimport*fromPyQt5importQtCore, QtGui, QtWidgetsfromPyQt5.QtWidgetsimportQMainWindow, QApplication, QPushButtonfromPyQt5.QtCoreimportQCoreApplication,QThreadfromPyQt5.QtWidgetsimport*importos,threadingfrommethod.ThreadClassimport*linux=Linux_Cmd(ip='192.168.253.131',port=22,username='root',password="123456")
classUi_MainWindow(object):
defsetupUi(self, MainWindow):
MainWindow.setObjectName("MainWindow")
MainWindow.resize(860, 580)
self.centralwidget=QtWidgets.QWidget(MainWindow)
self.centralwidget.setObjectName("centralwidget")
self.gridLayout=QtWidgets.QGridLayout(self.centralwidget)
self.gridLayout.setObjectName("gridLayout")
spacerItem=QtWidgets.QSpacerItem(40, 20, QtWidgets.QSizePolicy.Expanding, QtWidgets.QSizePolicy.Minimum)
self.gridLayout.addItem(spacerItem, 2, 0, 1, 1)
self.horizontalLayout=QtWidgets.QHBoxLayout()
self.horizontalLayout.setObjectName("horizontalLayout")
spacerItem1=QtWidgets.QSpacerItem(40, 20, QtWidgets.QSizePolicy.Expanding, QtWidgets.QSizePolicy.Minimum)
self.horizontalLayout.addItem(spacerItem1)
self.label=QtWidgets.QLabel(self.centralwidget)
self.label.setObjectName("label")
self.horizontalLayout.addWidget(self.label)
"""短输入框"""self.lineEdit=QtWidgets.QLineEdit(self.centralwidget)
self.lineEdit.setObjectName("lineEdit")
self.horizontalLayout.addWidget(self.lineEdit)
"""下载功能按钮"""self.pushButton=QtWidgets.QPushButton(self.centralwidget)
self.pushButton.setObjectName("pushButton")
self.horizontalLayout.addWidget(self.pushButton)
self.pushButton.clicked.connect(self.download_method)
spacerItem2=QtWidgets.QSpacerItem(40, 20, QtWidgets.QSizePolicy.Expanding, QtWidgets.QSizePolicy.Minimum)
self.horizontalLayout.addItem(spacerItem2)
self.gridLayout.addLayout(self.horizontalLayout, 1, 1, 1, 1)
spacerItem3=QtWidgets.QSpacerItem(40, 20, QtWidgets.QSizePolicy.Expanding, QtWidgets.QSizePolicy.Minimum)
self.gridLayout.addItem(spacerItem3, 2, 3, 1, 1)
"""文本输入框"""self.textEdit=QtWidgets.QTextEdit(self.centralwidget)
self.textEdit.setObjectName("textEdit")
self.gridLayout.addWidget(self.textEdit, 2, 1, 1, 1)
#self.textEdit.setFocusPolicy(QtCore.Qt.NoFocus)"""打开目录按钮"""self.pathButton=QtWidgets.QPushButton(self.centralwidget)
self.pathButton.setObjectName("pathButton")
self.pathButton.clicked.connect(self.open_path)
self.gridLayout.addWidget(self.pathButton, 1, 0, 1, 1)
"""关闭窗口按钮"""self.closeButton=QtWidgets.QPushButton(self.centralwidget)
self.closeButton.setObjectName("closeButton")
self.closeButton.clicked.connect(QCoreApplication.instance().quit)
self.gridLayout.addWidget(self.closeButton, 1, 3, 1, 1)
MainWindow.setCentralWidget(self.centralwidget)
self.statusbar=QtWidgets.QStatusBar(MainWindow)
self.statusbar.setObjectName("statusbar")
MainWindow.setStatusBar(self.statusbar)
self.retranslateUi(MainWindow)
QtCore.QMetaObject.connectSlotsByName(MainWindow)
defdownload_method(self):
value=self.lineEdit.text()
self.vlues_input=value.replace(" ","").replace("\n","")
print(value)
task=MyThread(linux.run,args=(self.vlues_input,))
task.start()
check=task.get_result()
self.messageDialog(check)
# 警告框defmessageDialog(self,check):
ifcheck==False:
msg_box=QMessageBox(QMessageBox.Warning, '警告', '你所输入的文件不存在')
msg_box.exec_()
elifcheck==True:
msg_box=QMessageBox(QMessageBox.Information, '提示', '下载完成')
msg_box.exec_()
defopen_path(self):
ifnotos.path.exists("./download_file"):  # os模块判断并创建os.mkdir("./download_file")
os.system("start explorer "+os.path.abspath(os.path.join(os.getcwd(), "./download_file")))
else:
os.system("start explorer "+os.path.abspath(os.path.join(os.getcwd(), "./download_file")))
defretranslateUi(self, MainWindow):
_translate=QtCore.QCoreApplication.translateMainWindow.setWindowTitle(_translate("MainWindow", "版本下载工具"))
self.label.setText(_translate("MainWindow", "输入文件名"))
self.pushButton.setText(_translate("MainWindow", "下载"))
self.textEdit.setHtml(_translate("MainWindow", "读取到的文件名："))
get_status=linux.get_info()
ifget_status=="无数据":
self.textEdit.append(get_status)
elifget_status=="数据读取失败请检查连接情况":
self.textEdit.append(get_status)
self.pushButton.setEnabled(False)
msg_box=QMessageBox(QMessageBox.Critical, 'error', '连接到服务器错误\n请检查环境是否正常')
msg_box.exec_()
else:
foriinget_status:
self.textEdit.append(i+"\n")
self.pathButton.setText(_translate("MainWindow", "打开目录"))
self.closeButton.setText(_translate("MainWindow", "关闭"))

# coding:utf-8importthreading, timeclassMyThread(threading.Thread):
def__init__(self, func, args=()):
super(MyThread, self).__init__()
self.func=funcself.args=argsdefrun(self):
self.result=self.func(*self.args)
defget_result(self):
threading.Thread.join(self) # 等待线程执行完毕try:
returnself.resultexceptException:
returnNone

#coding:utf-8importparamiko,osimportreip='10.20.12.103'port=22username='root'password="!QAZ2wsx#EDC"url="/home/version-management/"file_version="test_version.zip"classLinux_Cmd(object):
def__init__(self,ip,port,username,password):
self.ip=ipself.port=portself.username=usernameself.password=passworddeflinxu_ssh(self,cmd):
# 创建SSH对象client=paramiko.SSHClient()
# 自动添加策略，保存服务器的主机名和密钥信息client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器print(self.ip, self.port, self.username, self.password)
try:
client.connect(self.ip,self.port,self.username,self.password, compress=True,timeout=1)
# 执行linux命令cmd_info=client.exec_command(cmd)[1]
returncmd_infoexcept:
print("连接超时")
defremote_scp(self,remote_path, local_path):
client=paramiko.Transport((self.ip, self.port))
client.connect(username=self.username, password=self.password)
sftp=paramiko.SFTPClient.from_transport(client)
src=remote_pathdes=local_pathsftp.get(src, des)
client.close()
defrun(self,file_name):
cmd=Linux_Cmd(ip=ip, port=port, username=username, password=password)
stdout=cmd.linxu_ssh('ls '+url+file_name)
txt_list= []
try:
lines=stdout.readlines()
print(lines)
foriinlines:
txt_list.append(i.rstrip("\n"))
print(txt_list)
iffile_versionnotintxt_list:
returnFalseelse:
#dll_input = input()ifnotos.path.exists("./download_file/"):  # os模块判断并创建os.mkdir("./download_file")
cmd.remote_scp(remote_path=url+file_name+"/"+file_version,
local_path="./download_file/"+"/"+file_version)
returnTrueelse:
cmd.remote_scp(remote_path=url+file_name+"/"+file_version,
local_path="./download_file/"+"/"+file_version)
returnTrueexcept:
print("未获取到数据")
returnFalsedefget_info(self):
try:
cmd=Linux_Cmd(ip=ip, port=port, username=username, password=password)
stdout=cmd.linxu_ssh('ls '+url)
stdout_num=stdout.readlines()
print(len(stdout_num))
iflen(stdout_num) ==0:
print("无数据")
return"无数据"else:
txt_list= []
lines=stdout_numforiinlines:
txt_list.append(i.rstrip("\n\n"))
returntxt_listexcept:
return"数据读取失败请检查连接情况"#if __name__ == '__main__':#    print(get_info())