"""UIA / MSAA 后端（可选依赖 uiautomation；MSAA 用系统 oleacc）。""" from __future__ import annotations import ctypes import logging from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeout from ctypes import wintypes from typing import Callable, Optional, TypeVar _logger = logging.getLogger("jiangchang_desktop_wake.uia") T = TypeVar("T") # 企微 UI 被 N2 钩死后，UIA 会永久阻塞；必须限时 _UIA_TIMEOUT_S = 8.0 def _call_with_timeout(fn: Callable[[], T], timeout_s: float = _UIA_TIMEOUT_S) -> T: """在工作线程跑 UIA；线程内必须 CoInitialize（UIAutomationInitializerInThread）。""" def _threaded() -> T: import uiautomation as auto # 子线程用 UIA 必须初始化，否则：尚未调用 CoInitialize / UIAutomationCore.dll initializer = getattr(auto, "UIAutomationInitializerInThread", None) if initializer is not None: with initializer(debug=False): return fn() # 极老版本兜底 try: ctypes.windll.ole32.CoInitialize(None) except Exception: pass try: return fn() finally: try: ctypes.windll.ole32.CoUninitialize() except Exception: pass with ThreadPoolExecutor(max_workers=1) as pool: fut = pool.submit(_threaded) try: return fut.result(timeout=timeout_s) except FuturesTimeout as exc: raise TimeoutError(f"uia_timeout_after_{timeout_s}s") from exc WM_GETOBJECT = 0x003D OBJID_CLIENT = 0xFFFFFFFC OBJID_WINDOW = 0x00000000 def uia_available() -> bool: try: import uiautomation # noqa: F401 return True except ImportError: return False def control_from_hwnd(hwnd: int): import uiautomation as auto return auto.ControlFromHandle(int(hwnd)) def count_findall(hwnd: int, *, max_depth: int = 50) -> tuple[int, list[str]]: """返回 (遍历节点数, 采到的 Name 样本)。超时则 (0, [])。""" if not uia_available(): return 0, [] def _work() -> tuple[int, list[str]]: import uiautomation as auto root = control_from_hwnd(hwnd) names: list[str] = [] count = 0 try: if hasattr(auto, "WalkControl"): for c, _depth in auto.WalkControl( root, includeTop=True, maxDepth=max_depth ): count += 1 try: n = (c.Name or "").strip() if n and len(names) < 200: names.append(n) except Exception: pass return count, names except Exception as exc: _logger.debug("WalkControl failed: %s", exc) try: cond = ( auto.CreateTrueCondition() if hasattr(auto, "CreateTrueCondition") else None ) if cond is not None and hasattr(root, "FindAll"): controls = root.FindAll(max_depth, cond) count = len(controls or []) for c in (controls or [])[:200]: try: n = (c.Name or "").strip() if n: names.append(n) except Exception: pass return count, names except Exception as exc: _logger.debug("FindAll failed: %s", exc) queue = [root] depth_map = {id(root): 0} while queue: cur = queue.pop(0) count += 1 try: n = (cur.Name or "").strip() if n and len(names) < 200: names.append(n) except Exception: pass d = depth_map.get(id(cur), 0) if d >= max_depth: continue try: children = cur.GetChildren() except Exception: children = [] for ch in children or []: depth_map[id(ch)] = d + 1 queue.append(ch) return count, names try: return _call_with_timeout(_work) except TimeoutError: _logger.warning("count_findall timed out hwnd=%s", hwnd) return 0, [] def markers_present(names: list[str], markers: list[str]) -> tuple[list[str], list[str]]: name_set = set(names) # also allow substring match for AutomationId-like markers appearing in Name found: list[str] = [] missing: list[str] = [] for m in markers: if m in name_set or any(m == n or m in n for n in names): found.append(m) else: # try live Exists by Name via uiautomation if root available later missing.append(m) return found, missing def find_markers_on_hwnd(hwnd: int, markers: list[str], *, depth: int = 30) -> tuple[list[str], list[str]]: if not markers: return [], [] if not uia_available(): return [], list(markers) def _work() -> tuple[list[str], list[str]]: return _find_markers_on_hwnd_impl(hwnd, markers, depth=depth) try: return _call_with_timeout(_work, timeout_s=6.0) except TimeoutError: _logger.warning("find_markers timed out hwnd=%s", hwnd) return [], list(markers) def _find_markers_on_hwnd_impl( hwnd: int, markers: list[str], *, depth: int = 30 ) -> tuple[list[str], list[str]]: root = control_from_hwnd(hwnd) found: list[str] = [] missing: list[str] = [] for m in markers: try: c = root.Control(searchDepth=depth, Name=m) if c.Exists(0.4, 0.1): found.append(m) else: # AutomationId c2 = root.Control(searchDepth=depth, AutomationId=m) if c2.Exists(0.3, 0.1): found.append(m) else: missing.append(m) except Exception: missing.append(m) return found, missing def send_wm_getobject(hwnd: int) -> str: user32 = ctypes.windll.user32 SMTO_ABORTIFHUNG = 0x0002 result = ctypes.c_size_t(0) def _one(objid: int) -> int: ok = user32.SendMessageTimeoutW( ctypes.c_void_p(int(hwnd)), WM_GETOBJECT, 0, objid, SMTO_ABORTIFHUNG, 2000, ctypes.byref(result), ) return int(result.value) if ok else -1 r1 = _one(OBJID_CLIENT) r2 = _one(OBJID_WINDOW) return f"client_lresult={r1} window_lresult={r2}" def msaa_from_window(hwnd: int) -> str: """AccessibleObjectFromWindow(OBJID_CLIENT)；成功即说明 MSAA 通道有响应。""" oleacc = ctypes.windll.oleacc class GUID(ctypes.Structure): _fields_ = [ ("Data1", ctypes.c_ulong), ("Data2", ctypes.c_ushort), ("Data3", ctypes.c_ushort), ("Data4", ctypes.c_ubyte * 8), ] # IID_IAccessible iid = GUID( 0x618736E0, 0x3C3D, 0x11CF, (ctypes.c_ubyte * 8)(0x81, 0x0C, 0x00, 0xAA, 0x00, 0x38, 0x9B, 0x71), ) punk = ctypes.c_void_p() hr = oleacc.AccessibleObjectFromWindow( wintypes.HWND(int(hwnd)), ctypes.c_ulong(OBJID_CLIENT), ctypes.byref(iid), ctypes.byref(punk), ) if hr != 0 or not punk.value: return f"hr={hr} punk=null" try: # IUnknown::Release (vtable[2]) — best-effort，避免 COM 泄漏 pobj = ctypes.cast(punk, ctypes.POINTER(ctypes.c_void_p)) vtbl = ctypes.cast(pobj[0], ctypes.POINTER(ctypes.c_void_p)) release = ctypes.WINFUNCTYPE(ctypes.c_ulong, ctypes.c_void_p)(vtbl[2]) release(punk) except Exception as exc: return f"hr={hr} acquired=1 release_err={exc}" return f"hr={hr} acquired=1" def deep_uia_walk(hwnd: int, *, max_nodes: int = 2000) -> str: """主动以 UIA 客户端身份深遍历，刺激部分 App 挂 Provider。""" if not uia_available(): return "uia_unavailable" count, names = count_findall(hwnd) sample = ",".join(names[:8]) return f"walk_count={count} sample={sample!r}"