如何使用 `stdin=PIPE` 重现 `stdin=sys.stdin`？答案

【问题标题】：How do I reproduce `stdin=sys.stdin` with `stdin=PIPE`?如何使用 `stdin=PIPE` 重现 `stdin=sys.stdin`？
【发布时间】：2016-09-19 04:14:52
【问题描述】：

我的以下代码完全符合预期：

from subprocess import Popen

process = Popen(
    ["/bin/bash"],
    stdin=sys.stdin,
    stdout=sys.stdout,
    stderr=sys.stderr,
)
process.wait()

我可以交互地使用 bash、tab 作品等。

但是，我想控制我发送到标准输入的内容，所以我希望以下工作：

import os
import sys
from subprocess import Popen, PIPE
from select import select

process = Popen(
    ["/bin/bash"],
    stdin=PIPE,
    stdout=sys.stdout,
    stderr=sys.stderr,
)

while True:
    if process.poll() is not None:
        break

    r, _, _ = select([sys.stdin], [], [])

    if sys.stdin in r:
        stdin = os.read(sys.stdin.fileno(), 1024)
        # Do w/e I want with stdin
        os.write(process.stdin.fileno(), stdin)

process.wait()

但行为不一样。我尝试了另一种方法（通过 pty）：

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

master, slave = os.openpty()
stdin = sys.stdin.fileno()

try:
    tty.setraw(master)
    ttyname = os.ttyname(slave)

    def _preexec():
        os.setsid()
        open(ttyname, "r+")

    process = Popen(
        args=["/bin/bash"],
        preexec_fn=_preexec,
        stdin=slave,
        stdout=sys.stdout,
        stderr=sys.stderr,
        close_fds=True,
    )

    while True:
        if process.poll() is not None:
            break

        r, _, _ = select([sys.stdin], [], [])

        if sys.stdin in r:
            os.write(master, os.read(stdin, 1024))
finally:
    os.close(master)
    os.close(slave)

并且行为非常接近，除了选项卡仍然不起作用。好吧，标签已正确发送，但我的终端没有显示完成，即使它是由 bash 完成的。箭头还显示^[[A，而不是查看历史记录。

有什么想法吗？

【问题讨论】：

您知道（通常）您可以将您的第一个代码示例替换为：subprocess.call(['/bin/bash'])？你也可以直接使用（或者看看它是如何实现的）pty.spawn(['/bin/bash'], master_read, stdin_read)（你需要通过-i，-l等）吗？

标签： python linux subprocess

【解决方案1】：

我所需要的只是将我的 sys.stdout 设置为 raw。我还发现了 3 件事：

我需要在sys.stdout 上恢复终端设置
subprocess.Popen 有一个 start_new_session 参数，它执行我的 _preexec 函数正在执行的操作。
select.select 接受第四个参数，即放弃前的超时。它让我避免在退出后卡在选择循环中。

最终代码：

import os
import sys
import tty
import termios
import select
import subprocess

master, slave = os.openpty()
stdin = sys.stdin.fileno()

try:
    old_settings = termios.tcgetattr(sys.stdout)
    tty.setraw(sys.stdout)

    process = subprocess.Popen(
        args=["/bin/bash"],
        stdin=slave,
        stdout=sys.stdout,
        stderr=sys.stderr,
        close_fds=True,
        start_new_session=True,
    )

    while True:
        if process.poll() is not None:
            break

        r, _, _ = select.select([sys.stdin], [], [], 0.2)

        if sys.stdin in r:
            os.write(master, os.read(stdin, 1024))
finally:
    termios.tcsetattr(sys.stdout, termios.TCSADRAIN, old_settings)
    os.close(master)
    os.close(slave)

【讨论】：