从 python 子进程获取输出并向其发出命令

Question

我试图从子进程获取输出，然后根据前面的输出向该进程发出命令。当程序需要进一步输入时，我需要多次执行此操作。 （如果可能的话，我还需要能够隐藏子进程命令提示符）。

我认为这将是一项简单的任务，因为我在 2003 年的帖子中看到过这个问题，现在已经快到 2012 年了，而且它似乎是一个非常普遍的需求，而且看起来它确实应该成为任何编程语言的基本部分。显然我错了，不知何故，差不多 9 年后，仍然没有标准方法以稳定、非破坏性、独立于平台的方式完成这项任务！

我不太了解文件 i/o 和缓冲或线程，所以我更喜欢一个尽可能简单的解决方案。如果有一个与 python 3.x 兼容的模块可以完成此任务，我将非常愿意下载它。我意识到有多个问题基本上询问同一件事，但我尚未找到解决我试图完成的简单任务的答案。

这是我迄今为止基于各种来源的代码；但我完全不知道下一步该做什么。我所有的尝试都以失败告终，有些尝试使用了 100% 的 CPU（基本上什么也不做）并且不会退出。

import subprocess
from subprocess import Popen, PIPE
p = Popen(r'C:\postgis_testing\shellcomm.bat',stdin=PIPE,stdout=PIPE,stderr=subprocess.STDOUT shell=True)
stdout,stdin = p.communicate(b'command string')

如果我的问题不清楚，我将发布示例批处理文件的文本，该文件演示了需要向子进程发送多个命令的情况（如果您键入不正确的命令字符串，程序将循环）。

@echo off
:looper
set INPUT=
set /P INPUT=Type the correct command string:
if "%INPUT%" == "command string" (echo you are correct) else (goto looper)

如果有人能帮助我，我将非常感激，我相信许多其他人也会如此！

这里编辑是使用 eryksun 的代码（下一篇文章）的功能代码：

import subprocess
import threading
import time
import sys

try: 
    import queue
except ImportError:
    import Queue as queue

def read_stdout(stdout, q, p):
    it = iter(lambda: stdout.read(1), b'')
    for c in it:
        q.put(c)
        if stdout.closed:
            break

_encoding = getattr(sys.stdout, 'encoding', 'latin-1')
def get_stdout(q, encoding=_encoding):
    out = []
    while 1:
        try:
            out.append(q.get(timeout=0.2))
        except queue.Empty:
            break
    return b''.join(out).rstrip().decode(encoding)

def printout(q):
    outdata = get_stdout(q)
    if outdata:
        print('Output: %s' % outdata)

if __name__ == '__main__':
    #setup
    p = subprocess.Popen(['shellcomm.bat'], stdin=subprocess.PIPE, 
                     stdout=subprocess.PIPE, stderr=subprocess.PIPE, 
                     bufsize=0, shell=True) # I put shell=True to hide prompt
    q = queue.Queue()
    encoding = getattr(sys.stdin, 'encoding', 'utf-8')

    #for reading stdout
    t = threading.Thread(target=read_stdout, args=(p.stdout, q, p))
    t.daemon = True
    t.start()

    #command loop
    while p.poll() is None:
        printout(q)
        cmd = input('Input: ')
        cmd = (cmd + '\n').encode(encoding)
        p.stdin.write(cmd)
        time.sleep(0.1) # I added this to give some time to check for closure (otherwise it doesn't work)

    #tear down
    for n in range(4):
        rc = p.poll()
        if rc is not None:
            break
        time.sleep(0.25)
    else:
        p.terminate()
        rc = p.poll()
        if rc is None:
            rc = 1

    printout(q)
    print('Return Code: %d' % rc)

但是，当从命令提示符运行脚本时，会发生以下情况：

C:\Users\username>python C:\postgis_testing\shellcomm7.py
Input: sth
Traceback (most recent call last):
File "C:\postgis_testing\shellcomm7.py", line 51, in <module>
    p.stdin.write(cmd)
IOError: [Errno 22] Invalid argument

从命令提示符运行时，程序似乎会关闭。有什么想法吗？

Answer 1

该演示使用专用线程从标准输出读取。如果您四处搜索，我相信您可以找到用面向对象接口编写的更完整的实现。至少我可以说这对于我在 Python 2.7.2 和 3.2.2 中提供的批处理文件都有效。

shellcomm.bat：

@echo off
echo Command Loop Test
echo.
:looper
set INPUT=
set /P INPUT=Type the correct command string:
if "%INPUT%" == "command string" (echo you are correct) else (goto looper)

这是我根据命令“错误”、“仍然错误”和“命令字符串”序列得到的输出：

Output:
Command Loop Test

Type the correct command string:
Input: wrong
Output:
Type the correct command string:
Input: still wrong
Output:
Type the correct command string:
Input: command string
Output:
you are correct

Return Code: 0

为了读取管道输出，readline有时可能会起作用，但是批处理文件中的set /P INPUT自然不会写入行结尾。因此，我使用 lambda: stdout.read(1) 一次读取一个字节（效率不高，但有效）。读取函数将数据放入队列中。主线程在写入命令后从队列中获取输出。此处对 get 调用使用超时会使其等待一小段时间，以确保程序正在等待输入。相反，您可以检查输出中的提示，以了解程序何时需要输入。

尽管如此，您不能期望这样的设置能够普遍工作，因为您尝试与之交互的控制台程序可能会在通过管道传输时缓冲其输出。在 Unix 系统中，有一些可用的实用命令，您可以将它们插入管道中以将缓冲修改为非缓冲、行缓冲或给定大小 - 例如 stdbuf。还有一些方法可以欺骗程序认为它已连接到 pty（请参阅 pexpect）。但是，如果您无法访问程序的源代码来使用 setvbuf。

import subprocess
import threading
import time
import sys

if sys.version_info.major >= 3:
    import queue
else:
    import Queue as queue
    input = raw_input

def read_stdout(stdout, q):
    it = iter(lambda: stdout.read(1), b'')
    for c in it:
        q.put(c)
        if stdout.closed:
            break

_encoding = getattr(sys.stdout, 'encoding', 'latin-1')
def get_stdout(q, encoding=_encoding):
    out = []
    while 1:
        try:
            out.append(q.get(timeout=0.2))
        except queue.Empty:
            break
    return b''.join(out).rstrip().decode(encoding)

def printout(q):
    outdata = get_stdout(q)
    if outdata:
        print('Output:\n%s' % outdata)

if __name__ == '__main__':

    ARGS = ["shellcomm.bat"]   ### Modify this

    #setup
    p = subprocess.Popen(ARGS, bufsize=0, stdin=subprocess.PIPE, 
                         stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    q = queue.Queue()
    encoding = getattr(sys.stdin, 'encoding', 'utf-8')

    #for reading stdout
    t = threading.Thread(target=read_stdout, args=(p.stdout, q))
    t.daemon = True
    t.start()

    #command loop
    while 1:
        printout(q)
        if p.poll() is not None or p.stdin.closed:
            break
        cmd = input('Input: ') 
        cmd = (cmd + '\n').encode(encoding)
        p.stdin.write(cmd)

    #tear down
    for n in range(4):
        rc = p.poll()
        if rc is not None:
            break
        time.sleep(0.25)
    else:
        p.terminate()
        rc = p.poll()
        if rc is None:
            rc = 1

    printout(q)
    print('\nReturn Code: %d' % rc)