寻找node-pty的Python等效模块，实现远程Shell终端功能

Python标准库的pty模块没有像node-pty那样内置事件驱动的on_data回调机制，但可以通过线程监听伪终端输出的方式模拟相同效果，以下是具体实现方案：

一、基于标准库的实现 #

核心思路 #

通过pty.openpty()创建伪终端，将子进程的标准IO绑定到伪终端，再启动独立线程持续读取终端输出，触发自定义的输出处理函数（对应Node.js中的on('data')）。

import os
import pty
import subprocess
import threading
import fcntl
import termios
import struct

def handle_output(data):
    # 这里替换为将终端数据发送到Web应用的逻辑
    print(f"终端输出: {data.decode('utf-8', errors='ignore')}")

def set_terminal_size(master_fd, cols, rows):
    # 设置终端窗口大小，对应Node.js中的cols/rows参数
    winsize = struct.pack('HHHH', rows, cols, 0, 0)
    fcntl.ioctl(master_fd, termios.TIOCSWINSZ, winsize)

def read_terminal_output(master_fd, output_handler):
    # 持续读取终端输出，模拟on_data事件
    while True:
        try:
            data = os.read(master_fd, 1024)
            if not data:
                break
            output_handler(data)
        except OSError:
            break

# 1. 创建伪终端
master_fd, slave_fd = pty.openpty()

# 2. 启动Shell进程，绑定到伪终端
proc = subprocess.Popen(
    ['bash'],
    stdin=slave_fd,
    stdout=slave_fd,
    stderr=slave_fd,
    preexec_fn=os.setsid,  # 创建新会话，避免Shell退出问题
    cwd=os.environ['HOME'],
    env=os.environ
)

# 关闭不需要的从设备文件描述符
os.close(slave_fd)

# 3. 设置终端窗口大小
set_terminal_size(master_fd, cols=100, rows=100)

# 4. 启动线程监听终端输出
threading.Thread(
    target=read_terminal_output,
    args=(master_fd, handle_output),
    daemon=True  # 随主进程退出而终止
).start()

# 5. 向终端发送命令（对应Node.js中的write方法）
os.write(master_fd, b'ls -la\n')

# 保持主进程运行，避免线程退出
try:
    proc.wait()
except KeyboardInterrupt:
    proc.terminate()

二、使用第三方库简化实现 #

如果不想手动处理底层伪终端操作，可以用ptyprocess库（封装了标准库的pty功能），代码更简洁：

首先安装依赖：

pip install ptyprocess

实现代码：

import os
import threading
from ptyprocess import PtyProcess

def handle_output(data):
    # 处理终端输出的逻辑
    print(f"终端输出: {data.decode('utf-8', errors='ignore')}")

def read_terminal_output(proc, output_handler):
    while proc.isalive():
        try:
            data = proc.read(1024)
            if data:
                output_handler(data)
        except EOFError:
            break

# 1. 启动Shell进程
proc = PtyProcess.spawn(
    ['bash'],
    cwd=os.environ['HOME'],
    env=os.environ
)

# 2. 设置终端窗口大小
proc.setwinsize(rows=100, cols=100)

# 3. 启动线程监听输出
threading.Thread(
    target=read_terminal_output,
    args=(proc, handle_output),
    daemon=True
).start()

# 4. 发送命令
proc.write(b'ls -la\n')

# 保持进程运行
try:
    proc.wait()
except KeyboardInterrupt:
    proc.terminate()

关键说明 #

• Python没有原生的事件回调机制，因此用线程轮询读取伪终端输出的方式，模拟node-pty的on('data')行为；
• 线程设为daemon=True是为了让主进程退出时自动终止监听线程；
• 处理输出时注意编码问题，终端输出可能包含非UTF-8字符，建议用errors='ignore'避免解码异常。

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