pexpect笔记和应用 ~ 绯境之外~Outside of Scarlet

pexpect笔记和应用

本次notebook采用了.ipynb格式交互式文档，下载链接

pexpect包

本次实验要用到的类和方法

Pexpect简介

在讲解Pexpect之前，我们需要先了解一下Expect这个脚本语言，它是由TCL语言实现的，主要用于人机交互式对话的自动化控制，可以用来完成ssh、ftp、telnet等命令行程序的自动化交互。Pexpect其实就是一个用Python语言实现的类Expect功能的模块，通过它就可以在Python中完成Expect所完成的功能。
Pexpect的基本工作流程，基本可以分为以下三个步骤：

首先用spawn来执行一个程序；
然后用expect方法来等待指定的关键字，这个关键字是被执行的程序打印到标准输出上面的；
最后当发现这个关键字以后，使用send/sendline方法发送字符串给这个程序。

通常在程序中第一步只需要做一次，第二步和第三步会不停的循环来完成整个工作。当然在Pexpect中还有很多其他方法，编写程序时可以根据自己的需求选择使用。

作者：码道仕
链接：https://juejin.cn/post/6909406787045654536
来源：稀土掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

spawn 类

通过spawn()方法用来执行一个程序，返回程序的操作句柄，后续就可以通过操作句柄来与这个程序进行交互了。
这里列出了本次实验用到的参数

import pexpect
child = pexpect.spawn(command="ls -l") # 使用命令

child.expect(pexpect.EOF)
print(child.match)
print(child.before)
print(child.after)

<class 'pexpect.exceptions.EOF'>
b'total 96\r\n-rwxrwxrwx 1 scb scb 78586 Aug  4 13:28 log.txt\r\n-rwxrwxrwx 1 scb scb  1797 Aug  5 12:24 main2.py\r\n-rwxrwxrwx 1 scb scb  2155 Aug  4 12:59 main.py\r\n-rwxrwxrwx 1 scb scb  6203 Aug  5 22:47 README.ipynb\r\ndrwxrwxrwx 1 scb scb  4096 Aug  4 08:42 scapy\r\n'
<class 'pexpect.exceptions.EOF'>

# 注意这里是全部当作字符串执行，所以如果想要使用管道可能需要直接调用`bin/bash`程序
import pexpect
child = pexpect.spawn(command='/bin/bash',args=["-c","ls -l | grep main"])
child.expect(pexpect.EOF)
print(child.match)
print(child.before)
print(child.after)

<class 'pexpect.exceptions.EOF'>
b'-rwxrwxrwx 1 scb scb  1797 Aug  5 12:24 main2.py\r\n-rwxrwxrwx 1 scb scb  2155 Aug  4 12:59 main.py\r\n'
<class 'pexpect.exceptions.EOF'>

expect方法

当使用spawn()方法启动了一个程序并返回程序控制句柄后，就可以使用expect()方法来等待指定的关键字了。关键字可以是字符串、正则表达式、EOF、TIMEOUT或者以上类型组成的列表，用来匹配子程序返回的结果。如果只提供字符串等非列表，则匹配成功后返回0，如果提供列表，则返回匹配成功的列表元素的索引，匹配失败会抛出异常。

作者：码道仕
链接：https://juejin.cn/post/6909406787045654536
来源：稀土掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

通常为了避免匹配不到而timeout的error，在expect方法的pattern列表最后一个参数写为pexpect.EOF，判断返回值序号是否为列表的最后一个，或者设置timeout参数（当然默认情况下貌似是20s）来try从而except也是可以的。

child = pexpect.spawn(command="ls -l")
i = child.expect("main\d?\.py")
print(child.match)
i = child.expect("main\d?\.py")
print(child.match)

<re.Match object; span=(99, 107), match=b'main2.py'>
<re.Match object; span=(42, 49), match=b'main.py'>

除了expect方法之外，还有其他更加底层的方法。
child.expect_exact()只能匹配字符串
child.expect_list()只能匹配正则表达式，expect()方法就是继承于此

send方法

send(str)：将字符串输入终端
sendline(str):将字符串输入终端并在最后加入回车符（从而可以发送）,如linux就是\r\n

interact方法

interact()表示将终端控制权交给用户（或者说将标准输入交给用户）。通常情况下Pexpect会接管所有的输入和输出，如果需要用户介入完成部分工作的时候，interact()就派上用场了。

# 让出控制权给用户
process.interact()
# 通过设置escape_character的值定义返回码，当用户输入此值后，会将控制权重新交给pexpect
process.interact(escape_character='\x1d', input_filter=None, output_filter=None)

作者：码道仕
链接：https://juejin.cn/post/6909406787045654536
来源：稀土掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

应用

案例要求：终端中输入ssh noname.plus -p2023即可通过ssh链接到一个自动化程序，首先要求你输入id作为scoreboard的用户名。
接着输入1可以选择start game
游戏规则：Whenever numbers appear, remember them!!!Stick to it for at least 11 rounds!!!

在这里要注意终端中远程应用使用ansi终端输出，会有颜色字符串，所以读取缓冲区中的字符串后需要过滤成ascii的字符串再调用pexpect的send方法

import pexpect
import threading

# deposit func
def filterBuff(buf:str)->str:
    ret:str = ''
    buf_list:'list[str]' = buf.split("\\x1b[")[1::2]
    for word in buf_list:
        ret += word[-1]

    return ret

def filterBuffandSend(child:pexpect.spawn,buf:str):
    buf_list:'list[str]' = buf.split("\\x1b[")
    for word in buf_list[1::2]:
        child.send(word[-1])
    child.send("\n")

def func(your_name:str,times:int):
    #f = open("log.txt", "a")

    process = pexpect.spawn(command="ssh noname.plus -p2023")

    process.expect(pattern=["Input Your player ID:", pexpect.EOF])
    # print(process.match)
    process.sendline(your_name)


    
    nu = process.expect(["Choose an option:",pexpect.EOF])
    if nu == 0:
        process.sendline("1")

    for i in range(0,times):
        process.expect(["Enter numbers now:",pexpect.EOF])
        # print("get")
        filterBuffandSend(process,str(process.before))
        # print("send")
    
    process.expect(["Enter numbers now:",pexpect.EOF])
    process.sendline("0")

if __name__ == "__main__":
    your_name = input("type the ID you want:")
    mintimes = input("type the minus score you want:")
    maxtimes = input("type the max times you want:")

    mintimes = int(mintimes)
    maxtimes = int(maxtimes)

    Threadlist:list[threading.Thread] = list()
    for times in range(mintimes,maxtimes+1):
        newT = threading.Thread(target=func,args=(your_name +'|'+ str(times),times))
        Threadlist.append(newT)
        newT.setDaemon(True)
        newT.start()

    ii = 0
    for t in Threadlist:
        t.join()
        print(f"thread {mintimes + ii} end")
        ii += 1 
    
    print("All thread end")

绯境之外~Outside of Scarlet

2023年8月5日星期六