from PySide6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit, QPushButton, QGroupBox, QTextEdit, QMessageBox, QFormLayout, QDialog, QDialogButtonBox) from PySide6.QtCore import Qt, QThread, Signal from loguru import logger 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 RemoteCommandThread(QThread): output_signal = Signal(str) finished_signal = Signal(bool, str) password_request_signal = Signal() # 请求密码的信号 def __init__(self, ssh_client, command): super().__init__() self.ssh_client = ssh_client self.command = command self.password = None self.waiting_for_password = False def set_password(self, password): self.password = password self.waiting_for_password = False def run(self): try: logger.info(f"执行远程命令: {self.command}") # 检查SSH连接是否仍然有效 try: # 尝试执行一个简单的命令来检查连接 transport = self.ssh_client.get_transport() if self.ssh_client else None if not transport or not transport.is_active(): raise Exception("SSH连接已断开") except Exception as e: logger.error(f"SSH连接检查失败: {str(e)}") self.output_signal.emit(f"错误: SSH连接已断开，请重新连接服务器") self.finished_signal.emit(False, f"SSH连接已断开，请重新连接服务器") return # 如果命令包含sudo，修改为使用-S选项从标准输入读取密码 if "sudo" in self.command: command_with_sudo = self.command.replace("sudo", "sudo -S") stdin, stdout, stderr = self.ssh_client.exec_command(command_with_sudo) # 检查是否需要密码 password_prompt = False for line in stderr: self.output_signal.emit(line) if "password for" in line.lower() or "密码" in line: password_prompt = True break # 如果需要密码，请求用户输入 if password_prompt: self.waiting_for_password = True self.password_request_signal.emit() # 等待密码输入 while self.waiting_for_password: self.msleep(100) # 发送密码 if self.password: stdin.write(self.password + "\n") stdin.flush() else: stdin, stdout, stderr = self.ssh_client.exec_command(self.command) # 读取输出 output = "" for line in stdout: output += line self.output_signal.emit(line) # 读取错误 error = "" for line in stderr: error += line if "password for" not in line.lower() and "密码" not in line: # 避免重复显示密码提示 self.output_signal.emit(f"错误: {line}") # 检查退出状态 exit_status = stdout.channel.recv_exit_status() if exit_status == 0: logger.info(f"命令执行成功: {self.command}") self.finished_signal.emit(True, "命令执行成功") else: logger.error(f"命令执行失败，退出状态: {exit_status}") self.finished_signal.emit(False, f"命令执行失败，退出状态: {exit_status}") except Exception as e: logger.error(f"执行命令时发生错误: {str(e)}") error_msg = str(e) # 检查是否是Socket关闭错误 if "Socket is closed" in error_msg or "SSH session not active" in error_msg: self.output_signal.emit(f"错误: SSH连接已断开，请重新连接服务器") self.finished_signal.emit(False, "SSH连接已断开，请重新连接服务器") else: self.output_signal.emit(f"错误: {error_msg}") self.finished_signal.emit(False, f"执行命令时发生错误: {error_msg}") class RemoteCommandsTab(QWidget): def __init__(self): super().__init__() logger.info("初始化远程命令标签页") self.ssh_client = None self.command_thread = None self.init_ui() def init_ui(self): # 主布局 main_layout = QVBoxLayout() # Git操作组 git_group = QGroupBox("Git操作") git_layout = QVBoxLayout() # 安装Git按钮 install_git_layout = QHBoxLayout() self.install_git_button = QPushButton("安装Git") self.install_git_button.clicked.connect(self.install_git) install_git_layout.addWidget(self.install_git_button) install_git_layout.addStretch() git_layout.addLayout(install_git_layout) # 克隆项目 clone_layout = QHBoxLayout() # 左侧：仓库URL和目录信息 left_layout = QFormLayout() self.repo_url_input = QLineEdit() left_layout.addRow("仓库URL:", self.repo_url_input) self.remote_dir_display = QLineEdit() self.remote_dir_display.setReadOnly(True) left_layout.addRow("远程目录:", self.remote_dir_display) clone_layout.addLayout(left_layout) # 右侧：按钮 right_layout = QVBoxLayout() self.clone_button = QPushButton("克隆项目") self.clone_button.clicked.connect(self.clone_repository) right_layout.addWidget(self.clone_button) right_layout.addStretch() clone_layout.addLayout(right_layout) git_layout.addLayout(clone_layout) git_group.setLayout(git_layout) main_layout.addWidget(git_group) # 系统管理组 system_group = QGroupBox("系统管理") system_layout = QVBoxLayout() # 时区设置 timezone_layout = QHBoxLayout() self.set_timezone_button = QPushButton("设置时区为北京时区") self.set_timezone_button.clicked.connect(self.set_timezone) timezone_layout.addWidget(self.set_timezone_button) timezone_layout.addStretch() system_layout.addLayout(timezone_layout) # 服务器重启 reboot_layout = QHBoxLayout() self.reboot_button = QPushButton("重启服务器") self.reboot_button.clicked.connect(self.reboot_server) reboot_layout.addWidget(self.reboot_button) reboot_layout.addStretch() system_layout.addLayout(reboot_layout) # 配置sudo免密按钮 self.sudo_nopasswd_button = QPushButton("配置sudo免密") self.sudo_nopasswd_button.clicked.connect(self.configure_sudo_nopasswd) system_layout.addWidget(self.sudo_nopasswd_button) system_group.setLayout(system_layout) main_layout.addWidget(system_group) # 目录管理组 dir_group = QGroupBox("目录管理") dir_layout = QVBoxLayout() # 当前目录显示 current_dir_layout = QHBoxLayout() current_dir_layout.addWidget(QLabel("当前目录:")) self.current_dir_display = QLineEdit() # 允许用户输入当前目录路径 # self.current_dir_display.setReadOnly(True) # 添加回车键刷新目录功能 self.current_dir_display.returnPressed.connect(self.refresh_directory) current_dir_layout.addWidget(self.current_dir_display) # 刷新目录按钮 self.refresh_dir_button = QPushButton("刷新目录") self.refresh_dir_button.clicked.connect(self.refresh_directory) current_dir_layout.addWidget(self.refresh_dir_button) dir_layout.addLayout(current_dir_layout) # 目录列表 dir_list_layout = QHBoxLayout() dir_list_layout.addWidget(QLabel("目录内容:")) self.dir_list_text = QTextEdit() self.dir_list_text.setReadOnly(True) self.dir_list_text.setMaximumHeight(150) dir_list_layout.addWidget(self.dir_list_text) dir_layout.addLayout(dir_list_layout) # 删除目录 delete_dir_layout = QHBoxLayout() delete_dir_layout.addWidget(QLabel("删除目录:")) self.delete_dir_input = QLineEdit() delete_dir_layout.addWidget(self.delete_dir_input) self.delete_dir_button = QPushButton("删除目录") self.delete_dir_button.clicked.connect(self.delete_directory) delete_dir_layout.addWidget(self.delete_dir_button) dir_layout.addLayout(delete_dir_layout) dir_group.setLayout(dir_layout) main_layout.addWidget(dir_group) # 命令输出区域 output_group = QGroupBox("命令输出") output_layout = QVBoxLayout() self.output_text = QTextEdit() self.output_text.setReadOnly(True) output_layout.addWidget(self.output_text) output_group.setLayout(output_layout) main_layout.addWidget(output_group) # 状态栏 self.status_label = QLabel("就绪") self.status_label.setAlignment(Qt.AlignCenter) main_layout.addWidget(self.status_label) # 添加伸缩空间 main_layout.addStretch() self.setLayout(main_layout) logger.info("远程命令标签页UI初始化完成") def set_ssh_client(self, ssh_client): logger.info("设置SSH客户端") self.ssh_client = ssh_client if self.ssh_client: self.status_label.setText("已连接到服务器") self.status_label.setStyleSheet("color: green;") else: self.status_label.setText("未连接到服务器") self.status_label.setStyleSheet("color: red;") def set_server_config(self, git_url, remote_dir): logger.info(f"设置服务器配置: git_url={git_url}, remote_dir={remote_dir}") self.repo_url_input.setText(git_url) self.remote_dir_display.setText(remote_dir) # 如果设置了远程目录，也设置为当前目录 if remote_dir: self.current_dir_display.setText(remote_dir) self.refresh_directory() def install_git(self): logger.info("安装Git") if not self.ssh_client: QMessageBox.warning(self, "警告", "请先连接到服务器") return self.output_text.clear() self.status_label.setText("正在安装Git...") # 创建并启动线程执行命令 self.command_thread = RemoteCommandThread(self.ssh_client, "sudo apt update && sudo apt install -y git") self.command_thread.output_signal.connect(self.append_output) self.command_thread.finished_signal.connect(self.on_command_finished) self.command_thread.password_request_signal.connect(self.request_password) self.command_thread.start() def request_password(self): logger.info("请求输入sudo密码") # 创建密码输入对话框 dialog = PasswordDialog(self) if dialog.exec() == QDialog.Accepted: password = dialog.get_password() if password: self.command_thread.set_password(password) self.output_text.append("密码已发送\n") else: self.output_text.append("已取消输入密码\n") self.command_thread.set_password("") self.command_thread.waiting_for_password = False def clone_repository(self): logger.info("克隆仓库") if not self.ssh_client: QMessageBox.warning(self, "警告", "请先连接到服务器") return repo_url = self.repo_url_input.text().strip() if not repo_url: QMessageBox.warning(self, "警告", "请输入仓库URL") return remote_dir = self.remote_dir_display.text().strip() self.output_text.clear() self.status_label.setText("正在克隆仓库...") # 构建克隆命令 if remote_dir: # 如果指定了远程目录，先创建目录（如果不存在），然后克隆到指定目录 clone_command = f"mkdir -p {remote_dir} && cd {remote_dir} && git clone --verbose {repo_url}" else: clone_command = f"git clone --verbose {repo_url}" # 创建并启动线程执行命令 self.command_thread = RemoteCommandThread(self.ssh_client, clone_command) self.command_thread.output_signal.connect(self.append_output) self.command_thread.finished_signal.connect(self.on_command_finished) self.command_thread.start() def append_output(self, text): self.output_text.append(text) def on_command_finished(self, success, message): if success: self.status_label.setText(message) self.status_label.setStyleSheet("color: green;") QMessageBox.information(self, "成功", message) # 如果是目录相关操作，刷新目录列表 if "删除" in message or "目录" in message: self.refresh_directory() else: self.status_label.setText(message) self.status_label.setStyleSheet("color: red;") # 检查是否是SSH连接断开错误 if "SSH连接已断开" in message: reply = QMessageBox.question(self, "SSH连接已断开", f"{message}\n\n是否现在重新连接服务器？", QMessageBox.Yes | QMessageBox.No, QMessageBox.Yes) if reply == QMessageBox.Yes: self.reconnect_ssh() else: QMessageBox.warning(self, "错误", message) def refresh_directory(self): logger.info("刷新目录列表") if not self.ssh_client: QMessageBox.warning(self, "警告", "请先连接到服务器") return current_dir = self.current_dir_display.text().strip() if not current_dir: current_dir = "~" # 默认用户主目录 self.current_dir_display.setText(current_dir) logger.info(f"使用默认目录: {current_dir}") else: logger.info(f"使用用户输入目录: {current_dir}") self.status_label.setText("正在刷新目录列表...") # 创建并启动线程执行命令 command = f"cd {current_dir} && pwd && ls -la" self.command_thread = RemoteCommandThread(self.ssh_client, command) self.command_thread.output_signal.connect(self.append_dir_output) self.command_thread.finished_signal.connect(self.on_dir_refresh_finished) self.command_thread.start() def append_dir_output(self, text): self.dir_list_text.append(text) # 将目录信息输出到日志文件 logger.info(f"目录列表信息: {text.strip()}") def on_dir_refresh_finished(self, success, message): if success: self.status_label.setText("目录列表已刷新") self.status_label.setStyleSheet("color: green;") logger.info("目录列表刷新成功") else: self.status_label.setText("刷新目录列表失败") self.status_label.setStyleSheet("color: red;") logger.error(f"刷新目录列表失败: {message}") QMessageBox.warning(self, "错误", f"刷新目录列表失败: {message}") def set_timezone(self): logger.info("设置时区为北京时区") if not self.ssh_client: QMessageBox.warning(self, "警告", "请先连接到服务器") return # 确认操作 reply = QMessageBox.question(self, "确认设置时区", "确定要将服务器时区设置为北京时区吗？", QMessageBox.Yes | QMessageBox.No, QMessageBox.No) if reply == QMessageBox.No: return self.output_text.clear() self.status_label.setText("正在设置时区...") # 创建并启动线程执行命令 command = "sudo timedatectl set-timezone Asia/Shanghai" self.command_thread = RemoteCommandThread(self.ssh_client, command) self.command_thread.output_signal.connect(self.append_output) self.command_thread.finished_signal.connect(self.on_timezone_set) self.command_thread.password_request_signal.connect(self.request_password) self.command_thread.start() def on_timezone_set(self, success, message): if success: self.status_label.setText("时区设置成功") self.status_label.setStyleSheet("color: green;") QMessageBox.information(self, "成功", "服务器时区已设置为北京时区") # 显示当前时区 self.show_current_timezone() else: self.status_label.setText("时区设置失败") self.status_label.setStyleSheet("color: red;") QMessageBox.warning(self, "错误", f"设置时区失败: {message}") def show_current_timezone(self): logger.info("显示当前时区") if not self.ssh_client: return command = "timedatectl status" self.command_thread = RemoteCommandThread(self.ssh_client, command) self.command_thread.output_signal.connect(self.append_output) self.command_thread.finished_signal.connect(lambda success, message: None) self.command_thread.start() def reboot_server(self): logger.info("重启服务器") if not self.ssh_client: QMessageBox.warning(self, "警告", "请先连接到服务器") return # 确认操作 reply = QMessageBox.question(self, "确认重启", "确定要重启服务器吗？\n此操作将导致服务器临时不可用！", QMessageBox.Yes | QMessageBox.No, QMessageBox.No) if reply == QMessageBox.No: return self.output_text.clear() self.status_label.setText("正在重启服务器...") # 创建并启动线程执行命令 command = "sudo reboot" self.command_thread = RemoteCommandThread(self.ssh_client, command) self.command_thread.output_signal.connect(self.append_output) self.command_thread.finished_signal.connect(self.on_reboot_initiated) self.command_thread.password_request_signal.connect(self.request_password) self.command_thread.start() def on_reboot_initiated(self, success, message): if success: self.status_label.setText("服务器重启命令已发送") self.status_label.setStyleSheet("color: green;") QMessageBox.information(self, "重启已启动", "服务器重启命令已发送，服务器将在几秒后重启。\n请等待服务器重启完成后重新连接。") else: self.status_label.setText("服务器重启失败") self.status_label.setStyleSheet("color: red;") QMessageBox.warning(self, "错误", f"服务器重启失败: {message}") def delete_directory(self): logger.info("删除目录") if not self.ssh_client: QMessageBox.warning(self, "警告", "请先连接到服务器") return dir_to_delete = self.delete_dir_input.text().strip() if not dir_to_delete: QMessageBox.warning(self, "警告", "请输入要删除的目录名") return current_dir = self.current_dir_display.text().strip() if not current_dir: current_dir = "~" # 确认删除 reply = QMessageBox.question(self, "确认删除", f"确定要删除目录 '{current_dir}/{dir_to_delete}' 及其所有内容吗？\n此操作不可撤销！", QMessageBox.Yes | QMessageBox.No, QMessageBox.No) if reply == QMessageBox.No: return self.output_text.clear() self.status_label.setText("正在删除目录...") # 创建并启动线程执行命令 command = f"cd {current_dir} && rm -rf {dir_to_delete}" self.command_thread = RemoteCommandThread(self.ssh_client, command) self.command_thread.output_signal.connect(self.append_output) self.command_thread.finished_signal.connect(self.on_command_finished) self.command_thread.start() def configure_sudo_nopasswd(self): logger.info("配置sudo免密") if not self.ssh_client: QMessageBox.warning(self, "警告", "请先连接到服务器") return # 确认操作 reply = QMessageBox.question(self, "确认配置sudo免密", "确定要配置sudo免密吗？\n此操作将允许当前用户无需密码执行sudo命令，请谨慎操作！", QMessageBox.Yes | QMessageBox.No, QMessageBox.No) if reply == QMessageBox.No: return self.output_text.clear() self.status_label.setText("正在配置sudo免密...") # 创建并启动线程执行命令 # 首先获取当前用户名，然后配置sudo免密 command = "whoami && echo '\n' | sudo -S visudo -c && (echo '\n' | sudo -S visudo -f /etc/sudoers.d/nopasswd && echo '$(whoami) ALL=(ALL) NOPASSWD: ALL' | sudo -S tee /etc/sudoers.d/nopasswd && sudo -S chmod 440 /etc/sudoers.d/nopasswd)" self.command_thread = RemoteCommandThread(self.ssh_client, command) self.command_thread.output_signal.connect(self.append_output) self.command_thread.finished_signal.connect(self.on_sudo_nopasswd_configured) self.command_thread.password_request_signal.connect(self.request_password) self.command_thread.start() def on_sudo_nopasswd_configured(self, success, message): if success: self.status_label.setText("sudo免密配置成功") self.status_label.setStyleSheet("color: green;") QMessageBox.information(self, "成功", "sudo免密配置成功！\n当前用户现在可以无需密码执行sudo命令。") else: self.status_label.setText("sudo免密配置失败") self.status_label.setStyleSheet("color: red;") QMessageBox.warning(self, "错误", f"sudo免密配置失败: {message}") def check_ssh_connection(self): """检查SSH连接是否有效""" if not self.ssh_client: return False try: transport = self.ssh_client.get_transport() return transport and transport.is_active() except Exception as e: logger.error(f"检查SSH连接时发生错误: {str(e)}") return False def reconnect_ssh(self): """重新连接SSH服务器""" logger.info("尝试重新连接SSH服务器") # 关闭现有连接 if self.ssh_client: try: self.ssh_client.close() except: pass self.ssh_client = None # 切换到服务器连接标签页 main_window = self.parent().parent() if hasattr(main_window, 'tabs'): main_window.tabs.setCurrentIndex(0) # 切换到服务器连接标签页 # 显示提示信息 self.status_label.setText("请重新连接服务器") self.status_label.setStyleSheet("color: orange;") self.output_text.append("\n=== SSH连接已断开 ===\n") self.output_text.append("请切换到\"服务器连接\"标签页重新连接服务器\n") # 显示重新连接的对话框 reply = QMessageBox.question(self, "SSH连接已断开", "SSH连接已断开，是否现在重新连接？", QMessageBox.Yes | QMessageBox.No, QMessageBox.Yes) if reply == QMessageBox.Yes: # 模拟点击连接按钮 server_connection_tab = main_window.server_connection_tab server_connection_tab.connect_to_server()