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): """克隆Git仓库到远程服务器""" if not self.check_ssh_connection(): return git_url = self.git_url_input.text().strip() if not git_url: QMessageBox.warning(self, "警告", "请输入Git仓库地址") return # 获取当前服务器配置 config = None if self.parent and hasattr(self.parent, 'server_connection_tab'): config = self.parent.server_connection_tab.get_current_config() if config: # 使用remote_directory而不是django_path target_path = config.get('remote_directory', '/home/user') project_name = config.get('project_name', 'myproject') else: # 如果没有config，使用默认值 target_path = '/home/user' project_name = 'myproject' # 获取密码 password = self.get_password() if password is None: return self.output_text.append(f"正在克隆Git仓库 {git_url} 到 {target_path}...") self.clone_git_btn.setEnabled(False) self.progress_bar.setVisible(True) self.progress_bar.setValue(0) self.clone_thread = CloneGitThread(self.parent.ssh_client, git_url, target_path, project_name, password) self.clone_thread.progress_updated.connect(self.update_progress) self.clone_thread.result_ready.connect(self.on_clone_git_result) self.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 # 获取密码 password = self.get_password() if password is None: return self.output_text.append("正在设置时区为Asia/Shanghai并重启服务器...") self.set_timezone_btn.setEnabled(False) self.timezone_thread = SetTimezoneAndRestartThread(self.parent.ssh_client, password) 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 # 获取密码 password = self.get_password() if password is None: return self.output_text.append("正在检查防火墙状态...") self.check_firewall_btn.setEnabled(False) self.firewall_thread = CheckFirewallThread(self.parent.ssh_client, password) 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 # 获取密码 password = self.get_password() if password is None: return self.output_text.append(f"正在开放端口 {port}...") self.open_port_btn.setEnabled(False) self.open_port_thread = OpenPortThread(self.parent.ssh_client, port, password) 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 get_password(self): """获取服务器连接密码""" if not self.parent or not hasattr(self.parent, 'password_input'): QMessageBox.warning(self, "警告", "无法获取服务器密码") return None password = self.parent.password_input.text() if not password: QMessageBox.warning(self, "警告", "请先输入服务器密码") return None return password def check_ssh_connection(self): if not self.parent or not self.parent.ssh_client: QMessageBox.warning(self, "警告", "请先连接服务器") return False return True