Python实现带专属TTY的交互式Bash并捕获KeyboardInterrupt

我看了你的代码，没法捕获KeyboardInterrupt主要是几个终端模式和信号处理的问题在搞鬼，我给你一步步修复，同时保留异步读写的特性：

问题根源分析 #

1. 终端原始模式导致SIGINT无法被Python捕获：你调用了tty.setraw()把标准输入设为原始模式，这时候按下Ctrl+C产生的SIGINT信号不会被Python进程接收，而是直接作为字符传递给了bash子进程。
2. 子进程组未被正确终止：用os.setsid()创建了新的会话，bash成了会话组长，单独终止bash进程可能没法清理它启动的子进程。
3. select无阻塞等待：主循环里的select.select()没有超时，会一直阻塞，没法及时响应信号。

修改后的完整代码 #

import subprocess
import sys
import os
import signal
import atexit
import select
import termios
import tty
import pty

# 全局变量存储子进程组ID和原始终端设置
child_pgid = None
old_tty = None

class ProcessException(Exception):
    def __init__(self, command, exitCode, output):
        self.command = command
        self.exitCode = exitCode
        self.output = output
        super().__init__(f"Command '{command}' failed with exit code {exitCode}")

def signal_handler(signum, frame):
    """捕获SIGINT信号，终止子进程组并清理终端"""
    global child_pgid, old_tty
    if child_pgid:
        print("\nCleaning up all bash child jobs...")
        # 终止整个进程组，包括bash和它的子进程
        os.killpg(child_pgid, signal.SIGTERM)
    # 恢复原始终端模式
    if old_tty:
        termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
    sys.exit(0)

def exit_protocol():
    """程序退出时的清理逻辑"""
    global child_pgid, old_tty
    if old_tty:
        termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
    if child_pgid:
        print("\nShutting down bash process group", child_pgid)
        try:
            os.killpg(child_pgid, signal.SIGTERM)
        except OSError:
            # 进程组可能已经退出，忽略错误
            pass

def execute(command):
    global child_pgid, old_tty
    # 保存原始终端设置
    old_tty = termios.tcgetattr(sys.stdin)
    try:
        # 设置终端为cbreak模式（比raw模式更友好，保留部分信号处理）
        tty.setcbreak(sys.stdin.fileno())
        master_fd, slave_fd = pty.openpty()
        process = subprocess.Popen(
            command,
            preexec_fn=os.setsid,
            stdin=slave_fd,
            stdout=slave_fd,
            stderr=slave_fd,
            universal_newlines=True
        )
        # 获取子进程组ID（因为setsid后，进程组ID等于进程ID）
        child_pgid = process.pid

        while process.poll() is None:
            # 给select加超时，避免无限阻塞，让信号有机会被处理
            r, w, e = select.select([sys.stdin, master_fd], [], [], 0.1)
            if sys.stdin in r:
                d = os.read(sys.stdin.fileno(), 10240)
                os.write(master_fd, d)
            elif master_fd in r:
                o = os.read(master_fd, 10240)
                if o:
                    os.write(sys.stdout.fileno(), o)
        # 读取剩余输出
        output = process.communicate()[0]
        exit_code = process.returncode
        if exit_code == 0:
            return output
        else:
            raise ProcessException(command, exit_code, output)
    finally:
        # 确保终端设置被恢复
        if old_tty:
            termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)

# 注册信号处理器和退出清理函数
signal.signal(signal.SIGINT, signal_handler)
atexit.register(exit_protocol)

try:
    execute('bash')
except ProcessException as e:
    print(f"Error executing command: {e}")

关键改动说明 #

• 替换raw模式为cbreak模式：tty.setcbreak()相比raw模式，会保留SIGINT等信号的传递，让Python能捕获到Ctrl+C。
• 注册SIGINT信号处理器：专门处理Ctrl+C，终止整个子进程组（用os.killpg），确保bash和它启动的所有子进程都被清理。
• select添加超时：每次等待0.1秒，这样主循环能定期检查信号，不会一直阻塞。
• 完善终端恢复逻辑：在finally块和退出函数里都加入终端恢复，避免程序异常退出后终端处于异常状态。
• 使用进程组ID替代进程ID：因为os.setsid()创建了新会话，用进程组ID可以一次性终止所有相关子进程。

这样修改后，你按下Ctrl+C就能正常触发中断，程序会清理bash及其子进程，同时恢复终端的正常模式。

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