Python调用Bash：保持上下文、实时输出并返回结果的类实现阻塞问题

我有一个能向Bash发送指令的函数，它可以实时打印输出到控制台，同时返回执行结果，运行正常，代码如下：

import subprocess
import io
import selectors
import sys

def call_in_bash(statement): 
  process = subprocess.Popen(['bash','-c',statement],stdout=subprocess.PIPE, universal_newlines=True, bufsize=1)
  buffer = io.StringIO()
  selector = selectors.DefaultSelector()
  selector.register(process.stdout, selectors.EVENT_READ)
  line = ''
  while process.poll() is None:
    events = selector.select()
    for key, mask in events:
      line = key.fileobj.readline()
      buffer.write(line)
      sys.stdout.write(line)

  selector.close()
  output = buffer.getvalue()
  buffer.close()
  return output

现在我想把它改造成带构造函数和析构函数的类，让后续调用能保持Bash上下文（比如先执行cd folder，再在该目录下执行其他命令）。我尝试了下面的实现，但不管执行什么命令都会阻塞，无法结束：

from subprocess import Popen

def __init__(self):
  self.process: Popen(['bash'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, universal_newlines=True, bufsize=1)

def call_to_bash(self, statement):
  self.process.stdin.write(statement + "\n")
  self.process.stdin.flush()
  
  # 后续逻辑和原函数一致

我搞不懂直接调用popen(['bash'])写入标准输入，和用bash -c statement有什么本质区别，也没找到能同时满足保持上下文、实时输出、返回结果三者的方案。

bash -c statement是启动一个Bash进程，专门执行你传入的单条命令，执行完就立刻退出，所以原函数里process.poll()能检测到进程结束，循环正常终止。

而直接启动bash进程是开启一个交互式会话：它会一直运行，等待你输入命令，执行完命令后会输出提示符（比如user@host:~$），不会自动退出。你的代码里用process.poll()判断进程是否结束，但这个Bash进程一直活着，循环永远不会停，自然就阻塞了。

下面的实现同时满足三个需求：保持Bash上下文、实时打印输出、返回执行结果。核心思路是给每条命令加一个唯一的结束标记，用来判断命令执行完成，同时过滤掉Bash的提示符，避免干扰输出。

import subprocess
import io
import selectors
import sys
from typing import Optional

class BashSession:
    def __init__(self):
        # 启动交互式Bash进程，同时绑定stdin/stdout/stderr
        self.process = subprocess.Popen(
            ['bash'],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,  # 把stderr重定向到stdout，统一处理
            universal_newlines=True,
            bufsize=1,
            start_new_session=False
        )
        self.selector = selectors.DefaultSelector()
        self.selector.register(self.process.stdout, selectors.EVENT_READ)
        # 定义唯一的结束标记，避免和命令输出冲突
        self.done_marker = "__BASH_SESSION_COMMAND_COMPLETED__"
        # 先读取并丢弃初始提示符
        self._read_until_prompt()

    def _read_until_prompt(self):
        # 读取输出直到遇到提示符（简单处理，假设提示符以$或#结尾）
        buffer = []
        while True:
            events = self.selector.select(timeout=1)
            if not events:
                break
            for key, mask in events:
                line = key.fileobj.readline()
                if line.endswith('$ ') or line.endswith('# '):
                    break
                buffer.append(line)

    def call(self, statement: str) -> str:
        buffer = io.StringIO()
        # 构造命令：执行目标命令后，输出结束标记和退出码
        full_command = f"{statement}; echo '{self.done_marker}' $?\n"
        self.process.stdin.write(full_command)
        self.process.stdin.flush()

        done = False
        exit_code = 0
        while not done:
            events = self.selector.select(timeout=1)
            if not events:
                continue
            for key, mask in events:
                line = key.fileobj.readline()
                if not line:
                    done = True
                    break
                # 检查是否遇到结束标记
                if self.done_marker in line:
                    # 解析退出码
                    parts = line.strip().split()
                    if len(parts) >= 2:
                        exit_code = int(parts[1])
                    done = True
                    continue
                # 实时打印到控制台
                sys.stdout.write(line)
                sys.stdout.flush()
                # 写入结果缓存
                buffer.write(line)

        output = buffer.getvalue()
        buffer.close()
        # 再次读取并丢弃后续的提示符，避免影响下一次命令
        self._read_until_prompt()
        return output

    def __del__(self):
        # 析构函数：关闭进程和资源
        try:
            self.process.stdin.write("exit\n")
            self.process.stdin.flush()
            self.process.wait(timeout=2)
        except (subprocess.TimeoutExpired, BrokenPipeError):
            self.process.kill()
        finally:
            self.selector.close()

if __name__ == "__main__":
    bash = BashSession()
    # 切换目录
    bash.call("cd /tmp")
    # 在/tmp目录下创建文件
    output = bash.call("touch test_file && ls -l test_file")
    print("\n=== 命令返回结果 ===")
    print(output)
    # 销毁会话时自动关闭Bash进程
    del bash

1. 结束标记机制：每条命令后追加echo '{self.done_marker}' $?，既可以标记命令执行完成，还能获取命令的退出码（如果需要的话可以扩展返回值包含退出码）。
2. 提示符处理：初始化和每次命令执行后，都会读取并丢弃Bash的提示符，避免提示符混入命令结果。
3. stderr重定向：把stderr重定向到stdout，确保错误输出也能实时打印并被捕获。
4. 资源清理：析构函数里先尝试优雅退出Bash，超时则强制杀死进程，同时关闭选择器资源。

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