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

Posted 2023-02-19

【中文标题】尝试后如何正确等待检查 Excel 实例是不是已关闭？

【英文标题】：How to properly wait to check if an Excel instance has closed after attempting to?尝试后如何正确等待检查 Excel 实例是否已关闭？

【发布时间】：2020-12-23 14:07:21

【问题描述】：

我正在使用 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://***.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://***.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

【问题讨论】：

那么你的问题是什么？ 1）您在某些情况下存在 Excel 正在运行的进程而您没有检测到它们，或者 2）您可以 100% 正确识别所有 Excel 正在运行的进程，但您不知道如何将它们全部杀死。

@sancho.sReinstateMonicaCellio 第二个选项接近它。我可以识别所有正在运行的 Excel 实例。我可以终止任何进程。只是我只想将其作为最后一个资源，以防使用 Excel 的 Quit() 方法正确终止它不起作用。

我还是不明白你要完成什么目标，你做不到。那会是“确保退出 Excel 进程的唯一方法是杀死它”吗？

@sancho.sReinstateMonicaCellio 否。我当前的解决方案执行以下操作：遍历每个正在运行的实例，对它们做任何我想做的事情，然后当我想关闭它们时，我首先执行 Quit() 这通常关闭它。除了在某些极少数情况下不会，如给出的示例中所示。所以它会在一段时间（1 秒）后检查（实例的进程），看看它是否确实关闭了。如果它确实继续，否则它会通过终止进程来强制它关闭。我的问题是关于等待 1 秒的部分。

@sancho.sReinstateMonicaCellio 因为从 Quit 方法关闭可能需要少于或多于 1 秒。一个合适的解决方案是检测Quit 何时完成，然后检查它是否有效（关闭）。因为如果 Quit 花费的时间少于 1 秒，那么 Python 代码将不必要地等待整整一秒，如果花费的时间更长，那么代码会在 Quit 方法尚未完成运行时终止进程。 （我认为Quit 是同步的，问题是它没有返回关于它是否工作以及在实例的进程关闭之前是否工作的值）。

【参考方案1】：

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

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

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

【参考方案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

【讨论】：

关闭所有工作簿不会关闭 Excel 应用程序实例。 workbook 是 excel._excel_workbook_wrapper 类的对象，不具有 closed 属性（print(hasattr(workbook, 'closed')) 打印 False）。使用with 对我的情况也没有帮助。我认为您可能混淆了这个问题，或者只是没有测试您的解决方案尝试。