使用 subprocess.Popen 执行 bash答案

【问题标题】：Executing bash with subprocess.Popen使用 subprocess.Popen 执行 bash
【发布时间】：2011-08-26 14:50:21
【问题描述】：

我正在尝试使用 python 为 bash 会话编写一个包装器。我做的第一件事就是尝试生成一个 bash 进程，然后尝试读取它的输出。像这样：

from subprocess import Popen, PIPE
bash = Popen("bash", stdin = PIPE, stdout = PIPE, stderr = PIPE)
prompt = bash.stdout.read()
bash.stdin.write("ls\n")
ls_output = bash.stdout.read()

但这不起作用。首先，在创建进程后从 bash 的标准输出读取失败，当我尝试写入标准输入时，我得到了一个损坏的管道错误。我做错了什么？

再次澄清一下，我对通过 bash 运行单个命令然后检索其输出不感兴趣，我希望在某个进程中运行一个 bash 会话，我可以通过管道与之通信。

【问题讨论】：

听起来你需要一个 Expect 模块：noah.org/wiki/…
好的，我想我找到了我需要的东西。 Python 有一个名为 pty 的模块，它带有一个 fork，它在类似终端的环境中生成一个子进程并返回一个读/写 fd 给它。然后我可以在孩子中执行 bash。

标签： python bash subprocess

【解决方案1】：

这行得通：

import subprocess
command = "ls"
p = subprocess.Popen(command, shell=True, bufsize=0, stdout=subprocess.PIPE, universal_newlines=True)
p.wait()
output = p.stdout.read()
p.stdout.close()

【讨论】：

啊，但我不想运行单个命令。我希望 bash 进程在我与之交互时保持活跃。
您可能能够一个接一个地生成一个进程，分析前一个进程的输出。您想到了哪些步骤/交互顺序？
请注意，这是在 'sh' 而不是 'bash' 中运行的。存在令人讨厌的差异，例如“echo -n Hello”的工作方式不同。

【解决方案2】：

read() 调用被阻塞，因此它读取第一个命令的输出并挂起，因为更多数据仍可能到达（即bash 仍在运行）。要执行非阻塞读取，请参阅Non-blocking read on a subprocess.PIPE in python。

【讨论】：

【解决方案3】：

为什么不直接使用cmd module/raw_input 和 subprocess.Popen(shell = True)？

【讨论】：

【解决方案4】：

您可以参考我的另一个答案：https://stackoverflow.com/a/43012138/3555925，它使用伪终端并在句柄stdin/stdout中选择。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import sys
import select
import termios
import tty
import pty
from subprocess import Popen

command = 'bash'
# command = 'docker run -it --rm centos /bin/bash'.split()

# save original tty setting then set it to raw mode
old_tty = termios.tcgetattr(sys.stdin)
tty.setraw(sys.stdin.fileno())

# open pseudo-terminal to interact with subprocess
master_fd, slave_fd = pty.openpty()

# use os.setsid() make it run in a new process group, or bash job control will not be enabled
p = Popen(command,
          preexec_fn=os.setsid,
          stdin=slave_fd,
          stdout=slave_fd,
          stderr=slave_fd,
          universal_newlines=True)

while p.poll() is None:
    r, w, e = select.select([sys.stdin, master_fd], [], [])
    if sys.stdin in r:
        d = os.read(sys.stdin.fileno(), 10240)
        os.write(master_fd, d)
    elif master_fd in r:
        o = os.read(master_fd, 10240)
        if o:
            os.write(sys.stdout.fileno(), o)

# restore tty settings back
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)

【讨论】：