import os import json from loguru import logger from PySide6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit, QPushButton, QComboBox, QMessageBox, QTextEdit, QGroupBox, QGridLayout) from PySide6.QtCore import Qt from threads import (GitInstallThread, GitCloneThread, ListDirectoryThread, SetTimezoneAndRestartThread, CheckFirewallThread, OpenPortThread) class RemoteCommandTab(QWidget): def __init__(self, parent=None): super().__init__(parent) self.parent = parent self.init_ui() def init_ui(self): layout = QVBoxLayout() # Git管理组 git_group = QGroupBox("Git代码管理") git_layout = QGridLayout() git_layout.addWidget(QLabel("Git仓库URL:"), 0, 0) self.git_url_input = QLineEdit() self.git_url_input.setPlaceholderText("https://github.com/username/repo.git") git_layout.addWidget(self.git_url_input, 0, 1) git_layout.addWidget(QLabel("项目路径:"), 1, 0) self.project_path_input = QLineEdit() self.project_path_input.setPlaceholderText("/home/user/project") git_layout.addWidget(self.project_path_input, 1, 1) self.install_git_btn = QPushButton("安装Git") self.install_git_btn.clicked.connect(self.install_git) git_layout.addWidget(self.install_git_btn, 0, 2) self.clone_btn = QPushButton("拉取代码") self.clone_btn.clicked.connect(self.clone_git) git_layout.addWidget(self.clone_btn, 1, 2) self.list_dir_btn = QPushButton("列出目录") self.list_dir_btn.clicked.connect(self.list_directory) git_layout.addWidget(self.list_dir_btn, 2, 2) self.set_timezone_btn = QPushButton("设置时区并重启") self.set_timezone_btn.clicked.connect(self.set_timezone_and_restart) git_layout.addWidget(self.set_timezone_btn, 3, 0, 1, 2) git_group.setLayout(git_layout) layout.addWidget(git_group) # 防火墙管理组 firewall_group = QGroupBox("防火墙管理") firewall_layout = QGridLayout() self.check_firewall_btn = QPushButton("检查防火墙") self.check_firewall_btn.clicked.connect(self.check_firewall) firewall_layout.addWidget(self.check_firewall_btn, 0, 0) firewall_layout.addWidget(QLabel("端口:"), 0, 1) self.port_input = QLineEdit("8000") firewall_layout.addWidget(self.port_input, 0, 2) self.open_port_btn = QPushButton("开放端口") self.open_port_btn.clicked.connect(self.open_port) firewall_layout.addWidget(self.open_port_btn, 0, 3) firewall_group.setLayout(firewall_layout) layout.addWidget(firewall_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) output_group.setLayout(output_layout) layout.addWidget(output_group) layout.addStretch() self.setLayout(layout) # 从配置文件加载git配置 self.load_git_config() def load_git_config(self): try: # 从config.json文件读取配置 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] git_url = server_config.get('git_url', '') project_path = server_config.get('remote_directory', '') self.git_url_input.setText(git_url) self.project_path_input.setText(project_path) logger.info(f"从配置文件加载git配置: git_url={git_url}, project_path={project_path}") else: logger.warning("配置文件中未找到服务器配置") except Exception as e: logger.error(f"加载git配置失败: {str(e)}") QMessageBox.warning(self, "警告", f"加载git配置失败: {str(e)}") def install_git(self): if not self.check_ssh_connection(): return self.output_text.append("正在安装Git...") self.install_git_btn.setEnabled(False) self.git_install_thread = GitInstallThread(self.parent.ssh_client) self.git_install_thread.result_ready.connect(self.on_git_install_result) self.git_install_thread.start() def on_git_install_result(self, success, message): self.install_git_btn.setEnabled(True) if success: self.output_text.append(f"Git安装成功: {message}") logger.info(f"Git安装成功: {message}") else: self.output_text.append(f"Git安装失败: {message}") logger.error(f"Git安装失败: {message}") def clone_git(self): if not self.check_ssh_connection(): return git_url = self.git_url_input.text().strip() project_path = self.project_path_input.text().strip() if not git_url or not project_path: QMessageBox.warning(self, "警告", "请填写Git仓库URL和项目路径") return # 从config.json获取django_path作为克隆目标路径 try: 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] django_path = server_config.get('django_path', '') if django_path: project_path = django_path logger.info(f"使用django_path作为克隆目标路径: {django_path}") except Exception as e: logger.error(f"获取django_path失败: {str(e)}") self.output_text.append(f"正在克隆 {git_url} 到 {project_path}...") self.clone_btn.setEnabled(False) self.git_clone_thread = GitCloneThread(self.parent.ssh_client, git_url, project_path) self.git_clone_thread.result_ready.connect(self.on_git_clone_result) self.git_clone_thread.start() def on_git_clone_result(self, success, message): self.clone_btn.setEnabled(True) if success: self.output_text.append(f"Git克隆成功: {message}") logger.info(f"Git克隆成功: {message}") else: self.output_text.append(f"Git克隆失败: {message}") logger.error(f"Git克隆失败: {message}") def list_directory(self): if not self.check_ssh_connection(): return path = self.project_path_input.text().strip() if not path: QMessageBox.warning(self, "警告", "请输入要列出的目录路径") return self.output_text.append(f"正在列出目录 {path}...") self.list_dir_btn.setEnabled(False) self.list_dir_thread = ListDirectoryThread(self.parent.ssh_client, path) self.list_dir_thread.result_ready.connect(self.on_list_directory_result) self.list_dir_thread.start() def on_list_directory_result(self, success, message): self.list_dir_btn.setEnabled(True) if success: self.output_text.append(f"目录列表:\n{message}") logger.info(f"目录列表成功:\n{message}") else: self.output_text.append(f"列出目录失败: {message}") logger.error(f"列出目录失败: {message}") def set_timezone_and_restart(self): if not self.check_ssh_connection(): return self.output_text.append("正在设置时区为Asia/Shanghai并重启服务器...") self.set_timezone_btn.setEnabled(False) self.timezone_thread = SetTimezoneAndRestartThread(self.parent.ssh_client) self.timezone_thread.result_ready.connect(self.on_set_timezone_and_restart_result) self.timezone_thread.start() def on_set_timezone_and_restart_result(self, success, message): self.set_timezone_btn.setEnabled(True) 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_firewall(self): if not self.check_ssh_connection(): return self.output_text.append("正在检查防火墙状态...") self.check_firewall_btn.setEnabled(False) self.firewall_thread = CheckFirewallThread(self.parent.ssh_client) self.firewall_thread.result_ready.connect(self.on_check_firewall_result) self.firewall_thread.start() def on_check_firewall_result(self, success, message): self.check_firewall_btn.setEnabled(True) if success: self.output_text.append(f"防火墙状态:\n{message}") logger.info(f"防火墙状态检查成功") else: self.output_text.append(f"防火墙状态检查失败: {message}") logger.error(f"防火墙状态检查失败: {message}") def open_port(self): if not self.check_ssh_connection(): return port = self.port_input.text().strip() if not port: QMessageBox.warning(self, "警告", "请输入要开放的端口号") return self.output_text.append(f"正在开放端口 {port}...") self.open_port_btn.setEnabled(False) self.open_port_thread = OpenPortThread(self.parent.ssh_client, port) self.open_port_thread.result_ready.connect(self.on_open_port_result) self.open_port_thread.start() def on_open_port_result(self, success, message): self.open_port_btn.setEnabled(True) 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_ssh_connection(self): if not self.parent or not self.parent.ssh_client: QMessageBox.warning(self, "警告", "请先连接服务器") return False return True