Python实现类似expect脚本的远程交互式编程实现自动化需求

￡神魔★判官ぃ 2022-12-23 11:16 133阅读 0赞

### 文章目录 ###

*   *   *  需求说明
         *  实现
         *   *  依赖
             *  代码封装实现

### 需求说明 ###

通过python代码来实现类似expect shell的交互式能力，这样在python代码中能结合业务逻辑组装代码，具有比expect更强大的自动化处理能力。

如果能通过该交互能力，结合SFTP来实现文件相关的操作，那就更加的强大了。

该篇文章代码为交互式命令封装实现。

### 实现 ###

#### 依赖 ####

依赖paramiko库，所以需要执行`pip install paramiko`来安装。

#### 代码封装实现 ####

import time
    
    import paramiko
    from paramiko import SSHClient, Channel
    import re
    
    recv_len = 1024 * 10
    
    
    class PtySession(object):
    
        def __init__(self, session, ending_char, timeout=10) -> None:
            super().__init__()
            self.session = session  # type:Channel
            self.last_line = ""
            self.ending_char = ending_char  # 每执行一个命令之后，标记输出的结束字符
            self.clear_tail()
            self.timeout = timeout  # 超时时间，秒
    
        def clear_tail(self):
            """ 清理输出还处于缓冲区中未读取的流 """
            while True:
                time.sleep(0.2)
                # self.session.recv_ready()在读取过程中不一定总是True，只有当读取缓冲流中有字节读取时，才会为True。所以在读取头一次后获取下次流到缓冲区中前为False
                if self.session.recv_ready():
                    self.last_line = self.session.recv(recv_len)
                    self.last_line = self.last_line.decode('utf-8')
                    print(self.last_line, end="")
                if re.search(self.ending_char, self.last_line):
                    break
    
        def destroy(self):
            """ 销毁并关闭session :return: :rtype: """
            self.clear_tail()
            self.session.close()
    
        def exp(self, *exp_cmds):
            """ 期望并执行，与expect的用法类似。 :param exp_cmds: 第一个元素为获取的期望结束字符，第二个元素为需要执行的命令，如果传入的第三个元素，则第三个元素必须为元祖，并且也同父级一样，属递归结构。 类似GNU的递归缩写。 :type exp_cmds: tuple """
            interval = 0.2
            cur_time = 0.0
            while True:
                if self.session.recv_ready():
                    self.last_line = self.session.recv(recv_len).decode('utf-8')
                    print(self.last_line, end="")
                elif self.session.send_ready():
                    for exp_cmd in exp_cmds:
                        _cmd = exp_cmd[1]
                        if not _cmd.endswith("\r"):
                            _cmd += "\r"
                        match = re.search(exp_cmd[0], self.last_line)
                        if match and match.group():
                            self.session.send(_cmd)
                            # 清空最后一行数据缓存，便于下个命令的读取流输出。此行代码去除，会导致无法等待命令执行完毕提前执行后续代码的问题。
                            self.last_line = ""
                            if len(exp_cmd) == 3 and exp_cmd[2]:
                                self.exp(exp_cmd[2])
                            return
                cur_time += interval
                if cur_time >= self.timeout:
                    raise Exception("timeout...")
                time.sleep(interval)
    
        def send(self, cmd):
            """ 单纯的发送命令到目标服务器执行。 :param cmd: 命令 :type cmd: str """
            self.last_line = ""
            if not cmd.endswith("\r"):
                cmd += "\r"
            self.session.send(cmd)
            self.clear_tail()
    
    
    if __name__ == '__main__':
        """ ====示例==== """
        host = "127.0.0.1"  # 服务器ip
        port = 22  # ssh端口
        username = "xxx"  # 账号
        pwd = "xxx"  # 密码
    
        _ssh_client = SSHClient()
        _ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        _ssh_client.connect(hostname=host, timeout=60, port=port, username=username,
                            password=pwd)
        print("ssh连接成功.")
        _session = _ssh_client.get_transport().open_session(timeout=1 * 3600)  # type:Channel
        _session.get_pty()
        _session.invoke_shell()
    
        # ending_char为正则表达式,以"~]$ "结尾的字符作为结束字符，这里取的命令行标识符PS1的结尾
        pty_session = PtySession(_session, ending_char=r"]\$ $", timeout=10)
    
        pty_session.send("sleep 5\r")
        pty_session.send("touch /test_a.txt")  # 发送命令
        pty_session.exp(
            (
                r"]\$ $",  # 期望如果以$结束
                "scp -P%s %s %s@%s:%s" % (  # 执行scp命令
                    port,
                    "~/test_a.txt",
                    username,
                    host,
                    "~/test_b.txt"
                )
            )
        )
        """ 下方exp的执行伪代码类似： if(yes/no): send yes if (password): send pwd elif(passowrd): send pwd """
        pty_session.exp(
            (
                "yes/no",  # 元素1(期望结果) 如果返回的yes/no
                "yes",  # 元素2(命令) 那么发送yes
                (  # 元素3(上面的命令执行完毕后，接着这行该分支命令)
                    ("password", pwd)  # 如果返回password，则输入密码。与上面注释逻辑一模一样，只是没有了第三个元素作为后续执行分支
                )
            ),
            ("password", pwd)  # 如果直接返回的password，则输入密码
        )
    
        pty_session.destroy()  # 销毁session

该篇文章代码仅供参考。

后续有空再补结合SFTP自动化文件上传下载等通用处理封装。