交互式python子过程中的额外回车

编程问答 2022-08-16

问题描述

我正在编写一个基于subprocess的python程序，该程序充当用户输入和子进程之间的代理（试图超越pexpect）。我以this线程为参考，并且从pexpect（_read_incoming()的{​​{1}}方法）中读取了一些代码块来读取输出（popen_spawn方法有效，但不能令人满意）。

代码可以运行，但是有一个问题：似乎有一个附加的回车符正在发送到该进程。当我尝试执行将密码发送到ssh等操作时，这会导致我出现问题。

您能否调查可能是什么问题？谢谢！

代码如下：

fcntl

示例运行：

from queue import Queue,Empty
from threading import Thread
import subprocess
import signal
import fcntl
import os

global terminating
terminating = False

def setNonBlocking(fd):
    """
    Set the file description of the given file descriptor to non-blocking.
    """
    print(fd)
    flags = fcntl.fcntl(fd,fcntl.F_GETFL)
    flags = flags | os.O_NONBLOCK
    fcntl.fcntl(fd,fcntl.F_SETFL,flags)

def enqueue(out,q):
  fileno = out.fileno()
  while not terminating:
    buf = b''
    try:
      buf = os.read(fileno,1024)
      if buf and len(buf)>0:
        q.put(buf)
    except OSError as e:
      #print("OS error: {0}".format(e))
      pass
    if not buf:
      q.put(buf)
  # for line in iter(out.readline,b''):
  #   if len(line.strip()) > 0:
  #     print(line)
  #     q.put(line)
  out.close()
  print('Terminating')
  return

def get_output(q):
  out_str = bytes()
  while True:
    try:
      incoming = q.get_Nowait()
    except Empty:
      break
    else:
      if incoming is None:
        break
      else:
        out_str +=  incoming

  if out_str:
    return out_str
  else:
    return b''

def explore(cmd="/bin/bash"):
  global terminating
  universal_newlines = False
  p = subprocess.Popen([cmd],stdout=subprocess.PIPE,stdin=subprocess.PIPE,stderr=subprocess.STDOUT,bufsize=0,shell=False,universal_newlines=universal_newlines)
  #setNonBlocking(p.stdout)
  outQueue = Queue()

  outThread = Thread(target=enqueue,args=(p.stdout,outQueue))

  outThread.daemon = True

  outThread.start()

  while True:
    try:
      someInput = input()
      print('[In]:'+someInput)
      someInput += '\n'
      if not universal_newlines:
        p.stdin.write(someInput.encode('utf-8'))
      else:
        p.stdin.write(someInput)
      p.stdin.flush()
      out = get_output(outQueue).decode('utf-8')
      print('[Out]:'+out)
      #p.communicate(someInput+'\n')
    except KeyboardInterrupt:
      print('Interrupting')
      p.send_signal(signal.SIGINT)
      terminating = True
      outThread.join()
      break
  p.wait()

if __name__ == '__main__':
  explore()

第二个ls [In]:ls [Out]: [In]: [Out]:explorer.py __init__.py ^CInterrupting Terminating 是来自用户的回车。

更新：

使用In的popen_spawn模块测试了备用服务器。结果相同：

pexpect

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）

interactivepythonpython-3.xsubprocess