456 lines
18 KiB
Python
456 lines
18 KiB
Python
import os
|
||
import sys
|
||
import datetime
|
||
from PySide6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QPushButton,
|
||
QLabel, QTextEdit, QFileDialog, QMessageBox,
|
||
QLineEdit, QDialog, QDialogButtonBox)
|
||
from PySide6.QtCore import QThread, Signal
|
||
from loguru import logger
|
||
from django_threads import DjangoInstallThread, DjangoCommandThread, UploadSettingsThread
|
||
|
||
class PasswordDialog(QDialog):
|
||
def __init__(self, parent=None):
|
||
super().__init__(parent)
|
||
self.setWindowTitle("输入密码")
|
||
self.setMinimumWidth(300)
|
||
|
||
layout = QVBoxLayout()
|
||
|
||
# 提示标签
|
||
label = QLabel("请输入sudo密码:")
|
||
layout.addWidget(label)
|
||
|
||
# 密码输入框
|
||
self.password_input = QLineEdit()
|
||
self.password_input.setEchoMode(QLineEdit.Password)
|
||
layout.addWidget(self.password_input)
|
||
|
||
# 按钮
|
||
button_box = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
|
||
button_box.accepted.connect(self.accept)
|
||
button_box.rejected.connect(self.reject)
|
||
layout.addWidget(button_box)
|
||
|
||
self.setLayout(layout)
|
||
|
||
def get_password(self):
|
||
return self.password_input.text()
|
||
|
||
class DjangoTab(QWidget):
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.ssh_client = None
|
||
self.username = ""
|
||
self.manage_py_path = ""
|
||
self.settings_py_path = ""
|
||
self.init_ui()
|
||
|
||
def init_ui(self):
|
||
layout = QVBoxLayout()
|
||
|
||
# Django管理区域
|
||
manage_layout = QHBoxLayout()
|
||
|
||
# 安装Django按钮
|
||
self.install_django_btn = QPushButton("安装Django")
|
||
self.install_django_btn.clicked.connect(self.install_django)
|
||
manage_layout.addWidget(self.install_django_btn)
|
||
|
||
# 测试启动按钮
|
||
self.test_server_btn = QPushButton("测试启动")
|
||
self.test_server_btn.clicked.connect(self.test_server)
|
||
manage_layout.addWidget(self.test_server_btn)
|
||
|
||
# 检查Django状态按钮
|
||
self.check_django_btn = QPushButton("检查Django状态")
|
||
self.check_django_btn.clicked.connect(self.check_django_status)
|
||
manage_layout.addWidget(self.check_django_btn)
|
||
|
||
# 收集静态文件按钮
|
||
self.collect_static_btn = QPushButton("收集静态文件")
|
||
self.collect_static_btn.clicked.connect(self.collect_static)
|
||
manage_layout.addWidget(self.collect_static_btn)
|
||
|
||
layout.addLayout(manage_layout)
|
||
|
||
# 文件管理区域
|
||
file_layout = QHBoxLayout()
|
||
|
||
# 查找文件按钮
|
||
self.find_files_btn = QPushButton("查找manage.py和settings.py")
|
||
self.find_files_btn.clicked.connect(self.find_django_files)
|
||
file_layout.addWidget(self.find_files_btn)
|
||
|
||
# 下载settings.py按钮
|
||
self.download_settings_btn = QPushButton("下载settings.py")
|
||
self.download_settings_btn.clicked.connect(self.download_settings)
|
||
file_layout.addWidget(self.download_settings_btn)
|
||
|
||
# 上传settings.py按钮
|
||
self.upload_settings_btn = QPushButton("上传settings.py")
|
||
self.upload_settings_btn.clicked.connect(self.upload_settings)
|
||
file_layout.addWidget(self.upload_settings_btn)
|
||
|
||
layout.addLayout(file_layout)
|
||
|
||
# 文件路径显示区域
|
||
path_layout = QHBoxLayout()
|
||
|
||
# manage.py路径
|
||
manage_path_layout = QVBoxLayout()
|
||
manage_path_layout.addWidget(QLabel("manage.py路径:"))
|
||
self.manage_path_label = QLabel("未找到")
|
||
manage_path_layout.addWidget(self.manage_path_label)
|
||
path_layout.addLayout(manage_path_layout)
|
||
|
||
# settings.py路径
|
||
settings_path_layout = QVBoxLayout()
|
||
settings_path_layout.addWidget(QLabel("settings.py路径:"))
|
||
self.settings_path_label = QLabel("未找到")
|
||
settings_path_layout.addWidget(self.settings_path_label)
|
||
path_layout.addLayout(settings_path_layout)
|
||
|
||
layout.addLayout(path_layout)
|
||
|
||
# 文件编辑区域
|
||
edit_layout = QVBoxLayout()
|
||
edit_layout.addWidget(QLabel("settings.py编辑器:"))
|
||
|
||
# 文件编辑文本框
|
||
self.settings_editor = QTextEdit()
|
||
self.settings_editor.setReadOnly(True)
|
||
edit_layout.addWidget(self.settings_editor)
|
||
|
||
# 保存按钮
|
||
self.save_settings_btn = QPushButton("保存settings.py到服务器")
|
||
self.save_settings_btn.clicked.connect(self.save_settings_to_server)
|
||
self.save_settings_btn.setEnabled(False) # 初始状态禁用
|
||
edit_layout.addWidget(self.save_settings_btn)
|
||
|
||
layout.addLayout(edit_layout)
|
||
|
||
# 输出区域
|
||
self.output_text = QTextEdit()
|
||
self.output_text.setReadOnly(True)
|
||
layout.addWidget(self.output_text)
|
||
|
||
self.setLayout(layout)
|
||
|
||
def set_ssh_client(self, ssh_client):
|
||
"""设置SSH客户端"""
|
||
self.ssh_client = ssh_client
|
||
logger.info("Django标签页已设置SSH客户端")
|
||
|
||
def set_username(self, username):
|
||
"""设置用户名"""
|
||
self.username = username
|
||
logger.info(f"Django标签页已设置用户名: {username}")
|
||
|
||
def append_output(self, text):
|
||
"""添加输出到文本框"""
|
||
self.output_text.append(text)
|
||
|
||
def on_command_finished(self):
|
||
"""命令执行完成时的处理"""
|
||
logger.info("Django命令执行完成")
|
||
|
||
def install_django(self):
|
||
"""安装Django"""
|
||
if not self.ssh_client:
|
||
self.append_output("错误: 未连接到服务器")
|
||
return
|
||
|
||
# 请求用户输入sudo密码
|
||
dialog = PasswordDialog(self)
|
||
if dialog.exec_() == QDialog.Accepted:
|
||
password = dialog.get_password()
|
||
self.append_output("正在安装Django...")
|
||
|
||
# 创建并启动Django安装线程
|
||
self.install_thread = DjangoInstallThread(self.ssh_client, password)
|
||
self.install_thread.progress_updated.connect(self.on_install_progress)
|
||
self.install_thread.result_ready.connect(self.on_install_result)
|
||
self.install_thread.start()
|
||
else:
|
||
self.append_output("用户取消了密码输入")
|
||
|
||
def on_install_progress(self, progress):
|
||
"""处理安装进度更新"""
|
||
self.append_output(f"安装进度: {progress}%")
|
||
logger.info(f"Django安装进度: {progress}%")
|
||
|
||
def on_install_result(self, success, message):
|
||
"""处理安装结果"""
|
||
if success:
|
||
self.append_output(f"安装成功: {message}")
|
||
logger.info(f"Django安装成功: {message}")
|
||
QMessageBox.information(self, "成功", message)
|
||
else:
|
||
self.append_output(f"安装失败: {message}")
|
||
logger.error(f"Django安装失败: {message}")
|
||
QMessageBox.warning(self, "错误", f"Django安装失败: {message}")
|
||
|
||
def request_password(self):
|
||
"""请求用户输入密码"""
|
||
dialog = PasswordDialog(self)
|
||
if dialog.exec_() == QDialog.Accepted:
|
||
password = dialog.get_password()
|
||
self.thread.set_password(password)
|
||
logger.info("用户已输入密码")
|
||
else:
|
||
self.thread.set_password("")
|
||
logger.info("用户取消了密码输入")
|
||
|
||
def test_server(self):
|
||
"""测试启动Django服务器"""
|
||
if not self.ssh_client:
|
||
self.append_output("错误: 未连接到服务器")
|
||
return
|
||
|
||
if not self.manage_py_path:
|
||
self.append_output("错误: 未找到manage.py文件，请先查找文件")
|
||
return
|
||
|
||
# 切换到manage.py所在目录并执行命令
|
||
manage_dir = os.path.dirname(self.manage_py_path)
|
||
command = f"cd {manage_dir} && python3 manage.py runserver 0.0.0.0:8000"
|
||
self.append_output(f"执行命令: {command}")
|
||
|
||
# 创建并启动线程执行命令
|
||
self.thread = DjangoCommandThread(self.ssh_client, command)
|
||
self.thread.output_signal.connect(self.append_output)
|
||
self.thread.finished_signal.connect(self.on_command_finished)
|
||
self.thread.start()
|
||
|
||
def check_django_status(self):
|
||
"""检查Django安装状态"""
|
||
if not self.ssh_client:
|
||
self.append_output("错误: 未连接到服务器")
|
||
return
|
||
|
||
command = "pip3 list | grep Django"
|
||
self.append_output(f"执行命令: {command}")
|
||
|
||
# 创建并启动线程执行命令
|
||
self.thread = DjangoCommandThread(self.ssh_client, command)
|
||
self.thread.output_signal.connect(self.append_output)
|
||
self.thread.finished_signal.connect(self.on_command_finished)
|
||
self.thread.start()
|
||
|
||
def collect_static(self):
|
||
"""收集静态文件"""
|
||
if not self.ssh_client:
|
||
self.append_output("错误: 未连接到服务器")
|
||
return
|
||
|
||
if not self.manage_py_path:
|
||
self.append_output("错误: 未找到manage.py文件，请先查找文件")
|
||
return
|
||
|
||
# 切换到manage.py所在目录并执行命令
|
||
manage_dir = os.path.dirname(self.manage_py_path)
|
||
command = f"cd {manage_dir} && python3 manage.py collectstatic --noinput"
|
||
self.append_output(f"执行命令: {command}")
|
||
|
||
# 创建并启动线程执行命令
|
||
self.thread = DjangoCommandThread(self.ssh_client, command)
|
||
self.thread.output_signal.connect(self.append_output)
|
||
self.thread.finished_signal.connect(self.on_command_finished)
|
||
self.thread.start()
|
||
|
||
def find_django_files(self):
|
||
"""查找manage.py和settings.py文件"""
|
||
if not self.ssh_client:
|
||
self.append_output("错误: 未连接到服务器")
|
||
return
|
||
|
||
if not self.username:
|
||
self.append_output("错误: 未获取到用户名")
|
||
return
|
||
|
||
self.append_output("正在查找Django项目文件...")
|
||
|
||
# 查找manage.py文件
|
||
find_manage_cmd = f"find /home/{self.username} -name \"manage.py\" 2>/dev/null | head -5"
|
||
self.append_output(f"执行命令: {find_manage_cmd}")
|
||
|
||
# 创建并启动线程执行命令
|
||
self.thread = DjangoCommandThread(self.ssh_client, find_manage_cmd)
|
||
self.thread.output_signal.connect(self.process_manage_py_result)
|
||
self.thread.finished_signal.connect(self.find_settings_py)
|
||
self.thread.start()
|
||
|
||
def process_manage_py_result(self, text):
|
||
"""处理manage.py查找结果"""
|
||
if text.startswith("/") and text.endswith("manage.py"):
|
||
self.manage_py_path = text
|
||
self.manage_path_label.setText(text)
|
||
logger.info(f"找到manage.py文件: {text}")
|
||
|
||
def find_settings_py(self):
|
||
"""查找settings.py文件"""
|
||
if not self.ssh_client:
|
||
self.append_output("错误: 未连接到服务器")
|
||
return
|
||
|
||
if not self.username:
|
||
self.append_output("错误: 未获取到用户名")
|
||
return
|
||
|
||
# 查找settings.py文件
|
||
find_settings_cmd = f"find /home/{self.username} -name \"settings.py\" 2>/dev/null | head -5"
|
||
self.append_output(f"执行命令: {find_settings_cmd}")
|
||
|
||
# 创建并启动线程执行命令
|
||
self.thread = DjangoCommandThread(self.ssh_client, find_settings_cmd)
|
||
self.thread.output_signal.connect(self.process_settings_py_result)
|
||
self.thread.finished_signal.connect(self.on_command_finished)
|
||
self.thread.start()
|
||
|
||
def process_settings_py_result(self, text):
|
||
"""处理settings.py查找结果"""
|
||
if text.startswith("/") and text.endswith("settings.py"):
|
||
self.settings_py_path = text
|
||
self.settings_path_label.setText(text)
|
||
logger.info(f"找到settings.py文件: {text}")
|
||
|
||
def download_settings(self):
|
||
"""下载settings.py文件并在编辑器中显示"""
|
||
if not self.ssh_client:
|
||
self.append_output("错误: 未连接到服务器")
|
||
return
|
||
|
||
if not self.settings_py_path:
|
||
self.append_output("错误: 未找到settings.py文件，请先查找文件")
|
||
return
|
||
|
||
try:
|
||
# 使用SFTP下载文件内容
|
||
sftp = self.ssh_client.open_sftp()
|
||
with sftp.file(self.settings_py_path, 'r') as remote_file:
|
||
file_content = remote_file.read().decode('utf-8')
|
||
sftp.close()
|
||
|
||
# 在编辑器中显示文件内容
|
||
self.settings_editor.setPlainText(file_content)
|
||
self.settings_editor.setReadOnly(False) # 允许编辑
|
||
self.save_settings_btn.setEnabled(True) # 启用保存按钮
|
||
|
||
self.append_output(f"文件已加载到编辑器: {self.settings_py_path}")
|
||
logger.info(f"settings.py已加载到编辑器: {self.settings_py_path}")
|
||
|
||
except Exception as e:
|
||
error_msg = f"下载文件时出错: {str(e)}"
|
||
self.append_output(error_msg)
|
||
logger.error(error_msg)
|
||
|
||
def save_settings_to_server(self):
|
||
"""将编辑器中的settings.py内容保存到服务器"""
|
||
if not self.ssh_client:
|
||
self.append_output("错误: 未连接到服务器")
|
||
return
|
||
|
||
if not self.settings_py_path:
|
||
self.append_output("错误: 未找到settings.py文件，请先查找文件")
|
||
return
|
||
|
||
try:
|
||
# 获取编辑器中的内容
|
||
file_content = self.settings_editor.toPlainText()
|
||
|
||
# 使用SFTP上传文件内容
|
||
sftp = self.ssh_client.open_sftp()
|
||
with sftp.file(self.settings_py_path, 'w') as remote_file:
|
||
remote_file.write(file_content.encode('utf-8'))
|
||
sftp.close()
|
||
|
||
self.append_output(f"文件已保存到服务器: {self.settings_py_path}")
|
||
logger.info(f"settings.py已保存到服务器: {self.settings_py_path}")
|
||
|
||
# 显示成功消息
|
||
QMessageBox.information(self, "保存成功", "settings.py文件已成功保存到服务器")
|
||
|
||
except Exception as e:
|
||
error_msg = f"保存文件时出错: {str(e)}"
|
||
self.append_output(error_msg)
|
||
logger.error(error_msg)
|
||
|
||
def upload_settings(self):
|
||
"""上传settings.py文件"""
|
||
if not self.ssh_client:
|
||
self.append_output("错误: 未连接到服务器")
|
||
return
|
||
|
||
if not self.settings_py_path:
|
||
self.append_output("错误: 未找到settings.py文件，请先查找文件")
|
||
return
|
||
|
||
# 直接打开文件选择对话框，让用户选择本地settings.py文件
|
||
file_path, _ = QFileDialog.getOpenFileName(
|
||
self,
|
||
"选择要上传的settings.py文件",
|
||
"",
|
||
"Python文件 (*.py);;所有文件 (*)"
|
||
)
|
||
|
||
if not file_path:
|
||
self.append_output("用户取消了文件选择")
|
||
return
|
||
|
||
# 验证选择的文件是否为settings.py
|
||
if os.path.basename(file_path) != "settings.py":
|
||
reply = QMessageBox.question(
|
||
self,
|
||
"文件名验证",
|
||
"选择的文件不是settings.py，是否继续上传？",
|
||
QMessageBox.Yes | QMessageBox.No
|
||
)
|
||
if reply == QMessageBox.No:
|
||
self.append_output("用户取消了上传")
|
||
return
|
||
|
||
try:
|
||
# 读取文件内容
|
||
with open(file_path, 'r', encoding='utf-8') as f:
|
||
settings_content = f.read()
|
||
|
||
# 获取Django项目路径（settings.py所在目录的父目录）
|
||
django_path = os.path.dirname(os.path.dirname(self.settings_py_path))
|
||
|
||
# 创建并启动上传线程
|
||
self.upload_thread = UploadSettingsThread(self.ssh_client, django_path, settings_content)
|
||
self.upload_thread.result_ready.connect(self.on_upload_result)
|
||
self.upload_thread.password_request_signal.connect(self.request_upload_password)
|
||
self.upload_thread.start()
|
||
|
||
self.append_output(f"正在上传文件: {os.path.basename(file_path)}...")
|
||
logger.info(f"开始上传settings.py文件: {file_path}")
|
||
|
||
except Exception as e:
|
||
error_msg = f"读取文件时出错: {str(e)}"
|
||
self.append_output(error_msg)
|
||
logger.error(error_msg)
|
||
|
||
def request_upload_password(self):
|
||
"""请求用户输入密码（用于上传settings.py）"""
|
||
dialog = PasswordDialog(self)
|
||
if dialog.exec_() == QDialog.Accepted:
|
||
password = dialog.get_password()
|
||
self.upload_thread.set_password(password)
|
||
self.append_output("密码已发送")
|
||
logger.info("用户已输入上传密码")
|
||
else:
|
||
self.upload_thread.set_password("")
|
||
self.append_output("用户取消了密码输入")
|
||
logger.info("用户取消了上传密码输入")
|
||
|
||
def on_upload_result(self, success, message):
|
||
"""处理上传结果"""
|
||
if success:
|
||
self.append_output(f"上传成功: {message}")
|
||
logger.info(f"settings.py上传成功: {message}")
|
||
QMessageBox.information(self, "上传成功", message)
|
||
else:
|
||
self.append_output(f"上传失败: {message}")
|
||
logger.error(f"settings.py上传失败: {message}")
|
||
QMessageBox.warning(self, "上传失败", f"settings.py上传失败: {message}") |