我如何在Python中命名一个外部命令,就好像我把它写在一个<unk>或命令中?
当前回答
这里有很多答案,但没有一个满足了我所有的需求。
我需要运行命令并捕获输出和输出代码. 我需要时间out 执行的程序,并强迫它出去,如果时间out 达到,并杀死它的所有儿童过程. 我需要它在 Windows XP 和以后, Cygwin 和 Linux 工作。
def _run(command, timeout_s=False, shell=False):
### run a process, capture the output and wait for it to finish. if timeout is specified then Kill the subprocess and its children when the timeout is reached (if parent did not detach)
## usage: _run(arg1, arg2, arg3)
# arg1: command + arguments. Always pass a string; the function will split it when needed
# arg2: (optional) timeout in seconds before force killing
# arg3: (optional) shell usage. default shell=False
## return: a list containing: exit code, output, and if timeout was reached or not
# - Tested on Python 2 and 3 on Windows XP, Windows 7, Cygwin and Linux.
# - preexec_fn=os.setsid (py2) is equivalent to start_new_session (py3) (works on Linux only), in Windows and Cygwin we use TASKKILL
# - we use stderr=subprocess.STDOUT to merge standard error and standard output
import sys, subprocess, os, signal, shlex, time
def _runPY3(command, timeout_s=None, shell=False):
# py3.3+ because: timeout was added to communicate() in py3.3.
new_session=False
if sys.platform.startswith('linux'): new_session=True
p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, start_new_session=new_session, shell=shell)
try:
out = p.communicate(timeout=timeout_s)[0].decode('utf-8')
is_timeout_reached = False
except subprocess.TimeoutExpired:
print('Timeout reached: Killing the whole process group...')
killAll(p.pid)
out = p.communicate()[0].decode('utf-8')
is_timeout_reached = True
return p.returncode, out, is_timeout_reached
def _runPY2(command, timeout_s=0, shell=False):
preexec=None
if sys.platform.startswith('linux'): preexec=os.setsid
p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, preexec_fn=preexec, shell=shell)
start_time = time.time()
is_timeout_reached = False
while timeout_s and p.poll() == None:
if time.time()-start_time >= timeout_s:
print('Timeout reached: Killing the whole process group...')
killAll(p.pid)
is_timeout_reached = True
break
time.sleep(1)
out = p.communicate()[0].decode('utf-8')
return p.returncode, out, is_timeout_reached
def killAll(ParentPid):
if sys.platform.startswith('linux'):
os.killpg(os.getpgid(ParentPid), signal.SIGTERM)
elif sys.platform.startswith('cygwin'):
# subprocess.Popen(shlex.split('bash -c "TASKKILL /F /PID $(</proc/{pid}/winpid) /T"'.format(pid=ParentPid)))
winpid=int(open("/proc/{pid}/winpid".format(pid=ParentPid)).read())
subprocess.Popen(['TASKKILL', '/F', '/PID', str(winpid), '/T'])
elif sys.platform.startswith('win32'):
subprocess.Popen(['TASKKILL', '/F', '/PID', str(ParentPid), '/T'])
# - In Windows, we never need to split the command, but in Cygwin and Linux we need to split if shell=False (default), shlex will split the command for us
if shell==False and (sys.platform.startswith('cygwin') or sys.platform.startswith('linux')):
command=shlex.split(command)
if sys.version_info >= (3, 3): # py3.3+
if timeout_s==False:
returnCode, output, is_timeout_reached = _runPY3(command, timeout_s=None, shell=shell)
else:
returnCode, output, is_timeout_reached = _runPY3(command, timeout_s=timeout_s, shell=shell)
else: # Python 2 and up to 3.2
if timeout_s==False:
returnCode, output, is_timeout_reached = _runPY2(command, timeout_s=0, shell=shell)
else:
returnCode, output, is_timeout_reached = _runPY2(command, timeout_s=timeout_s, shell=shell)
return returnCode, output, is_timeout_reached
然后用它如下:
总是将命令作为一个行(它更容易)。你不需要分开它;函数在需要时将分开它。
所以我们可以用这个类似的时间:
a=_run('cmd /c echo 11111 & echo 22222 & calc',3)
for i in a[1].splitlines(): print(i)
或者没有时间:
b=_run('cmd /c echo 11111 & echo 22222 & calc')
更多例子:
b=_run('''wmic nic where 'NetConnectionID="Local Area Connection"' get NetConnectionStatus /value''')
print(b)
c=_run('cmd /C netsh interface ip show address "Local Area Connection"')
print(c)
d=_run('printf "<%s>\n" "{foo}"')
print(d)
你也可以指定Shell=True,但在大多数情况下,这个功能是无用的。我宁愿自己选择我想要的Shell,但这里是,如果你也需要它:
# windows
e=_run('echo 11111 & echo 22222 & calc',3, shell=True)
print(e)
# Cygwin/Linux:
f=_run('printf "<%s>\n" "{foo}"', shell=True)
print(f)
為什麼我沒有使用更簡單的新方法 subprocess.run()?
因为它支持在Python 3.7+,但最后支持的Python版本在Windows XP是3.4. 而且由于这个功能的Timeout论点在Windows中是无用的,它不会杀死执行命令的儿童过程. 如果你使用 capture_output + Timeout论点,它会挂在如果有一个儿童过程仍然运行。
其他回答
您可以使用Popen,然后您可以检查程序的状态:
from subprocess import Popen
proc = Popen(['ls', '-l'])
if proc.poll() is None:
proc.kill()
查看 subprocess.Popen。
典型的实施:
import subprocess
p = subprocess.Popen('ls', shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
for line in p.stdout.readlines():
print line,
retval = p.wait()
事实上,你可以简单地忽略这些参数(stdout=和stderr=)并将像os.system()一样行事。
下面是如何呼叫外部程序的概述,包括其优点和缺点:
os.system 将命令和论点转移到您的系统的阴道. 这很好,因为您实际上可以以这种方式同时运行多个命令,并设置管道和输入/输出重定向。 例如: os.system(“some_command < input_file♰ another_command > output_file”) 但是,虽然这是方便的,您必须手动处理阴道字符的逃避,如空间,和
副过程模块应该是你所使用的。
最后,请注意,对于你通过的所有方法,最终命令将由丝带执行,你负责逃避它。 有严重的安全影响,如果你通过的丝带的任何部分不能完全信任。 例如,如果用户进入某些 / 任何部分的丝带. 如果你不确定,只使用这些方法的连续。
print subprocess.Popen("echo %s " % user_input, stdout=PIPE).stdout.read()
想象一下,用户输入了一些“我的妈妈不喜欢我&rm -rf /”可以删除整个文件系统。
这里有很多答案,但没有一个满足了我所有的需求。
我需要运行命令并捕获输出和输出代码. 我需要时间out 执行的程序,并强迫它出去,如果时间out 达到,并杀死它的所有儿童过程. 我需要它在 Windows XP 和以后, Cygwin 和 Linux 工作。
def _run(command, timeout_s=False, shell=False):
### run a process, capture the output and wait for it to finish. if timeout is specified then Kill the subprocess and its children when the timeout is reached (if parent did not detach)
## usage: _run(arg1, arg2, arg3)
# arg1: command + arguments. Always pass a string; the function will split it when needed
# arg2: (optional) timeout in seconds before force killing
# arg3: (optional) shell usage. default shell=False
## return: a list containing: exit code, output, and if timeout was reached or not
# - Tested on Python 2 and 3 on Windows XP, Windows 7, Cygwin and Linux.
# - preexec_fn=os.setsid (py2) is equivalent to start_new_session (py3) (works on Linux only), in Windows and Cygwin we use TASKKILL
# - we use stderr=subprocess.STDOUT to merge standard error and standard output
import sys, subprocess, os, signal, shlex, time
def _runPY3(command, timeout_s=None, shell=False):
# py3.3+ because: timeout was added to communicate() in py3.3.
new_session=False
if sys.platform.startswith('linux'): new_session=True
p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, start_new_session=new_session, shell=shell)
try:
out = p.communicate(timeout=timeout_s)[0].decode('utf-8')
is_timeout_reached = False
except subprocess.TimeoutExpired:
print('Timeout reached: Killing the whole process group...')
killAll(p.pid)
out = p.communicate()[0].decode('utf-8')
is_timeout_reached = True
return p.returncode, out, is_timeout_reached
def _runPY2(command, timeout_s=0, shell=False):
preexec=None
if sys.platform.startswith('linux'): preexec=os.setsid
p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, preexec_fn=preexec, shell=shell)
start_time = time.time()
is_timeout_reached = False
while timeout_s and p.poll() == None:
if time.time()-start_time >= timeout_s:
print('Timeout reached: Killing the whole process group...')
killAll(p.pid)
is_timeout_reached = True
break
time.sleep(1)
out = p.communicate()[0].decode('utf-8')
return p.returncode, out, is_timeout_reached
def killAll(ParentPid):
if sys.platform.startswith('linux'):
os.killpg(os.getpgid(ParentPid), signal.SIGTERM)
elif sys.platform.startswith('cygwin'):
# subprocess.Popen(shlex.split('bash -c "TASKKILL /F /PID $(</proc/{pid}/winpid) /T"'.format(pid=ParentPid)))
winpid=int(open("/proc/{pid}/winpid".format(pid=ParentPid)).read())
subprocess.Popen(['TASKKILL', '/F', '/PID', str(winpid), '/T'])
elif sys.platform.startswith('win32'):
subprocess.Popen(['TASKKILL', '/F', '/PID', str(ParentPid), '/T'])
# - In Windows, we never need to split the command, but in Cygwin and Linux we need to split if shell=False (default), shlex will split the command for us
if shell==False and (sys.platform.startswith('cygwin') or sys.platform.startswith('linux')):
command=shlex.split(command)
if sys.version_info >= (3, 3): # py3.3+
if timeout_s==False:
returnCode, output, is_timeout_reached = _runPY3(command, timeout_s=None, shell=shell)
else:
returnCode, output, is_timeout_reached = _runPY3(command, timeout_s=timeout_s, shell=shell)
else: # Python 2 and up to 3.2
if timeout_s==False:
returnCode, output, is_timeout_reached = _runPY2(command, timeout_s=0, shell=shell)
else:
returnCode, output, is_timeout_reached = _runPY2(command, timeout_s=timeout_s, shell=shell)
return returnCode, output, is_timeout_reached
然后用它如下:
总是将命令作为一个行(它更容易)。你不需要分开它;函数在需要时将分开它。
所以我们可以用这个类似的时间:
a=_run('cmd /c echo 11111 & echo 22222 & calc',3)
for i in a[1].splitlines(): print(i)
或者没有时间:
b=_run('cmd /c echo 11111 & echo 22222 & calc')
更多例子:
b=_run('''wmic nic where 'NetConnectionID="Local Area Connection"' get NetConnectionStatus /value''')
print(b)
c=_run('cmd /C netsh interface ip show address "Local Area Connection"')
print(c)
d=_run('printf "<%s>\n" "{foo}"')
print(d)
你也可以指定Shell=True,但在大多数情况下,这个功能是无用的。我宁愿自己选择我想要的Shell,但这里是,如果你也需要它:
# windows
e=_run('echo 11111 & echo 22222 & calc',3, shell=True)
print(e)
# Cygwin/Linux:
f=_run('printf "<%s>\n" "{foo}"', shell=True)
print(f)
為什麼我沒有使用更簡單的新方法 subprocess.run()?
因为它支持在Python 3.7+,但最后支持的Python版本在Windows XP是3.4. 而且由于这个功能的Timeout论点在Windows中是无用的,它不会杀死执行命令的儿童过程. 如果你使用 capture_output + Timeout论点,它会挂在如果有一个儿童过程仍然运行。
更新:
subprocess.run 是关于 Python 3.5 的推荐方法,如果您的代码不需要与以前的 Python 版本保持兼容性。
下面是文档的一些例子。
运行一个过程:
>>> subprocess.run(["ls", "-l"]) # Doesn't capture output
CompletedProcess(args=['ls', '-l'], returncode=0)
上一篇: 失败的跑步:
>>> subprocess.run("exit 1", shell=True, check=True)
Traceback (most recent call last):
...
subprocess.CalledProcessError: Command 'exit 1' returned non-zero exit status 1
捕获输出:
>>> subprocess.run(["ls", "-l", "/dev/null"], stdout=subprocess.PIPE)
CompletedProcess(args=['ls', '-l', '/dev/null'], returncode=0,
stdout=b'crw-rw-rw- 1 root root 1, 3 Jan 23 16:23 /dev/null\n')
原始答案:
我建议尝试发送,这是一个子过程的插槽,其目的是取代旧的模块和功能。
使用 README 的例子:
>>> r = envoy.run('git config', data='data to pipe in', timeout=2)
>>> r.status_code
129
>>> r.std_out
'usage: git config [options]'
>>> r.std_err
''
周围的管道也:
>>> r = envoy.run('uptime | pbcopy')
>>> r.command
'pbcopy'
>>> r.status_code
0
>>> r.history
[<Response 'uptime'>]
推荐文章
- 在python shell中按方向键时看到转义字符
- Shell命令查找两个文件中的公共行
- 在pip install中方括号是什么意思?
- 使用Matplotlib以非阻塞的方式绘图
- 使用sklearn缩放Pandas数据框架列
- 当存储命令输出到变量时，如何保存换行符?
- 如何从shell执行XPath一行程序?
- 如何创建关键或附加一个元素的关键?
- virtualenv的问题-无法激活
- 是否可以使用scikit-learn K-Means聚类来指定自己的距离函数?
- 如何在Python中删除文本文件的文件内容?
- 一个干净、轻量级的Python扭曲的替代品?
- 如何使用Bash递归创建不存在的子目录?
- 在Python中从字符串中移除所有非数字字符
- 在Python中，如何以排序的键顺序遍历字典?
