运行shell命令并捕获输出

问题描述 投票:732回答:17

我想编写一个函数来执行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'
python shell subprocess
17个回答
927
投票

这个问题的答案取决于您正在使用的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写入输入,请跳至runPopen部分。如果要执行复杂的shell命令,请参阅本答案末尾的shell=True上的注释。

check_output函数适用于几乎所有仍在广泛使用的Python版本(2.7 +)。2但对于更新版本,它不再是推荐的方法。

Modern versions of Python (3.5 or higher): run

如果您使用的是Python 3.5或更高版本,并且不需要向后兼容性,则建议使用new run function。它为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'

如果要将输入传递给进程qazxsw poi,请将qazxsw poi对象传递给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来运行更复杂的shell命令,如下面的注释所述。

与旧的做事方式相比,这增加了一点复杂性。但我认为这是值得的回报:现在你几乎可以做任何你需要做的result.stdout功能。

Older versions of Python (2.7-3.4): shell=True

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

check_output

它需要与check_output相同的参数(见下文),并返回包含程序输出的字符串。这个答案的开头有一个更详细的用法示例。在Python 3.5及更高版本中,subprocess.check_output(*popenargs, **kwargs) 相当于使用Popencheck_output执行run,并返回check=True属性。

您可以传递stdout=PIPE以确保错误消息包含在返回的输出中 - 但在某些版本的Python中,将stdout传递给stderr=subprocess.STDOUT会导致stderr=subprocess.PIPE。如果不担心安全性,您还可以通过传递check_output来运行更复杂的shell命令,如下面的注释所述。

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

Complex applications & legacy versions of Python (2.6 and below): check_output

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

check_output构造函数接受不带参数的单个命令,或者包含命令作为其第一项的列表,后跟任意数量的参数,每个参数作为列表中的单独项。 Popen可以帮助将字符串解析为格式正确的列表。 Popen对象也接受shlex.split进行进程IO管理和低级配置。

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

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 answerstdout全部设置为stderr(或stdin)以使PIPE完全正常工作。

在极少数情况下,您可能需要复杂的实时输出捕获。 DEVNULL的回答表明了前进的方向,但是如果不仔细使用,communicate以外的方法很容易出现死锁。

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

Notes

1.运行shell命令:communicate参数

通常,每次调用shell=Trueshell=Truerun构造函数都会执行一个程序。这意味着没有花哨的bash式管道。如果要运行复杂的shell命令,可以传递check_output,这三个函数都支持。

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

shell=True

要么

security concerns

直接连接管道的诱惑力很强;抵制它。否则,你可能会看到死锁或者必须做像run(cmd, [stdout=etc...], input=other_output) 这样的hacky事情。

2. Unicode考虑因素

Popen(cmd, [stdout=etc...]).communicate(other_output) 在Python 2中返回一个字符串,但在Python 3中返回一个this对象。如果你还没有,那么值得花一点时间来讨论check_output


3
投票

如果你需要在多个文件上运行shell命令,这对我来说就是这个诀窍。

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

编辑:刚刚看到马克斯佩尔森的解决方案与J.F.塞巴斯蒂安的建议。走在前面,并将其纳入其中。


2
投票

拆分for line in run_command(cmd): print(line) 的初始命令可能会非常棘手和繁琐。

使用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 来帮助自己。

示例命令

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

1
投票

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

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

就像你在bash中运行命令一样


1
投票

如果使用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") python模块,则可以单独处理STDOUT,STDERR和返回命令代码。您可以看到完整命令调用程序实现的示例。当然,如果你愿意,你可以用sh("ls -a") 扩展它。

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

subprocess

0
投票

例如,执行('ls -ahl')区分三个/四个可能的返回和操作系统平台:

  1. 没有输出,但运行成功
  2. 输出空行,运行成功
  3. 运行失败
  4. 输出一些东西,运行成功

功能如下

try..except

0
投票

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

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

181
投票

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

bytes

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

对于适用于Python2和Python3的解决方案,请使用learn about unicode模块:

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

105
投票

像这样的东西:

subprocess

注意,我正在将stderr重定向到stdout,它可能不是你想要的,但我也想要错误消息。

这个函数在它们到来时逐行产生(通常你必须等待子进程完成以获得整个输出)。

对于您的情况,用法将是:

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

63
投票

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 答案没有读取所有行,所以我做了一个版本:

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

用法与接受的答案相同:

Vartec's

44
投票

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

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

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

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

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

19
投票

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

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

希望它有所帮助

注意:此解决方案是Python3特定的,因为os.remove('tmp') 在Python2中不起作用


15
投票

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

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

注意:自python 2.6以来不推荐使用它。现在你必须使用subprocess.getoutput()。以下是示例

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

11
投票

你的里程可能会变化,我试图在@ 2.6中使用@remitle在Windows上的Vartec解决方案,但我遇到了错误,没有其他解决方案有效。我的错误是:subprocess.Popen

我发现我必须将PIPE分配给每个句柄以使其返回我期望的输出 - 以下对我有效。

import subprocess

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

并且这样调用,(WindowsError: [Error 6] The handle is invalid获取元组的第一个元素,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]

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

要停止控制台弹出窗口(使用Windows),请执行以下操作:

stdout

9
投票

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

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

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

run_command('tracert 11.1.0.1')[0]

此代码将与先前的答案执行相同:

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')
© www.soinside.com 2019 - 2024. All rights reserved.