如何在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管道。你要求他们提供命令作为一个字符串列表或一个单独的字符串，然后你将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. 泄漏zombies进程/打开的进程句柄，因为没有进程的退出代码，但收集了最后一个；操作系统保持资源打开，以便您这样做
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

Answer 2

不幸的是，它有一些边缘情况，shell会为您处理，或者shell会完全忽略您。一些问题：

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

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

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

• 子进程中的错误应作为异常引发，除了SIGPIPE。读者可以在最后一个进程上为SIGPIPE提出异常，这是一个练习，因为SIGPIPE不是这里所期望的，但是忽略它是无害的。

请注意，subprocess.DEVNULL在python3.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