如何保持Python pexpect启动的Shell持续运行并返回主脚本控制权

尝试用pexpect实现主循环按顺序向多个子进程发送输入，但运行后进程会卡在启动的Shell中，只能靠Ctrl+C或超时解决。需要在维持Shell运行的前提下，将逻辑控制权返回给主脚本。

相关代码 #

pexpect_test.py #

import pexpect
import sys
from time import sleep

if __name__ == '__main__':
    count = 3
    processes = []
    for x in range(count):
        processes.append(pexpect.spawn('/bin/bash', logfile=sys.stdout.buffer))
    print(f'Processes Created: {len(processes)}')

    inputs = ['a', 'b', 'c']
    for input in inputs:
        print(f'Processing input: {input}')
        for process in processes:
            line = f'python test.py {input}'
            print(f'Sending Line: {line}')
            process.sendline(line)
            print('reading')
            res = process.read()
            print(f'res: {res}')

test.py #

import sys
from time import sleep

if __name__ == '__main__':
    print(f'Process Started: {sys.argv[1]} ')
    sleep(2)
    print(f'Process Complete: {sys.argv[1]}')

运行结果 #

$ python pexpect_test.py 
Processes Created: 3
Processing input: a
Sending Line: python test.py a
python test.py a
reading
$ python test.py a
Process Started: a 
Process Complete: a
$ ^CTraceback (most recent call last):

简化测试示例 #

import pexpect
import sys

if __name__ == '__main__':
    ps = pexpect.spawn('/bin/bash', logfile=sys.stdout.buffer)
    ps.sendline('python test.py 1 a')

    # Expect EOF：卡住直到超时，无法返回控制权
    # ps.expect(pexpect.EOF)

    # Expect终端字符'$ '：直接跳过等待，Shell无法保持活跃
    # ps.expect_exact('$ ')

    # Expect进程输出的特定字符串：能等待并返回控制权，但要求被执行进程输出固定标识，不可行（不管理这些进程的代码）
    # ps.expect_exact('Complete')

    # Expect两次终端字符：恰好符合需求，但原理不明
    # ps.expect('\$.*\$ ')

备注：subprocess.run(...)性能不足，实际场景中命令初始化需2-3秒，且要处理10000+输入，必须复用Shell进程。

解决方案 #

核心问题分析 #

使用process.read()会卡住，是因为read()会持续等待子进程输出流结束（EOF），但复用的Shell进程不会退出，因此永远无法等到EOF。正确的做法是等待Shell提示符出现——这代表当前命令执行完成，Shell回到可接收新命令的状态。

正确实现方式 #

1. 匹配Shell提示符的通用正则
   不同Shell的提示符可能为$（普通用户）或#（root），部分还会包含用户名、主机名，可用通用正则匹配：

   prompt_pattern = r'[$#] '
   

   发送命令后，用expect等待该提示符，即可准确判断命令执行完成，同时维持Shell活跃。

2. 修改主循环代码
   将process.read()替换为process.expect(prompt_pattern)，同时可添加超时、异常处理逻辑：

   修改后的pexpect_test.py核心部分：

   import pexpect
   import sys
   
   if __name__ == '__main__':
       count = 3
       processes = []
       prompt_pattern = r'[$#] '
       # 初始化Shell时先等待初始提示符
       for x in range(count):
           p = pexpect.spawn('/bin/bash', logfile=sys.stdout.buffer)
           p.expect(prompt_pattern)
           processes.append(p)
       print(f'Processes Created: {len(processes)}')
   
       inputs = ['a', 'b', 'c']
       for input in inputs:
           print(f'Processing input: {input}')
           for process in processes:
               line = f'python test.py {input}'
               print(f'Sending Line: {line}')
               process.sendline(line)
               # 等待命令执行完成，回到Shell提示符
               idx = process.expect([prompt_pattern, pexpect.TIMEOUT, pexpect.EOF])
               if idx == 0:
                   print(f'Command completed for input {input}')
               elif idx == 1:
                   print(f'Timeout waiting for input {input}')
               elif idx == 2:
                   print(f'Shell exited for input {input}')
   

关于ps.expect('\$.*\$ ')的原理 #

这个正则匹配的是两次Shell提示符之间的内容：第一个$是输入命令前的初始提示符，第二个$是命令执行完成后的新提示符。整个匹配过程会等待命令执行完毕、Shell回到就绪状态，因此能满足需求。但该写法依赖Shell回显开启（默认开启），不如直接匹配最终提示符可靠。

额外注意事项 #

• 若Shell提示符包含用户名、主机名（如user@host:~$ ），可调整正则为r'\w+@\w+:\S+[$#] '适配。
• 建议设置合理超时时间，如process.expect(prompt_pattern, timeout=30)，避免因命令异常导致永久卡住。
• 如需捕获命令输出，可使用process.before，它包含两次expect之间的所有输出（需注意过滤Shell回显内容）。

内容的提问来源于stack exchange，提问作者Cogito Ergo Sum