如何在 Python 中创建一个通用方法来执行多个管道 shell 命令？

Question

我有许多 shell 命令需要在我的 python 脚本中执行。我知道我不应该像here 中提到的那样使用 shell=true，并且我可以使用 std 输出和输入，以防我在提到的命令中有管道 here。

但问题是我的 shell 命令很复杂并且充满了管道，所以我想创建一个通用方法供我的脚本使用。

我在下面做了一个小测试，但打印结果后挂了（我简化了只是放在这里）。谁能告诉我：

1. 为什么会挂起。
2. 如果有更好的方法来做到这一点。

谢谢。

PS：这只是一个大型 python 项目的一小部分，我尝试这样做是有商业原因的。谢谢。

#!/usr/bin/env python3
import subprocess as sub
from subprocess import Popen, PIPE
import shlex

def exec_cmd(cmd,p=None,isFirstLoop=True):
   if not isFirstLoop and not p:
       print("Error, p is null")
       exit()
   if "|" in cmd:
       cmds = cmd.split("|")
       while "|" in cmd:
           # separates what is before and what is after the first pipe
           now_cmd = cmd.split('|',1)[0].strip()
           next_cmd = cmd.split('|',1)[-1].strip()
           try:
               if isFirstLoop:
                   p1 = sub.Popen(shlex.split(now_cmd), stdout=PIPE)
                   exec_cmd(next_cmd,p1,False)
               else:
                   p2 = sub.Popen(shlex.split(now_cmd),stdin=p.stdout, stdout=PIPE)
                   exec_cmd(next_cmd,p2,False)
           except Exception as e:
               print("Error executing command '{0}'.\nOutput:\n:{1}".format(cmd,str(e)))
               exit()
           # Adjust cmd to execute the next part
           cmd = next_cmd
   else:
       proc = sub.Popen(shlex.split(cmd),stdin=p.stdout, stdout=PIPE, universal_newlines=True)
       (out,err) = proc.communicate()
       if err:
           print(str(err).strip())
       else:
           print(out)



exec_cmd("ls -ltrh | awk '{print $9}' | wc -l ")

Answer 1

不幸的是，其中有一些边缘情况，shell 会为您处理，或者shell 会完全忽略您。一些担忧：

• 该函数应始终wait() 以完成每个进程，否则您将得到所谓的zombie processes。

• 这些命令应该使用真正的管道相互连接，这样就不需要一次将整个输出读入内存。这是管道的正常工作方式。

• 每个管道的读取端都应该在父进程中关闭，这样子进程才能在下一个进程关闭其输入时正确地SIGPIPE。没有这个，父进程可以保持管道打开，子进程不知道退出，它可能永远运行。

• 子进程中的错误应作为异常引发，SIGPIPE 除外。在最终流程中为SIGPIPE 提出异常作为练习留给读者，因为SIGPIPE 不是预期的，但忽略它并无害。

请注意，subprocess.DEVNULL 在 Python 3.3 之前不存在。我知道你们中的一些人仍然使用 2.x，您必须手动打开 /dev/null 的文件，或者只是决定管道中的第一个进程与父进程共享 stdin。

代码如下：

import signal
import subprocess

def run_pipe(*cmds):
    """Run a pipe that chains several commands together."""
    pipe = subprocess.DEVNULL
    procs = []
    try:
        for cmd in cmds:
            proc = subprocess.Popen(cmd, stdin=pipe,
                                    stdout=subprocess.PIPE)
            procs.append(proc)
            if pipe is not subprocess.DEVNULL:
                pipe.close()
            pipe = proc.stdout
        stdout, _ = proc.communicate()
    finally:
        # Must call wait() on every process, otherwise you get
        # zombies.
        for proc in procs:
            proc.wait()
    # Fail if any command in the pipe failed, except due to SIGPIPE
    # which is expected.
    for proc in procs:
        if (proc.returncode
            and proc.returncode != -signal.SIGPIPE):
            raise subprocess.CalledProcessError(
                proc.returncode, proc.args)
    return stdout

在这里我们可以看到它的实际效果。您可以看到管道以yes 正确终止（一直运行到SIGPIPE）并以false 正确失败（始终失败）。

In [1]: run_pipe(["yes"], ["head", "-n", "1"])
Out[1]: b'y\n'

In [2]: run_pipe(["false"], ["true"])
---------------------------------------------------------------------------
CalledProcessError                        Traceback (most recent call last)
<ipython-input-2-db97c6876cd7> in <module>()
----> 1 run_pipe(["false"], ["true"])

~/test.py in run_pipe(*cmds)
     22     for proc in procs:
     23         if proc.returncode and proc.returncode != -signal.SIGPIPE:
---> 24             raise subprocess.CalledProcessError(proc.returncode, proc.args)
     25     return stdout

CalledProcessError: Command '['false']' returned non-zero exit status 1

Answer 2

我会要求用户自己提供命令作为单独的实体，而不是使用 shell 字符串并尝试用自己的方式解析它。这避免了检测| 的明显陷阱，该| 是命令的一部分并且不用作shell 管道。您要求他们提供命令作为字符串列表或单个字符串，然后您将shlex.split 取决于您要公开的接口。在下面的示例中，我会选择第一个，因为它很简单。

一旦您有了单独的命令，一个简单的for 循环就足以将先前命令的输出通过管道传递到下一个命令的输入，例如you have found yourself：

def pipe_subprocesses(*commands):
    if not commands:
        return

    next_input = None
    for command in commands:
        p = subprocess.Popen(command, stdin=next_input, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        next_input = p.stdout

    out, err = p.communicate()
    if err:
        print(err.decode().strip())
    else:
        print(out.decode())

用法是：

>>> pipe_subprocesses(['ls', '-lhtr'], ['awk', '{print $9}'], ['wc', '-l'])
25

---

现在这是一种快速而肮脏的方式来设置它，并且看起来像你想要的那样工作。但是这段代码至少有两个问题：

1. 您泄漏了僵尸进程/打开的进程句柄，因为没有进程的退出代码但最后一个被收集；并且操作系统会保持资源开放供您这样做；
2. 您无法访问中途失败的进程的信息。

为避免这种情况，您需要维护一个已打开进程的列表，并为每个进程明确地wait。而且由于我不知道您的确切用例，我将只返回第一个失败的进程（如果有）或最后一个进程（如果没有），以便您采取相应的行动：

def pipe_subprocesses(*commands):
    if not commands:
        return

    processes = []
    next_input = None
    for command in commands:
        if isinstance(command, str):
            command = shlex.split(command)
        p = subprocess.Popen(command, stdin=next_input, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        next_input = p.stdout
        processes.append(p)

    for p in processes:
        p.wait()

    for p in processes:
        if p.returncode != 0:
            return p
    return p  # return the last process in case everything went well

我还提供了一些 shlex 作为示例，以便您可以混合原始字符串和已解析的列表：

>>> pipe_subprocesses('ls -lhtr', ['awk', '{print $9}'], 'wc -l')
25