Discord Bot实现Shell连续命令执行的技术方案问询

我正在开发一款Discord Bot，希望为其添加模拟Shell的功能，允许用户在Bot的主机上执行ls、cd等命令。我通过Python官方文档学习使用subprocess模块，编写了如下类：

class CommandLine:
    def __init__(self, command: str):
       self.process = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    
    def run(self):
       self.process.wait()
       # some processes don't have an exit code, so i check if there's an error instead
       if len(self.process.stderr) == 0:
           return self.process.stdout
       else:
           return self.process.stderr

    def close(self):
       self.process.stdin.close()

该类可单次执行命令，但无法连续执行依赖前序命令上下文的操作（例如cd命令无法生效，后续依赖当前目录的命令无法正常工作）。请问如何无需每次创建新子进程，连续执行多条命令？注：我的目标仅为模拟Shell，不限制使用的模块，若有更合适的模块或实现方式，我更倾向于采用。

要解决上下文保持的问题，核心是启动一个持久化的Shell进程（比如bash或cmd.exe，根据操作系统选择），所有命令都在同一个Shell进程中执行，这样cd、export这类Shell内置命令的上下文会被保留。

实现思路 #

1. 初始化时启动一个交互式Shell进程，而非单次命令的进程。
2. 通过该Shell的stdin管道持续发送命令，通过stdout和stderr读取输出。
3. 注意处理命令的换行符、编码和解码，避免输出乱码或命令执行不完整。

改进后的代码示例 #

import subprocess
import os

class PersistentShell:
    def __init__(self, shell_path=None):
        # 自动选择系统默认Shell，Windows用cmd，Linux/macOS用bash
        self.shell_path = shell_path or (os.environ.get('COMSPEC', 'cmd.exe') if os.name == 'nt' else '/bin/bash')
        # 启动交互式Shell，启用文本模式自动处理编码，行缓冲实现实时输出
        self.process = subprocess.Popen(
            [self.shell_path],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=1,
            universal_newlines=True
        )

    def run_command(self, command: str) -> tuple[str, str]:
        if not self.process.stdin:
            raise RuntimeError("Shell process has been closed")
        
        # 发送命令，必须加换行符触发Shell执行
        self.process.stdin.write(command + '\n')
        self.process.stdin.flush()

        # 读取输出（基础实现，长时间运行命令需用select做非阻塞读取避免死锁）
        stdout = []
        stderr = []
        while True:
            line = self.process.stdout.readline()
            if line:
                stdout.append(line)
            err_line = self.process.stderr.readline()
            if err_line:
                stderr.append(err_line)
            # 检查进程是否结束或无输出
            if not line and not err_line and self.process.poll() is not None:
                break
        
        return ''.join(stdout), ''.join(stderr)

    def close(self):
        if self.process.stdin:
            self.process.stdin.close()
        self.process.wait()

使用示例 #

# 创建持久化Shell实例
shell = PersistentShell()

# 执行cd命令切换目录
stdout, stderr = shell.run_command('cd /tmp')
print("cd执行结果:", stdout, stderr)

# 执行ls命令，此时会显示/tmp目录内容
stdout, stderr = shell.run_command('ls')
print("ls执行结果:", stdout, stderr)

# 关闭Shell进程
shell.close()

关键注意事项 #

• 安全风险：允许用户在主机上执行Shell命令极其危险，必须添加严格的权限验证（比如仅允许指定Discord用户ID使用），禁止公开开放该功能，否则可能导致主机被恶意入侵。
• 输出读取优化：上述代码的读取逻辑适用于短时间完成的命令，对于长时间运行的命令（如top），需要用select模块实现非阻塞读取，避免死锁。
• 跨平台兼容：不同系统的Shell语法存在差异（Windows的dir vs Linux的ls），若需跨平台支持，要做命令适配处理。

替代方案 #

如果不想直接操作subprocess，可以使用第三方库pexpect，它专门用于处理交互式进程，简化命令发送和输出读取逻辑：

import pexpect

class PexpectShell:
    def __init__(self):
        self.child = pexpect.spawn('/bin/bash', encoding='utf-8')
        # 等待Shell提示符出现，确认进程启动完成
        self.child.expect(r'\$ ')

    def run_command(self, command: str) -> str:
        self.child.sendline(command)
        # 等待命令执行完成，匹配提示符
        self.child.expect(r'\$ ')
        # 返回命令输出（剔除命令本身和提示符）
        return self.child.before.strip()

    def close(self):
        self.child.sendline('exit')
        self.child.wait()

内容的提问来源于stack exchange，提问作者user21815993