130 lines
3.8 KiB
Python
130 lines
3.8 KiB
Python
"""Windows 顶层窗口作用域：枚举、唯一性校验和激活。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import ctypes
|
||
import ntpath
|
||
import time
|
||
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
|
||
SW_MAXIMIZE = 3
|
||
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, *, maximize: bool = False) -> None:
|
||
"""激活窗口。
|
||
|
||
maximize=True：主窗进入依赖完整侧栏的流程时使用。
|
||
maximize=False：只前置；**禁止**对已最大化窗口无条件 SW_RESTORE。
|
||
"""
|
||
hwnd = window.hwnd
|
||
iconic = bool(user32.IsIconic(hwnd))
|
||
zoomed = bool(user32.IsZoomed(hwnd))
|
||
if maximize:
|
||
if iconic:
|
||
user32.ShowWindow(hwnd, SW_RESTORE)
|
||
if not zoomed:
|
||
user32.ShowWindow(hwnd, SW_MAXIMIZE)
|
||
else:
|
||
if iconic:
|
||
user32.ShowWindow(hwnd, SW_RESTORE)
|
||
if not user32.SetForegroundWindow(hwnd):
|
||
raise DesktopRpaError("WINDOW_ACTIVATE_FAILED", f"无法激活 hwnd={hwnd}")
|
||
time.sleep(0.35 if maximize else 0.2)
|