使用 python 的 pty 创建实时控制台

Question

我正在尝试创建一个执行环境/shell，它将在服务器上远程执行，它将 stdout、err、in 流式传输到套接字上以在浏览器中呈现。我目前已经尝试过使用subprocess.run 和PIPE 的方法。问题是该过程完成后我得到了标准输出。我想要实现的是逐行、伪终端的实现。

我目前的实现

test.py

def greeter():
    for _ in range(10):
        print('hello world')

greeter()

在外壳中

>>> import subprocess
>>> result = subprocess.run(['python3', 'test.py'], stdout=subprocess.PIPE)
>>> print(result.stdout.decode('utf-8'))
hello world
hello world
hello world
hello world
hello world
hello world
hello world
hello world
hello world
hello world

如果我尝试使用pty 来尝试这个简单的实现，那该怎么做？

Answer 1

我确定某个地方有骗子，但我找不到它

process = subprocess.Popen(cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE,bufsize=0)

for out in iter(process.stdout.readline, b""):
    print(out)

Answer 2

如果您在 Windows 上，那么您将在很长一段时间内进行艰苦的战斗，我很抱歉您将忍受的痛苦（一直在那里）。但是，如果您使用的是 Linux，则可以使用 pexpect 模块。 Pexpect 允许您生成一个可以与之执行双向通信的后台子进程。这对所有类型的系统自动化都很有用，但一个非常常见的用例是 ssh。

import pexpect

child   = pexpect.spawn('python3 test.py')
message = 'hello world'

while True:
    try:
        child.expect(message)
    except pexpect.exceptions.EOF:
        break
    input('child sent: "%s"\nHit enter to continue: ' %
         (message + child.before.decode()))

print('reached end of file!')

我发现创建一个类来处理诸如 ssh 连接之类的复杂事物非常有用，但如果您的用例足够简单，则可能不合适或没有必要。 pexpect.before 是字节类型并省略您正在搜索的模式的方式可能很尴尬，因此创建一个至少为您处理此问题的函数可能是有意义的。

def get_output(child, message):
    return(message + child.before.decode())

如果要向子进程发送消息，可以使用 child.sendline(line)。有关更多详细信息，请查看我链接的文档。

希望能帮到你！

Answer 3

我不知道你是否可以在浏览器中渲染它，但你可以运行一个类似模块的程序，这样你就可以立即得到标准输出：

import importlib
from importlib.machinery import SourceFileLoader

class Program:

    def __init__(self, path, name=''):
        self.path = path
        self.name = name
        if self.path:
            if not self.name:
                self.get_name()
            self.loader = importlib.machinery.SourceFileLoader(self.name, self.path)
            self.spec = importlib.util.spec_from_loader(self.loader.name, self.loader)
            self.mod = importlib.util.module_from_spec(self.spec)
        return

    def get_name(self):
        extension = '.py' #change this if self.path is not python program with extension .py
        self.name = self.path.split('\\')[-1].strip('.py')
        return

    def load(self):
        self.check()
        self.loader.exec_module(self.mod)
        return

    def check(self):
        if not self.path:
            Error('self.file is NOT defined.'.format(path)).throw()
        return

file_path = 'C:\\Users\\RICHGang\\Documents\\projects\\stackoverflow\\ptyconsole\\test.py'
file_name = 'test'
prog = Program(file_path, file_name)   
prog.load()

你可以在 test.py 中添加 sleep 看看有什么区别：

from time import sleep

def greeter():
    for i in range(10):
        sleep(0.3)
        print('hello world')

greeter()

Answer 4

如果您的应用程序要异步处理多个任务，例如从标准输出读取数据然后将其写入 websocket，我建议使用asyncio。

这是一个运行进程并将其输出重定向到 websocket 的示例：

import asyncio.subprocess
import os

from aiohttp.web import (Application, Response, WebSocketResponse, WSMsgType,
                         run_app)


async def on_websocket(request):
    # Prepare aiohttp's websocket...
    resp = WebSocketResponse()
    await resp.prepare(request)
    # ... and store in a global dictionary so it can be closed on shutdown
    request.app['sockets'].append(resp)

    process = await asyncio.create_subprocess_exec(sys.executable,
                                                   '/tmp/test.py',
                                                    stdout=asyncio.subprocess.PIPE,
                                                    stderr=asyncio.subprocess.PIPE,
                                                    bufsize=0)
    # Schedule reading from stdout and stderr as asynchronous tasks.
    stdout_f = asyncio.ensure_future(p.stdout.readline())
    stderr_f = asyncio.ensure_future(p.stderr.readline())

    # returncode will be set upon process's termination.
    while p.returncode is None:
        # Wait for a line in either stdout or stderr.
        await asyncio.wait((stdout_f, stderr_f), return_when=asyncio.FIRST_COMPLETED)

        # If task is done, then line is available.
        if stdout_f.done():
            line = stdout_f.result().encode()
            stdout_f = asyncio.ensure_future(p.stdout.readline())
            await ws.send_str(f'stdout: {line}')

        if stderr_f.done():
            line = stderr_f.result().encode()
            stderr_f = asyncio.ensure_future(p.stderr.readline())
            await ws.send_str(f'stderr: {line}')

    return resp


async def on_shutdown(app):
    for ws in app['sockets']:
        await ws.close()    


async def init(loop):
    app = Application()
    app['sockets'] = []
    app.router.add_get('/', on_websocket)
    app.on_shutdown.append(on_shutdown)
    return app


loop = asyncio.get_event_loop()
app = loop.run_until_complete(init())
run_app(app)

它使用aiohttp 并基于web_ws 和subprocess streams 示例。