如何在Python脚本中维持两个Bash终端会话存活？适配方案咨询

需求概述 #

需按以下步骤自动化执行脚本，且维持终端会话状态：

1. 在terminal_1中执行source script1.sh，成功后执行python script2.py
2. 开启terminal_2，执行python script3.py，成功后执行python script4.py
3. 回到维持会话的terminal_1，执行python script5.py，完成流程

现有调研方法对比 #

• 方法1：subprocess.Popen结合线程读取输出
  通过subprocess.Popen(["bash"], stderr=subprocess.PIPE, shell=False, stdin=subprocess.PIPE, stdout=subprocess.PIPE)启动bash进程，用线程实时读取输出。需手动处理进程输入输出交互，维持会话状态的逻辑较繁琐。
• 方法2：pexpect.spawn
  利用pexpect库的交互式进程控制能力，直接模拟终端会话，支持维持进程上下文（如环境变量、工作状态），适配多步骤会话需求。
• 方法3：脚本执行结果写入文件验证
  每步执行后将输出写入文件，校验成功再执行下一步。缺点是流程冗余，耗时较长，且无法维持终端会话状态。

适配性结论 #

方法2（pexpect）是最适配需求的方案：

• 原生支持交互式会话维持，能轻松保留terminal_1的环境变量（比如source script1.sh后的环境）
• 内置进程状态检测、输出匹配功能，无需手动处理线程和IO读写的复杂逻辑
• 在Cygwin环境下兼容性良好，可直接模拟bash终端的交互行为

方法1也可实现，但需要手动处理stdin/stdout的同步、会话状态保存，代码复杂度更高；方法3效率低下，不推荐。

推荐方案示例 #

方案1：pexpect实现示例 #

import pexpect

def run_in_session(session, cmd, success_pattern):
    """在指定会话中执行命令，等待成功标识"""
    session.sendline(cmd)
    # 等待命令执行完成，根据实际输出调整匹配模式（如bash提示符或自定义成功关键词）
    index = session.expect([success_pattern, pexpect.TIMEOUT, pexpect.EOF])
    if index != 0:
        raise RuntimeError(f"执行命令`{cmd}`失败：超时或进程退出")
    return session.before

# 启动terminal_1会话，替换为你的工作目录
terminal1 = pexpect.spawn('bash', cwd='/path/to/your/scripts', encoding='utf-8')
# 等待bash初始提示符
terminal1.expect(r'\$ ')

# 步骤1：执行source script1.sh
run_in_session(terminal1, 'source script1.sh', r'\$ ')
# 步骤2：执行script2.py，替换为实际成功输出关键词
run_in_session(terminal1, 'python script2.py', 'script2 executed successfully')

# 步骤3：启动terminal_2会话
terminal2 = pexpect.spawn('bash', cwd='/path/to/your/scripts', encoding='utf-8')
terminal2.expect(r'\$ ')
run_in_session(terminal2, 'python script3.py', 'script3 executed successfully')

# 步骤4：执行script4.py
run_in_session(terminal2, 'python script4.py', 'script4 executed successfully')

# 步骤5：回到terminal_1执行script5.py
run_in_session(terminal1, 'python script5.py', 'script5 executed successfully')

# 关闭会话
terminal1.sendline('exit')
terminal2.sendline('exit')
terminal1.wait()
terminal2.wait()

方案2：subprocess实现示例（手动维护会话） #

import subprocess
import threading
import queue

def read_output(pipe, output_queue):
    """读取进程输出并放入队列"""
    for line in iter(pipe.readline, b''):
        output_queue.put(line.decode('utf-8'))
    pipe.close()

def run_cmd(session, cmd, success_keyword, output_queue):
    """在subprocess会话中执行命令，等待成功标识"""
    session.stdin.write(f"{cmd}\n".encode('utf-8'))
    session.stdin.flush()
    
    while True:
        line = output_queue.get()
        print(line.strip())
        if success_keyword in line:
            break
        if session.poll() is not None:
            raise RuntimeError(f"进程已退出，命令`{cmd}`执行失败")

# 启动terminal_1
terminal1 = subprocess.Popen(
    ["bash"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    cwd='/path/to/your/scripts',
    shell=False
)
# 启动线程读取输出
output_queue1 = queue.Queue()
thread1 = threading.Thread(target=read_output, args=(terminal1.stdout, output_queue1))
thread1.daemon = True
thread1.start()
# 跳过初始bash提示符
output_queue1.get()

# 步骤1和2
run_cmd(terminal1, 'source script1.sh', '$', output_queue1)
run_cmd(terminal1, 'python script2.py', 'script2 executed successfully', output_queue1)

# 启动terminal_2
terminal2 = subprocess.Popen(
    ["bash"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    cwd='/path/to/your/scripts',
    shell=False
)
output_queue2 = queue.Queue()
thread2 = threading.Thread(target=read_output, args=(terminal2.stdout, output_queue2))
thread2.daemon = True
thread2.start()
output_queue2.get()

# 步骤3和4
run_cmd(terminal2, 'python script3.py', 'script3 executed successfully', output_queue2)
run_cmd(terminal2, 'python script4.py', 'script4 executed successfully', output_queue2)

# 步骤5
run_cmd(terminal1, 'python script5.py', 'script5 executed successfully', output_queue1)

# 关闭会话
terminal1.stdin.write(b"exit\n")
terminal1.stdin.flush()
terminal2.stdin.write(b"exit\n")
terminal2.stdin.flush()
terminal1.wait()
terminal2.wait()

其他可行方案 #

• Cygwin shell脚本结合状态同步：编写bash脚本，用&后台启动terminal_2进程，通过临时文件记录执行状态，实现流程同步。适合熟悉shell脚本的场景。
• tmux/screen会话管理：在Cygwin中安装tmux或screen，用Python调用命令创建会话、发送命令，同样能维持终端上下文，需提前安装对应工具。

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