问题:运行shell命令并捕获输出
我想编写一个函数,它将执行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'
回答 0
这个问题的答案取决于您使用的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
运行一个仅接受参数作为输入的程序。1它完全返回打印到的结果stdout
。如果您需要将输入内容写入stdin
,请跳至run
或Popen
部分。如果要执行复杂的Shell命令,请参阅shell=True
此答案末尾的注释。
该check_output
功能适用于仍在广泛使用的几乎所有版本的Python(2.7+)。2但对于较新的版本,不再推荐使用此方法。
现代版本的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
(请参见下文)相同的参数,并返回一个包含程序输出的字符串。该答案的开头有一个更详细的用法示例。在Python 3.5及更高版本中,check_output
等效于run
使用check=True
和stdout=PIPE
,仅返回stdout
属性。
您可以通过stderr=subprocess.STDOUT
确保错误信息包含在返回的输出-但在Python中通过一些版本stderr=subprocess.PIPE
,以check_output
可引起死锁。如果不关心安全性,您还可以shell=True
按照下面的说明通过传递来运行更复杂的Shell命令。
如果您需要通过管道stderr
传递输入或将输入传递给流程,check_output
则将无法完成任务。Popen
在这种情况下,请参见下面的示例。
复杂的应用程序和Python的旧版(2.6及以下版本): Popen
如果需要深入的向后兼容性,或者需要比check_output
提供的功能更复杂的功能,则必须直接使用Popen
对象,这些对象封装了用于子流程的低级API。
所述Popen
构造器接受单个命令没有参数,或列表包含指令作为其第一项,其次是任意数量的参数,每个作为列表一个单独的项目。shlex.split
可以帮助将字符串解析为格式正确的列表。Popen
对象还接受用于进程IO管理和低级配置的许多不同参数。
发送输入和捕获输出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
注艾伦·霍尔的回答,这表明在某些系统上,你可能需要设置stdout
,stderr
以及stdin
所有PIPE
(或DEVNULL
)得到communicate
工作的。
在极少数情况下,您可能需要复杂的实时输出捕获。Vartec的答案提出了一条前进的道路,但是communicate
如果不谨慎使用,则其他方法都容易出现死锁。
与上述所有功能一样,当不考虑安全性时,可以通过传递运行更复杂的Shell命令shell=True
。
笔记
1.运行shell命令:shell=True
参数
通常,对run
,check_output
或Popen
构造函数的每次调用都会执行一个程序。这意味着没有花哨的bash风格的管道。如果要运行复杂的Shell命令,则可以传递shell=True
,这三个功能都支持。
但是,这样做会引起安全问题。如果您要做的不仅仅是轻脚本编写,那么最好单独调用每个进程,并将每个进程的输出作为输入通过以下方式传递给下一个进程:
run(cmd, [stdout=etc...], input=other_output)
要么
Popen(cmd, [stdout=etc...]).communicate(other_output)
直接连接管道的诱惑力很强;抵抗它。否则,您很可能会遇到僵局,或者不得不执行类似此类的骇人行为。
2. Unicode注意事项
check_output
在Python 2中返回一个字符串,但bytes
在Python 3中返回一个对象。如果您还没有花时间学习unicode,那么值得花一点时间。
回答 1
这很容易,但仅适用于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
回答 2
像这样:
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,
回答 3
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)
回答 4
这是一个棘手但超级简单的解决方案,可在许多情况下使用:
import os
os.system('sample_cmd > tmp')
print open('tmp', 'r').read()
使用命令的输出创建一个临时文件(这里是tmp),您可以从中读取所需的输出。
注释中的额外说明:如果是一次性作业,则可以删除tmp文件。如果您需要多次执行此操作,则无需删除tmp。
os.remove('tmp')
回答 5
我遇到了同样的问题,但是想出了一种非常简单的方法:
import subprocess
output = subprocess.getoutput("ls -l")
print(output)
希望能帮上忙
注意:此解决方案特定subprocess.getoutput()
于Python3,因为在Python2中不起作用
回答 6
您可以使用以下命令来运行任何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")
回答 7
您的里程可能会有所不同,我尝试使用@senderle在Windows 2.6.5上的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')
回答 8
对于以下问题,我对同一问题的口味略有不同:
- 当STDOUT消息在STDOUT缓冲区中累积时(即实时)捕获并返回它们。
- @vartec通过使用生成器和
上面的’yield’ 关键字以Python方式解决了这个问题
- @vartec通过使用生成器和
- 打印所有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)
回答 9
拆分初始命令 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...'
回答 10
如果您需要在多个文件上运行一个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的解决方案。继续前进,并纳入。
回答 11
根据@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中运行命令一样
回答 12
如果您使用 subprocess
python模块,则可以分别处理STDOUT,STDERR和命令的返回代码。您可以看到完整的命令调用程序实现的示例。当然,您可以根据需要扩展它try..except
。
下面的函数返回STDOUT,STDERR和Return代码,因此您可以在其他脚本中处理它们。
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
回答 13
例如,execute(’ls -ahl’)区分了三种/四种可能的收益和OS平台:
- 无输出,但运行成功
- 输出空行,运行成功
- 运行失败
- 输出一些东西,成功运行
功能如下
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 ""
回答 14
可以将输出重定向到文本文件,然后将其读回。
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))
回答 15
刚刚写了一个小的bash脚本来使用curl做到这一点
https://gist.github.com/harish2704/bfb8abece94893c53ce344548ead8ba5
#!/usr/bin/env bash
# Usage: gdrive_dl.sh <url>
urlBase='https://drive.google.com'
fCookie=tmpcookies
curl="curl -L -b $fCookie -c $fCookie"
confirm(){
$curl "$1" | grep jfk-button-action | sed -e 's/.*jfk-button-action" href="\(\S*\)".*/\1/' -e 's/\&/\&/g'
}
$curl -O -J "${urlBase}$(confirm $1)"