运行shell命令并捕获输出

Question

我想编写一个函数来执行shell命令并将其输出作为字符串返回,无论是错误还是成功消息.我只想获得与命令行相同的结果.

什么是代码示例会做这样的事情？

例如:

def run_command(cmd):
    # ??????

print run_command('mysqladmin create test -uroot -pmysqladmin12')
# Should output something like:
# mysqladmin: CREATE DATABASE failed; error: 'Can't create database 'test'; database exists'

Answer 1

这个问题的答案取决于您正在使用的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版本(2.7+).²但是对于更新的版本,它不再是推荐的方法.

现代版本的Python(3.5或更高版本): run

如果您使用的是Python 3.5或更高版本,并且不需要向后兼容性,则建议使用新run功能.它为subprocess模块提供了非常通用的高级API .要捕获程序的输出,请将subprocess.PIPE标志传递给stdout关键字参数.然后访问stdout返回CompletedProcess对象的属性:

>>> 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']
>>> input = 'foo\nfoofoo\n'.encode('utf-8')
>>> result = subprocess.run(cmd, stdout=subprocess.PIPE, input=input)
>>> result.stdout.decode('utf-8')
'foofoo\n'

您可以通过传递stderr=subprocess.PIPE(捕获到result.stderr)或stderr=subprocess.STDOUT(捕获到result.stdout常规输出)来捕获错误.如果不担心安全性,您还可以shell=True按照下面的说明中的说明运行更复杂的shell命令.

与旧的做事方式相比,这增加了一点复杂性.但我认为值得付出代价:现在你几乎可以做任何你需要做的run事情.

旧版本的Python(2.7-3.4): check_output

如果您使用的是旧版本的Python,或者需要适度的向后兼容性,则可以使用check_output上面简要描述的功能.它自Python 2.7开始提供.

subprocess.check_output(*popenargs, **kwargs)  

它需要与Popen(见下文)相同的参数,并返回包含程序输出的字符串.这个答案的开头有一个更详细的用法示例.

您可以传递check_output以确保返回的输出中包含错误消息 - 但不传递run给check=True.它可能导致死锁.如果不担心安全性,您还可以stdout=PIPE按照下面的说明中的说明运行更复杂的shell命令.

如果您需要管道stdout输入或将输入传递给流程,stderr=subprocess.STDOUT则无法完成任务.stderr=subprocess.PIPE在这种情况下,请参阅下面的示例.

复杂的应用程序和Python的旧版本(2.6及以下版本): check_output

如果您需要深度向后兼容性,或者您需要比shell=True提供的功能更复杂的功能,则必须直接使用stderr对象,这些对象封装了子进程的低级API.

所述check_output构造器接受单个命令没有参数,或列表包含指令作为其第一项,其次是任意数量的参数,每个作为列表一个单独的项目.Popen可以帮助将字符串解析为适当格式化的列表.Popen对象还接受一系列不同的参数,用于进程IO管理和低级配置.

要发送输入和捕获输出,check_output几乎总是首选方法.如:

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

如果设置Popen,Popen还允许您通过以下方式将数据传递给流程shlex.split:

>>> 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

注意艾伦·霍尔的回答,这表明在某些系统上,你可能需要设置Popen,communicate以及stdin=PIPE所有communicate(或stdin)来获得stdout在所有的工作.

在极少数情况下,您可能需要复杂的实时输出捕获.Vartec的回答表明了前进的方向,但stderr如果不仔细使用,除了容易出现死锁之外的其他方法.

与上述所有功能一样,当安全性不是问题时,您可以通过传递来运行更复杂的shell命令stdin.

笔记

1.运行shell命令:PIPE参数

通常情况下,每次调用DEVNULL,communicate或communicate构造函数执行一个程序.这意味着没有花哨的bash式管道.如果要运行复杂的shell命令,可以传递shell=True,这三个函数都支持.

但是,这样做会引起安全问题.如果你做的不仅仅是轻量级脚本,你可能最好分别调用每个进程,并将每个进程的输出作为输入传递给下一个进程,通过

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

要么

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

直接连接管道的诱惑力很强; 抵制它.否则,你可能会看到死锁或有做这样的事情哈克这样.

2. Unicode考虑因素

shell=True在Python 2中返回一个字符串,但run在Python 3中返回一个对象.如果你还没有,那么值得花一点时间来学习unicode.

Answer 2

这样更容易,但只适用于Unix(包括Cygwin).

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

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

这仅适用于subprocess:它不可用subprocess.对于兼容两者的解决方案,请使用subprocess模块:

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

Answer 3

像这样的东西:

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,

Answer 4

Vartec的答案并没有读取所有行,所以我做了一个版本:

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)

Answer 5

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

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

使用命令的输出创建临时文件(此处为tmp),您可以从中读取所需的输出.

注释中的额外注释:您可以在一次性作业的情况下删除tmp文件.如果您需要多次执行此操作,则无需删除tmp.

os.remove('tmp')

Answer 6

我遇到了同样的问题但是想出了一个非常简单的方法

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

希望它有所帮助

注意:此解决方案是特定于python3的,因为subprocess.getoutput()在python2中不起作用

Answer 7

在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))

Answer 8

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

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

Answer 9

你的里程可能会变化,我试图在@ 2.6中使用@remitle在Windows上的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]

在了解更多之后,我相信我需要这些管道参数,因为我正在使用不同句柄的自定义系统,所以我必须直接控制所有std.

要停止控制台弹出窗口(使用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')

Answer 10

我对以下要求的问题略有不同:

1. 捕获并返回STDOUT消息,因为它们在STDOUT缓冲区中累积(即实时).
   • @vartec通过使用生成器和
     上面的'yield' 关键字来解决这个问题
2. 打印所有STDOUT行(即使在完全读取STDOUT缓冲区之前进程退出)
3. 不要浪费CPU周期以高频率轮询过程
4. 检查子进程的返回码
5. 如果我们得到非零错误返回码,则打印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)

Answer 11

拆分初始命令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()代码将如下所示

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...'

Answer 12

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

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

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

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

Answer 13

由于某种原因，这个可以在 Python 2.7 上运行，你只需要导入 os!

import os 

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

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

Answer 14

如果您需要在多个文件上运行 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

编辑：刚刚根据 JF Sebastian 的建议看到了 Max Persson 的解决方案。继续前进并将其纳入其中。

Answer 15

根据@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 中运行命令完全相同