112 lines
3.2 KiB
Python
112 lines
3.2 KiB
Python
"""Windows 顶层窗口作用域：枚举、唯一性校验和激活。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ctypes
|
|
import ntpath
|
|
from ctypes import wintypes
|
|
from dataclasses import dataclass
|
|
from typing import Iterable
|
|
|
|
from util.selectors import WindowSpec
|
|
|
|
|
|
class DesktopRpaError(RuntimeError):
|
|
def __init__(self, code: str, message: str) -> None:
|
|
super().__init__(f"ERROR:{code} {message}")
|
|
self.code = code
|
|
self.message = message
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WindowCandidate:
|
|
hwnd: int
|
|
pid: int
|
|
process_name: str
|
|
class_name: str
|
|
title: str
|
|
visible: bool
|
|
|
|
|
|
user32 = ctypes.windll.user32
|
|
kernel32 = ctypes.windll.kernel32
|
|
SW_RESTORE = 9
|
|
PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
|
|
|
|
|
|
def _process_name(pid: int) -> str:
|
|
handle = kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
|
|
if not handle:
|
|
return ""
|
|
try:
|
|
size = wintypes.DWORD(32768)
|
|
buffer = ctypes.create_unicode_buffer(size.value)
|
|
if not kernel32.QueryFullProcessImageNameW(
|
|
handle,
|
|
0,
|
|
buffer,
|
|
ctypes.byref(size),
|
|
):
|
|
return ""
|
|
return ntpath.splitext(ntpath.basename(buffer.value))[0]
|
|
finally:
|
|
kernel32.CloseHandle(handle)
|
|
|
|
|
|
def enumerate_top_windows() -> list[WindowCandidate]:
|
|
found: list[WindowCandidate] = []
|
|
|
|
@ctypes.WINFUNCTYPE(ctypes.c_bool, wintypes.HWND, wintypes.LPARAM)
|
|
def callback(hwnd, _lparam):
|
|
pid = wintypes.DWORD()
|
|
user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
|
|
cls = ctypes.create_unicode_buffer(256)
|
|
user32.GetClassNameW(hwnd, cls, 256)
|
|
length = user32.GetWindowTextLengthW(hwnd)
|
|
title = ctypes.create_unicode_buffer(length + 1)
|
|
user32.GetWindowTextW(hwnd, title, length + 1)
|
|
found.append(
|
|
WindowCandidate(
|
|
hwnd=int(hwnd),
|
|
pid=int(pid.value),
|
|
process_name=_process_name(int(pid.value)),
|
|
class_name=cls.value,
|
|
title=title.value,
|
|
visible=bool(user32.IsWindowVisible(hwnd)),
|
|
)
|
|
)
|
|
return True
|
|
|
|
user32.EnumWindows(callback, 0)
|
|
return found
|
|
|
|
|
|
def resolve_unique_window(
|
|
spec: WindowSpec,
|
|
*,
|
|
candidates: Iterable[WindowCandidate] | None = None,
|
|
) -> WindowCandidate:
|
|
pool = list(candidates) if candidates is not None else enumerate_top_windows()
|
|
matched = [
|
|
item
|
|
for item in pool
|
|
if item.visible
|
|
and item.process_name.casefold() == spec.process_name.casefold()
|
|
and item.class_name == spec.class_name
|
|
and (spec.title is None or item.title == spec.title)
|
|
]
|
|
if not matched:
|
|
raise DesktopRpaError("WINDOW_NOT_FOUND", f"{spec.logical_name} 未找到")
|
|
if len(matched) != 1:
|
|
raise DesktopRpaError(
|
|
"ELEMENT_AMBIGUOUS",
|
|
f"{spec.logical_name} 匹配 {len(matched)} 个窗口，请按 PID/账号补充作用域",
|
|
)
|
|
return matched[0]
|
|
|
|
|
|
def activate_window(window: WindowCandidate) -> None:
|
|
user32.ShowWindow(window.hwnd, SW_RESTORE)
|
|
if not user32.SetForegroundWindow(window.hwnd):
|
|
raise DesktopRpaError("WINDOW_ACTIVATE_FAILED", f"无法激活 hwnd={window.hwnd}")
|