"""Thư viện dùng chung quản lý cửa sổ Win32 cho Path of Exile 2. Cung cấp các hàm tương tác cửa sổ cấp thấp, chuẩn hóa tọa độ Client Rectangle, tính toán hệ số DPI Scale và gắn luồng vào Desktop tương tác của người dùng. Tuân thủ nghiêm ngặt chuẩn tài liệu Google Docstrings theo Rule 16. """ import ctypes import ctypes.wintypes import os import time from typing import List, Optional, Tuple from src.common.win32_process import is_poe2_pid _global_default_desk: Optional[int] = None def attach_to_default_desktop() -> Optional[int]: """Gắn luồng hiện tại vào Desktop 'Default' tương tác của phiên người dùng. Hàm này mở trạm làm việc 'WinSta0' và desktop 'Default', sau đó gọi SetThreadDesktop để cho phép tiến trình chạy ngầm hoặc service có thể tìm thấy và chụp cửa sổ Win32 của game Path of Exile 2 mà không bị lỗi Access Denied. Preconditions: Hệ điều hành Windows (nt). Luồng gọi có quyền truy cập desktop tương tác. Postconditions: Nếu thành công, luồng hiện tại được liên kết với 'Default' desktop. Returns: Optional[int]: Handle của desktop nếu thành công, hoặc None nếu thất bại / không chạy trên Windows. Thread-safety: Thread-safe. Có thể gọi an toàn từ bất kỳ worker thread nào. Example: >>> hdesk = attach_to_default_desktop() >>> if hdesk: ... print("Đã gắn luồng vào Default Desktop thành công.") """ if os.name != "nt": return None try: user32 = ctypes.windll.user32 hwinsta = user32.OpenWindowStationW("WinSta0", False, 0x10000000) if hwinsta: user32.SetProcessWindowStation(hwinsta) # Luôn ưu tiên mở Active Input Desktop động (Interactive Input Session) # để đảm bảo SendInput và keybd_event không bị chặn bởi ERROR_ACCESS_DENIED (5) h_input = user32.OpenInputDesktop(0, False, 0x10000000) if h_input: user32.SetThreadDesktop(h_input) return int(h_input) # Fallback sang desktop Default tĩnh nếu OpenInputDesktop không mở được h_default = user32.OpenDesktopW("Default", 0, False, 0x10000000) if h_default: user32.SetThreadDesktop(h_default) return int(h_default) except Exception: pass return None _cached_window_desktops = {} def attach_to_window_desktop(hwnd: int) -> bool: """Gắn luồng hiện tại vào Desktop chứa cửa sổ HWND chỉ định. Tự động truy vấn Thread ID sở hữu HWND qua GetWindowThreadProcessId, lấy handle Desktop qua GetThreadDesktop hoặc duyệt EnumDesktopsW với quyền MAXIMUM_ALLOWED. Đảm bảo 100% tương thích bất kể cửa sổ nằm trên Default desktop hay sandbox desktop (exebox). """ if os.name != "nt" or not hwnd: return False user32 = ctypes.windll.user32 if user32.IsWindow(hwnd): return True cached = _cached_window_desktops.get(hwnd) if cached and user32.SetThreadDesktop(cached): if user32.IsWindow(hwnd): return True try: tid = user32.GetWindowThreadProcessId(hwnd, None) if tid: hdesk = user32.GetThreadDesktop(tid) if hdesk and user32.SetThreadDesktop(hdesk): if user32.IsWindow(hwnd): _cached_window_desktops[hwnd] = hdesk return True except Exception: pass try: hwinsta = user32.GetProcessWindowStation() if hwinsta: DESKTOPENUMPROCW = ctypes.WINFUNCTYPE(ctypes.wintypes.BOOL, ctypes.c_wchar_p, ctypes.wintypes.LPARAM) desktops = [] def _d_cb(d, lp): if d: desktops.append(str(d)) return True user32.EnumDesktopsW(hwinsta, DESKTOPENUMPROCW(_d_cb), 0) for d in desktops: hd = user32.OpenDesktopW(d, 0, False, 0x10000000) if hd: if user32.SetThreadDesktop(hd): if user32.IsWindow(hwnd): _cached_window_desktops[hwnd] = hd return True user32.CloseDesktop(hd) except Exception: pass return False def find_poe2_window(title_candidates: Optional[List[str]] = None) -> int: """Tìm handle (HWND) của cửa sổ game Path of Exile 2. Thực hiện dò tìm thông qua lớp cửa sổ đã biết (POEWindowClass, POEC2_MainWindow) và danh sách tiêu đề ứng viên (title candidates). Tự động duyệt qua toàn bộ các cửa sổ trên Default Desktop nếu EnumDesktopWindows khả dụng. Args: title_candidates: Danh sách chuỗi tiêu đề cửa sổ cần khớp (không phân biệt hoa thường). Nếu None, mặc định sử dụng ["path of exile 2", "path of exile"]. Returns: int: HWND của cửa sổ game nếu tìm thấy, hoặc 0 nếu không tìm thấy. Thread-safety: Thread-safe. Example: >>> hwnd = find_poe2_window() >>> if hwnd != 0: ... print(f"Tìm thấy cửa sổ game: HWND={hex(hwnd)}") """ if os.name != "nt": return 0 attach_to_default_desktop() if title_candidates is None: title_candidates = ["path of exile 2", "path of exile"] normalized_titles = [t.lower() for t in title_candidates] target_classes = ["poewindowclass", "poec2_mainwindow"] user32 = ctypes.windll.user32 found_hwnd = 0 def _is_valid_poe2_hwnd(h: int) -> bool: if not h or not user32.IsWindow(h) or not user32.IsWindowVisible(h): return False w_pid = ctypes.c_ulong() user32.GetWindowThreadProcessId(h, ctypes.byref(w_pid)) if w_pid.value == 0: return False return is_poe2_pid(w_pid.value) # 1. Thử FindWindow trực tiếp theo Class kèm xác thực Process Name for cls in ["POEWindowClass", "POEC2_MainWindow"]: hwnd = user32.FindWindowW(cls, None) if hwnd and _is_valid_poe2_hwnd(hwnd): return int(hwnd) # 2. Thử FindWindow trực tiếp theo Title kèm xác thực Process Name for title in title_candidates: hwnd = user32.FindWindowW(None, title) if hwnd and _is_valid_poe2_hwnd(hwnd): return int(hwnd) # 3. Quét EnumWindows / EnumDesktopWindows toàn diện (Two-Factor Window Validation) hdesk = attach_to_default_desktop() WNDENUMPROC = ctypes.WINFUNCTYPE(ctypes.wintypes.BOOL, ctypes.wintypes.HWND, ctypes.wintypes.LPARAM) # Bỏ qua các class phổ biến của trình duyệt và IDE để chống dương tính giả (False Positive) ignored_classes = { "cabinetwclass", "chrome_widgetwin_1", "applicationframewindow", "mozilla_window_class", "window_class", } def _enum_proc(hwnd, lparam): nonlocal found_hwnd if not user32.IsWindow(hwnd) or not user32.IsWindowVisible(hwnd): return True cbuf = ctypes.create_unicode_buffer(256) user32.GetClassNameW(hwnd, cbuf, 256) cls_name = cbuf.value.strip().lower() if cls_name in ignored_classes: return True tbuf = ctypes.create_unicode_buffer(256) user32.GetWindowTextW(hwnd, tbuf, 256) win_title = tbuf.value.strip().lower() # Kiểm tra lớp cửa sổ kết hợp xác thực PID hình ảnh tiến trình if cls_name in target_classes: if _is_valid_poe2_hwnd(hwnd): found_hwnd = int(hwnd) return False # Kiểm tra tiêu đề cửa sổ kết hợp xác thực PID hình ảnh tiến trình for target_title in normalized_titles: if target_title in win_title: if _is_valid_poe2_hwnd(hwnd): found_hwnd = int(hwnd) return False return True cb = WNDENUMPROC(_enum_proc) try: if hdesk: user32.EnumDesktopWindows(hdesk, cb, 0) else: user32.EnumWindows(cb, 0) except Exception: pass # Nếu chưa tìm thấy trên Default desktop, tự động quét các desktop khác trên WinSta0 (Multi-Desktop Fallback) if not found_hwnd: try: DESKTOPENUMPROCW = ctypes.WINFUNCTYPE(ctypes.wintypes.BOOL, ctypes.c_wchar_p, ctypes.wintypes.LPARAM) desktops_list = [] def _d_cb(dname, lp): if dname: desktops_list.append(str(dname)) return True hwinsta = user32.GetProcessWindowStation() if hwinsta: user32.EnumDesktopsW(hwinsta, DESKTOPENUMPROCW(_d_cb), 0) for d in desktops_list: if d.lower() == "default": continue hd = user32.OpenDesktopW(d, 0, False, 0x10000000) if hd: try: user32.EnumDesktopWindows(hd, cb, 0) if found_hwnd: user32.SetThreadDesktop(hd) break finally: user32.CloseDesktop(hd) except Exception: pass return found_hwnd def get_client_rect(hwnd: int) -> Tuple[int, int, int, int]: """Lấy tọa độ vùng vẽ thực tế (Client Rectangle) của cửa sổ game theo pixel màn hình. Tự động bù trừ thanh tiêu đề (title bar) và viền cửa sổ thông qua Win32 API GetClientRect và ClientToScreen, trả về tọa độ tuyệt đối chuẩn xác cho việc tính toán giao diện, quét mẫu pixel và bắt sự kiện chuột. Args: hwnd: Handle cửa sổ Win32 của tiến trình Path of Exile 2. Returns: Tuple[int, int, int, int]: Bộ tứ (left, top, width, height) theo tọa độ màn hình. Trả về (0, 0, 0, 0) nếu HWND không hợp lệ hoặc gọi Win32 API thất bại. Thread-safety: Thread-safe. Example: >>> left, top, width, height = get_client_rect(hwnd) >>> print(f"Client Area: {width}x{height} tại ({left}, {top})") """ if os.name != "nt" or not hwnd: return (0, 0, 0, 0) attach_to_window_desktop(hwnd) user32 = ctypes.windll.user32 if not user32.IsWindow(hwnd): return (0, 0, 0, 0) rect = ctypes.wintypes.RECT() if not user32.GetClientRect(hwnd, ctypes.byref(rect)): return (0, 0, 0, 0) point = ctypes.wintypes.POINT(0, 0) if not user32.ClientToScreen(hwnd, ctypes.byref(point)): return (0, 0, 0, 0) width = rect.right - rect.left height = rect.bottom - rect.top return (point.x, point.y, width, height) def get_dpi_scale(hwnd: int) -> float: """Tính toán hệ số phóng đại DPI (DPI Scale Factor) của cửa sổ chỉ định. Sử dụng GetDpiForWindow (Windows 10 v1607+ / Windows 11) với cơ chế fallback về Device Context GetDeviceCaps(LOGPIXELSX) để đảm bảo độ co giãn giao diện và tọa độ chuột luôn chính xác trên mọi thiết lập màn hình (HiDPI / 4K / Ultrawide). Args: hwnd: Handle cửa sổ Win32 cần đo lường. Returns: float: Hệ số tỉ lệ DPI (ví dụ 1.0 cho 100% 96 DPI, 1.25 cho 125% 120 DPI, 1.5 cho 150%). Trả về 1.0 nếu không xác định được. Thread-safety: Thread-safe. Example: >>> scale = get_dpi_scale(hwnd) >>> print(f"DPI Scale: {scale * 100:.0f}%") """ if os.name != "nt" or not hwnd: return 1.0 user32 = ctypes.windll.user32 # 1. Thử GetDpiForWindow (Win10 build 14393 trở lên) if hasattr(user32, "GetDpiForWindow"): try: dpi = user32.GetDpiForWindow(hwnd) if dpi > 0: return float(dpi) / 96.0 except Exception: pass # 2. Fallback qua Device Context GetDeviceCaps try: gdi32 = ctypes.windll.gdi32 hdc = user32.GetDC(hwnd) if hdc: try: LOGPIXELSX = 88 dpi = gdi32.GetDeviceCaps(hdc, LOGPIXELSX) if dpi > 0: return float(dpi) / 96.0 finally: user32.ReleaseDC(hwnd, hdc) except Exception: pass return 1.0 def ensure_window_focus(hwnd: int, timeout_ms: int = 500) -> bool: """Tự động chuyển tab và đưa cửa sổ chỉ định lên foreground với quyền tương tác an toàn. Sử dụng kỹ thuật AttachThreadInput kết hợp SetForegroundWindow và BringWindowToTop để đảm bảo quyền kích hoạt cửa sổ ngay cả khi tiến trình gọi đang chạy nền hoặc không có quyền input focus trực tiếp. Tự động khôi phục nếu cửa sổ đang bị thu nhỏ (minimized). Args: hwnd: Handle cửa sổ Win32 cần đưa lên tiêu điểm. timeout_ms: Thời gian chờ tối đa tính bằng mili-giây để xác nhận focus (mặc định 500ms). Returns: bool: True nếu cửa sổ đã nằm ở foreground thành công, False nếu thất bại hoặc không hợp lệ. Thread-safety: Thread-safe. Có thể gọi từ bất kỳ worker thread nào. Example: >>> hwnd = find_poe2_window() >>> if hwnd and ensure_window_focus(hwnd): ... print("Cửa sổ POE2 đã sẵn sàng nhận phím/chuột.") """ if os.name != "nt" or not hwnd: return False user32 = ctypes.windll.user32 kernel32 = ctypes.windll.kernel32 attach_to_window_desktop(hwnd) if not user32.IsWindow(hwnd): return False # Virtual Display Safe-Guard: Nếu cửa sổ đang nằm trên màn hình ảo IddCx, # KHÔNG gọi SetForegroundWindow hay Alt-bypass để tránh cướp tiêu điểm của người dùng trên Monitor 1. try: from src.common.virtual_display_manager import is_window_on_virtual_display if is_window_on_virtual_display(hwnd): if user32.IsWindowVisible(hwnd) and not user32.IsIconic(hwnd): return True except Exception: pass # Fast-path: Nếu cửa sổ POE2 đã thực sự có Foreground Focus thì trả về True ngay, # không làm gián đoạn trạng thái chuột hoặc phím WASD đang nhấn if user32.GetForegroundWindow() == hwnd and is_poe2_window_focused(hwnd): return True try: # 1. Khôi phục nếu cửa sổ đang bị thu nhỏ và đưa lên TOPMOST if user32.IsIconic(hwnd): user32.ShowWindow(hwnd, 9) # SW_RESTORE else: user32.ShowWindow(hwnd, 9) # SW_RESTORE user32.SetWindowPos(hwnd, -1, 0, 0, 0, 0, 0x0001 | 0x0002 | 0x0040) # HWND_TOPMOST # 2. Đồng bộ Input Thread và Alt-Key Bypass để đảm bảo quyền SetForegroundWindow cur_thread = kernel32.GetCurrentThreadId() fg_hwnd = user32.GetForegroundWindow() fg_thread = user32.GetWindowThreadProcessId(fg_hwnd, None) if fg_hwnd else 0 target_thread = user32.GetWindowThreadProcessId(hwnd, None) attached_fg = False attached_target = False if fg_thread and cur_thread != fg_thread: attached_fg = bool(user32.AttachThreadInput(cur_thread, fg_thread, True)) if target_thread and cur_thread != target_thread: attached_target = bool(user32.AttachThreadInput(cur_thread, target_thread, True)) # Unlocks SetForegroundWindow if blocked (Windows 10/11 ForegroundLockTimeout bypass) try: if hasattr(user32, "LockSetForegroundWindow"): user32.LockSetForegroundWindow(2) # LSFW_UNLOCK = 2 SPI_SETFOREGROUNDLOCKTIMEOUT = 0x2001 SPIF_SENDCHANGE = 0x0002 user32.SystemParametersInfoW(SPI_SETFOREGROUNDLOCKTIMEOUT, 0, 0, SPIF_SENDCHANGE) except Exception: pass try: if hasattr(user32, "AllowSetForegroundWindow"): user32.AllowSetForegroundWindow(-1) # ASFW_ANY = -1 if hasattr(user32, "SwitchToThisWindow"): user32.SwitchToThisWindow(hwnd, True) user32.keybd_event(0x12, 0, 0, 0) # Alt down user32.BringWindowToTop(hwnd) user32.SetForegroundWindow(hwnd) user32.SetActiveWindow(hwnd) user32.SetFocus(hwnd) user32.keybd_event(0x12, 0, 2, 0) # Alt up finally: if attached_target and target_thread: user32.AttachThreadInput(cur_thread, target_thread, False) if attached_fg and fg_thread: user32.AttachThreadInput(cur_thread, fg_thread, False) # 3. Chờ xác nhận tiêu điểm deadline = time.time() + (timeout_ms / 1000.0) while time.time() < deadline: fg = user32.GetForegroundWindow() if fg == hwnd: return True target_pid = ctypes.c_ulong() user32.GetWindowThreadProcessId(hwnd, ctypes.byref(target_pid)) fg_pid = ctypes.c_ulong() user32.GetWindowThreadProcessId(fg, ctypes.byref(fg_pid)) if target_pid.value > 0 and fg_pid.value == target_pid.value: return True if is_poe2_window_focused(hwnd): return True time.sleep(0.02) return is_poe2_window_focused(hwnd) except Exception as e: import traceback print(f"[ensure_window_focus Exception] {e}", flush=True) traceback.print_exc() return is_poe2_window_focused(hwnd) def snap_window_to_half(hwnd: int, side: str = "right") -> bool: """Tự động đặt cửa sổ game chiếm nửa màn hình (trái hoặc phải) để người dùng làm việc bên còn lại. Args: hwnd: Handle cửa sổ Win32 của game. side: 'right' (nửa phải) hoặc 'left' (nửa trái). Returns: bool: True nếu thành công. """ if os.name != "nt" or not hwnd: return False user32 = ctypes.windll.user32 if not user32.IsWindow(hwnd): return False rect = ctypes.wintypes.RECT() SPI_GETWORKAREA = 0x0030 if not user32.SystemParametersInfoW(SPI_GETWORKAREA, 0, ctypes.byref(rect), 0): rect.left = 0 rect.top = 0 rect.right = user32.GetSystemMetrics(0) rect.bottom = user32.GetSystemMetrics(1) screen_w = rect.right - rect.left screen_h = rect.bottom - rect.top half_w = screen_w // 2 new_x = rect.left + half_w if side == "right" else rect.left new_y = rect.top new_w = half_w new_h = screen_h SWP_NOZORDER = 0x0004 SWP_SHOWWINDOW = 0x0040 return bool(user32.SetWindowPos(hwnd, 0, new_x, new_y, new_w, new_h, SWP_NOZORDER | SWP_SHOWWINDOW)) def position_poe2_window_safely(hwnd: int, target_x: int = 850, target_y: int = 60) -> bool: """Tự động định vị cửa sổ POE2 tại vùng an toàn trên desktop (tránh che khuất bởi ứng dụng khác).""" if os.name != "nt" or not hwnd: return False user32 = ctypes.windll.user32 if not user32.IsWindow(hwnd): return False rect = ctypes.wintypes.RECT() if not user32.GetWindowRect(hwnd, ctypes.byref(rect)): return False w = rect.right - rect.left h = rect.bottom - rect.top HWND_TOP = 0 SWP_SHOWWINDOW = 0x0040 res = user32.SetWindowPos(hwnd, HWND_TOP, target_x, target_y, w, h, SWP_SHOWWINDOW) ensure_window_focus(hwnd, timeout_ms=500) return bool(res) def is_poe2_window_focused(hwnd: Optional[int] = None) -> bool: """Kiểm tra xem cửa sổ game Path of Exile 2 hiện tại có đang giữ Foreground Focus hay không. Bảo vệ Focus Interlock: Đảm bảo chuột và bàn phím vật lý không bao giờ bị bơm rò rỉ sang màn hình desktop hoặc các ứng dụng khác của người dùng khi game bị unfocused. Args: hwnd: Handle cửa sổ Win32 tùy chọn để kiểm tra cụ thể. Nếu None, tự động kiểm tra GetForegroundWindow() với các ClassName đã biết ('POEWindowClass', 'POEC2_MainWindow') hoặc HWND từ find_poe2_window(). Returns: bool: True nếu cửa sổ POE2 đang có Focus trên Foreground, ngược lại False. """ if os.name != "nt": return False try: user32 = ctypes.windll.user32 target_hwnd = hwnd if (hwnd and hwnd > 0) else find_poe2_window() # Virtual Display Safe-Guard: Nếu cửa sổ nằm trên màn hình ảo IddCx và đang hiển thị, # coi như đã có focus an toàn để pipeline tự hành ngầm hoạt động mà không cướp chuột/focus của người dùng. if target_hwnd: try: from src.common.virtual_display_manager import is_window_on_virtual_display if is_window_on_virtual_display(target_hwnd): attach_to_window_desktop(target_hwnd) if user32.IsWindow(target_hwnd) and user32.IsWindowVisible(target_hwnd) and not user32.IsIconic(target_hwnd): return True except Exception: pass attach_to_default_desktop() fg_hwnd = user32.GetForegroundWindow() if not fg_hwnd: # Fallback khi chạy ngầm / service / DirectX full-screen nơi GetForegroundWindow() trả về 0: # Nếu target_hwnd hợp lệ, đang hiển thị và không bị thu nhỏ (Iconic) if target_hwnd and user32.IsWindow(target_hwnd) and user32.IsWindowVisible(target_hwnd): if not user32.IsIconic(target_hwnd): return True return False # 1. Nếu có hwnd chỉ định, kiểm tra trực tiếp if hwnd is not None and hwnd > 0: if int(fg_hwnd) == int(hwnd): return True # 2. Kiểm tra cùng Process ID (hỗ trợ các sub-windows, modals, DX12 swapchains) target_hwnd = hwnd if (hwnd and hwnd > 0) else find_poe2_window() if target_hwnd and target_hwnd > 0: target_pid = ctypes.c_ulong() user32.GetWindowThreadProcessId(target_hwnd, ctypes.byref(target_pid)) fg_pid = ctypes.c_ulong() user32.GetWindowThreadProcessId(fg_hwnd, ctypes.byref(fg_pid)) if target_pid.value > 0 and fg_pid.value == target_pid.value: return True # 3. Kiểm tra ClassName của Foreground Window class_buf = ctypes.create_unicode_buffer(256) user32.GetClassNameW(fg_hwnd, class_buf, 256) cls_name = class_buf.value.lower() if cls_name in ("poewindowclass", "poec2_mainwindow"): return True # 4. Đối chiếu với find_poe2_window() if target_hwnd > 0 and int(fg_hwnd) == target_hwnd: return True # 5. Kiểm tra tiêu đề kết hợp bắt buộc xác thực Process ID (Two-Factor Validation) # Loại trừ 100% trường hợp tab trình duyệt hoặc IDE có chứa chữ "path of exile" title_buf = ctypes.create_unicode_buffer(512) user32.GetWindowTextW(fg_hwnd, title_buf, 512) title_val = title_buf.value.lower() if "path of exile" in title_val: fg_pid = ctypes.c_ulong() user32.GetWindowThreadProcessId(fg_hwnd, ctypes.byref(fg_pid)) if fg_pid.value > 0 and is_poe2_pid(fg_pid.value): return True return False except Exception: return False