运行 shell 命令并捕获输出

在所有官方维护的 Python 版本中，最简单的方法是使用 subprocess.check_output 函数：

>>> subprocess.check_output(['ls', '-l'])
b'total 0\n-rw-r--r--  1 memyself  staff  0 Mar 14 11:04 files\n'

check_output 运行一个仅接受参数作为输入的程序。¹它返回的结果与打印到 stdout 的结果完全相同。如果您需要将输入写入 stdin，请跳至 run 或 Popen 部分。如果要执行复杂的 shell 命令，请参阅本答案末尾的 shell=True 注释。

check_output 函数适用于所有官方维护的 Python 版本。但是对于更新的版本，可以使用更灵活的方法。

现代版本的 Python（3.5 或更高版本）：运行

如果您使用的是 Python 3.5+，并且不需要向后兼容，官方文档推荐新的 run 函数用于大多数任务。它为 subprocess 模块提供了一个非常通用的高级 API。要捕获程序的输出，请将 subprocess.PIPE 标志传递给 stdout 关键字参数。然后访问返回的 CompletedProcess 对象的 stdout 属性：

>>> import subprocess
>>> result = subprocess.run(['ls', '-l'], stdout=subprocess.PIPE)
>>> result.stdout
b'total 0\n-rw-r--r--  1 memyself  staff  0 Mar 14 11:04 files\n'

返回值是一个 bytes 对象，因此如果您想要一个正确的字符串，则需要 decode 它。假设被调用的进程返回一个 UTF-8 编码的字符串：

>>> result.stdout.decode('utf-8')
'total 0\n-rw-r--r--  1 memyself  staff  0 Mar 14 11:04 files\n'

如果需要，这可以全部压缩为单行：

>>> subprocess.run(['ls', '-l'], stdout=subprocess.PIPE).stdout.decode('utf-8')
'total 0\n-rw-r--r--  1 memyself  staff  0 Mar 14 11:04 files\n'

如果要将输入传递给进程的 stdin，可以将 bytes 对象传递给 input 关键字参数：

>>> cmd = ['awk', 'length($0) > 5']
>>> ip = 'foo\nfoofoo\n'.encode('utf-8')
>>> result = subprocess.run(cmd, stdout=subprocess.PIPE, input=ip)
>>> result.stdout.decode('utf-8')
'foofoo\n'

您可以通过传递 stderr=subprocess.PIPE（捕获到 result.stderr）或 stderr=subprocess.STDOUT（捕获到 result.stdout 以及常规输出）来捕获错误。如果您希望 run 在进程返回非零退出代码时引发异常，则可以传递 check=True。 （或者您可以检查上面 result 的 returncode 属性。）当安全不是问题时，您还可以通过传递 shell=True 来运行更复杂的 shell 命令，如本答案末尾所述。

更高版本的 Python 进一步简化了上述内容。在 Python 3.7+ 中，上面的 one-liner 可以这样拼写：

>>> subprocess.run(['ls', '-l'], capture_output=True, text=True).stdout
'total 0\n-rw-r--r--  1 memyself  staff  0 Mar 14 11:04 files\n'

与旧的做事方式相比，以这种方式使用 run 只会增加一点复杂性。但是现在您几乎可以仅使用 run 函数来完成任何您需要做的事情。

旧版本的 Python (3-3.4)：更多关于 check_output

如果您使用的是旧版本的 Python，或者需要适度的向后兼容性，则可以使用上面简要描述的 check_output 函数。它从 Python 2.7 开始可用。

subprocess.check_output(*popenargs, **kwargs)  

它采用与 Popen 相同的参数（见下文），并返回一个包含程序输出的字符串。这个答案的开头有一个更详细的用法示例。在 Python 3.5+ 中，check_output 等效于使用 check=True 和 stdout=PIPE 执行 run，并且只返回 stdout 属性。

您可以传递 stderr=subprocess.STDOUT 以确保返回的输出中包含错误消息。当安全不是问题时，您还可以通过传递 shell=True 来运行更复杂的 shell 命令，如本答案末尾所述。

如果您需要从 stderr 进行管道传输或将输入传递给进程，check_output 将无法完成任务。在这种情况下，请参阅下面的 Popen 示例。

复杂的应用程序和 Python 的旧版本（2.6 及以下）：Popen

如果您需要深度向后兼容性，或者如果您需要比 check_output 或 run 提供的更复杂的功能，则必须直接使用 Popen 对象，它封装了子进程的低级 API。

Popen 构造函数接受不带参数的单个命令，或包含命令作为其第一项的列表，后跟任意数量的参数，每个参数作为单独的项在列表中。 shlex.split 可以帮助将字符串解析为格式适当的列表。 Popen 对象还接受用于进程 IO 管理和低级配置的 host of different arguments。

要发送输入和捕获输出，communicate 几乎总是首选方法。如：

output = subprocess.Popen(["mycmd", "myarg"], 
                          stdout=subprocess.PIPE).communicate()[0]

或者

>>> import subprocess
>>> p = subprocess.Popen(['ls', '-a'], stdout=subprocess.PIPE, 
...                                    stderr=subprocess.PIPE)
>>> out, err = p.communicate()
>>> print out
.
..
foo

如果您设置 stdin=PIPE，则 communicate 还允许您通过 stdin 将数据传递给进程：

>>> cmd = ['awk', 'length($0) > 5']
>>> p = subprocess.Popen(cmd, stdout=subprocess.PIPE,
...                           stderr=subprocess.PIPE,
...                           stdin=subprocess.PIPE)
>>> out, err = p.communicate('foo\nfoofoo\n')
>>> print out
foofoo

注意 Aaron Hall's answer，它表示在某些系统上，您可能需要将 stdout、stderr 和 stdin all 设置为 PIPE（或 DEVNULL）才能使 communicate 完全正常工作。

在极少数情况下，您可能需要复杂的实时输出捕获。 Vartec 的回答提出了一条前进的道路，但如果不小心使用，communicate 以外的方法很容易出现死锁。

与上述所有函数一样，当不考虑安全性时，您可以通过传递 shell=True 来运行更复杂的 shell 命令。

笔记

<强> 1。运行 shell 命令：shell=True 参数

通常，对 run、check_output 或 Popen 构造函数的每次调用都会执行一个单个程序。这意味着没有花哨的 bash 风格的管道。如果要运行复杂的 shell 命令，可以传递 shell=True，这三个函数都支持。例如：

>>> subprocess.check_output('cat books/* | wc', shell=True, text=True)
' 1299377 17005208 101299376\n'

但是，这样做会引发 security concerns。如果你做的不仅仅是简单的脚本，你最好分别调用每个进程，并将每个进程的输出作为输入传递给下一个，通过

run(cmd, [stdout=etc...], input=other_output)

或者

Popen(cmd, [stdout=etc...]).communicate(other_output)

直接连接管道的诱惑力很强；抵制它。否则，您可能会看到死锁或不得不做 this 之类的骇人听闻的事情。

---

这更容易，但仅适用于 Unix（包括 Cygwin）和 Python2.7。

import commands
print commands.getstatusoutput('wc -l file')

它返回一个带有 (return_value, output) 的元组。

对于同时适用于 Python2 和 Python3 的解决方案，请改用 subprocess 模块：

from subprocess import Popen, PIPE
output = Popen(["date"],stdout=PIPE)
response = output.communicate()
print response

---

像这样的东西：

def runProcess(exe):    
    p = subprocess.Popen(exe, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    while(True):
        # returns None while subprocess is running
        retcode = p.poll() 
        line = p.stdout.readline()
        yield line
        if retcode is not None:
            break

请注意，我将 stderr 重定向到 stdout，它可能不是您想要的，但我也想要错误消息。

此函数会在它们出现时逐行生成（通常您必须等待子进程完成才能获得整个输出）。

对于您的情况，用法是：

for line in runProcess('mysqladmin create test -uroot -pmysqladmin12'.split()):
    print line,

---

我有同样的问题，但想出了一个非常简单的方法来做到这一点：

import subprocess
output = subprocess.getoutput("ls -l")
print(output)

希望对您有所帮助

注意：此解决方案是 Python3 特定的，因为 subprocess.getoutput() 在 Python2 中不起作用

---

这是一个棘手但超级简单的解决方案，适用于许多情况：

import os
os.system('sample_cmd > tmp')
print(open('tmp', 'r').read())

使用命令的输出创建一个临时文件（这里是 tmp），您可以从中读取所需的输出。

评论中的额外说明：您可以在一次性作业的情况下删除 tmp 文件。如果您需要多次执行此操作，则无需删除 tmp。

os.remove('tmp')

---

Vartec's 答案没有读完所有行，所以我制作了一个版本：

def run_command(command):
    p = subprocess.Popen(command,
                         stdout=subprocess.PIPE,
                         stderr=subprocess.STDOUT)
    return iter(p.stdout.readline, b'')

用法与接受的答案相同：

command = 'mysqladmin create test -uroot -pmysqladmin12'.split()
for line in run_command(command):
    print(line)

---

您可以使用以下命令运行任何 shell 命令。我在 ubuntu 上使用过它们。

import os
os.popen('your command here').read()

注意： 自 python 2.6 起已弃用。现在您必须使用 subprocess.Popen。下面是示例

import subprocess

p = subprocess.Popen("Your command", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).communicate()[0]
print p.split("\n")

---

对于以下要求，我对同一问题的看法略有不同：

捕获并返回 STDOUT 消息，因为它们在 STDOUT 缓冲区中累积（即实时）。 @vartec 通过使用生成器和上面的 'yield' 关键字以 Python 方式解决了这个问题 打印所有 STDOUT 行（即使进程在 STDOUT 缓冲区可以完全读取之前退出）不要浪费 CPU 周期以高频轮询进程 检查返回如果我们得到一个非零错误返回码，则打印子进程的代码打印 STDERR（与 STDOUT 分开）。

我结合并调整了以前的答案，得出以下结论：

import subprocess
from time import sleep

def run_command(command):
    p = subprocess.Popen(command,
                         stdout=subprocess.PIPE,
                         stderr=subprocess.PIPE,
                         shell=True)
    # Read stdout from subprocess until the buffer is empty !
    for line in iter(p.stdout.readline, b''):
        if line: # Don't print blank lines
            yield line
    # This ensures the process has completed, AND sets the 'returncode' attr
    while p.poll() is None:                                                                                                                                        
        sleep(.1) #Don't waste CPU-cycles
    # Empty STDERR buffer
    err = p.stderr.read()
    if p.returncode != 0:
       # The run_command() function is responsible for logging STDERR 
       print("Error: " + str(err))

此代码的执行方式与之前的答案相同：

for line in run_command(cmd):
    print(line)

---

您的里程可能会有所不同，我在 Python 2.6.5 的 Windows 中尝试了 @senderle 的 Vartec 解决方案，但我遇到了错误，并且没有其他解决方案有效。我的错误是：WindowsError: [Error 6] The handle is invalid。

我发现我必须将 PIPE 分配给每个句柄才能让它返回我期望的输出 - 以下对我有用。

import subprocess

def run_command(cmd):
    """given shell command, returns communication tuple of stdout and stderr"""
    return subprocess.Popen(cmd, 
                            stdout=subprocess.PIPE, 
                            stderr=subprocess.PIPE, 
                            stdin=subprocess.PIPE).communicate()

并像这样调用，（[0] 获取元组的第一个元素，stdout）：

run_command('tracert 11.1.0.1')[0]

在了解更多之后，我相信我需要这些管道参数，因为我正在开发一个使用不同句柄的自定义系统，所以我必须直接控制所有标准。

要停止控制台弹出窗口（使用 Windows），请执行以下操作：

def run_command(cmd):
    """given shell command, returns communication tuple of stdout and stderr"""
    # instantiate a startupinfo obj:
    startupinfo = subprocess.STARTUPINFO()
    # set the use show window flag, might make conditional on being in Windows:
    startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
    # pass as the startupinfo keyword argument:
    return subprocess.Popen(cmd,
                            stdout=subprocess.PIPE, 
                            stderr=subprocess.PIPE, 
                            stdin=subprocess.PIPE, 
                            startupinfo=startupinfo).communicate()

run_command('tracert 11.1.0.1')

---

在 Python 3.7+ 上，使用 subprocess.run 并传递 capture_output=True：

import subprocess
result = subprocess.run(['echo', 'hello', 'world'], capture_output=True)
print(repr(result.stdout))

这将返回字节：

b'hello world\n'

如果您希望它将字节转换为字符串，请添加 text=True：

result = subprocess.run(['echo', 'hello', 'world'], capture_output=True, text=True)
print(repr(result.stdout))

这将使用您的默认编码读取字节：

'hello world\n'

如果您需要手动指定不同的编码，请使用 encoding="your encoding" 而不是 text=True：

result = subprocess.run(['echo', 'hello', 'world'], capture_output=True, encoding="utf8")
print(repr(result.stdout))

---

拆分 subprocess 的初始命令可能既棘手又麻烦。

使用 shlex.split() 帮助自己。

示例命令

git log -n 5 --since "5 years ago" --until "2 year ago"

编码

from subprocess import check_output
from shlex import split

res = check_output(split('git log -n 5 --since "5 years ago" --until "2 year ago"'))
print(res)
>>> b'commit 7696ab087a163e084d6870bb4e5e4d4198bdc61a\nAuthor: Artur Barseghyan...'

如果没有 shlex.split()，代码将如下所示

res = check_output([
    'git', 
    'log', 
    '-n', 
    '5', 
    '--since', 
    '5 years ago', 
    '--until', 
    '2 year ago'
])
print(res)
>>> b'commit 7696ab087a163e084d6870bb4e5e4d4198bdc61a\nAuthor: Artur Barseghyan...'

---

这是一个解决方案，如果您想在进程运行时打印输出，则可以使用。

我还添加了当前工作目录，它不止一次对我有用。

希望解决方案能帮助某人:)。

import subprocess

def run_command(cmd_and_args, print_constantly=False, cwd=None):
"""Runs a system command.

:param cmd_and_args: the command to run with or without a Pipe (|).
:param print_constantly: If True then the output is logged in continuous until the command ended.
:param cwd: the current working directory (the directory from which you will like to execute the command)
:return: - a tuple containing the return code, the stdout and the stderr of the command
"""
output = []

process = subprocess.Popen(cmd_and_args, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd)

while True:
    next_line = process.stdout.readline()
    if next_line:
        output.append(str(next_line))
        if print_constantly:
            print(next_line)
    elif not process.poll():
        break

error = process.communicate()[1]

return process.returncode, '\n'.join(output), error

---

出于某种原因，这个适用于 Python 2.7，您只需要导入 os!

import os 

def bash(command):
    output = os.popen(command).read()
    return output

print_me = bash('ls -l')
print(print_me)

---

如果您需要对多个文件运行 shell 命令，这对我有用。

import os
import subprocess

# Define a function for running commands and capturing stdout line by line
# (Modified from Vartec's solution because it wasn't printing all lines)
def runProcess(exe):    
    p = subprocess.Popen(exe, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    return iter(p.stdout.readline, b'')

# Get all filenames in working directory
for filename in os.listdir('./'):
    # This command will be run on each file
    cmd = 'nm ' + filename

    # Run the command and capture the output line by line.
    for line in runProcess(cmd.split()):
        # Eliminate leading and trailing whitespace
        line.strip()
        # Split the output 
        output = line.split()

        # Filter the output and print relevant lines
        if len(output) > 2:
            if ((output[2] == 'set_program_name')):
                print filename
                print line

编辑：刚刚看到 Max Persson 在 JF Sebastian 的建议下的解决方案。继续并将其纳入。

---

根据@senderle，如果你像我一样使用 python3.6：

def sh(cmd, input=""):
    rst = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, input=input.encode("utf-8"))
    assert rst.returncode == 0, rst.stderr.decode("utf-8")
    return rst.stdout.decode("utf-8")

sh("ls -a")

将完全像您在 bash 中运行命令一样

---

改进以获得更好的日志记录。为了获得更好的输出，您可以使用迭代器。从下面，我们变得更好

from subprocess import Popen, getstatusoutput, PIPE
def shell_command(cmd):
    result = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE)

    output = iter(result.stdout.readline, b'')
    error = iter(result.stderr.readline, b'')
    print("##### OutPut ###")
    for line in output:
        print(line.decode("utf-8"))
    print("###### Error ########")
    for line in error:
        print(error.decode("utf-8")) # Convert bytes to str

    status, terminal_output = run_command(cmd)
    print(terminal_output)

shell_command("ls") # this will display all the files & folders in directory

使用 getstatusoutput 的其他方法（易于理解）

from subprocess import Popen, getstatusoutput, PIPE

status_Code, output = getstausoutput(command)
print(output) # this will give the terminal output

# status_code, output = getstatusoutput("ls") # this will print the all files & folder available in the directory

---

如果您使用subprocess python 模块，您可以分别处理 STDOUT、STDERR 和命令返回码。您可以查看完整的命令调用程序实现示例。当然，如果需要，您可以使用 try..except 对其进行扩展。

下面的函数返回 STDOUT、STDERR 和返回代码，以便您可以在其他脚本中处理它们。

import subprocess

def command_caller(command=None)
    sp = subprocess.Popen(command, stderr=subprocess.PIPE, stdout=subprocess.PIPE, shell=False)
    out, err = sp.communicate()
    if sp.returncode:
        print(
            "Return code: %(ret_code)s Error message: %(err_msg)s"
            % {"ret_code": sp.returncode, "err_msg": err}
            )
    return sp.returncode, out, err

---

我想建议将 simppl 作为一个可供考虑的选项。它是一个可通过 pypi: pip install simppl 获得的模块，并在 python3 上运行。

simppl 允许用户运行 shell 命令并从屏幕读取输出。

开发人员建议了三种类型的用例：

最简单的用法如下所示： from simppl.simple_pipeline import SimplePipeline sp = SimplePipeline(start=0, end=100): sp.print_and_run('') sp.print_and_run('') ```

要同时运行多个命令，请使用： commands = ['', ''] max_number_of_processes = 4 sp.run_parallel(commands, max_number_of_processes) ```

最后，如果您的项目使用 cli 模块，您可以直接运行另一个 command_line_tool 作为管道的一部分。另一个工具将在同一进程中运行，但它将作为管道中的另一个命令出现在日志中。这使得调用其他工具的工具能够更顺畅地调试和重构。 from example_module import example_tool sp.print_and_run_clt(example_tool.run, ['first_number', 'second_nmber'], {'-key1': 'val1', '-key2': 'val2'}, {'--flag'}) ```

请注意，打印到 STDOUT/STDERR 是通过 python 的 logging 模块。

这是一个完整的代码来展示 simppl 是如何工作的：

import logging
from logging.config import dictConfig

logging_config = dict(
    version = 1,
    formatters = {
        'f': {'format':
              '%(asctime)s %(name)-12s %(levelname)-8s %(message)s'}
        },
    handlers = {
        'h': {'class': 'logging.StreamHandler',
              'formatter': 'f',
              'level': logging.DEBUG}
        },
    root = {
        'handlers': ['h'],
        'level': logging.DEBUG,
        },
)
dictConfig(logging_config)

from simppl.simple_pipeline import SimplePipeline
sp = SimplePipeline(0, 100)
sp.print_and_run('ls')

---

这是一个简单灵活的解决方案，适用于各种操作系统版本，包括 Python 2 和 3，在 shell 模式下使用 IPython：

from IPython.terminal.embed import InteractiveShellEmbed
my_shell = InteractiveShellEmbed()
result = my_shell.getoutput("echo hello world")
print(result)

Out: ['hello world']

它有几个优点

它只需要一个 IPython 安装，所以你在使用它时不需要担心你的特定 Python 或操作系统版本，它带有 Jupyter - 它具有广泛的支持它默认需要一个简单的字符串 - 所以不需要使用 shell 模式 arg 或字符串拆分，使其更简洁 IMO 它还使得在字符串本身中轻松替换变量甚至整个 Python 命令变得更简洁

展示：

var = "hello world "
result = my_shell.getoutput("echo {var*2}")
print(result)

Out: ['hello world hello world']

只是想给你一个额外的选择，特别是如果你已经安装了 Jupyter

当然，如果您使用的是实际的 Jupyter notebook 而不是 .py 脚本，您也可以随时执行以下操作：

result = !echo hello world
print(result)

要完成同样的事情。

---

输出可以重定向到文本文件，然后再读回。

import subprocess
import os
import tempfile

def execute_to_file(command):
    """
    This function execute the command
    and pass its output to a tempfile then read it back
    It is usefull for process that deploy child process
    """
    temp_file = tempfile.NamedTemporaryFile(delete=False)
    temp_file.close()
    path = temp_file.name
    command = command + " > " + path
    proc = subprocess.run(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
    if proc.stderr:
        # if command failed return
        os.unlink(path)
        return
    with open(path, 'r') as f:
        data = f.read()
    os.unlink(path)
    return data

if __name__ == "__main__":
    path = "Somepath"
    command = 'ecls.exe /files ' + path
    print(execute(command))

---

例如，execute('ls -ahl') 区分了三个/四个可能的返回和操作系统平台：

没有输出，但运行成功输出空行，运行成功运行失败输出一些东西，运行成功

下面的函数

def execute(cmd, output=True, DEBUG_MODE=False):
"""Executes a bash command.
(cmd, output=True)
output: whether print shell output to screen, only affects screen display, does not affect returned values
return: ...regardless of output=True/False...
        returns shell output as a list with each elment is a line of string (whitespace stripped both sides) from output
        could be 
        [], ie, len()=0 --> no output;    
        [''] --> output empty line;     
        None --> error occured, see below

        if error ocurs, returns None (ie, is None), print out the error message to screen
"""
if not DEBUG_MODE:
    print "Command: " + cmd

    # https://stackoverflow.com/a/40139101/2292993
    def _execute_cmd(cmd):
        if os.name == 'nt' or platform.system() == 'Windows':
            # set stdin, out, err all to PIPE to get results (other than None) after run the Popen() instance
            p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
        else:
            # Use bash; the default is sh
            p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, executable="/bin/bash")

        # the Popen() instance starts running once instantiated (??)
        # additionally, communicate(), or poll() and wait process to terminate
        # communicate() accepts optional input as stdin to the pipe (requires setting stdin=subprocess.PIPE above), return out, err as tuple
        # if communicate(), the results are buffered in memory

        # Read stdout from subprocess until the buffer is empty !
        # if error occurs, the stdout is '', which means the below loop is essentially skipped
        # A prefix of 'b' or 'B' is ignored in Python 2; 
        # it indicates that the literal should become a bytes literal in Python 3 
        # (e.g. when code is automatically converted with 2to3).
        # return iter(p.stdout.readline, b'')
        for line in iter(p.stdout.readline, b''):
            # # Windows has \r\n, Unix has \n, Old mac has \r
            # if line not in ['','\n','\r','\r\n']: # Don't print blank lines
                yield line
        while p.poll() is None:                                                                                                                                        
            sleep(.1) #Don't waste CPU-cycles
        # Empty STDERR buffer
        err = p.stderr.read()
        if p.returncode != 0:
            # responsible for logging STDERR 
            print("Error: " + str(err))
            yield None

    out = []
    for line in _execute_cmd(cmd):
        # error did not occur earlier
        if line is not None:
            # trailing comma to avoid a newline (by print itself) being printed
            if output: print line,
            out.append(line.strip())
        else:
            # error occured earlier
            out = None
    return out
else:
    print "Simulation! The command is " + cmd
    print ""

---