gpt4 book ai didi

python - 为Python脚本提供远程shell

转载 作者:太空宇宙 更新时间:2023-11-04 05:53:09 25 4
gpt4 key购买 nike

我想创建一种方便简单的方法来远程连接到我正在运行的 Python 脚本(通过文件套接字、TCP 或其他方式)以获取远程交互式 shell。

我认为这可以通过 IPython 等方式轻松实现。但是,我并没有真正找到任何好的例子。我尝试启动 IPython.embed_kernel(),但这是阻塞的。所以我尝试在另一个线程中运行它,但这对我的脚本的其余部分有很多奇怪的副作用,我不想要任何副作用(不替换 sys.stdoutsys .stderrsys.excepthook 或其他),它也没有用——我无法连接。我找到了 this related bug reportthis code snippet建议使用 mock.patch('signal.signal') 但这也没有用。另外,我为什么需要它——我也不希望 IPython 注册任何信号处理程序。

还有像pyringe这样的hack和我自己的 pydbattach附加到一些正在运行的 Python 实例,但它们似乎太老套了。

也许 QdbRemotePythonDebugger可以帮我吗?

最佳答案

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

IPython.embed_kernel()

因为这有很多副作用,比如搞乱 sys.stdoutsys.stderrsys.excepthooksignal.signal 等,我不想要这些副作用。此外,embed_kernel() 是阻塞的,并不能在单独的线程中开箱即用(参见 here)。

所以,我想出了这段代码,在我看来它太复杂了。 (这就是我创建功能请求 here 的原因。)

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()

请注意,此代码现在已过时。我做了一个包here它应该包含更新的版本,并且可以通过 pip 安装。


无需事先准备就可以附加到正在运行的 CPython 进程的其他替代方法。那些通常使用操作系统调试功能(或使用 gdb/lldb)附加到 native CPython 进程,然后注入(inject)一些代码或仅分析 native CPython 线程堆栈。

这里有其他备选方案,您可以预先准备 Python 脚本以监听某些(tcp/文件)套接字,以提供用于远程调试的接口(interface)和/或仅提供 Python shell/REPL。

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

(此概述来自 here。)

关于python - 为Python脚本提供远程shell,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/29148319/

25 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com