为Python脚本提供远程shell

4
我希望创建一种方便简单的方法,通过文件套接字、TCP或其他方式远程连接到我的正在运行的Python脚本,以获得远程交互式shell。
我认为这可以通过IPython等轻松实现。然而,我没有找到任何好的例子。我尝试启动IPython.embed_kernel(),但那是阻塞的。所以我尝试在另一个线程中运行它,但这对我的脚本的其余部分产生了许多奇怪的副作用,我不想要任何副作用(不替换sys.stdoutsys.stderrsys.excepthook或其他)而且它也没有起作用——我无法连接。我发现了这个相关的bug报告这个代码片段,建议使用mock.patch('signal.signal'),但这也没有起作用。此外,我为什么需要这个——我也不希望IPython注册任何信号处理程序。
还有一些hack,比如pyringe和我自己的pydbattach来附加到一些正在运行的Python实例,但它们似乎太hacky了。
也许QdbRemotePythonDebugger可以帮助我?

有时候阻塞是可以接受的,而外部线程不起作用,可以参考我的问题如何使gdb的Python交互式(pi)shell中的tab补全工作?-堆栈溢出了解一些替代方法。 - user202729
1个回答

4

我的当前解决方案是设置一个IPython ZMQ内核。我不只是使用

IPython.embed_kernel()

因为这会产生许多副作用,例如干扰sys.stdoutsys.stderrsys.excepthooksignal.signal等,我不想要这些副作用。另外,embed_kernel()是阻塞的,在单独的线程中并不能真正地开箱即用(请参见此处)。所以,我想出了下面这个代码,但我认为它太过复杂了(这也是我在这里创建功能请求的原因)。
def initIPythonKernel():
  # You can remotely connect to this kernel. See the output on stdout.
  try:
    import IPython.kernel.zmq.ipkernel
    from IPython.kernel.zmq.ipkernel import Kernel
    from IPython.kernel.zmq.heartbeat import Heartbeat
    from IPython.kernel.zmq.session import Session
    from IPython.kernel import write_connection_file
    import zmq
    from zmq.eventloop import ioloop
    from zmq.eventloop.zmqstream import ZMQStream
    IPython.kernel.zmq.ipkernel.signal = lambda sig, f: None  # Overwrite.
  except ImportError, e:
    print "IPython import error, cannot start IPython kernel. %s" % e
    return
  import atexit
  import socket
  import logging
  import threading

  # Do in mainthread to avoid history sqlite DB errors at exit.
  # https://github.com/ipython/ipython/issues/680
  assert isinstance(threading.currentThread(), threading._MainThread)
  try:
    connection_file = "kernel-%s.json" % os.getpid()
    def cleanup_connection_file():
      try:
        os.remove(connection_file)
      except (IOError, OSError):
        pass
    atexit.register(cleanup_connection_file)

    logger = logging.Logger("IPython")
    logger.addHandler(logging.NullHandler())
    session = Session(username=u'kernel')

    context = zmq.Context.instance()
    ip = socket.gethostbyname(socket.gethostname())
    transport = "tcp"
    addr = "%s://%s" % (transport, ip)
    shell_socket = context.socket(zmq.ROUTER)
    shell_port = shell_socket.bind_to_random_port(addr)
    iopub_socket = context.socket(zmq.PUB)
    iopub_port = iopub_socket.bind_to_random_port(addr)
    control_socket = context.socket(zmq.ROUTER)
    control_port = control_socket.bind_to_random_port(addr)

    hb_ctx = zmq.Context()
    heartbeat = Heartbeat(hb_ctx, (transport, ip, 0))
    hb_port = heartbeat.port
    heartbeat.start()

    shell_stream = ZMQStream(shell_socket)
    control_stream = ZMQStream(control_socket)

    kernel = Kernel(session=session,
                    shell_streams=[shell_stream, control_stream],
                    iopub_socket=iopub_socket,
                    log=logger)

    write_connection_file(connection_file,
                          shell_port=shell_port, iopub_port=iopub_port, control_port=control_port, hb_port=hb_port,
                          ip=ip)

    print "To connect another client to this IPython kernel, use:", \
          "ipython console --existing %s" % connection_file
  except Exception, e:
    print "Exception while initializing IPython ZMQ kernel. %s" % e
    return

  def ipython_thread():
    kernel.start()
    try:
      ioloop.IOLoop.instance().start()
    except KeyboardInterrupt:
      pass

  thread = threading.Thread(target=ipython_thread, name="IPython kernel")
  thread.daemon = True
  thread.start()

请注意,这段代码现已过时。我制作了一个,其中应该包含更近期的版本,并且可以通过pip进行安装。
其他附加到运行中的CPython进程而不需要事先准备的替代方法。它们通常使用操作系统的调试功能(或使用gdb/lldb)附加到本地CPython进程,然后注入一些代码或仅分析本地CPython线程堆栈。 以下是其他替代方法,您可以事先准备好Python脚本来侦听某个(tcp/file)套接字,以提供远程调试和/或Python shell / REPL界面。

一些概述和收集的代码示例:

(此概述来自这里。)


干得好!这将对我很有帮助(嵌入在tkinter应用程序中)。我将添加from jupyter_core.paths import jupyter_runtime_dir以便在更易访问的目录中编写文件。 - Tinmarino

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接