django.remote/remote_commands_tab.py

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
            
            # 在执行命令前，先输出当前目录
            try:
                pwd_stdin, pwd_stdout, pwd_stderr = self.ssh_client.exec_command("pwd")
                pwd_output = pwd_stdout.read().decode().strip()
                if pwd_output:
                    self.output_signal.emit(f"当前目录: {pwd_output}")
                    logger.info(f"当前目录: {pwd_output}")
            except Exception as e:
                logger.error(f"获取当前目录失败: {str(e)}")
            
            # 如果命令包含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)
                
                # 如果预先设置了密码，直接发送
                if self.password:
                    logger.info("使用预先设置的密码")
                    stdin.write(self.password + "\n")
                    stdin.flush()
                else:
                    # 检查是否需要密码
                    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()
        
        # 自定义命令执行区域
        custom_command_layout = QVBoxLayout()
        custom_command_layout.addWidget(QLabel("自定义命令:"))
        
        # 命令输入框
        self.custom_command_input = QTextEdit()
        self.custom_command_input.setMaximumHeight(100)
        custom_command_layout.addWidget(self.custom_command_input)
        
        # 执行按钮
        execute_button_layout = QHBoxLayout()
        self.execute_command_button = QPushButton("执行命令")
        self.execute_command_button.clicked.connect(self.execute_custom_command)
        execute_button_layout.addWidget(self.execute_command_button)
        execute_button_layout.addStretch()
        custom_command_layout.addLayout(execute_button_layout)
        
        system_layout.addLayout(custom_command_layout)
        
        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.on_current_dir_entered)
        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_server_info(self, server_info):
        """设置服务器信息"""
        logger.info(f"设置服务器信息: {server_info}")
        self.server_info = server_info
        
        # 更新主窗口状态栏显示当前目录
        try:
            main_window = self.parent().parent()
            if hasattr(main_window, 'status_bar') and self.ssh_client:
                current_dir = self.current_dir_display.text().strip()
                server_host = self.server_info.get('host', '未知')
                if current_dir:
                    main_window.status_bar.showMessage(f"远程服务器 {server_host}: {current_dir}")
                    logger.info(f"主窗口状态栏更新为远程服务器目录: {server_host}: {current_dir}")
                else:
                    main_window.status_bar.showMessage(f"远程服务器: {server_host}")
                    logger.info(f"主窗口状态栏更新为远程服务器: {server_host}")
        except Exception as e:
            logger.error(f"更新主窗口状态栏失败: {str(e)}")
    
    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;")
            
            # 更新主窗口状态栏显示当前目录
            try:
                main_window = self.parent().parent()
                if hasattr(main_window, 'status_bar'):
                    current_dir = self.current_dir_display.text().strip()
                    server_host = self.server_info.get('host', '未知') if hasattr(self, 'server_info') else '未知'
                    if current_dir:
                        main_window.status_bar.showMessage(f"远程服务器 {server_host}: {current_dir}")
                        logger.info(f"主窗口状态栏更新为远程服务器目录: {server_host}: {current_dir}")
                    else:
                        main_window.status_bar.showMessage(f"远程服务器: {server_host}")
                        logger.info(f"主窗口状态栏更新为远程服务器: {server_host}")
            except Exception as e:
                logger.error(f"更新主窗口状态栏失败: {str(e)}")
        else:
            self.status_label.setText("未连接到服务器")
            self.status_label.setStyleSheet("color: red;")
            
            # 更新主窗口状态栏显示未连接状态
            try:
                main_window = self.parent().parent()
                if hasattr(main_window, 'status_bar'):
                    main_window.status_bar.showMessage("未连接到远程服务器")
                    logger.info("主窗口状态栏更新为未连接状态")
            except Exception as e:
                logger.error(f"更新主窗口状态栏失败: {str(e)}")
    
    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()
            
            # 更新主窗口状态栏显示当前目录
            try:
                main_window = self.parent().parent()
                if hasattr(main_window, 'status_bar'):
                    server_host = self.server_info['host'] if hasattr(self, 'server_info') and 'host' in self.server_info else '未知'
                    main_window.status_bar.showMessage(f"远程服务器 {server_host}: {remote_dir}")
                    logger.info(f"主窗口状态栏更新为远程服务器目录: {server_host}: {remote_dir}")
            except Exception as e:
                logger.error(f"更新主窗口状态栏失败: {str(e)}")
    
    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 on_current_dir_entered(self):
        """处理用户输入目录路径并按回车键的情况"""
        logger.info("用户输入目录路径并按回车键")
        
        # 更新主窗口状态栏显示当前目录
        try:
            main_window = self.parent().parent()
            if hasattr(main_window, 'status_bar'):
                current_dir = self.current_dir_display.text().strip()
                server_host = self.server_info['host'] if hasattr(self, 'server_info') and 'host' in self.server_info else '未知'
                if current_dir:
                    main_window.status_bar.showMessage(f"远程服务器 {server_host}: {current_dir}")
                    logger.info(f"主窗口状态栏更新为远程服务器目录: {server_host}: {current_dir}")
        except Exception as e:
            logger.error(f"更新主窗口状态栏失败: {str(e)}")
        
        # 刷新目录列表
        self.refresh_directory()
    
    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("目录列表刷新成功")
            
            # 更新主窗口状态栏显示当前目录
            try:
                main_window = self.parent().parent()
                if hasattr(main_window, 'status_bar'):
                    current_dir = self.current_dir_display.text()
                    server_host = self.server_info['host'] if hasattr(self, 'server_info') and 'host' in self.server_info else '未知'
                    main_window.status_bar.showMessage(f"远程服务器 {server_host}: {current_dir}")
                    logger.info(f"主窗口状态栏更新为远程服务器目录: {server_host}: {current_dir}")
            except Exception as e:
                logger.error(f"更新主窗口状态栏失败: {str(e)}")
        else:
            self.status_label.setText("刷新目录列表失败")
            self.status_label.setStyleSheet("color: red;")
            logger.error(f"刷新目录列表失败: {message}")
            QMessageBox.warning(self, "错误", f"刷新目录列表失败: {message}")
    
    def execute_custom_command(self):
        logger.info("执行自定义命令")
        
        if not self.ssh_client:
            QMessageBox.warning(self, "警告", "请先连接到服务器")
            return
            
        command = self.custom_command_input.toPlainText().strip()
        if not command:
            QMessageBox.warning(self, "警告", "请输入要执行的命令")
            return
            
        self.output_text.clear()
        self.status_label.setText("正在执行命令...")
        
        # 如果命令以sudo开头，预先请求密码
        if command.startswith("sudo "):
            logger.info("检测到sudo命令，预先请求密码")
            # 创建密码输入对话框
            dialog = PasswordDialog(self)
            if dialog.exec() == QDialog.Accepted:
                password = dialog.get_password()
                if not password:
                    QMessageBox.warning(self, "警告", "未输入密码，取消执行命令")
                    return
            else:
                QMessageBox.warning(self, "警告", "已取消输入密码")
                return
        
        # 创建并启动线程执行命令
        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_custom_command_finished)
        self.command_thread.password_request_signal.connect(self.request_password)
        
        # 如果是sudo命令且已获取密码，预先设置密码
        if command.startswith("sudo ") and password:
            self.command_thread.set_password(password)
            self.output_text.append("密码已设置\n")
        
        self.command_thread.start()
    
    def on_custom_command_finished(self, success, message):
        if success:
            self.status_label.setText("命令执行成功")
            self.status_label.setStyleSheet("color: green;")
            self.output_text.append(f"\n=== 命令执行成功 ===\n{message}")
        else:
            self.status_label.setText("命令执行失败")
            self.status_label.setStyleSheet("color: red;")
            self.output_text.append(f"\n=== 命令执行失败 ===\n{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 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()