python 控制台延迟 window 关闭
python console delay window shutdown
我在python 3.6 中编写了一个数据收集器，它将一些数据保存在RAM 中并每分钟将其发送到云端，或者在没有互联网连接时将其保存到磁盘。该应用程序在控制台 window 中是 运行，因此每个人都可以查看它是 运行 还是抛出一些异常。
为了防止数据丢失，我想在 Windows 关机时保存数据。我发现了几个声明使用 win32api.SetConsoleCtrlHandler 的来源（例如 SetConsoleCtrlHandler does not get called on shutdown) or a hidden window and listen to WM_QUERYENDSESSION (for example: Prevent windows shutdown from python）
但这两种方法都没有达到预期的效果。如果控制台 window 关闭，SetConsoleCtrlHandler 会收到信号，但如果整个系统关闭，则不会收到信号。
WM_QUERYENDSESSION 的消息循环只有在我使用没有控制台 window 而不是 python.exe 的 pythonw.exe 时才有效，但我想要一个控制台 window。我想在 python 控制台打开时，控制台会在消息循环执行我的正常关闭之前杀死我的进程。
有没有人有关于如何在 python 控制台内防止 windows 关机的工作示例？
我想我找到了一个合适的解决方案：
我创建了自己的小型控制台应用程序并连接到它的消息队列中以捕获关闭事件。
我还没有对其进行太多测试，我也不知道这是否是一个好的解决方案，但也许它对某人有帮助。
首先是我基于 tkinter 的简单控制台的代码。它以黑色显示标准输出，以红色显示标准错误：
# a simple console based on tkinter to display stdout and stderr
class SimpleConsole(object):
def __init__(self, name):
self.root = Tk()
self.root.title(name)
self.init_ui()
def init_ui(self):
self.text_box = Text(self.root, wrap='word', height = 11, width=50)
self.text_box.grid(column=0, row=0, columnspan = 2, sticky='NSWE', padx=5, pady=5)
self.text_box.tag_config('std', foreground="black")
self.text_box.tag_config('err', foreground="red")
self.text_box.pack(side=LEFT, fill=BOTH, expand = YES)
self.text_box.yview()
self.yscrollbar = Scrollbar(self.root, orient=VERTICAL, command=self.text_box.yview)
self.yscrollbar.pack(side=RIGHT, fill=Y)
self.text_box["yscrollcommand"] = self.yscrollbar.set
sys.stdout = SimpleConsole.StdRedirector(self.text_box, "std")
sys.stderr = SimpleConsole.StdRedirector(self.text_box, "err")
self.update()
class StdRedirector(object):
def __init__(self, text_widget, tag):
self.text_space = text_widget
self.tag = tag
def write(self, string):
self.text_space.insert('end', string, self.tag)
self.text_space.see('end')
def flush(self):
pass
def update(self):
self.root.update()
def get_window_handle(self):
return int(self.root.wm_frame(), 16)
然后我创建了一个 class，它连接到我的控制台的消息队列并管理关闭：
#class to handle a graceful shutdown by hooking into windows message queue
class GracefulShutdown:
def __init__(self, handle):
self.shutdown_requested = False
self._shutdown_functions = []
self.handle = handle
try:
if os.name == 'nt':
# Make a dictionary of message names to be used for printing below
self.msgdict = {}
for name in dir(win32con):
if name.startswith("WM_"):
value = getattr(win32con, name)
self.msgdict[value] = name
# Set the WndProc to our function
self.oldWndProc = win32gui.SetWindowLong(self.handle, win32con.GWL_WNDPROC, self.my_wnd_proc)
if self.oldWndProc == 0:
raise NameError("wndProc override failed!")
self.message_map = {win32con.WM_QUERYENDSESSION: self.hdl_query_end_session,
win32con.WM_ENDSESSION: self.hdl_end_session,
win32con.WM_QUIT: self.hdl_quit,
win32con.WM_DESTROY: self.hdl_destroy,
win32con.WM_CLOSE: self.hdl_close}
# pass a shutdown message to windows
retval = windll.user32.ShutdownBlockReasonCreate(self.handle,c_wchar_p("I'm still saving data!"))
if retval == 0:
raise NameError("shutdownBlockReasonCreate failed!")
except Exception as e:
logging.exception("something went wrong during win32 shutdown detection setup")
#catches all close signals and passes it to our own functions; all other signals are passed to the original function
def my_wnd_proc(self, hwnd, msg, w_param, l_param):
# Display what we've got.
logging.debug(self.msgdict.get(msg), msg, w_param, l_param)
# Restore the old WndProc. Notice the use of wxin32api
# instead of win32gui here. This is to avoid an error due to
# not passing a callable object.
if msg == win32con.WM_DESTROY:
win32api.SetWindowLong(self.handle,
win32con.GWL_WNDPROC,
self.oldWndProc)
#simplify function for calling
def call_window_proc_old():
return win32gui.CallWindowProc(self.oldWndProc, hwnd, msg, w_param, l_param)
#either call our handle functions or call the original wndProc
return self.message_map.get(msg, call_window_proc_old)()
def hdl_query_end_session(self):
logging.info("WM_QUERYENDSESSION received")
self.shutdown_requested = True
#we have to return 0 here to prevent the windows shutdown until our application is closed
return 0
def hdl_end_session(self):
logging.info("WM_ENDSESSION received")
self.exit_gracefully()
return 0
def hdl_quit(self):
logging.info("WM_QUIT received")
self.shutdown_requested = True
return 0
def hdl_destroy(self):
logging.info("WM_DESTROY received")
return 0
def hdl_close(self):
logging.info("WM_CLOSE received")
self.shutdown_requested = True
return 0
def exit_gracefully(self):
logging.info("shutdown request received")
self.shutdown_requested = True
for func in self._shutdown_functions:
try:
func()
except:
logging.exception("Exception during shutdown function:")
logging.info("shutdown request done, bye!")
exit(0)
def add_cleanup_function(self, function):
self._shutdown_functions.append(function)
这里有一些 "main" 代码来启动两个 classes 并测试它：
if __name__ == "__main__":
import time
from logging.handlers import RotatingFileHandler
#setup own console window
console = SimpleConsole("Test Shutdown")
#setup 3 loggers:
#log debug and info to stdout
#log warning and above to stderr
#log info and above to a file
logging.getLogger().setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logging_path = 'graceful_shutdown_test.log'
rot_file_handler = RotatingFileHandler(logging_path, maxBytes=50 * 1024 * 1024, backupCount=5)
rot_file_handler.setFormatter(formatter)
rot_file_handler.setLevel(logging.INFO)
logging.getLogger().addHandler(rot_file_handler)
log_to_stdout = logging.StreamHandler(sys.stdout)
log_to_stdout.setLevel(logging.INFO)
log_to_stdout.addFilter(lambda record: record.levelno <= logging.INFO)
log_to_stdout.setFormatter(formatter)
logging.getLogger().addHandler(log_to_stdout)
log_to_stderr = logging.StreamHandler()
log_to_stderr.setLevel(logging.WARNING)
log_to_stderr.setFormatter(formatter)
logging.getLogger().addHandler(log_to_stderr)
logging.info("start shutdown test")
#init graceful shutdown with tkinter window handle
shutdown = GracefulShutdown(console.get_window_handle())
counter = 0
counterError = 0
#test cleanup function which runs if shutdown is requested
def graceful_shutdown():
logging.info("start shutdown")
time.sleep(15)
logging.info("stop shutdown")
shutdown.add_cleanup_function(graceful_shutdown)
#main test loop
while not shutdown.shutdown_requested:
console.update()
counter += 1
if counter > 50:
logging.info("still alive")
counter = 0
counterError += 1
if counterError > 150:
logging.error("error for test")
try:
raise NameError("i'm a exception")
except:
logging.exception("exception found!")
counterError = 0
time.sleep(0.1)
shutdown.exit_gracefully()
我在python 3.6 中编写了一个数据收集器，它将一些数据保存在RAM 中并每分钟将其发送到云端，或者在没有互联网连接时将其保存到磁盘。该应用程序在控制台 window 中是 运行，因此每个人都可以查看它是 运行 还是抛出一些异常。
为了防止数据丢失，我想在 Windows 关机时保存数据。我发现了几个声明使用 win32api.SetConsoleCtrlHandler 的来源（例如 SetConsoleCtrlHandler does not get called on shutdown) or a hidden window and listen to WM_QUERYENDSESSION (for example: Prevent windows shutdown from python）
但这两种方法都没有达到预期的效果。如果控制台 window 关闭，SetConsoleCtrlHandler 会收到信号，但如果整个系统关闭，则不会收到信号。
WM_QUERYENDSESSION 的消息循环只有在我使用没有控制台 window 而不是 python.exe 的 pythonw.exe 时才有效，但我想要一个控制台 window。我想在 python 控制台打开时，控制台会在消息循环执行我的正常关闭之前杀死我的进程。
有没有人有关于如何在 python 控制台内防止 windows 关机的工作示例？
我想我找到了一个合适的解决方案： 我创建了自己的小型控制台应用程序并连接到它的消息队列中以捕获关闭事件。 我还没有对其进行太多测试，我也不知道这是否是一个好的解决方案，但也许它对某人有帮助。
首先是我基于 tkinter 的简单控制台的代码。它以黑色显示标准输出，以红色显示标准错误：
# a simple console based on tkinter to display stdout and stderr
class SimpleConsole(object):
def __init__(self, name):
self.root = Tk()
self.root.title(name)
self.init_ui()
def init_ui(self):
self.text_box = Text(self.root, wrap='word', height = 11, width=50)
self.text_box.grid(column=0, row=0, columnspan = 2, sticky='NSWE', padx=5, pady=5)
self.text_box.tag_config('std', foreground="black")
self.text_box.tag_config('err', foreground="red")
self.text_box.pack(side=LEFT, fill=BOTH, expand = YES)
self.text_box.yview()
self.yscrollbar = Scrollbar(self.root, orient=VERTICAL, command=self.text_box.yview)
self.yscrollbar.pack(side=RIGHT, fill=Y)
self.text_box["yscrollcommand"] = self.yscrollbar.set
sys.stdout = SimpleConsole.StdRedirector(self.text_box, "std")
sys.stderr = SimpleConsole.StdRedirector(self.text_box, "err")
self.update()
class StdRedirector(object):
def __init__(self, text_widget, tag):
self.text_space = text_widget
self.tag = tag
def write(self, string):
self.text_space.insert('end', string, self.tag)
self.text_space.see('end')
def flush(self):
pass
def update(self):
self.root.update()
def get_window_handle(self):
return int(self.root.wm_frame(), 16)
然后我创建了一个 class，它连接到我的控制台的消息队列并管理关闭：
#class to handle a graceful shutdown by hooking into windows message queue
class GracefulShutdown:
def __init__(self, handle):
self.shutdown_requested = False
self._shutdown_functions = []
self.handle = handle
try:
if os.name == 'nt':
# Make a dictionary of message names to be used for printing below
self.msgdict = {}
for name in dir(win32con):
if name.startswith("WM_"):
value = getattr(win32con, name)
self.msgdict[value] = name
# Set the WndProc to our function
self.oldWndProc = win32gui.SetWindowLong(self.handle, win32con.GWL_WNDPROC, self.my_wnd_proc)
if self.oldWndProc == 0:
raise NameError("wndProc override failed!")
self.message_map = {win32con.WM_QUERYENDSESSION: self.hdl_query_end_session,
win32con.WM_ENDSESSION: self.hdl_end_session,
win32con.WM_QUIT: self.hdl_quit,
win32con.WM_DESTROY: self.hdl_destroy,
win32con.WM_CLOSE: self.hdl_close}
# pass a shutdown message to windows
retval = windll.user32.ShutdownBlockReasonCreate(self.handle,c_wchar_p("I'm still saving data!"))
if retval == 0:
raise NameError("shutdownBlockReasonCreate failed!")
except Exception as e:
logging.exception("something went wrong during win32 shutdown detection setup")
#catches all close signals and passes it to our own functions; all other signals are passed to the original function
def my_wnd_proc(self, hwnd, msg, w_param, l_param):
# Display what we've got.
logging.debug(self.msgdict.get(msg), msg, w_param, l_param)
# Restore the old WndProc. Notice the use of wxin32api
# instead of win32gui here. This is to avoid an error due to
# not passing a callable object.
if msg == win32con.WM_DESTROY:
win32api.SetWindowLong(self.handle,
win32con.GWL_WNDPROC,
self.oldWndProc)
#simplify function for calling
def call_window_proc_old():
return win32gui.CallWindowProc(self.oldWndProc, hwnd, msg, w_param, l_param)
#either call our handle functions or call the original wndProc
return self.message_map.get(msg, call_window_proc_old)()
def hdl_query_end_session(self):
logging.info("WM_QUERYENDSESSION received")
self.shutdown_requested = True
#we have to return 0 here to prevent the windows shutdown until our application is closed
return 0
def hdl_end_session(self):
logging.info("WM_ENDSESSION received")
self.exit_gracefully()
return 0
def hdl_quit(self):
logging.info("WM_QUIT received")
self.shutdown_requested = True
return 0
def hdl_destroy(self):
logging.info("WM_DESTROY received")
return 0
def hdl_close(self):
logging.info("WM_CLOSE received")
self.shutdown_requested = True
return 0
def exit_gracefully(self):
logging.info("shutdown request received")
self.shutdown_requested = True
for func in self._shutdown_functions:
try:
func()
except:
logging.exception("Exception during shutdown function:")
logging.info("shutdown request done, bye!")
exit(0)
def add_cleanup_function(self, function):
self._shutdown_functions.append(function)
这里有一些 "main" 代码来启动两个 classes 并测试它：
if __name__ == "__main__":
import time
from logging.handlers import RotatingFileHandler
#setup own console window
console = SimpleConsole("Test Shutdown")
#setup 3 loggers:
#log debug and info to stdout
#log warning and above to stderr
#log info and above to a file
logging.getLogger().setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logging_path = 'graceful_shutdown_test.log'
rot_file_handler = RotatingFileHandler(logging_path, maxBytes=50 * 1024 * 1024, backupCount=5)
rot_file_handler.setFormatter(formatter)
rot_file_handler.setLevel(logging.INFO)
logging.getLogger().addHandler(rot_file_handler)
log_to_stdout = logging.StreamHandler(sys.stdout)
log_to_stdout.setLevel(logging.INFO)
log_to_stdout.addFilter(lambda record: record.levelno <= logging.INFO)
log_to_stdout.setFormatter(formatter)
logging.getLogger().addHandler(log_to_stdout)
log_to_stderr = logging.StreamHandler()
log_to_stderr.setLevel(logging.WARNING)
log_to_stderr.setFormatter(formatter)
logging.getLogger().addHandler(log_to_stderr)
logging.info("start shutdown test")
#init graceful shutdown with tkinter window handle
shutdown = GracefulShutdown(console.get_window_handle())
counter = 0
counterError = 0
#test cleanup function which runs if shutdown is requested
def graceful_shutdown():
logging.info("start shutdown")
time.sleep(15)
logging.info("stop shutdown")
shutdown.add_cleanup_function(graceful_shutdown)
#main test loop
while not shutdown.shutdown_requested:
console.update()
counter += 1
if counter > 50:
logging.info("still alive")
counter = 0
counterError += 1
if counterError > 150:
logging.error("error for test")
try:
raise NameError("i'm a exception")
except:
logging.exception("exception found!")
counterError = 0
time.sleep(0.1)
shutdown.exit_gracefully()