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

Question

我正在使用 PyWin32 包中的 Python standard library modules 以及 pythoncom 和 win32com.client 模块与 Microsoft Excel 进行交互。

我得到一个正在运行的 Excel 实例列表作为 COM 对象引用，然后当我想关闭 Excel 实例时，我首先遍历工作簿并关闭它们。然后我执行Quit method 并在尝试终止 Excel 进程（如果它没有终止）之后。

我进行检查 (_is_process_running) 因为 Excel 实例可能无法成功关闭，例如，如果 Excel 进程是僵尸进程 (information on how one can be created) 或 VBA 侦听 before close event 并取消它。

我目前知道何时检查它是否关闭的古怪解决方案是使用sleep function。它似乎确实有效，但在某些情况下它可能会失败，例如它需要的时间比睡眠功能等待的时间长。

我认为如果Quit 方法确实成功，则清除所有 COM 引用并收集垃圾就足以让 Excel 进程终止，但异步仍需要一些时间。

检查在excel.pyw 文件中_excel_application_wrapper 类的close 方法中。

---

生成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

在我看来，您知道如何检测 Excel 实例的当前状态。 您唯一缺少的一点是检测Quitting 操作的事件。

AFAIK，没有办法按您的意思检测事件。 但是一个（可能非常好的）解决方法是设置时间点，例如在列表中，并检查这些点的状态。 如果您担心浪费 1000 毫秒，同时执行过多的检查，您可以将您的列表设置为 [1, 3, 10, 30, ...]，即在 log(time) 中等间距。

即使有可用的事件，我猜你的代码会更“优雅”，但你不会得到比上面的建议更好的性能（除非等待时间在几分钟的范围内或以上）。

Answer 2

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