python - 尝试后如何正确等待检查 Excel 实例是否已关闭？

Question

我正在使用 Python standard library modules和 pythoncom和 win32com.client来自 PyWin32 的模块包以与 Microsoft Excel 交互。
我得到一个正在运行的 Excel 实例列表作为 COM 对象引用，然后当我想关闭 Excel 实例时，我首先遍历工作簿并关闭它们。然后我执行 Quit method并且在我尝试终止 Excel 进程后，如果它没有终止。
我进行检查( _is_process_running )，因为如果 Excel 进程是僵尸进程( information on how one can be created )或 VBA 监听 before close event，则 Excel 实例可能无法成功关闭。并取消它。
我目前知道何时检查它是否关闭的古怪解决方案是使用 sleep function .它似乎确实有效，但在某些情况下它可能会失败，例如它需要的时间比 sleep 功能等待的时间长。
我认为清除所有 COM 引用并收集垃圾就足以让 Excel 进程终止，如果 Quit方法确实成功了，但它仍然需要一些异步时间。
支票在 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, name, value)

class _excel_workbook_wrapper(_object_wrapper_base_class):
  # __setattr__ takes precedence over properties with setters
  # https://stackoverflow.com/a/15751159
  def __setattr__(self, name, 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, name, 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, False, 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

This answer建议通过从 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

Answer 1

您可以使用 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