import sys
import ftplib
import shutil
from datetime import datetime
from PySide6.QtWidgets import (
QApplication, QWidget, QVBoxLayout, QHBoxLayout, QLabel,
QLineEdit, QPushButton, QProgressBar, QMessageBox, QFrame, QFileDialog, QComboBox, QPlainTextEdit, QDialog,
)
from db import Sqlite3Database
from PySide6.QtGui import QIcon, QFont
from PySide6.QtCore import QThread, Signal, Qt, QTimer, QPropertyAnimation, QRect
import os
def ftp_mkdir_recursive(ftp: ftplib.FTP, remote_directory: str):
"""
递归创建远程 FTP 目录，如果不存在则创建
:param ftp: FTP 对象
:param remote_directory: 远程目录路径
"""
dirs = remote_directory.split('/')
current_path = ''
for directory in dirs:
if directory not in ["",".","./"]: # 跳过空字符串
current_path += '/' + directory
try:
ftp.cwd(current_path) # 尝试切换到该目录
except ftplib.error_perm:
ftp.mkd(current_path)
ftp.cwd(current_path) # 切换到刚创建的目录
def ftp_upload_recursive(ftp: ftplib.FTP, local_dir: str, remote_dir: str):
"""
递归上传本地目录到 FTP 服务器，并创建不存在的远程目录
:param ftp: FTP 对象
:param local_dir: 本地目录路径
:param remote_dir: 远程 FTP 目录路径
"""
ftp_mkdir_recursive(ftp, remote_dir)
for root, dirs, files in os.walk(local_dir):
rel_path = os.path.relpath(root, local_dir)
rel_remote_path = os.path.join(remote_dir, rel_path).replace('\\', '/')
for dir_name in dirs:
ftp_mkdir_recursive(ftp, f"{rel_remote_path}/{dir_name}")
for file_name in files:
local_file_path = os.path.join(root, file_name)
remote_file_path = f"{rel_remote_path}/{file_name}".replace('\\', '/')
with open(local_file_path, 'rb') as file:
print(f"Uploading {local_file_path} to {remote_file_path}")
ftp.storbinary(f'STOR {remote_file_path}', file)
def ftp_directory_exists(ftp: ftplib.FTP, directory: str) -> bool:
"""
检查 FTP 服务器上的目录是否存在
:param ftp: FTP 对象
:param directory: 需要检查的目录路径
:return: 如果目录存在则返回 True，否则返回 False
"""
current_dir = ftp.pwd() # 记住当前所在目录
try:
ftp.cwd(directory) # 尝试切换到目标目录
ftp.cwd(current_dir) # 成功后返回原始目录
return True
except ftplib.error_perm as e:
if str(e).startswith('550'):
print(f"Error:{e}")
return False
else:
raise e # 如果是其他错误，抛出异常
class LogWindow(QDialog):
def __init__(self):
super().__init__()
self.setWindowTitle("Upgrade Progress")
self.resize(500, 350)
# 设置样式表美化界面
self.setStyleSheet("""
QDialog {
background-color: #f2f2f2;
border-radius: 10px;
}
QPlainTextEdit {
background-color: #1e1e1e;
color: #dcdcdc;
border-radius: 5px;
padding: 10px;
font-family: Consolas, monospace;
font-size: 14px;
}
QPushButton {
background-color: #4caf50;
color: white;
border-radius: 5px;
padding: 8px 15px;
font-size: 14px;
}
QPushButton:hover {
background-color: #45a049;
}
""")
# 布局
main_layout = QVBoxLayout()
# 添加日志显示窗口
self.log_edit = QPlainTextEdit(self)
self.log_edit.setReadOnly(True)
main_layout.addWidget(self.log_edit)
# 底部的关闭按钮
self.close_button = QPushButton("Close", self)
self.close_button.clicked.connect(self.close)
# 添加按钮布局
button_layout = QHBoxLayout()
button_layout.addStretch(1) # 左侧扩展
button_layout.addWidget(self.close_button) # 右侧关闭按钮
main_layout.addLayout(button_layout)
self.setLayout(main_layout)
def clear(self):
self.log_edit.clear()
def append_log(self, text):
self.log_edit.appendPlainText(text)
self.log_edit.verticalScrollBar().setValue(self.log_edit.verticalScrollBar().maximum())
class FTPDownloadThread(QThread):
progress = Signal(int)
finished = Signal(bool)
progressMsg = Signal(str)
def __init__(self, host, username, password, file_path, save_path):
super().__init__()
self.host = host
self.username = username
self.password = password
self.file_path = file_path
self.save_path = save_path
def run(self):
try:
ftp = ftplib.FTP(self.host)
ftp.login(self.username, self.password)
total_size = ftp.size(self.file_path)
with open(self.save_path, 'wb') as f:
def handle_block(block):
f.write(block)
self.progress.emit(int(f.tell() / total_size * 100))
ftp.retrbinary(f"RETR {self.file_path}", handle_block)
ftp.quit()
self.finished.emit(True)
except Exception as e:
print(f"Error: {e}")
self.finished.emit(False)
class FTPUploadThread(QThread):
progress = Signal(int)
finished = Signal(bool)
progressMsg = Signal(str)
def __init__(self, host: str, username: str, password: str, remotePath: str, save_path: str):
super().__init__()
self.host = host
self.username = username
self.password = password
self.file_path = remotePath
self.save_path = save_path
def run(self):
try:
ftp = ftplib.FTP(self.host,timeout=3)
ftp.login(self.username, self.password)
current_dir = os.getcwd()
os.chdir(self.save_path)
self.progressMsg.emit("开始扫描升级文件夹")
for check in ["/mnt/flash","/"]:
if ftp_directory_exists(ftp, check):
self.progressMsg.emit("开始检查文件是否合法")
for dirpath, dirnames, filenames in os.walk("."):
for filename in filenames:
total_size = os.path.getsize(dirpath + "/" + filename)
tmp = dirpath
self.progress.emit(tmp)
if tmp != ".":
self.progressMsg.emit(f"开始创建合法目录:{tmp.replace('\\','/')}")
ftp_mkdir_recursive(ftp,f"{check}/{tmp}".replace("\\","/"))
self.progressMsg.emit(f"完成创建合法目录:{tmp.replace('\\','/')}")
self.progressMsg.emit(f"{dirpath + "/" + filename}")
if filename.find(".db") != -1:
self.progressMsg.emit(f"{dirpath}/{filename} 复写数据库开始")
try:
with open(current_dir+"/"+filename, 'wb') as f:
def handle_block(block):
f.write(block)
ftp.retrbinary(f"RETR {check}/config/{filename}", callback=handle_block)
newDB = Sqlite3Database(dirpath + "/" + filename)
oldDB = Sqlite3Database(filename)
oldDB.patch_database("V0",6,"ID",0,newDB)
self.progressMsg.emit(f"{dirpath}/{filename} 复写数据库完成")
dbk = filename + "." + datetime.now().strftime("%Y%m%d%H%M%S")
self.progressMsg.emit(f"备份数据库名称: {dbk}")
newDB.close()
oldDB.close()
shutil.copy(current_dir+"/"+filename,current_dir+"/"+dbk)
except ftplib.error_perm as e:
self.progressMsg.emit(f"Error: {e}")
self.progressMsg.emit(f"开始上传:{dirpath}/{filename}")
with open(f"{dirpath + "/" + filename}", 'rb') as f:
track_length = 0
def _handle_block(block):
nonlocal track_length
track_length = track_length + len(block)
self.progress.emit((track_length / total_size) * 100)
ftp.storbinary(f"STOR {check+"/"+tmp+"/"+filename}", f, callback=_handle_block)
self.progressMsg.emit(f"完成上传:{dirpath}/{filename}")
break
os.chdir(current_dir)
ftp.quit()
self.finished.emit(True)
except Exception as e:
self.progressMsg.emit(f"Error: {e}")
self.finished.emit(False)
def progress_bar_style():
return """
QProgressBar {
border: 1px solid #007BFF; /* 简单的边框 */
border-radius: 0; /* 去掉圆角 */
text-align: center;
}
QProgressBar::chunk {
background-color: #007BFF; /* 进度条填充色 */
}
"""
def button_style():
return """
QPushButton {
background-color: #007BFF; /* 按钮背景色 */
color: white; /* 按钮文字颜色 */
font-size: 16px;
padding: 8px; /* 按钮内边距 */
border: none; /* 去掉边框 */
}
QPushButton:hover {
background-color: #0056b3; /* 鼠标悬停时的背景色 */
}
"""
def input_style():
return """
QLineEdit {
border: 1px solid gray; /* 简单的边框 */
border-radius: 0; /* 去掉圆角 */
padding: 5px;
font-size: 14px;
}
QLineEdit:focus {
border-color: #007BFF; /* 聚焦时的边框颜色 */
}
"""
def combobox_style():
return """
QComboBox {
font-size: 14px;
padding: 5px;
border: 2px solid #007BFF;
border-radius: 5px;
background-color: white;
}
QComboBox QAbstractItemView {
border: 1px solid #007BFF;
background-color: #f9f9f9;
selection-background-color: #007BFF;
selection-color: white;
}
QComboBox::drop-down {
subcontrol-origin: padding;
subcontrol-position: top right;
width: 30px;
border-left-width: 1px;
border-left-color: #007BFF;
border-left-style: solid;
border-top-right-radius: 5px;
border-bottom-right-radius: 5px;
background-color: #007BFF;
}
"""
class FTPUpgradeWindow(QWidget):
def __init__(self):
super().__init__()
self.downloadThread = None
self.upgrade_button = None
self.progress_bar = None
self.rollback_button = None
self.browse_button = None
self.file_input = None
self.password_input = None
self.user_input = None
self.host_input = None
self.uploadThread = None
self.log_window = LogWindow()
self.log_window.setWindowModality(Qt.WindowModality.NonModal)
self.log_window.setMaximumSize(800,600)
self.log_window.setMinimumSize(800,600)
self.init_ui()
def init_ui(self):
self.setWindowTitle('FTP Upgrade')
self.setWindowIcon(QIcon('ftp_icon.png'))
layout = QVBoxLayout()
frame = QFrame(self)
frame.setFrameShape(QFrame.NoFrame) # 不使用边框
frame_layout = QVBoxLayout()
self.host_input = QComboBox(self)
self.host_input.addItems(["近端机 AU ftp://192.168.1.2", "远端机器 RU ftp://192.168.98.98","测试机器 ftp://localhost"])
self.host_input.setPlaceholderText('Hostname (ftp://hostname)')
self.host_input.setStyleSheet(combobox_style())
self.host_input.setEditable(True)
self.user_input = QLineEdit(self)
self.user_input.setPlaceholderText('Username')
self.user_input.setStyleSheet(input_style())
self.user_input.setText("root")
self.password_input = QLineEdit(self)
self.password_input.setPlaceholderText('Password')
self.password_input.setEchoMode(QLineEdit.Password) # 隐藏密码输入
self.password_input.setStyleSheet(input_style())
self.password_input.setText("root")
self.file_input = QLineEdit(self)
self.file_input.setPlaceholderText('File Path (dir)')
self.file_input.setStyleSheet(input_style())
# 文件选择按钮
self.browse_button = QPushButton('Browse...', self)
self.browse_button.setStyleSheet(button_style())
self.browse_button.clicked.connect(self.browse_file)
self.progress_bar = QProgressBar(self)
self.progress_bar.setValue(0)
self.progress_bar.setStyleSheet(progress_bar_style())
self.upgrade_button = QPushButton('Start Upgrade', self)
self.upgrade_button.setStyleSheet(button_style())
self.upgrade_button.clicked.connect(self.start_upgrade)
self.rollback_button = QPushButton('Rollback', self)
self.rollback_button.setStyleSheet(button_style())
self.rollback_button.clicked.connect(self.rollback)
self.rollback_button.setEnabled(False) # 初始禁用
button_layout = QHBoxLayout()
button_layout.addWidget(self.upgrade_button)
button_layout.addWidget(self.rollback_button)
frame_layout.addWidget(QLabel('Enter FTP Details:'))
frame_layout.addWidget(self.host_input)
frame_layout.addWidget(self.user_input)
frame_layout.addWidget(self.password_input)
frame_layout.addWidget(self.file_input)
frame_layout.addWidget(self.browse_button) # 添加浏览按钮
frame_layout.addWidget(self.progress_bar)
frame_layout.addLayout(button_layout)
frame.setLayout(frame_layout)
layout.addWidget(frame)
self.setLayout(layout)
self.setMaximumSize(400, 300)
self.setMinimumSize(400, 300)
font = QFont("Arial", 10)
self.setFont(font)
def browse_file(self):
dirName = QFileDialog.getExistingDirectory(self, "Select Dir", "")
if dirName:
self.file_input.setText(dirName) # 设置选择的文件路径
def start_upgrade(self):
try:
host = self.host_input.currentText().strip().split('://')[1]
username = self.user_input.text().strip()
password = self.password_input.text().strip()
file_path = self.file_input.text().strip()
except Exception as e:
QMessageBox.warning(self, 'Input Error', 'Please enter valid FTP details! {}'.format(e))
return
main_window_rect = self.geometry()
log_window_x = main_window_rect.x() + main_window_rect.width() + 10 # 在主窗口右侧10像素
log_window_y = main_window_rect.y()
self.log_window.move(log_window_x, log_window_y)
self.log_window.show()
self.rollback_button.setEnabled(True)
self.upgrade_button.setEnabled(False)
self.uploadThread = FTPUploadThread(host, username, password, "/", file_path) # 你可以选择保存路径
self.uploadThread.progress.connect(self.progress_bar.setValue)
self.uploadThread.finished.connect(self.on_upgrade_finished)
self.uploadThread.progressMsg.connect(self.log)
self.uploadThread.start()
def log(self,msg:str):
self.log_window.append_log(msg)
def rollback(self):
try:
host = self.host_input.currentText().strip().split('://')[1]
username = self.user_input.text().strip()
password = self.password_input.text().strip()
file_path = self.file_input.text().strip()
except Exception as e:
QMessageBox.warning(self, 'Input Error', 'Please enter valid FTP details! {}'.format(e))
return
main_window_rect = self.geometry()
log_window_x = main_window_rect.x() + main_window_rect.width() + 10
log_window_y = main_window_rect.y()
self.log_window.move(log_window_x, log_window_y)
self.log_window.clear()
self.log_window.show()
self.rollback_button.setEnabled(False)
self.upgrade_button.setEnabled(True)
self.downloadThread = FTPDownloadThread(host, username, password, "/", file_path)
self.downloadThread.progress.connect(self.progress_bar.setValue)
self.downloadThread.finished.connect(self.on_upgrade_finished)
self.downloadThread.progressMsg.connect(self.log)
self.downloadThread.start()
def on_upgrade_finished(self, success):
if success:
QMessageBox.information(self, 'Upgrade Complete', 'File has been upgrade successfully.')
else:
QMessageBox.critical(self, 'Upgrade Failed', 'An error occurred during the upgrade process.')
self.upgrade_button.setEnabled(True)
self.progress_bar.setValue(0)
self.log_window.close()
if __name__ == "__main__":
app = QApplication(sys.argv)
window = FTPUpgradeWindow()
window.show()
sys.exit(app.exec())
正常使用用
import sys
import ftplib
import shutil
from datetime import datetime
from PySide6.QtWidgets import (
QApplication, QWidget, QVBoxLayout, QHBoxLayout, QLabel,
QLineEdit, QPushButton, QProgressBar, QMessageBox, QFrame, QFileDialog, QComboBox, QPlainTextEdit, QDialog,
)
from db import Sqlite3Database
from PySide6.QtGui import QIcon, QFont
from PySide6.QtCore import QThread, Signal, Qt, QTimer, QPropertyAnimation, QRect
import os
from ftp import FTPWalk
def ftp_mkdir_recursive(ftp: ftplib.FTP, remote_directory: str):
"""
递归创建远程 FTP 目录，如果不存在则创建
:param ftp: FTP 对象
:param remote_directory: 远程目录路径
"""
dirs = remote_directory.split('/')
current_path = ''
for directory in dirs:
if directory not in ["",".","./"]: # 跳过空字符串
current_path += '/' + directory
try:
ftp.cwd(current_path) # 尝试切换到该目录
except ftplib.error_perm:
ftp.mkd(current_path)
ftp.cwd(current_path) # 切换到刚创建的目录
def ftp_upload_recursive(ftp: ftplib.FTP, local_dir: str, remote_dir: str):
"""
递归上传本地目录到 FTP 服务器，并创建不存在的远程目录
:param ftp: FTP 对象
:param local_dir: 本地目录路径
:param remote_dir: 远程 FTP 目录路径
"""
ftp_mkdir_recursive(ftp, remote_dir)
for root, dirs, files in os.walk(local_dir):
rel_path = os.path.relpath(root, local_dir)
rel_remote_path = os.path.join(remote_dir, rel_path).replace('\\', '/')
for dir_name in dirs:
ftp_mkdir_recursive(ftp, f"{rel_remote_path}/{dir_name}")
for file_name in files:
local_file_path = os.path.join(root, file_name)
remote_file_path = f"{rel_remote_path}/{file_name}".replace('\\', '/')
with open(local_file_path, 'rb') as file:
print(f"Uploading {local_file_path} to {remote_file_path}")
ftp.storbinary(f'STOR {remote_file_path}', file)
def ftp_directory_exists(ftp: ftplib.FTP, directory: str) -> bool:
"""
检查 FTP 服务器上的目录是否存在
:param ftp: FTP 对象
:param directory: 需要检查的目录路径
:return: 如果目录存在则返回 True，否则返回 False
"""
current_dir = ftp.pwd() # 记住当前所在目录
try:
ftp.cwd(directory) # 尝试切换到目标目录
ftp.cwd(current_dir) # 成功后返回原始目录
return True
except ftplib.error_perm as e:
if str(e).startswith('550'):
return False
else:
raise e # 如果是其他错误，抛出异常
class LogWindow(QDialog):
def __init__(self):
super().__init__()
self.setWindowTitle("Upgrade Progress")
self.resize(500, 350)
# 设置样式表美化界面
self.setStyleSheet("""
QDialog {
background-color: #f2f2f2;
border-radius: 10px;
}
QPlainTextEdit {
background-color: #1e1e1e;
color: #dcdcdc;
border-radius: 5px;
padding: 10px;
font-family: Consolas, monospace;
font-size: 14px;
}
QPushButton {
background-color: #4caf50;
color: white;
border-radius: 5px;
padding: 8px 15px;
font-size: 14px;
}
QPushButton:hover {
background-color: #45a049;
}
""")
# 布局
main_layout = QVBoxLayout()
# 添加日志显示窗口
self.log_edit = QPlainTextEdit(self)
self.log_edit.setReadOnly(True)
main_layout.addWidget(self.log_edit)
# 底部的关闭按钮
self.close_button = QPushButton("Close", self)
self.close_button.clicked.connect(self.close)
# 添加按钮布局
button_layout = QHBoxLayout()
button_layout.addStretch(1) # 左侧扩展
button_layout.addWidget(self.close_button) # 右侧关闭按钮
main_layout.addLayout(button_layout)
self.setLayout(main_layout)
def clear(self):
self.log_edit.clear()
def append_log(self, text):
self.log_edit.appendPlainText(text)
self.log_edit.verticalScrollBar().setValue(self.log_edit.verticalScrollBar().maximum())
class FTPDownloadThread(QThread):
progress = Signal(int)
finished = Signal(bool)
progressMsg = Signal(str)
def __init__(self, host, username, password, file_path, save_path):
super().__init__()
self.host = host
self.username = username
self.password = password
self.file_path = file_path
self.save_path = save_path
def run(self):
try:
ftp = ftplib.FTP(self.host,timeout=3)
ftp.login(self.username, self.password)
bk = datetime.now().strftime("%Y%m%d%H%M%S")
for check in [ "/mnt/flash","/"] :
if not ftp_directory_exists(ftp,check):
continue
for path, _, file in FTPWalk(ftp).walk(check):
self.progress.emit(file)
for _file in file:
Path = path + "/" if path != "/" else "/"
os.makedirs("app/" + bk+"/" + Path,exist_ok=True)
self.progressMsg.emit(f"Create Backup Path: app/{datetime.now().strftime('%Y%m%d%H%M%S')}"
f"/{Path}")
File = Path + _file
size = 0
def callback(line):
nonlocal size
file_info = line.split(maxsplit=8)
if len(file_info) >= 4:
size = int(file_info[4])
ftp.retrlines(f"LIST {File}",callback)
self.progressMsg.emit(f"{File} Size: {size}")
with open("app/"+bk+"/"+File, 'wb') as f:
def handle_block(block):
nonlocal size
f.write(block)
self.progress.emit(int(f.tell() / size * 100))
ftp.retrbinary(f"RETR {File}", handle_block)
break
ftp.quit()
self.finished.emit(True)
except Exception as e:
print(f"Error: {e}")
self.finished.emit(False)
class FTPUploadThread(QThread):
progress = Signal(int)
finished = Signal(bool)
progressMsg = Signal(str)
def __init__(self, host: str, username: str, password: str, remotePath: str, save_path: str):
super().__init__()
self.host = host
self.username = username
self.password = password
self.file_path = remotePath
self.save_path = save_path
def run(self):
try:
ftp = ftplib.FTP(self.host,timeout=3)
ftp.login(self.username, self.password)
current_dir = os.getcwd()
os.chdir(self.save_path)
self.progressMsg.emit("开始扫描升级文件夹")
for check in ["/mnt/flash","/"]:
if ftp_directory_exists(ftp, check):
self.progressMsg.emit("开始检查文件是否合法")
for dirpath, dirnames, filenames in os.walk("."):
for filename in filenames:
total_size = os.path.getsize(dirpath + "/" + filename)
tmp = dirpath
self.progress.emit(tmp)
if tmp != ".":
self.progressMsg.emit(f"开始创建合法目录:{tmp.replace('\\','/')}")
ftp_mkdir_recursive(ftp,f"{check}/{tmp}".replace("\\","/"))
self.progressMsg.emit(f"完成创建合法目录:{tmp.replace('\\','/')}")
self.progressMsg.emit(f"{dirpath + "/" + filename}")
if filename.find(".db") != -1:
self.progressMsg.emit(f"{dirpath}/{filename} 复写数据库开始")
try:
with open(current_dir+"/"+filename, 'wb') as f:
def handle_block(block):
f.write(block)
ftp.retrbinary(f"RETR {check}/config/{filename}", callback=handle_block)
newDB = Sqlite3Database(dirpath + "/" + filename)
oldDB = Sqlite3Database(current_dir+"/"+filename)
oldDB.patch_database("V0",6,"ID",0,newDB)
self.progressMsg.emit(f"{dirpath}/{filename} 复写数据库完成")
dbk = filename + "." + datetime.now().strftime("%Y%m%d%H%M%S")
self.progressMsg.emit(f"备份数据库名称: {dbk}")
newDB.close()
oldDB.close()
shutil.copy(current_dir+"/"+filename,current_dir+"/"+dbk)
except ftplib.error_perm as e:
self.progressMsg.emit(f"Error: {e}")
self.progressMsg.emit(f"开始上传:{dirpath}/{filename}")
with open(f"{dirpath + "/" + filename}", 'rb') as f:
track_length = 0
def _handle_block(block):
nonlocal track_length
track_length = track_length + len(block)
self.progress.emit((track_length / total_size) * 100)
ftp.storbinary(f"STOR "
f"{check+"/" if check != "/" else check +tmp+"/"+filename}".replace(
"\\",
"/"), f,
callback=_handle_block)
self.progressMsg.emit(f"完成上传:{dirpath}/{filename}")
break
os.chdir(current_dir)
ftp.quit()
self.finished.emit(True)
except Exception as e:
self.progressMsg.emit(f"Error: {e}")
self.finished.emit(False)
def progress_bar_style():
return """
QProgressBar {
border: 1px solid #007BFF; /* 简单的边框 */
border-radius: 0; /* 去掉圆角 */
text-align: center;
}
QProgressBar::chunk {
background-color: #007BFF; /* 进度条填充色 */
}
"""
def button_style():
return """
QPushButton {
background-color: #007BFF; /* 按钮背景色 */
color: white; /* 按钮文字颜色 */
font-size: 16px;
padding: 8px; /* 按钮内边距 */
border: none; /* 去掉边框 */
}
QPushButton:hover {
background-color: #0056b3; /* 鼠标悬停时的背景色 */
}
"""
def input_style():
return """
QLineEdit {
border: 1px solid gray; /* 简单的边框 */
border-radius: 0; /* 去掉圆角 */
padding: 5px;
font-size: 14px;
}
QLineEdit:focus {
border-color: #007BFF; /* 聚焦时的边框颜色 */
}
"""
def combobox_style():
return """
QComboBox {
font-size: 14px;
padding: 5px;
border: 2px solid #007BFF;
border-radius: 5px;
background-color: white;
}
QComboBox QAbstractItemView {
border: 1px solid #007BFF;
background-color: #f9f9f9;
selection-background-color: #007BFF;
selection-color: white;
}
QComboBox::drop-down {
subcontrol-origin: padding;
subcontrol-position: top right;
width: 30px;
border-left-width: 1px;
border-left-color: #007BFF;
border-left-style: solid;
border-top-right-radius: 5px;
border-bottom-right-radius: 5px;
background-color: #007BFF;
}
"""
class FTPUpgradeWindow(QWidget):
def __init__(self):
super().__init__()
self.downloadThread = None
self.upgrade_button = None
self.progress_bar = None
self.rollback_button = None
self.browse_button = None
self.file_input = None
self.password_input = None
self.user_input = None
self.host_input = None
self.uploadThread = None
self.log_window = LogWindow()
self.log_window.setWindowModality(Qt.WindowModality.NonModal)
self.log_window.setMaximumSize(800,600)
self.log_window.setMinimumSize(800,600)
self.init_ui()
def init_ui(self):
self.setWindowTitle('FTP Upgrade')
self.setWindowIcon(QIcon('ftp_icon.png'))
layout = QVBoxLayout()
frame = QFrame(self)
frame.setFrameShape(QFrame.NoFrame) # 不使用边框
frame_layout = QVBoxLayout()
self.host_input = QComboBox(self)
self.host_input.addItems(["近端机 AU ftp://192.168.1.2",
"近端机 AU ftp://192.168.1.113",
"远端机 RU ftp://192.168.98.98",
"测试机 ftp://localhost"])
self.host_input.setPlaceholderText('Hostname (ftp://hostname)')
self.host_input.setStyleSheet(combobox_style())
self.host_input.setEditable(True)
self.user_input = QLineEdit(self)
self.user_input.setPlaceholderText('Username')
self.user_input.setStyleSheet(input_style())
self.user_input.setText("root")
self.password_input = QLineEdit(self)
self.password_input.setPlaceholderText('Password')
self.password_input.setEchoMode(QLineEdit.Password) # 隐藏密码输入
self.password_input.setStyleSheet(input_style())
self.password_input.setText("root")
self.file_input = QLineEdit(self)
self.file_input.setPlaceholderText('File Path (dir)')
self.file_input.setStyleSheet(input_style())
self.file_input.setText(os.getcwd()+"\\upgrade")
# 文件选择按钮
self.browse_button = QPushButton('Browse...', self)
self.browse_button.setStyleSheet(button_style())
self.browse_button.clicked.connect(self.browse_file)
self.progress_bar = QProgressBar(self)
self.progress_bar.setValue(0)
self.progress_bar.setStyleSheet(progress_bar_style())
self.upgrade_button = QPushButton('Start Upgrade', self)
self.upgrade_button.setStyleSheet(button_style())
self.upgrade_button.clicked.connect(self.start_upgrade)
self.rollback_button = QPushButton('Start Backup', self)
self.rollback_button.setStyleSheet(button_style())
self.rollback_button.clicked.connect(self.rollback)
self.rollback_button.setEnabled(True)
button_layout = QHBoxLayout()
button_layout.addWidget(self.upgrade_button)
button_layout.addWidget(self.rollback_button)
frame_layout.addWidget(QLabel('Enter FTP Details:'))
frame_layout.addWidget(self.host_input)
frame_layout.addWidget(self.user_input)
frame_layout.addWidget(self.password_input)
frame_layout.addWidget(self.file_input)
frame_layout.addWidget(self.browse_button) # 添加浏览按钮
frame_layout.addWidget(self.progress_bar)
frame_layout.addLayout(button_layout)
frame.setLayout(frame_layout)
layout.addWidget(frame)
self.setLayout(layout)
self.setMaximumSize(400, 300)
self.setMinimumSize(400, 300)
font = QFont("Arial", 10)
self.setFont(font)
def browse_file(self):
dirName = QFileDialog.getExistingDirectory(self, "Select Dir", "")
if dirName:
self.file_input.setText(dirName) # 设置选择的文件路径
def start_upgrade(self):
try:
host = self.host_input.currentText().strip().split('://')[1]
username = self.user_input.text().strip()
password = self.password_input.text().strip()
file_path = self.file_input.text().strip()
except Exception as e:
QMessageBox.warning(self, 'Input Error', 'Please enter valid FTP details! {}'.format(e))
return
main_window_rect = self.geometry()
log_window_x = main_window_rect.x() + main_window_rect.width() + 10 # 在主窗口右侧10像素
log_window_y = main_window_rect.y()
self.log_window.move(log_window_x, log_window_y)
self.log_window.show()
self.rollback_button.setEnabled(False)
self.upgrade_button.setEnabled(False)
self.uploadThread = FTPUploadThread(host, username, password, "/", file_path) # 你可以选择保存路径
self.uploadThread.progress.connect(self.progress_bar.setValue)
self.uploadThread.finished.connect(self.on_upgrade_finished)
self.uploadThread.progressMsg.connect(self.log)
self.uploadThread.start()
def log(self,msg:str):
self.log_window.append_log(msg)
def rollback(self):
try:
host = self.host_input.currentText().strip().split('://')[1]
username = self.user_input.text().strip()
password = self.password_input.text().strip()
file_path = self.file_input.text().strip()
except Exception as e:
QMessageBox.warning(self, 'Input Error', 'Please enter valid FTP details! {}'.format(e))
return
main_window_rect = self.geometry()
log_window_x = main_window_rect.x() + main_window_rect.width() + 10
log_window_y = main_window_rect.y()
self.log_window.move(log_window_x, log_window_y)
self.log_window.clear()
self.log_window.show()
self.rollback_button.setEnabled(False)
self.upgrade_button.setEnabled(False)
self.downloadThread = FTPDownloadThread(host, username, password, "/", file_path)
self.downloadThread.progress.connect(self.progress_bar.setValue)
self.downloadThread.finished.connect(self.backup_finished)
self.downloadThread.progressMsg.connect(self.log)
self.downloadThread.start()
def backup_finished(self,success:bool):
if success:
QMessageBox.information(self, 'backup Complete', 'File has been backup successfully.')
else:
QMessageBox.critical(self, 'backup Failed', 'An error occurred during the backup process.')
self.rollback_button.setEnabled(True)
self.upgrade_button.setEnabled(True)
self.progress_bar.setValue(0)
self.log_window.close()
def on_upgrade_finished(self, success):
if success:
QMessageBox.information(self, 'Upgrade Complete', 'File has been upgrade successfully.')
else:
QMessageBox.critical(self, 'Upgrade Failed', 'An error occurred during the upgrade process.')
self.upgrade_button.setEnabled(True)
self.rollback_button.setEnabled(True)
self.progress_bar.setValue(0)
self.log_window.close()
if __name__ == "__main__":
app = QApplication(sys.argv)
window = FTPUpgradeWindow()
window.show()
sys.exit(app.exec())
调整一下效果