用Python创建你自己的Shel

　　介绍很多人讨厌bash脚本。每当我要做最简单的事情时，我都必须查阅文档。如何将函数的参数转发给子命令？如何将字符串分配给变量，然后作为命令调用该字符串？如何检查两个字符串变量是否相等？如何分割字符串并获得后半部分？等等。不是我找不到这些答案，而是每次都必须查找它们。但是，我们不能否认将整个程序当作纯粹的功能发挥作用的能力，以及将一个程序的输出传递到另一个程序的自然程度。因此，我想知道，我们能否将bash的某些功能与Python结合起来？基础知识让我们从一个类开始。这是一个简单的方法，将其初始化参数保存到局部变量，然后使用subprocess.run对其自身进行延迟求值并保存结果。

importsubprocessclassPipePy:def__init__(self,*args):self._args=argsself._result=Nonedef_evaluate(self):ifself._resultisnotNone:returnself._result=subprocess.run(self._args,capture_output=True,text=True)

propertydefreturncode(self):self._evaluate()returnself._result.returncode

propertydefstdout(self):self._evaluate()returnself._result.stdoutdef__str__(self):returnself.stdout

propertydefstderr(self):self._evaluate()returnself._result.stderr我们让它旋转一下：

ls=PipePy(ls)ls_l=PipePy(ls,-l)print(ls)#files.txt#...main.py#...tagsprint(ls_l)#total16#...-rw-r--r--1kbairakkbairakJan:53files.txt#...-rw-r--r--1kbairakkbairakFeb:54main.py#...-rw-r--r--1kbairakkbairakFeb:54tags使其看起来更像“命令式”不用每次我们要自定义命令时都去调用PipePy。

ls_l=PipePy(ls,-l)print(ls_l)相当于

ls=PipePy(ls)print(ls(-l))换句话说，我们要使：

PipePy(ls,-l)相当于

PipePy(ls)(-l)值得庆幸的是，我们的类创建了惰性对象这一事实在很大程度上帮助了我们：

classPipePy:#__init__,etcdef__call__(self,*args):args=self._args+argsreturnself.__class__(*args)ls=PipePy(ls)print(ls(-l))#total16#...-rw-r--r--1kbairakkbairakJan:53files.txt#...-rw-r--r--1kbairakkbairakFeb:54main.py#...-rw-r--r--1kbairakkbairakFeb:54tags关键字参数如果要向ls传递更多参数，则可能会遇到--sort=size。我们可以轻松地执行ls（-l，--sort=size）。我们可以做得更好吗？

classPipePy:-def__init__(self,*args):+def__init__(self,*args,**kwargs):self._args=args+self._kwargs=kwargsself._result=Nonedef_evaluate(self):ifself._resultisnotNone: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)forarginself._args]+forkey,valueinself._kwargs.items():+key=key.replace(_,-)+args.append(f"--{key}={value}")+returnargs-def__call__(self,*args):+def__call__(self,*args,**kwargs):args=self._args+args+kwargs={**self._kwargs,**kwargs}-returnself.__class__(*args)+returnself.__class__(*args,**kwargs)#returncode,etc让我们来旋转一下：

print(ls(-l))#total16#...-rw-r--r--1kbairakkbairakJan:53files.txt#...-rw-r--r--1kbairakkbairakFeb:54main.py#...-rw-r--r--1kbairakkbairakFeb:54tagsprint(ls(-l,sort="size"))#total16#...-rw-r--r--1kbairakkbairakFeb:54main.py#...-rw-r--r--1kbairakkbairakFeb:54tags#...-rw-r--r--1kbairakkbairakJan:53files.txtPiping事情开始变得有趣起来。我们的最终目标是能够做到：

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输入传递给右操作数。

classPipePy:-def__init__(self,*args,**kwargs):+def__init__(self,*args,_pipe_input=None,**kwargs):self._args=argsself._kwargs=kwargs+self._pipe_input=_pipe_inputself._result=None-def__call__(self,*args,**kwargs):+def__call__(self,*args,_pipe_input=None,**kwargs):args=self._args+argskwargs={**self._kwargs,**kwargs}-returnself.__class__(*args,**kwargs)+returnself.__class__(*args,_pipe_input=_pipe_input,**kwargs)def_evaluate(self):ifself._resultisnotNone:returnself._result=subprocess.run(self._convert_args(),+input=self._pipe_input,capture_output=True,text=True)+def__or__(left,right):+returnright(_pipe_input=left.stdout)让我们尝试一下（从之前稍微修改命令以证明它确实有效）：

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

grep(tags))#-rw-r--r--1kbairakkbairakFeb:54tags让我们添加一些简单的东西1、真实性：

classPipePy:#__init__,etcdef__bool__(self):returnself.returncode==0现在我们可以作出如下处理：

git=PipePy(git)grep=PipePy(grep)ifgit(branch)

grep(my_feature):print("Branchmy_featurefound")2、读取/写入文件：

classPipePy:#__init__,etcdef__gt__(self,filename):withopen(filename,w)asf:f.write(self.stdout)def__rshift__(self,filename):withopen(filename,a)asf:f.write(self.stdout)def__lt__(self,filename):withopen(filename)asf:returnself(_pipe_input=f.read())现在可以作出如下操作：

ls=PipePy(ls)grep=PipePy(grep)cat=PipePy(cat)lsfiles.txtprint(grep(main)files.txt)#main.pylsfiles.txtprint(cat(files.txt))#files.txt#...main.py#...tags#...files.txt#...main.py#...tags3、迭代

classPipePy:#__init__,etcdef__iter__(self):returniter(self.stdout.split())现在可以作出如下操作：

ls=PipePy(ls)fornameinls:print(name.upper())#FILES.TXT#...MAIN.PY#...TAGS4、表格：

classPipePy:#__init__,etcdefas_table(self):lines=self.stdout.splitlines()fields=lines[0].split()result=[]forlineinlines[1:]:item={}fori,valueinenumerate(line.split(maxsplit=len(fields)-1)):item[fields[i]]=valueresult.append(item)returnresult现在可以作出下面操作：

ps=PipePy(ps)print(ps)#PIDTTYTIMECMD#...pts/:00:00zsh#...pts/:00:22ptipython#...pts/:00:00psps.as_table()#[{PID:,TTY:pts/4,TIME:00:00:00,CMD:zsh},#...{PID:,TTY:pts/4,TIME:00:00:22,CMD:ptipython},#...{PID:,TTY:pts/4,TIME:00:00:00,CMD:ps}]5、普通bash实用程序：在子进程中更改工作目录不会影响当前的脚本或pythonshell。与更改环境变量相同，以下内容不是PipePy的补充，但很不错：

importoscd=os.chdirexport=os.environ.__setitem__pwd=PipePy(pwd)pwd#/home/kbairak/prog/python/pipepycd(..)pwd#/home/kbairak/prog/python使事情看起来更shell-like如果我在交互式shell中，则希望能够简单地键入ls并完成它。

classPipePy:#__init__,etcdef__repr__(self):returnself.stdout+self.stderr交互式shell

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

frompipepyimportPipePytar=PipePy(tar)tar(-xf,some_archive.tar)print("Fileextracted")该脚本实际上不会执行任何操作，因为tar调用实际上并未得到评估。我认为一个不错的惯例是，如果不带参数调用__call__强制求值：

classPipePy:def__call__(self,*args,_pipe_input=None,**kwargs):args=self._args+argskwargs={**self._kwargs,**kwargs}-returnself.__class__(*args,_pipe_input=_pipe_input,**kwargs)+result=self.__class__(*args,_pipe_input=_pipe_input,**kwargs)+ifnotargsandnot_pipe_inputandnotkwargs:+result._evaluate()+returnresult因此在编写脚本时，如果要确保实际上已调用命令，则必须用一对括号来调用它：

frompipepyimportPipePytar=PipePy(tar)-tar(-xf,some_archive.tar)+tar(-xf,some_archive.tar)()print("Fileextracted")但是，我们还没有解决问题。考虑一下：

date=PipePy(date)date#MonFeb:43:08PMEET#Wait5secondsdate#MonFeb:43:08PMEET不好！date没有改变。date对象将其_result保留在内存中。随后的评估实际上不会调用该命令，而只是返回存储的值。一种解决方案是通过使用空括号来强制创建副本：

date=PipePy(date)date()#MonFeb:45:09PMEET#Wait5secondsdate()#MonFeb:45:14PMEET另一个解决方案是：由PipePy构造函数返回的实例不应该是惰性的，但由__call__调用返回的实例将是惰性的。

classPipePy:-def__init__(self,*args,_pipe_input=None,**kwargs):+def__init__(self,*args,_pipe_input=None,_lazy=False,**kwargs):self._args=argsself._kwargs=kwargsself._pipe_input=_pipe_input+self._lazy=_lazyself._result=Nonedef__call__(self,*args,_pipe_input=None,**kwargs):args=self._args+argskwargs={**self._kwargs,**kwargs}-result=self.__class__(*args,_pipe_input=_pipe_input,**kwargs)+result=self.__class__(*args,+_pipe_input=_pipe_input,+_lazy=True,+**kwargs)ifnotargsandnot_pipe_inputandnotkwargs:result._evaluate()returnresultdef_evaluate(self):-ifself._resultisnotNone:+ifself._resultisnotNoneandself._lazy:returnself._result=subprocess.run(self._convert_args(),input=self._pipe_input,capture_output=True,text=True)旋转一下：

date=PipePy(date)date#MonFeb:54:09PMEET#Wait5secondsdate#MonFeb:54:14PMEET并且可以预见的是，使用空调用的返回值将具有之前的行为：

date=PipePy(date)d=date()d#MonFeb:56:21PMEET#Wait5secondsd#MonFeb:56:21PMEET没关系您不会期望d会更新其值。越来越危险好吧，ls（-l）不错，但是如果我们像人类一样简单地做ls-l，那就太好了。嗯，我有个主意：

classPipePy:#__init__,etcdef__sub__(left,right):returnleft(f"-{right}")现在可以作如下操作：

ls=PipePy(ls)ls-l#total16#...-rw-r--r--1kbairakkbairak46Feb:04files.txt#...-rw-r--r--1kbairakkbairakFeb:54main.py#...-rw-r--r--1kbairakkbairakFeb:54tags我们还有一步：

l=lls-l现在无济于事：

importstringforcharinstring.ascii_letters:ifcharinlocals():continuelocals()[char]=charclassPipePy:#__init__,etc更危险的事情用locals（）给了我一个灵感。为什么我们必须一直实例化PipePy？我们无法在路径中找到所有可执行文件，并根据它们创建PipePy实例吗？我们当然可以！

importosimportstatforpathinos.get_exec_path():try:names=os.listdir(path)exceptFileNotFoundError:continuefornameinnames:ifnameinlocals():continueifxinstat.filemode(os.lstat(os.path.join(path,name)).st_mode):locals()[name]=PipePy(name)因此，现在，将我们拥有的所有内容都放在一个python文件中，并删除脚本（这是实际bash脚本的转录）：

frompipepyimportmysqladmin,sleep,drush,grepforiinrange(10):ifmysqladmin(ping,host="mysql_drupal7",user="user",password="password"):breaksleep(1)()#Remembertoactuallyinvokeifnotdrush(status,bootstrap)

grep(-q,Successful):drush(-y,site-install,standard,db_url="mysql://user:password

mysql_drupal7:/drupal",acount_pass="kbairak")()#Remembertoactuallyinvokedrush(en,tmgmt_ui,tmgmt_entity_ui,tmgmt_node_ui)()

更多阅读

年最佳流行Python库Top10

Python中文社区热门文章Top10

5分钟快速掌握Python定时任务框架

特别推荐

点击下方阅读原文加入社区会员

预览时标签不可点收录于话题#个上一篇下一篇

转载请注明：http://www.lwblm.com/bzbk/12254.html