使用 bash 通过 Python3 自调用交互式 shell

Question

我正在使用 python3 和 subprocess.Popen 生成一个

bash

的进程并通过标准解释器再次调用 Python3 解释器。

bash -i

状态：

  -s        If the -s option is present, or if no arguments remain after
            option processing, then commands are read from the standard
            input.  This option allows the positional parameters to be
            set when invoking an interactive shell.

这是一个最小化的示例，但它主要归结为以下代码：

import subprocess
import sys

p = subprocess.Popen(["bash", "-s"], stdin=subprocess.PIPE,stderr=sys.stderr, stdout=sys.stdout)

p.stdin.write(b"python3\n")
p.stdin.flush()

print("Done")

输出只是“完成”。有什么建议我需要如何处理标准输入管道，以便让交互式 shell 在新执行的 python3 解释器中弹出？

实际产量

% python3 test.py 
Done

预期输出：

% python3 test.py
Python 3.10.8 (main, Oct 13 2022, 10:17:43) [Clang 14.0.0 (clang-1400.0.29.102)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> 

Answer 1

听起来你想要一个伪终端。看看pty模块。

pty.fork()

给你一个 pid 和一个文件描述符（子进程得到 pid=0）。从父进程中，您使用文件描述符从终端读取数据并将数据发送到终端。最好的方法可能是让一个线程从终端读取并对输出做一些事情，比如写入主进程的标准输出。

这是一个可以实现您的目标的示例。我已经向终端添加了一些额外的输入，这样它就不会永远挂起。

import pty
import os
import sys
import time
from select import select
import signal
from threading import Thread

class TerminalEmulator:

    def __init__(self, command):
        self.pid = None
        self.fd = None
        self.command = command
        self.thread = None

    def __enter__(self):
        if self.pid is not None:
            raise RuntimerError('Cannot enter multiple times')
        self.pid, self.fd = pty.fork()
        if self.pid:    # Parent
            self.thread = Thread(target=self.communicate)
            self.thread.start()
            return self
        else:           # Child
            os.execlp(*self.command)
            os.exit(1)

    def __exit__(self, *args):
        if self.pid:
            try:
                os.waitpid(self.pid, os.WNOHANG)
            except ChildProcessError:
                pass
            else:
                try:
                    os.kill(self.pid, signal.SIGHUP)
                    os.kill(self.pid, signal.SIGTERM)
                    # Note: process could still be running after this.
                    # Some better handling would be a good idea.
                except ProcessLookupError:
                    pass
            self.pid = False
            self.thread.join()
            self.thread = None

    def send(self, data):
        os.write(self.fd, data)

    def communicate(self):
        while self.pid:
            rfds, _, _ = select([self.fd], [], [], 1)
            if self.fd in rfds:
                try:
                    data = os.read(self.fd, 10240)
                except OSError:
                    return
                os.write(1, data)

with TerminalEmulator(['bash', '-i']) as terminal:
    terminal.send(b'python3\n')
    terminal.send(b'import time, sys\n')
    terminal.send(b'time.sleep(2)\n')
    terminal.send(b'sys.exit(0)\n')
    terminal.send(b'exit\n')
    os.waitpid(terminal.pid, 0)