如何解析pexpect asyncio生成器以获取ping命令输出？

问题本质 #

当使用pexpect的expect方法并指定async=True时，返回的是asyncio异步生成器，不能直接打印或赋值获取结果，必须通过异步迭代或官方推荐的异步API来处理。

常见错误示例 #

初始错误代码（未处理异步生成器） #

import pexpect

ssh = pexpect.spawn('ssh user@server')
ssh.expect('password:')
ssh.sendline('your_password')
ssh.expect('$')
# 错误：直接获取生成器但未迭代处理
result = ssh.expect('ping -c 4 8.8.8.8', async=True)
print(result)  # 输出仅为生成器对象，无法拿到实际输出

更新后错误代码（协程中错误处理生成器） #

import asyncio
import pexpect

async def run_ping():
    ssh = pexpect.spawn('ssh user@server')
    ssh.expect('password:')
    ssh.sendline('your_password')
    ssh.expect('$')
    # 错误：异步生成器不能直接用await调用
    result = await ssh.expect('ping -c 4 8.8.8.8', async=True)
    print(result)

asyncio.run(run_ping())

可行解决方案 #

方案1：迭代异步生成器获取输出 #

针对async=True返回的生成器，使用async for循环迭代，每次迭代后通过ssh.before收集输出内容：

import asyncio
import pexpect

async def run_async_ping():
    # 启动SSH进程，指定编码避免字节/字符串问题
    ssh = pexpect.spawn('ssh user@your_server', encoding='utf-8')
    
    # 处理SSH密码验证
    async for match_idx in ssh.expect(['password:', pexpect.EOF, pexpect.TIMEOUT], async=True):
        if match_idx == 0:
            ssh.sendline('your_password')
            break
        elif match_idx in (1, 2):
            raise Exception('SSH连接失败：超时或连接断开')
    
    # 等待命令提示符，确认登录成功
    async for match_idx in ssh.expect(['$', '#', pexpect.EOF, pexpect.TIMEOUT], async=True):
        if match_idx in (0, 1):
            break
        else:
            raise Exception('未检测到服务器命令提示符')
    
    # 发送ping命令
    ssh.sendline('ping -c 4 8.8.8.8')
    
    # 收集ping输出直到再次出现提示符
    ping_output = []
    async for match_idx in ssh.expect(['$', '#', pexpect.EOF, pexpect.TIMEOUT], async=True, timeout=30):
        ping_output.append(ssh.before)
        if match_idx in (0, 1):
            break
        elif match_idx in (2, 3):
            raise Exception('ping命令执行超时或连接中断')
    
    # 拼接并输出完整结果
    full_output = ''.join(ping_output)
    print('Ping命令输出:\n', full_output)

# 启动异步事件循环
asyncio.run(run_async_ping())

方案2：使用官方推荐的async_spawn（更简洁） #

pexpect提供了专门的异步APIasync_spawn，所有操作均为协程，无需手动处理生成器，更贴合asyncio编程习惯：

import asyncio
from pexpect import async_spawn

async def run_async_ping_v2():
    # 创建异步SSH进程
    ssh = await async_spawn('ssh user@your_server', encoding='utf-8')
    
    # 处理密码验证
    match_idx = await ssh.expect(['password:', pexpect.EOF, pexpect.TIMEOUT])
    if match_idx == 0:
        await ssh.sendline('your_password')
    else:
        raise Exception('SSH连接失败')
    
    # 等待命令提示符
    match_idx = await ssh.expect(['$', '#', pexpect.EOF, pexpect.TIMEOUT])
    if match_idx not in (0, 1):
        raise Exception('未找到服务器命令提示符')
    
    # 发送ping命令
    await ssh.sendline('ping -c 4 8.8.8.8')
    
    # 捕获ping输出直到提示符出现
    match_idx = await ssh.expect(['$', '#', pexpect.EOF, pexpect.TIMEOUT], timeout=30)
    ping_output = ssh.before
    
    print('Ping命令输出:\n', ping_output)
    
    # 关闭SSH连接
    await ssh.close()

asyncio.run(run_async_ping_v2())

关键注意事项 #

• 务必处理pexpect.EOF和pexpect.TIMEOUT异常，避免程序因连接中断或超时崩溃。
• 给expect方法设置合理的timeout参数（如ping命令建议设为30秒以上），适配命令执行时长。
• 使用encoding='utf-8'参数，避免处理字节串和字符串的转换问题。

内容的提问来源于stack exchange，提问作者Saikat Dasgupta