如何在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