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