迹忆客 专注技术分享

本文的主要目的是演示如何读取在 Python 中执行的子进程的标准输出。

Python 子进程在运行时读取标准输出

与许多其他内置模块一样，Subprocess 也是一个内置模块，预装了“正常”Python 安装。

它主要用于在新进程中运行任务、进程和程序，执行一组特定任务并返回结果时。

它被广泛使用的众多原因之一是它允许作为一个单独的进程直接从程序中执行外部程序和可执行文件。

使用 Subprocess 库执行程序时，可能需要实时显示该外部程序的输出。 出于多种原因，这可能是一项要求，例如当程序可能是实时的并且依赖于短时间后的计算时。

考虑以下程序：

import subprocess

def execute(command):
    process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    output = process.communicate()[0]
    exitCode = process.returncode

    if (exitCode == 0):
        return output
    else:
        raise Exception(command, exitCode, output)

if __name__ == '__main__':
    print(execute("cd C:\\ && C: && tree").decode('unicode_escape'))

输出:

Folder PATH listing
Volume serial number is 0213-B7F2
C:.
+---DRIVERS
¦   +---bluetooth
¦       +---U1BLT07AVVASDSAP
¦           +---Custom
¦           ¦   +---EULA
¦           +---Win64
¦               +---LD
¦               +---svcpack
+---flutter
¦   +---.git
¦   ¦   +---hooks
¦   ¦   +---info
¦   ¦   +---logs
¦   ¦   ¦   +---refs
¦   ¦   ¦       +---heads
¦   ¦   ¦       +---remotes
¦   ¦   ¦           +---origin
¦   ¦   +---objects
¦   ¦   ¦   +---info
¦   ¦   ¦   +---pack
¦   ¦   +---refs
¦   ¦       +---heads
¦   ¦       +---remotes
¦   ¦       ¦   +---origin
¦   ¦       +---tags
¦   +---.github
¦   ¦   +---ISSUE_TEMPLATE
¦   ¦   +---workflows
¦   +---.idea
¦   ¦   +---runConfigurations
...

我们可以在程序中看到输出，但不是实时的。 只有在整个命令（在我们的例子中是树）执行完毕后才会显示输出。

在使用 Subprocess 命令在单独的进程中执行程序（或命令）之前，输出不会显示任何内容。

在我们的例子中，由于我们需要实时获取输出，因此我们需要创建另一个解决方案，即显示写入标准输出的程序输出的解决方案。

可以通过多种方式来解决该解决方案，下面将提到其中一些。

方法 1：在 Python 中运行时使用 check_call 读取子进程的标准输出

考虑以下代码：

import subprocess
import sys

def execute(command):
    subprocess.check_call(command, shell=True, stdout=sys.stdout, stderr=subprocess.STDOUT)

if __name__ == '__main__':
    print(execute("cd C:\\ && C: && tree").decode('unicode_escape'))

输出:

Folder PATH listing
Volume serial number is 0213-B7F2
C:.
├───DRIVERS
│   └───bluetooth
│       └───U1BLT0720US14CMP
│           ├───Custom
│           │   └───EULA
│           └───Win64
│               ├───LD
│               └───svcpack
├───flutter
│   ├───.git
│   │   ├───hooks
│   │   ├───info
│   │   ├───logs
│   │   │   └───refs
│   │   │       ├───heads
│   │   │       └───remotes
│   │   │           └───origin
│   │   ├───objects
│   │   │   ├───info
│   │   │   └───pack
│   │   └───refs
│   │       ├───heads
│   │       ├───remotes
│   │       │   └───origin
│   │       └───tags
│   ├───.github
│   │   ├───ISSUE_TEMPLATE
│   │   └───workflows
│   ├───.idea
│   │   └───runConfigurations
│   ├───.pub-cache
│   │   ├───hosted
│   │   │   └───pub.dartlang.org
│   │   │       ├───.cache
...

如果主要需求是实时打印程序的输出，可以使用 check_call 来实现。 这个简单、干净和优雅的解决方案对于非常简单的程序来说是简洁和“完美”的，比如只需要打印输出的时候。

check_call 还支持传递参数，因此如果您的程序需要任何参数才能工作，可以轻松地将它们传递给函数而不会遇到任何麻烦。 此方法等待程序完成。

基于程序的完成，方法返回； 否则，它会引发 CalledProcessError 异常。 CalledProcessError 将具有失败返回代码，可以使用 returncode 属性访问该代码。

在上面提到的代码中，命令被传递给 check_call 方法，其中包含要执行的命令（或程序）。

shell 参数设置为 true 以使用 shell 执行进程，进程的 stdout 设置为我们程序的 stdout，因此它直接写入我们的 stdout，我们可以在我们的 stdout 中看到发生的变化。

最后，stderr 被设置为衍生进程的标准输出。

方法 2：在 Python 中运行时轮询进程以读取子进程的标准输出

考虑以下代码：

import subprocess
import sys

def execute(command):
    process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)

    # Poll process for new output until finished
    while True:
        nextline = process.stdout.readline().decode('unicode_escape')
        if nextline == '' and process.poll() is not None:
            break
            sys.stdout.write(nextline)
            sys.stdout.flush()

        output = process.communicate()[0]
        exitCode = process.returncode

        if (exitCode == 0):
            return output
        else:
            raise Exception(command, exitCode, output)

if __name__ == '__main__':
    print(execute("cd C:\\ && C: && tree").decode('unicode_escape'))

输出:

...
¦   ¦   ¦       ¦   ¦       +---UuidUtil
¦   ¦   ¦       ¦   +---example
¦   ¦   ¦       ¦   +---lib
¦   ¦   ¦       ¦   +---test
¦   ¦   ¦       +---vector_math-2.1.2
¦   ¦   ¦       ¦   +---benchmark
¦   ¦   ¦       ¦   +---bin
¦   ¦   ¦       ¦   +---lib
¦   ¦   ¦       ¦   ¦   +---src
¦   ¦   ¦       ¦   ¦       +---vector_math
¦   ¦   ¦       ¦   ¦       ¦   +---third_party
¦   ¦   ¦       ¦   ¦       +---vector_math_64
¦   ¦   ¦       ¦   ¦       ¦   +---third_party
¦   ¦   ¦       ¦   ¦       +---vector_math_geometry
¦   ¦   ¦       ¦   ¦       ¦   +---filters
¦   ¦   ¦       ¦   ¦       ¦   +---generators
¦   ¦   ¦       ¦   ¦       +---vector_math_lists
¦   ¦   ¦       ¦   ¦       +---vector_math_operations
¦   ¦   ¦       ¦   +---test
¦   ¦   ¦       ¦   +---tool
¦   ¦   ¦       +---video_player-2.2.11
¦   ¦   ¦       ¦   +---android
¦   ¦   ¦       ¦   ¦   +---gradle
...

为了确保使用 Subprocess 产生的程序输出在写入标准输出后立即打印出来，我们需要轮询输出过程并继续读取程序标准输出的最后一行标准输出。

在“无限”循环中，我们继续使用 readline() 读取生成进程的输出。 由于输出是编码的，我们需要对其进行解码（在我们的例子中是 utf-escape）以适当地表示。

这可以使用解码方法并传递相关的编码方案来完成。

一旦程序执行完毕，我们还必须跳出“无限”循环。 我们可以通过检查两件事来做到这一点：

1. 当前的标准输出行是空的。
2. 该进程已终止或死亡。

这可以非常简单地完成，对第一个条件进行简单的字符串比较，并使用 poll() 方法检查进程是否已终止。 如果程序已经终止，它返回返回码； 否则，它返回 None。

与 None 的简单比较，也可以告诉我们程序是否已经执行完毕。