323 lines
14 KiB
Python
323 lines
14 KiB
Python
import os
|
|
import json
|
|
from loguru import logger
|
|
from PySide6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit,
|
|
QPushButton, QComboBox, QMessageBox, QTextEdit,
|
|
QGroupBox, QGridLayout, QProgressBar)
|
|
from PySide6.QtCore import Qt
|
|
|
|
from threads import (DjangoInstallThread, DjangoTestThread, DownloadSettingsThread,
|
|
UploadSettingsThread, CollectStaticThread, CheckDjangoStatusThread)
|
|
|
|
|
|
class DjangoTab(QWidget):
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self.parent = parent
|
|
self.init_ui()
|
|
|
|
# 连接服务器切换信号
|
|
if self.parent and hasattr(self.parent, 'server_changed'):
|
|
self.parent.server_changed.connect(self.on_server_changed)
|
|
logger.info("Django标签已连接到服务器切换信号")
|
|
def init_ui(self):
|
|
layout = QVBoxLayout()
|
|
|
|
# Django项目配置组
|
|
config_group = QGroupBox("Django项目配置")
|
|
config_layout = QGridLayout()
|
|
|
|
config_layout.addWidget(QLabel("Django项目路径:"), 0, 0)
|
|
self.django_path_input = QLineEdit()
|
|
self.django_path_input.setPlaceholderText("/home/user/django_project")
|
|
config_layout.addWidget(self.django_path_input, 0, 1)
|
|
|
|
self.load_django_path_btn = QPushButton("加载路径")
|
|
self.load_django_path_btn.clicked.connect(self.load_django_path)
|
|
config_layout.addWidget(self.load_django_path_btn, 0, 2)
|
|
|
|
config_group.setLayout(config_layout)
|
|
layout.addWidget(config_group)
|
|
|
|
# Django操作组
|
|
django_group = QGroupBox("Django操作")
|
|
django_layout = QGridLayout()
|
|
|
|
self.install_django_btn = QPushButton("安装Django")
|
|
self.install_django_btn.clicked.connect(self.install_django)
|
|
django_layout.addWidget(self.install_django_btn, 0, 0)
|
|
|
|
self.test_django_btn = QPushButton("测试启动")
|
|
self.test_django_btn.clicked.connect(self.test_django)
|
|
django_layout.addWidget(self.test_django_btn, 0, 1)
|
|
|
|
self.download_settings_btn = QPushButton("下载settings.py")
|
|
self.download_settings_btn.clicked.connect(self.download_settings)
|
|
django_layout.addWidget(self.download_settings_btn, 1, 0)
|
|
|
|
self.upload_settings_btn = QPushButton("上传settings.py")
|
|
self.upload_settings_btn.clicked.connect(self.upload_settings)
|
|
django_layout.addWidget(self.upload_settings_btn, 1, 1)
|
|
|
|
self.collect_static_btn = QPushButton("收集静态文件")
|
|
self.collect_static_btn.clicked.connect(self.collect_static)
|
|
django_layout.addWidget(self.collect_static_btn, 2, 0)
|
|
|
|
self.check_status_btn = QPushButton("检查Django状态")
|
|
self.check_status_btn.clicked.connect(self.check_django_status)
|
|
django_layout.addWidget(self.check_status_btn, 2, 1)
|
|
|
|
django_group.setLayout(django_layout)
|
|
layout.addWidget(django_group)
|
|
|
|
# Settings.py编辑器
|
|
settings_group = QGroupBox("Settings.py编辑器")
|
|
settings_layout = QVBoxLayout()
|
|
|
|
self.settings_editor = QTextEdit()
|
|
self.settings_editor.setPlaceholderText("settings.py内容将在这里显示...")
|
|
settings_layout.addWidget(self.settings_editor)
|
|
|
|
settings_group.setLayout(settings_layout)
|
|
layout.addWidget(settings_group)
|
|
|
|
# 操作输出
|
|
output_group = QGroupBox("操作输出")
|
|
output_layout = QVBoxLayout()
|
|
|
|
self.output_text = QTextEdit()
|
|
self.output_text.setReadOnly(True)
|
|
self.output_text.setPlaceholderText("操作结果将在这里显示...")
|
|
output_layout.addWidget(self.output_text)
|
|
|
|
# 进度条
|
|
self.progress_bar = QProgressBar()
|
|
self.progress_bar.setVisible(False)
|
|
output_layout.addWidget(self.progress_bar)
|
|
|
|
output_group.setLayout(output_layout)
|
|
layout.addWidget(output_group)
|
|
|
|
layout.addStretch()
|
|
self.setLayout(layout)
|
|
|
|
# 加载Django路径
|
|
self.load_django_path()
|
|
|
|
def load_django_path(self):
|
|
"""从config.json加载Django路径"""
|
|
try:
|
|
if self.parent and hasattr(self.parent, 'get_current_config'):
|
|
config = self.parent.get_current_config()
|
|
if config:
|
|
# 使用remote_directory而不是django_path
|
|
django_path = config.get('remote_directory', '')
|
|
self.django_path_input.setText(django_path)
|
|
logger.info(f"从当前服务器配置加载Django路径: {django_path}")
|
|
else:
|
|
logger.warning("未找到当前服务器配置")
|
|
else:
|
|
# 兼容旧的加载方式
|
|
config_path = os.path.join(os.path.dirname(__file__), 'config.json')
|
|
with open(config_path, 'r', encoding='utf-8') as f:
|
|
config = json.load(f)
|
|
|
|
if config and 'servers' in config and len(config['servers']) > 0:
|
|
server_config = config['servers'][0]
|
|
# 使用remote_directory
|
|
django_path = server_config.get('remote_directory', '')
|
|
self.django_path_input.setText(django_path)
|
|
logger.info(f"从配置文件加载Django路径: {django_path}")
|
|
except Exception as e:
|
|
logger.error(f"加载Django路径失败: {str(e)}")
|
|
# 不显示警告，避免影响用户体验
|
|
QMessageBox.warning(self, "警告", f"加载django路径失败: {str(e)}")
|
|
|
|
def get_password(self):
|
|
"""从父类获取密码"""
|
|
if hasattr(self.parent, 'password_input'):
|
|
password = self.parent.password_input.text()
|
|
if not password:
|
|
QMessageBox.warning(self, "警告", "请输入服务器密码")
|
|
return None
|
|
return password
|
|
else:
|
|
QMessageBox.warning(self, "警告", "无法获取服务器密码")
|
|
return None
|
|
|
|
def check_ssh_connection(self):
|
|
if not self.parent or not self.parent.ssh_client:
|
|
QMessageBox.warning(self, "警告", "请先连接服务器")
|
|
return False
|
|
return True
|
|
|
|
def install_django(self):
|
|
if not self.check_ssh_connection():
|
|
return
|
|
|
|
# 获取密码
|
|
password = self.get_password()
|
|
if password is None:
|
|
return
|
|
|
|
self.output_text.append("正在安装Django...")
|
|
self.install_django_btn.setEnabled(False)
|
|
self.progress_bar.setVisible(True)
|
|
self.progress_bar.setValue(0)
|
|
|
|
self.django_install_thread = DjangoInstallThread(self.parent.ssh_client, password)
|
|
self.django_install_thread.progress_updated.connect(self.update_progress)
|
|
self.django_install_thread.result_ready.connect(self.on_install_django_result)
|
|
self.django_install_thread.start()
|
|
|
|
def update_progress(self, value):
|
|
self.progress_bar.setValue(value)
|
|
|
|
def on_install_django_result(self, success, message):
|
|
self.install_django_btn.setEnabled(True)
|
|
self.progress_bar.setVisible(False)
|
|
if success:
|
|
self.output_text.append(f"Django安装成功: {message}")
|
|
logger.info(f"Django安装成功: {message}")
|
|
else:
|
|
self.output_text.append(f"Django安装失败: {message}")
|
|
logger.error(f"Django安装失败: {message}")
|
|
|
|
def test_django(self):
|
|
if not self.check_ssh_connection():
|
|
return
|
|
|
|
django_path = self.django_path_input.text().strip()
|
|
if not django_path:
|
|
QMessageBox.warning(self, "警告", "请输入Django项目路径")
|
|
return
|
|
|
|
self.output_text.append(f"正在测试启动Django项目 {django_path}...")
|
|
self.test_django_btn.setEnabled(False)
|
|
self.progress_bar.setVisible(True)
|
|
self.progress_bar.setValue(0)
|
|
|
|
self.django_test_thread = DjangoTestThread(self.parent.ssh_client, django_path)
|
|
self.django_test_thread.progress_updated.connect(self.update_progress)
|
|
self.django_test_thread.result_ready.connect(self.on_test_django_result)
|
|
self.django_test_thread.start()
|
|
|
|
def on_test_django_result(self, success, message):
|
|
self.test_django_btn.setEnabled(True)
|
|
self.progress_bar.setVisible(False)
|
|
if success:
|
|
self.output_text.append(f"Django测试启动成功: {message}")
|
|
logger.info(f"Django测试启动成功: {message}")
|
|
else:
|
|
self.output_text.append(f"Django测试启动失败: {message}")
|
|
logger.error(f"Django测试启动失败: {message}")
|
|
|
|
def download_settings(self):
|
|
if not self.check_ssh_connection():
|
|
return
|
|
|
|
django_path = self.django_path_input.text().strip()
|
|
if not django_path:
|
|
QMessageBox.warning(self, "警告", "请输入Django项目路径")
|
|
return
|
|
|
|
self.output_text.append(f"正在下载settings.py从 {django_path}...")
|
|
self.download_settings_btn.setEnabled(False)
|
|
|
|
self.download_thread = DownloadSettingsThread(self.parent.ssh_client, django_path)
|
|
self.download_thread.result_ready.connect(self.on_download_settings_result)
|
|
self.download_thread.start()
|
|
|
|
def on_download_settings_result(self, success, message):
|
|
self.download_settings_btn.setEnabled(True)
|
|
if success:
|
|
self.settings_editor.setText(message)
|
|
self.output_text.append("settings.py下载成功")
|
|
logger.info("settings.py下载成功")
|
|
else:
|
|
self.output_text.append(f"settings.py下载失败: {message}")
|
|
logger.error(f"settings.py下载失败: {message}")
|
|
|
|
def upload_settings(self):
|
|
if not self.check_ssh_connection():
|
|
return
|
|
|
|
django_path = self.django_path_input.text().strip()
|
|
settings_content = self.settings_editor.toPlainText()
|
|
|
|
if not django_path or not settings_content:
|
|
QMessageBox.warning(self, "警告", "请输入Django项目路径并编辑settings.py内容")
|
|
return
|
|
|
|
self.output_text.append(f"正在上传settings.py到 {django_path}...")
|
|
self.upload_settings_btn.setEnabled(False)
|
|
|
|
self.upload_thread = UploadSettingsThread(self.parent.ssh_client, django_path, settings_content)
|
|
self.upload_thread.result_ready.connect(self.on_upload_settings_result)
|
|
self.upload_thread.start()
|
|
|
|
def on_upload_settings_result(self, success, message):
|
|
self.upload_settings_btn.setEnabled(True)
|
|
if success:
|
|
self.output_text.append("settings.py上传成功")
|
|
logger.info("settings.py上传成功")
|
|
else:
|
|
self.output_text.append(f"settings.py上传失败: {message}")
|
|
logger.error(f"settings.py上传失败: {message}")
|
|
|
|
def collect_static(self):
|
|
if not self.check_ssh_connection():
|
|
return
|
|
|
|
django_path = self.django_path_input.text().strip()
|
|
if not django_path:
|
|
QMessageBox.warning(self, "警告", "请输入Django项目路径")
|
|
return
|
|
|
|
self.output_text.append(f"正在收集 {django_path} 的静态文件...")
|
|
self.collect_static_btn.setEnabled(False)
|
|
self.progress_bar.setVisible(True)
|
|
self.progress_bar.setValue(0)
|
|
|
|
self.collect_thread = CollectStaticThread(self.parent.ssh_client, django_path)
|
|
self.collect_thread.progress_updated.connect(self.update_progress)
|
|
self.collect_thread.result_ready.connect(self.on_collect_static_result)
|
|
self.collect_thread.start()
|
|
|
|
def on_collect_static_result(self, success, message):
|
|
self.collect_static_btn.setEnabled(True)
|
|
self.progress_bar.setVisible(False)
|
|
if success:
|
|
self.output_text.append(f"静态文件收集成功: {message}")
|
|
logger.info(f"静态文件收集成功: {message}")
|
|
else:
|
|
self.output_text.append(f"静态文件收集失败: {message}")
|
|
logger.error(f"静态文件收集失败: {message}")
|
|
|
|
def check_django_status(self):
|
|
if not self.check_ssh_connection():
|
|
return
|
|
|
|
django_path = self.django_path_input.text().strip()
|
|
if not django_path:
|
|
QMessageBox.warning(self, "警告", "请输入Django项目路径")
|
|
return
|
|
|
|
self.output_text.append(f"正在检查 {django_path} 的Django状态...")
|
|
self.check_status_btn.setEnabled(False)
|
|
|
|
self.check_status_thread = CheckDjangoStatusThread(self.parent.ssh_client, django_path)
|
|
self.check_status_thread.result_ready.connect(self.on_check_django_status_result)
|
|
self.check_status_thread.start()
|
|
|
|
def on_check_django_status_result(self, success, message):
|
|
self.check_status_btn.setEnabled(True)
|
|
if success:
|
|
self.output_text.append(f"Django状态检查结果:\n{message}")
|
|
logger.info("Django状态检查成功")
|
|
else:
|
|
self.output_text.append(f"Django状态检查失败: {message}")
|
|
logger.error(f"Django状态检查失败: {message}")
|
|
|
|
def on_server_changed(self):
|
|
self.load_django_path() |