用 Python 创建你自己的 Shell

Python中文社区

共 1138字,需浏览 3分钟

 ·

2021-02-05 08:58


介绍
很多人讨厌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():
+            key = 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, **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, **kwargs)
+        result = self.__class__(*args, _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, **kwargs)
+        result = self.__class__(*args,
+                                _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')()


更多阅读



2020 年最佳流行 Python 库 Top 10


2020 Python中文社区热门文章 Top 10


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

特别推荐




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

浏览 51
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报
评论
图片
表情
推荐
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报