如何在Python中模拟运行Python文件并实现交互控制？

需求 #

• 编程运行Python文件，支持随时传入stdin输入
• 定期轮询获取stdout输出
• 可随时终止子进程

原尝试代码 #

Runner控制代码 #

import subprocess
from threading import Thread


class Runner:
    def __init__(self):
        self.process = subprocess.Popen(
            ["python", "x.py"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

    def run(self):
        self.process.wait()

    def poll(self):
        print("got stdout:", self.process.stdout.readline().decode(), end="")

    def give_input(self, text=""):
        return self.process.stdin.write(bytes(text))

    def kill(self):
        self.process.kill()


r = Runner()
t = Thread(target=r.run)
t.start()
r.poll() # should be "hi"
r.poll() # should be "your name:"
r.give_input("hi\n")
r.kill()
t.join()

被调用的x.py代码 #

print("hi")
input("Your name: ")

问题现象 #

启动线程后，第一次poll()能正常获取hi，但第二次poll()直接卡住冻结，无法获取Your name:提示。

问题原因 #

1. input("Your name: ")的提示输出不包含换行符，而stdout.readline()会一直阻塞，直到读到换行符或流结束，所以第二次调用会卡住。
2. Python子进程的stdout默认是行缓冲模式，没有换行的输出会暂时存在缓冲区，不会立即写入管道。

解决方案 #

下面是修改后的完整代码，解决了阻塞问题，同时满足所有需求：

import subprocess
from threading import Thread
import queue

class Runner:
    def __init__(self):
        self.process = subprocess.Popen(
            ["python", "x.py"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,  # 启用文本模式，自动处理编码，不用手动decode
            bufsize=0,   # 无缓冲模式，子进程输出实时写入管道
        )
        self.stdout_queue = queue.Queue()
        # 启动独立线程监听stdout，避免轮询阻塞
        self.stdout_listener = Thread(target=self._listen_stdout)
        self.stdout_listener.daemon = True  # 随主进程退出
        self.stdout_listener.start()

    def _listen_stdout(self):
        # 逐字符读取，不管有没有换行，确保捕获所有输出
        while True:
            char = self.process.stdout.read(1)
            if not char:
                break
            self.stdout_queue.put(char)

    def poll(self):
        # 一次性读取队列中所有积累的输出
        output = ""
        while not self.stdout_queue.empty():
            output += self.stdout_queue.get()
        if output:
            print("got stdout:", output, end="")
        return output

    def give_input(self, text=""):
        self.process.stdin.write(text)
        self.process.stdin.flush()  # 强制刷新输入缓冲，确保子进程立刻收到输入

    def kill(self):
        self.process.kill()
        self.stdout_listener.join(timeout=1)

# 测试逻辑
r = Runner()
# 启动线程等待子进程结束
wait_thread = Thread(target=r.process.wait)
wait_thread.start()

# 第一次轮询获取"hi\n"
r.poll()
# 第二次轮询获取"Your name: "
r.poll()
# 传入输入
r.give_input("hi\n")
# 等待子进程自然结束，再kill确保彻底终止
wait_thread.join(timeout=1)
r.kill()

关键改进说明 #

• text=True+bufsize=0：让子进程的输出实时刷新到管道，不需要等待换行触发缓冲刷新
• 独立监听线程+队列：用逐字符读取替代readline()，彻底避免换行阻塞问题，所有输出（包括无换行的提示）都会被实时捕获
• stdin.flush()：确保输入立刻发送给子进程，不会留在缓冲区中

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