有一些现成的库, 比如WExpect, 是开源的, 在github上可以搜索到. 但是, 不知道为什么, 在我自己的笔记本上不能正常工作. 而其源码也比较多, 懒得定位了.
于是自己实现了一个, 用法如下.
- 启动和停止命令行
import my_cmd as cmd
cmd.start()
cmd.stop()
- prompt命令行提示符匹配字符串
import my_cmd as cmd
print(cmd.get_prompt()) # 输出: `[A-Za-z]:\\.*>$`
# cmd.set_prompt('>') # 设置提示符匹配字符串, 需要是正则表达式
- 轮询等待expect预期输出
功能: 轮询等待expect预期输出, 如果没有预期输出, 则一直等待下去.
原型:expect(*regex_infos:str)->int:
expect()
: 如果不用参数, 则默认等待prompt提示符出现.
如果参数有一个或者多个正则表达式, 则返回第一个匹配的正则表达式在列表中的索引. 如果匹配到prompt, 则返回0.
import my_cmd as cmd
cmd.start()
cmd.expect()
cmd.stop()
输出:
Microsoft Windows [版本 10.0.22621.3155]
(c) Microsoft Corporation。保留所有权利。
d:\MyCode\Libs\Python\Test\my_cmd>
cmd.exe process exited.
- 查看和清除输出
import my_cmd as cmd
cmd.start()
cmd.expect()
print(cmd.get_output()) # 获取所有的输出
print(cmd.get_last_line()) # 获取最后一行的输出
print(cmd.clear_output()) # 清除输出信息
cmd.stop()
- 输入命令
write_line(cmd:str): 输入命令, 自动添加换行符. 为防止输出信息干扰, 会先清除之前的输出信息.
import my_cmd as cmd
cmd.start()
cmd.expect()
cmd.write_line('pwd')
cmd.expect()
cmd.stop()
输出:
Microsoft Windows [版本 10.0.22621.3155]
(c) Microsoft Corporation。保留所有权利。
d:\MyCode\Libs\Python\Test\my_cmd>pwd
/d/MyCode/Libs/Python/Test/my_cmd
d:\MyCode\Libs\Python\Test\my_cmd>
cmd.exe process exited.
源代码
import Common
import io
import re
import time
import subprocess
# region 全局变量
_buffer = io.StringIO()
'''存放输出字符串的缓冲区'''
_last_line_buffer = io.StringIO()
'''最后一行的输出缓冲区. 主要是为了提高expect函数的匹配效率.'''
_proc:subprocess.Popen = None
'''cmd.exe进程对象'''
_prompt:str = r'[A-Za-z]:\\.*>$' # 有时候盘符也可能是小写
r'''默认提示信息. 形如: C:\Users\huzho>'''
# endregion
def clear_output() -> None:
'''清除输出'''
global _buffer,_last_line_buffer
_buffer.close()
_buffer = io.StringIO()
_last_line_buffer.close()
_last_line_buffer = io.StringIO()
def expect(*regex_infos:str)->int:
'''
等待预期的字符串出现. 为提高效率, 默认只匹配输出的最后一行.
返回值: 匹配到的预期字符串的索引, 如果为0, 表示prompt, 其余的按顺序为1,2,3...
'''
regex_infos = list(regex_infos)
regex_infos.insert(0, _prompt) # 增加上提示符
while True: # 循环匹配, 直到匹配成功为止
for i,reg_info in enumerate(regex_infos):
if re.search(reg_info, get_last_line()):
return i
time.sleep(0.1) # 防止死循环
def get_last_line() -> str:
'''获取输出的最后一行字符串'''
return _last_line_buffer.getvalue()
def get_output() -> str:
'''获取当前的输出'''
return _buffer.getvalue()
def get_prompt() -> str:
'''获取当前的提示符'''
return _prompt
@Common.run_in_thread
def _output():
'''轮询输出'''
global _last_line_buffer
while True:
output = _proc.stdout.read(1) # 读取一个字符, 保证了只要有输出, 就会及时响应
if output == '' and _proc.poll() is not None: # poll函数有值, 表示进程已退出
_buffer.close() # 清空缓冲区
_last_line_buffer.close()
print('cmd.exe process exited.')
break
print(output,end='')
_buffer.write(output) # 写入缓冲区中
# 处理最后一行的缓冲区
if output == '\n': # 如果换行, 则重置最后一行
_last_line_buffer.close()
_last_line_buffer = io.StringIO()
else:
_last_line_buffer.write(output)
def set_prompt(prompt:str) -> None:
'''设置提示符'''
global _prompt
_prompt = prompt
def start():
'''启动cmd进程并轮询输出'''
global _proc,_buffer,_last_line_buffer
_buffer = io.StringIO()
_last_line_buffer = io.StringIO()
_proc = subprocess.Popen('cmd.exe',
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
#stderr=subprocess.PIPE,
stderr=subprocess.STDOUT, # 错误信息也同时输出
#encoding='gbk', # windows下的编码, 可以打印汉字信息, 是不是`text=True`就可以了?
cwd=None, # 工作目录, 后续可以增加设置工作目录的功能
bufsize=1,
text=True)
_output() # 在另一个线程中轮询输出
def stop():
'''停止cmd进程'''
#send('exit')
_proc.terminate()
def write_line(line:str) -> None:
'''写入命令. 写入命令之前先清除之前的输出. 以免有干扰.'''
clear_output() # 先清除之前的输出
_proc.stdin.write(line + '\n')
_proc.stdin.flush()
BTW: 虽然主要是针对windows系统的, 但是改造一下, 应该也可以用于linux系统, 因为是纯python语言开发的.
标签:输出,last,python,cmd,windows,buffer,命令行,prompt,line From: https://www.cnblogs.com/huzhongqiang/p/18057198