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()