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

Question

我试图创建一个执行环境/ shell，它将在服务器上远程执行，它通过套接字将stdout，err流式传输到浏览器中。我目前已经尝试了使用subprocess.run和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

我敢肯定，那里有一个欺骗周围的地方，但我不能很快找到它

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

我不知道你是否能在浏览器中渲染，但你可以像模块，让您得到的stdout立即像这样运行程序：

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来补充睡眠看到其中的差别：

from time import sleep 

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

greeter() 

Answer 4

如果你的应用程序要与多任务异步工作，喜欢读书，从标准输出数据，然后将其写入到WebSocket，我建议使用asyncio。

这里是一个运行过程和它的输出重定向到一个网页套接字的例子：

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例子。