如何执行程序或调用系统命令？

这里有很多答案，但没有一个能够满足我所有的需求。

• 我需要运行命令并捕获输出和退出代码。
• 我需要超时执行的程序并在达到超时时间时强制退出，并杀死它的所有子进程。
• 我需要它可以在Windows XP及以后版本、Cygwin和Linux中运行。在Python 2和3中都可以使用。

所以我创建了这个：

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

然后像这样使用：

始终将命令作为一个字符串传递（这样更容易）。您不需要拆分它；函数将在需要时拆分它。

如果您的命令在您的 shell 中有效，则会在此函数中有效，因此请先在 cmd/Bash 中测试您的命令。

因此，我们可以像这样使用它，并设置超时：

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+，而Windows XP上最后支持的Python版本是3.4。
• 并且由于该函数的超时参数在Windows中无用，它不能终止执行命令的子进程。
• 如果您使用capture_output + timeout参数，则会在仍有子进程运行时挂起。并且在Windows中仍然存在问题，其中问题31447仍然未解决。

我编写了一个包装器来处理错误、重定向输出和其他一些东西。

import shlex
import psutil
import subprocess

def call_cmd(cmd, stdout=sys.stdout, quiet=False, shell=False, raise_exceptions=True, use_shlex=True, timeout=None):
    """Exec command by command line like 'ln -ls "/var/log"'
    """
    if not quiet:
        print("Run %s", str(cmd))
    if use_shlex and isinstance(cmd, (str, unicode)):
        cmd = shlex.split(cmd)
    if timeout is None:
        process = subprocess.Popen(cmd, stdout=stdout, stderr=sys.stderr, shell=shell)
        retcode = process.wait()
    else:
        process = subprocess.Popen(cmd, stdout=stdout, stderr=sys.stderr, shell=shell)
        p = psutil.Process(process.pid)
        finish, alive = psutil.wait_procs([p], timeout)
        if len(alive) > 0:
            ps = p.children()
            ps.insert(0, p)
            print('waiting for timeout again due to child process check')
            finish, alive = psutil.wait_procs(ps, 0)
        if len(alive) > 0:
            print('process {} will be killed'.format([p.pid for p in alive]))
            for p in alive:
                p.kill()
            if raise_exceptions:
                print('External program timeout at {} {}'.format(timeout, cmd))
                raise CalledProcessTimeout(1, cmd)
        retcode = process.wait()
    if retcode and raise_exceptions:
        print("External program failed %s", str(cmd))
        raise subprocess.CalledProcessError(retcode, cmd)

你可以这样调用它：

cmd = 'ln -ls "/var/log"'
stdout = 'out.txt'
call_cmd(cmd, stdout)

Sultan 是一个近期的软件包，旨在实现这一目的。它提供了一些有关管理用户权限和添加有用的错误消息的便利。

from sultan.api import Sultan

with Sultan.load(sudo=True, hostname="myserver.com") as sultan:
  sultan.yum("install -y tree").run()

这是一个Python脚本，可以在Ubuntu上运行命令，并实时显示日志：

command = 'your command here'    
process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    while True:
        output = process.stdout.readline().decode()
        if output == '' and process.poll() is not None:
            break
        if output:
            print(output.strip())
    rc = process.poll()
    if rc == 0:
        print("Command succeeded.")
       
    else:
        print("Command failed.")

使用Python的subprocess模块来执行shell命令并将输出写入文件。
下面的脚本将运行"ps -ef"命令，过滤包含"python3"的行，并将它们写入名为"python_processes.txt"的文件中。请注意，该代码不处理在执行过程中可能发生的任何异常情况。

import subprocess

# Command to execute
cmd = ["ps", "-ef"]

# Execute the command
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
output, error = process.communicate()

# Check if the command was executed without errors
if error is None:
    # Filter lines with 'python3'
    python_processes = [line for line in output.decode('utf-8').split('\n') if 'python3' in line]

    # Write the output to a file
    with open('python_processes.txt', 'w') as f:
        for process in python_processes:
            f.write(process + '\n')
else:
    print(f"Error occurred while executing command: {error}")

我使用这个适用于Python 3.6+：

import subprocess
def execute(cmd):
    """
        Purpose  : To execute a command and return exit status
        Argument : cmd - command to execute
        Return   : result, exit_code
    """
    process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    (result, error) = process.communicate()
    rc = process.wait()
    if rc != 0:
        print ("Error: failed to execute command: ", cmd)
        print (error.rstrip().decode("utf-8"))
    return result.rstrip().decode("utf-8"), serror.rstrip().decode("utf-8")
# def