用Python创建你自己的Shell

介绍

很多人讨厌bash脚本。建自己每当我要做最简单的建自己事情时，我都必须查阅文档。建自己如何将函数的建自己参数转发给子命令？如何将字符串分配给变量，然后作为命令调用该字符串？建自己如何检查两个字符串变量是否相等？如何分割字符串并获得后半部分？等等。不是建自己我找不到这些答案，而是建自己每次都必须查找它们。

但是建自己，我们不能否认将整个程序当作纯粹的建自己功能发挥作用的能力，以及将一个程序的建自己输出传递到另一个程序的自然程度。因此，建自己我想知道，建自己我们能否将bash的建自己某些功能与Python结合起来？

基础知识

让我们从一个类开始。这是建自己一个简单的方法，将其初始化参数保存到局部变量，建自己然后使用subprocess.run对其自身进行延迟求值并保存结果。 

import subprocess  class PipePy:      def __init__(self, *args):          self._args = args          self._result = None      def _evaluate(self):          if self._result is not None:              return          self._result = subprocess.run(self._args,                                        capture_output=True,                                        text=True)       @property      def returncode(self):          self._evaluate()          return self._result.returncode      @property      def stdout(self):          self._evaluate()          return self._result.stdout      def __str__(self):          return self.stdout      @property      def stderr(self):          self._evaluate()          return self._result.stderr 

我们让它旋转一下： 

ls = PipePy(ls)  ls_l = PipePy(ls, -l)  print(ls)  # <<< files.txt  # ... main.py  # ... tags  print(ls_l)  # <<< total 16  # ... -rw-r--r-- 1 kbairak kbairak  125 Jan 22 08:53 files.txt  # ... -rw-r--r-- 1 kbairak kbairak 5425 Feb  1 21:54 main.py  # ... -rw-r--r-- 1 kbairak kbairak 1838 Feb  1 21:54 tags 

使其看起来更像“命令式”

不用每次我们要自定义命令时都去调用PipePy。 

ls_l = PipePy(ls, -l)  print(ls_l) 

相当于 

ls = PipePy(ls)  print(ls(-l)) 

换句话说，我们要使： 

PipePy(ls, -l) 

相当于 

PipePy(ls)(-l) 

值得庆幸的是，服务器租用我们的类创建了惰性对象这一事实在很大程度上帮助了我们： 

class PipePy:      # __init__, etc      def __call__(self, *args):          args = self._args + args          return self.__class__(*args)  ls = PipePy(ls)  print(ls(-l))  # <<< total 16  # ... -rw-r--r-- 1 kbairak kbairak  125 Jan 22 08:53 files.txt  # ... -rw-r--r-- 1 kbairak kbairak 5425 Feb  1 21:54 main.py # ... -rw-r--r-- 1 kbairak kbairak 1838 Feb  1 21:54 tags 

关键字参数

如果要向ls传递更多参数，则可能会遇到--sort = size。我们可以轻松地执行ls（-l，--sort = size）。我们可以做得更好吗？ 

 class PipePy:  -    def __init__(self, *args):  +    def __init__(self, *args, **kwargs):           self._args = args  +        self._kwargs = kwargs           self._result = None       def _evaluate(self):           if self._result is not None:               return  -        self._result = subprocess.run(self._args,  +        self._result = subprocess.run(self._convert_args(),                                         capture_output=True,                                         text=True)  +    def _convert_args(self):  +        args = [str(arg) for arg in self._args]  +        for key, value in self._kwargs.items(): +            keykey = key.replace(_, -)  +            args.append(f"--{ key}={ value}")  +        return args  -    def __call__(self, *args):  +    def __call__(self, *args, **kwargs):           args = self._args + args  +        kwargs = { **self._kwargs, **kwargs}  -        return self.__class__(*args)  +        return self.__class__(*args, **kwargs)       # returncode, etc 

让我们来旋转一下： 

print(ls(-l))  # <<< total 16  # ... -rw-r--r-- 1 kbairak kbairak  125 Jan 22 08:53 files.txt  # ... -rw-r--r-- 1 kbairak kbairak 5425 Feb  1 21:54 main.py  # ... -rw-r--r-- 1 kbairak kbairak 1838 Feb  1 21:54 tags  print(ls(-l, sort="size"))  # <<< total 16  # ... -rw-r--r-- 1 kbairak kbairak 5425 Feb  1 21:54 main.py  # ... -rw-r--r-- 1 kbairak kbairak 1838 Feb  1 21:54 tags  # ... -rw-r--r-- 1 kbairak kbairak  125 Jan 22 08:53 files.txt 

Piping

事情开始变得有趣起来。我们的最终目标是能够做到： 

ls = PipePy(ls)  grep = PipePy(grep)  print(ls | grep(tags))  # <<< tags 

我们的过程是：

1、让__init__和__call__方法接受一个仅用于关键字的新_pipe_input关键字参数，该参数将保存在self上。

2、在评估期间，如果设置了_pipe_input，它将作为输入参数传递给subprocess.run。

3、重写__or__方法以将左操作数的结果作为pipe输入传递给右操作数。 

 class PipePy:  -    def __init__(self, *args, **kwargs):  +    def __init__(self, *args, _pipe_input=None, **kwargs):           self._args = args           self._kwargs = kwargs  +        self._pipe_input = _pipe_input           self._result = None   -    def __call__(self, *args, **kwargs):  +    def __call__(self, *args, _pipe_input=None, **kwargs):           args = self._args + args           kwargs = { **self._kwargs, **kwargs}  -        return self.__class__(*args, **kwargs)  +        return self.__class__(*args, _pipe_input_pipe_input=_pipe_input, **kwargs)       def _evaluate(self):           if self._result is not None:               return           self._result = subprocess.run(self._convert_args(),  +                                      input=self._pipe_input,                                         capture_output=True,                                         text=True)  +    def __or__(left, right):  +        return right(_pipe_input=left.stdout) 

让我们尝试一下（从之前稍微修改命令以证明它确实有效）： 

ls = PipePy(ls)  grep = PipePy(grep) print(ls(-l) | grep(tags))  # <<< -rw-r--r-- 1 kbairak kbairak 1838 Feb  1 21:54 tags 

让我们添加一些简单的东西

1、真实性： 

class PipePy:      # __init__, etc      def __bool__(self):          return self.returncode == 0 

现在我们可以作出如下处理： 

git = PipePy(git)  grep = PipePy(grep)  if git(branch) | grep(my_feature):      print("Branch my_feature found") 

2、读取/写入文件： 

class PipePy:      # __init__, etc      def __gt__(self, filename):          with open(filename, w) as f:              f.write(self.stdout)      def __rshift__(self, filename):          with open(filename, a) as f:             f.write(self.stdout)      def __lt__(self, filename):          with open(filename) as f:              return self(_pipe_input=f.read()) 

现在可以作出如下操作： 

ls = PipePy(ls)  grep = PipePy(grep)  cat = PipePy(cat)  ls > files.txt  print(grep(main) < files.txt)  # <<< main.py  ls >> files.txt  print(cat(files.txt))  # <<< files.txt  # ... main.py  # ... tags  # ... files.txt  # ... main.py  # ... tags 

3、迭代 

class PipePy:      # __init__, etc      def __iter__(self):          return iter(self.stdout.split()) 

现在可以作出如下操作： 

ls = PipePy(ls)  for name in ls:      print(name.upper())  # <<< FILES.TXT  # ... MAIN.PY # ... TAGS 

4、表格： 

class PipePy:      # __init__, etc      def as_table(self):          lines = self.stdout.splitlines()          fields = lines[0].split()          result = []          for line in lines[1:]:              item = { }              for i, value in enumerate(line.split(maxsplit=len(fields) - 1)):                  item[fields[i]] = value              result.append(item)          return result 

现在可以作出下面操作： 

ps = PipePy(ps)  print(ps)  # <<<     PID TTY          TIME CMD  # ...    4205 pts/4    00:00:00 zsh  # ...   13592 pts/4    00:00:22 ptipython  # ...   16253 pts/4    00:00:00 ps  ps.as_table()  # <<< [{ PID: 4205, TTY: pts/4, TIME: 00:00:00, CMD: zsh},  # ...  { PID: 13592, TTY: pts/4, TIME: 00:00:22, CMD: ptipython},  # ...  { PID: 16208, TTY: pts/4, TIME: 00:00:00, CMD: ps}] 

5、普通bash实用程序：

在子进程中更改工作目录不会影响当前的脚本或python shell。与更改环境变量相同，以下内容不是PipePy的补充，亿华云计算但很不错： 

import os  cd = os.chdir  export = os.environ.__setitem__  pwd = PipePy(pwd)  pwd  # <<< /home/kbairak/prog/python/pipepy  cd(..)  pwd  # <<< /home/kbairak/prog/python 

使事情看起来更shell-like

如果我在交互式shell中，则希望能够简单地键入ls并完成它。 

class PipePy:      # __init__, etc      def __repr__(self):          return self.stdout + self.stderr 

交互式shell 

>>> ls = PipePy(ls)  >>> ls  files.txt  main.py  tags 

我们的实例是惰性的，这意味着如果我们对它们的结果感兴趣，则将对它们进行评估，此后不再进行评估。如果我们只是想确保已执行该操作怎么办？例如，假设我们有以下脚本： 

from pipepy import PipePy  tar = PipePy(tar)  tar(-xf, some_archive.tar)  print("File extracted") 

该脚本实际上不会执行任何操作，因为tar调用实际上并未得到评估。我认为一个不错的惯例是，如果不带参数调用__call__强制求值： 

 class PipePy:       def __call__(self, *args, _pipe_input=None, **kwargs):           args = self._args + args           kwargs = { **self._kwargs, **kwargs}  -        return self.__class__(*args, _pipe_input_pipe_input=_pipe_input, **kwargs)  +        result = self.__class__(*args, _pipe_input_pipe_input=_pipe_input, **kwargs)  +        if not args and not _pipe_input and not kwargs:  +            result._evaluate()  +        return result 

因此在编写脚本时，如果要确保实际上已调用命令，则必须用一对括号来调用它： 

 from pipepy import PipePy   tar = PipePy(tar)  -tar(-xf, some_archive.tar)  +tar(-xf, some_archive.tar)()   print("File extracted") 

但是，我们还没有解决问题。考虑一下： 

date = PipePy(date)  date  # <<< Mon Feb  1 10:43:08 PM EET 2021  # Wait 5 seconds  date  # <<< Mon Feb  1 10:43:08 PM EET 2021 

不好！date没有改变。date对象将其_result保留在内存中。随后的评估实际上不会调用该命令，而只是返回存储的值。

一种解决方案是通过使用空括号来强制创建副本： 

date = PipePy(date)  date()  # <<< Mon Feb  1 10:45:09 PM EET 2021  # Wait 5 seconds  date()  # <<< Mon Feb  1 10:45:14 PM EET 2021 

另一个解决方案是：由PipePy构造函数返回的高防服务器实例不应该是惰性的，但由__call__调用返回的实例将是惰性的。 

 class PipePy:  -    def __init__(self, *args, _pipe_input=None, **kwargs):  +    def __init__(self, *args, _pipe_input=None, _lazy=False, **kwargs):           self._args = args           self._kwargs = kwargs           self._pipe_input = _pipe_input  +        self._lazy = _lazy           self._result = None       def __call__(self, *args, _pipe_input=None, **kwargs):           args = self._args + args           kwargs = { **self._kwargs, **kwargs}  -        result = self.__class__(*args, _pipe_input_pipe_input=_pipe_input, **kwargs)  +        result = self.__class__(*args, +                                _pipe_input_pipe_input=_pipe_input,  +                                _lazy=True,  +                                **kwargs)          if not args and not _pipe_input and not kwargs:               result._evaluate()           return result       def _evaluate(self):  -        if self._result is not None:  +        if self._result is not None and self._lazy:               return           self._result = subprocess.run(self._convert_args(),                                         input=self._pipe_input,                                         capture_output=True,                                         text=True) 

旋转一下： 

date = PipePy(date)  date  # <<< Mon Feb  1 10:54:09 PM EET 2021  # Wait 5 seconds  date  # <<< Mon Feb  1 10:54:14 PM EET 2021 

并且可以预见的是，使用空调用的返回值将具有之前的行为： 

date = PipePy(date) d = date()  d  # <<< Mon Feb  1 10:56:21 PM EET 2021  # Wait 5 seconds  d  # <<< Mon Feb  1 10:56:21 PM EET 2021 

没关系 您不会期望d会更新其值。

越来越危险

好吧，ls（-l）不错，但是如果我们像人类一样简单地做ls -l，那就太好了。嗯，我有个主意： 

class PipePy:      # __init__, etc      def __sub__(left, right):          return left(f"-{ right}") 

现在可以作如下操作： 

ls = PipePy(ls)  ls - l  # <<< total 16  # ... -rw-r--r-- 1 kbairak kbairak   46 Feb  1 23:04 files.txt  # ... -rw-r--r-- 1 kbairak kbairak 5425 Feb  1 21:54 main.py  # ... -rw-r--r-- 1 kbairak kbairak 1838 Feb  1 21:54 tags 

我们还有一步： 

l = l  ls -l 

现在无济于事： 

import string  for char in string.ascii_letters:      if char in locals():          continue     locals()[char] = char  class PipePy:      # __init__, etc 

更危险的事情

用locals（）给了我一个灵感。为什么我们必须一直实例化PipePy？我们无法在路径中找到所有可执行文件，并根据它们创建PipePy实例吗？我们当然可以！ 

import os  import stat  for path in os.get_exec_path():      try:          names = os.listdir(path)      except FileNotFoundError:          continue      for name in names:          if name in locals():              continue          if x in stat.filemode(os.lstat(os.path.join(path, name)).st_mode):              locals()[name] = PipePy(name) 

因此，现在，将我们拥有的所有内容都放在一个python文件中，并删除脚本（这是实际bash脚本的转录）： 

from pipepy import mysqladmin, sleep, drush, grep  for i in range(10):      if mysqladmin(ping,                    host="mysql_drupal7",                    user="user",                    password="password"):          break      sleep(1)()  # Remember to actually invoke  if not drush(status, bootstrap) | grep(-q, Successful):      drush(-y, site-install, standard,            db_url="mysql://user:password@mysql_drupal7:3306/drupal",            acount_pass="kbairak")()  # Remember to actually invoke  drush(en, tmgmt_ui, tmgmt_entity_ui, tmgmt_node_ui)()  

很赞哦!（6）