使用python的pty创建一个live cons

Question

如果要在服务器上远程执行shell/streams，将在该环境中创建一个streams。我目前尝试过将subprocess.run与PIPE一起使用的方法。问题是我在进程完成后得到stdout。我想实现的是得到一个逐行的，伪终端类的实现。在

我当前的实现

test.py

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

greeter()

在壳里呢

^{pr2}$

如果我试着用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

如果您的应用程序要异步处理多个任务，比如从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和{a4}示例。在

Answer 3

如果你在窗户上，那么你将在很长一段时间内与一场艰苦的战斗，我为你所承受的痛苦感到抱歉。但是，如果您在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连接这样复杂的事情是非常有用的，但是如果您的用例足够简单，那么这可能是不合适的或不必要的。路途预期之前类型为bytes，并且忽略了您正在搜索的模式可能会很尴尬，因此创建一个至少为您处理此问题的函数可能是有意义的。在

^{pr2}$

如果要向子进程发送消息，可以使用子级.sendline（行）。有关更多详细信息，请查看我链接的文档。在

我希望我能帮忙！在