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

Question

我正在尝试创建一个将在服务器上远程执行的执行环境/ shell,它将stdout,err,流入套接字以在浏览器中呈现.我目前已经尝试过使用subprocess.runa 的方法PIPE.问题是我在进程完成后得到了stdout.我想要实现的是获得逐行,伪终端类型的实现.

我目前的实施

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

如果您的应用程序要与多个任务异步工作，例如从 stdout 读取数据然后将其写入 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和子流程流示例。

Answer 2

我确定某个地方周围有骗子，但我很快找不到

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

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