如何解决尝试后如何正确等待检查Excel实例是否已关闭?
我正在使用pythoncom
包中的Python standard library modules和win32com.client
和PyWin32
模块与Microsoft Excel进行交互。
我获得了作为COM对象引用的正在运行的Excel实例的列表,然后当我想关闭Excel实例时,我首先遍历工作簿并关闭它们。然后,我执行Quit method,并尝试终止Excel进程(如果尚未终止)。
我进行检查(_is_process_running
),因为例如如果Excel进程是僵尸进程(information on how one can be created)或VBA监听{{3},则Excel实例可能无法成功关闭}并取消。
我目前最奇怪的解决方案是使用before close event,以了解何时检查其是否关闭。它似乎确实可以工作,但是在某些情况下可能会失败,例如花费的时间比睡眠功能等待的时间长。
我认为,如果Quit
方法确实成功,但清除所有COM引用并收集垃圾足以使Excel进程终止,但是异步处理仍需要一些时间。
检查位于close
文件中_excel_application_wrapper
类的excel.pyw
方法中。
用于生成Excel僵尸进程的简单代码(您可以在任务管理器中看到该进程):
from os import getpid,kill
from win32com.client import DispatchEx
_ = DispatchEx('Excel.Application')
kill(getpid(),9)
这仅用于测试目的,以帮助重现调用Quit
时不会关闭的Excel实例。
使Quit
无法关闭的另一种方法是将此VBA代码添加到Excel中的工作簿中:
Private Sub Workbook_BeforeClose(Cancel As Boolean)
Cancel = True
End Sub
excel_test.py
文件上的代码:
import excel
from traceback import print_exc as print_exception
try:
excel_application_instances = excel.get_application_instances()
for excel_application_instance in excel_application_instances:
# use excel_application_instance here before closing it
# ...
excel_application_instance.close()
except Exception:
print('An exception has occurred. Details of the exception:')
print_exception()
finally:
input('Execution finished.')
excel.pyw
文件上的代码:
from ctypes import byref as by_reference,c_ulong as unsigned_long,windll as windows_dll
from gc import collect as collect_garbage
from pythoncom import CreateBindCtx as create_bind_context,GetRunningObjectTable as get_running_object_table,\
IID_IDispatch as dispatch_interface_iid,_GetInterfaceCount as get_interface_count
from win32com.client import Dispatch as dispatch
class _object_wrapper_base_class():
def __init__(self,object_to_be_wrapped):
# self.__dict__['_wrapped_object'] instead of
# self._wrapped_object to prevent recursive calling of __setattr__
# https://stackoverflow.com/a/12999019
self.__dict__['_wrapped_object'] = object_to_be_wrapped
def __getattr__(self,name):
return getattr(self._wrapped_object,name)
def __setattr__(self,name,value):
setattr(self._wrapped_object,value)
class _excel_workbook_wrapper(_object_wrapper_base_class):
# __setattr__ takes precedence over properties with setters
# https://stackoverflow.com/a/15751159
def __setattr__(self,value):
# raises AttributeError if the attribute doesn't exist
getattr(self,name)
if name in vars(_excel_workbook_wrapper):
attribute = vars(_excel_workbook_wrapper)[name]
# checks if the attribute is a property with a setter
if isinstance(attribute,property) and attribute.fset is not None:
attribute.fset(self,value)
return
setattr(self._wrapped_object,value)
@property
def saved(self):
return self.Saved
@saved.setter
def saved(self,value):
self.Saved = value
def close(self):
self.Close()
class _excel_workbooks_wrapper(_object_wrapper_base_class):
def __getitem__(self,key):
return _excel_workbook_wrapper(self._wrapped_object[key])
class _excel_application_wrapper(_object_wrapper_base_class):
@property
def workbooks(self):
return _excel_workbooks_wrapper(self.Workbooks)
def _get_process(self):
window_handle = self.hWnd
process_identifier = unsigned_long()
windows_dll.user32.GetWindowThreadProcessId(window_handle,by_reference(process_identifier))
return process_identifier.value
def _is_process_running(self,process_identifier):
SYNCHRONIZE = 0x00100000
process_handle = windows_dll.kernel32.OpenProcess(SYNCHRONIZE,False,process_identifier)
returned_value = windows_dll.kernel32.WaitForSingleObject(process_handle,0)
windows_dll.kernel32.CloseHandle(process_handle)
WAIT_TIMEOUT = 0x00000102
return returned_value == WAIT_TIMEOUT
def _terminate_process(self,process_identifier):
PROCESS_TERMINATE = 0x0001
process_handle = windows_dll.kernel32.OpenProcess(PROCESS_TERMINATE,process_identifier)
process_terminated = windows_dll.kernel32.TerminateProcess(process_handle,0)
windows_dll.kernel32.CloseHandle(process_handle)
return process_terminated != 0
def close(self):
for workbook in self.workbooks:
workbook.saved = True
workbook.close()
del workbook
process_identifier = self._get_process()
self.Quit()
del self._wrapped_object
# 0 COM references
print(f'{get_interface_count()} COM references.')
collect_garbage()
# quirky solution to wait for the Excel process to
# terminate if it did closed successfully from self.Quit()
windows_dll.kernel32.Sleep(1000)
# check if the Excel instance closed successfully
# it may not close for example if the Excel process is a zombie process
# or if the VBA listens to the before close event and cancels it
if self._is_process_running(process_identifier=process_identifier):
print('Excel instance failed to close.')
# if the process is still running then attempt to terminate it
if self._terminate_process(process_identifier=process_identifier):
print('The process of the Excel instance was successfully terminated.')
else:
print('The process of the Excel instance failed to be terminated.')
else:
print('Excel instance closed successfully.')
def get_application_instances():
running_object_table = get_running_object_table()
bind_context = create_bind_context()
excel_application_class_clsid = '{00024500-0000-0000-C000-000000000046}'
excel_application_clsid = '{000208D5-0000-0000-C000-000000000046}'
excel_application_instances = []
for moniker in running_object_table:
display_name = moniker.GetDisplayName(bind_context,None)
if excel_application_class_clsid not in display_name:
continue
unknown_com_interface = running_object_table.GetObject(moniker)
dispatch_interface = unknown_com_interface.QueryInterface(dispatch_interface_iid)
dispatch_clsid = str(dispatch_interface.GetTypeInfo().GetTypeAttr().iid)
if dispatch_clsid != excel_application_clsid:
continue
excel_application_instance_com_object = dispatch(dispatch=dispatch_interface)
excel_application_instance = _excel_application_wrapper(excel_application_instance_com_object)
excel_application_instances.append(excel_application_instance)
return excel_application_instances
sleep function建议通过从COM对象中调用某些内容来检查远程过程调用(RPC)服务器是否不可用。我以不同的方式尝试了反复试验而没有成功。例如在self.Quit()
之后添加以下代码。
from pythoncom import com_error,CoUninitialize as co_uninitialize
from traceback import print_exc as print_exception
co_uninitialize()
try:
print(self._wrapped_object)
except com_error as exception:
if exception.hresult == -2147023174: # "The RPC server is unavailable."
print_exception()
else:
raise
解决方法
您可以使用object_name.close
,如果文件未正确关闭,则返回False。
使用您的代码:
def close(self):
for workbook in self.workbooks:
workbook.saved = True
workbook.close()
if workbook.closed:
del workbook
else:
print("Lookout,it's a zombie! Workbook was not deleted")
但是,我还应该提到Pep 343使用Python的with
上下文管理器有更好的解决方案。这样可以确保在进一步执行之前关闭文件。
示例:
with open("file_name","w") as openfile:
# do some work
# "file_name" is now closed
,
在我看来,您知道如何检测Excel实例的当前状态。
您唯一缺少的一点是检测到Quit
ting操作的事件。
AFAIK,无法按照您的意图检测事件。 但是(可能非常好)的解决方法是设置时间点,例如在列表中,然后检查这些点的状态。 如果您担心浪费1000毫秒,并且同时执行过多的检查,则可以将列表设置为[1、3、10、30,...],即以log(time)等距。 / p>
即使有可用的事件,我想您的代码也会更“优雅”,但是您不会获得比上面的建议更好的性能(除非等待时间在几分钟之内)或更高)。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。