""" Module Khởi Chạy & Quản Lý Client Path of Exile 2 (Standalone & Steam Launcher) Tự động tìm kiếm, kích hoạt PathOfExile.exe và chờ cửa sổ game sẵn sàng. """ import ctypes import os import subprocess import sys import time from ctypes import wintypes from typing import Optional, Any try: import psutil _HAS_PSUTIL = True except ImportError: _HAS_PSUTIL = False try: getattr(sys.stdout, "reconfigure", lambda **kw: None)(encoding="utf-8", errors="replace") except Exception: pass from src.common.win32_process import is_poe2_running, find_process_by_name from src.common.win32_window import find_poe2_window # Windows API Constants & Structures TH32CS_SNAPPROCESS = 0x00000002 class PROCESSENTRY32W(ctypes.Structure): _fields_ = [ ("dwSize", wintypes.DWORD), ("cntUsage", wintypes.DWORD), ("th32ProcessID", wintypes.DWORD), ("th32DefaultHeapID", ctypes.c_size_t), ("th32ModuleID", wintypes.DWORD), ("cntThreads", wintypes.DWORD), ("th32ParentProcessID", wintypes.DWORD), ("pcPriClassBase", wintypes.LONG), ("dwFlags", wintypes.DWORD), ("szExeFile", wintypes.WCHAR * 260), ] class RECT(ctypes.Structure): _fields_ = [ ("left", wintypes.LONG), ("top", wintypes.LONG), ("right", wintypes.LONG), ("bottom", wintypes.LONG), ] class Poe2ClientLauncher: DEFAULT_STANDALONE_PATHS = [ r"C:\Program Files (x86)\Grinding Gear Games\Path of Exile 2\PathOfExile.exe", r"C:\Program Files (x86)\Grinding Gear Games\Path of Exile 2\PathOfExile_x64.exe", r"C:\Program Files\Grinding Gear Games\Path of Exile 2\PathOfExile.exe", r"C:\Program Files (x86)\Steam\steamapps\common\Path of Exile 2\PathOfExile.exe", r"C:\Program Files (x86)\Steam\steamapps\common\Path of Exile 2\PathOfExileSteam.exe", r"D:\Games\Path of Exile 2\PathOfExile.exe", r"E:\Games\Path of Exile 2\PathOfExile.exe", ] TARGET_PROCESS_NAMES = { "pathofexile.exe", "pathofexilesteam.exe", "pathofexile_x64.exe", "pathofexile_x64steam.exe", "pathofexile2.exe", } TARGET_WINDOW_CLASSES = [ "POEWindowClass", "POEC2_MainWindow", ] TARGET_WINDOW_TITLES = [ "Path of Exile 2", "Path of Exile", ] @classmethod def is_process_running(cls) -> bool: """ Kiểm tra xem một trong các tiến trình game POE2 có đang chạy hay không. Sử dụng thư viện dùng chung src.common.win32_process. """ return is_poe2_running() @classmethod def find_client_hwnd(cls) -> Optional[int]: """Tìm HWND của cửa sổ POE2 theo class hoặc window title.""" hwnd = find_poe2_window(cls.TARGET_WINDOW_TITLES) return hwnd if hwnd != 0 else None @classmethod def is_client_window_ready(cls) -> bool: """ Kiểm tra cửa sổ game đã hiển thị (IsWindowVisible) và có kích thước thực tế (> 200x200). """ if os.name != "nt": return False try: hwnd = cls.find_client_hwnd() if not hwnd: return False user32 = ctypes.windll.user32 if not user32.IsWindow(hwnd): return False if not user32.IsWindowVisible(hwnd): return False rect = RECT() if user32.GetWindowRect(hwnd, ctypes.byref(rect)): width = rect.right - rect.left height = rect.bottom - rect.top if width > 200 and height > 200: return True except Exception: pass return False @classmethod def is_client_running(cls) -> bool: """ Kiểm tra xem client POE2 có đang chạy hay không. 1. Quét danh sách tiến trình (Process Name). 2. Quét tìm HWND cửa sổ theo Window Class và Window Title. """ # 1. Kiểm tra Process Name if cls.is_process_running(): return True # 2. Kiểm tra HWND cửa sổ if cls.find_client_hwnd() is not None: return True return False @classmethod def find_standalone_exe(cls) -> Optional[str]: """Tự động tìm đường dẫn file thực thi PathOfExile.exe.""" # 1. Kiểm tra các đường dẫn mặc định phổ biến for path in cls.DEFAULT_STANDALONE_PATHS: if os.path.isfile(path) and os.path.getsize(path) > 1000000: return path # 2. Kiểm tra Registry Windows if os.name == "nt": try: import winreg for root_key in (winreg.HKEY_LOCAL_MACHINE, winreg.HKEY_CURRENT_USER): for sub_path in ( r"SOFTWARE\Grinding Gear Games\Path of Exile 2", r"SOFTWARE\WOW6432Node\Grinding Gear Games\Path of Exile 2" ): try: with winreg.OpenKey(root_key, sub_path) as key: install_dir, _ = winreg.QueryValueEx(key, "InstallLocation") if install_dir: exe = os.path.join(install_dir, "PathOfExile.exe") if os.path.isfile(exe): return exe except Exception: continue except Exception: pass return None @classmethod def kill_client_processes(cls) -> None: """Dọn sạch mọi tiến trình POE2 đang chạy bằng taskkill / psutil.""" print("[ClientLauncher] Dọn sạch các tiến trình POE2 zombie...", flush=True) if _HAS_PSUTIL: for proc in psutil.process_iter(["pid", "name"]): try: pname = proc.info.get("name") if pname and pname.lower() in cls.TARGET_PROCESS_NAMES: proc.kill() except Exception: pass if os.name == "nt": for name in cls.TARGET_PROCESS_NAMES: try: subprocess.run( f'cmd /c "taskkill /F /IM {name} >nul 2>&1 || exit 0"', shell=True, capture_output=True, timeout=5 ) except Exception: pass time.sleep(1.0) @classmethod def launch_client(cls, exe_path: Optional[str] = None, wait_timeout_sec: int = 90) -> bool: """ Khởi chạy POE2 Standalone/Steam Client và chờ cửa sổ game sẵn sàng. - Nếu game đã đang chạy và có cửa sổ sẵn sàng: Trả về True ngay, TUYỆT ĐỐI không gọi mở thêm instance mới. - Nếu có tiến trình nhưng không có HWND cửa sổ: * Chờ tối đa wait_timeout_sec xem cửa sổ có xuất hiện không (tránh kill nhầm game đang nạp shader 30-45s). * Nếu sau wait_timeout_sec vẫn không có HWND (zombie process), mới dọn dẹp và khởi chạy fresh instance. - Thực thi khởi chạy ưu tiên subprocess.Popen([target_exe], cwd=working_dir) và fallback sang ShellExecuteW. - Vòng lặp chờ (wait_timeout_sec): kiểm tra cls.is_client_running() và cls.is_client_window_ready(). """ if cls.is_client_running(): if cls.is_process_running(): if cls.find_client_hwnd() is not None or cls.is_client_window_ready(): print("[ClientLauncher] Client POE2 đã đang chạy và có cửa sổ. Không khởi chạy thêm instance mới.", flush=True) return True # Tiến trình đang chạy nhưng chưa có HWND, chờ tiến trình nạp xong (POE2 cần 30-45s cold start) print(f"[ClientLauncher] Phát hiện tiến trình POE2 đang chạy nhưng chưa có HWND. Đang chờ tối đa {wait_timeout_sec}s cho cửa sổ xuất hiện...", flush=True) start_wait_hwnd = time.time() while time.time() - start_wait_hwnd < wait_timeout_sec: if cls.find_client_hwnd() is not None or cls.is_client_window_ready(): print(f"[ClientLauncher] Cửa sổ POE2 đã xuất hiện sau {time.time() - start_wait_hwnd:.1f}s. Không khởi chạy thêm instance mới.", flush=True) time.sleep(2.0) return True time.sleep(1.0) # Zombie process thực sự không có HWND cửa sổ sau full timeout print(f"[ClientLauncher Warning] Phát hiện tiến trình zombie không có HWND sau {wait_timeout_sec}s. Đang dọn dẹp để khởi chạy fresh instance...", flush=True) cls.kill_client_processes() else: print("[ClientLauncher] Client POE2 đã đang chạy. Không khởi chạy thêm instance mới.", flush=True) return True target_exe = exe_path or cls.find_standalone_exe() if not target_exe or not os.path.isfile(target_exe): print(f"[ClientLauncher Error] Không tìm thấy PathOfExile.exe tại: {target_exe}", flush=True) return False working_dir = os.path.dirname(target_exe) try: print(f"[ClientLauncher] Đang khởi chạy POE2 từ: {target_exe}...", flush=True) launched = False # 1. Khởi chạy ưu tiên bằng subprocess.Popen trực tiếp trên Desktop tương tác try: flags = 0 if os.name == "nt": # CREATE_NEW_PROCESS_GROUP (0x200) flags = subprocess.CREATE_NEW_PROCESS_GROUP subprocess.Popen([target_exe], cwd=working_dir, shell=False, creationflags=flags) launched = True except Exception as ex_sub: print(f"[ClientLauncher Warning] subprocess.Popen thất bại: {ex_sub}, fallback sang ShellExecuteW...", flush=True) # 2. Fallback sang ShellExecuteW trên Windows nếu Popen thất bại if not launched and os.name == "nt": try: # Thử mở asInvoker thông thường ('open') ret = ctypes.windll.shell32.ShellExecuteW( None, "open", target_exe, None, working_dir, 1 # SW_SHOWNORMAL ) if ret <= 32: # Fallback cuối cùng sang runas nếu quyền thông thường bị từ chối ret = ctypes.windll.shell32.ShellExecuteW( None, "runas", target_exe, None, working_dir, 1 ) if ret > 32: launched = True else: print(f"[ClientLauncher Warning] ShellExecuteW trả về mã: {ret}", flush=True) except Exception as ex_shell: print(f"[ClientLauncher Warning] ShellExecuteW thất bại: {ex_shell}", flush=True) if not launched: return False except Exception as e: print(f"[ClientLauncher Error] Lỗi khi khởi chạy tiến trình: {e}", flush=True) return False # Vòng lặp chờ: kiểm tra cls.is_client_running() và cls.is_client_window_ready() start_time = time.time() process_detected_time: Optional[float] = None while time.time() - start_time < wait_timeout_sec: if cls.is_client_running(): if process_detected_time is None: process_detected_time = time.time() print(f"[ClientLauncher] Tiến trình POE2 đã phát hiện sau {process_detected_time - start_time:.1f}s. Đang chờ cửa sổ game sẵn sàng...", flush=True) # Kiểm tra cửa sổ game đã hiển thị (>200x200) if cls.is_client_window_ready(): print(f"[ClientLauncher] Cửa sổ POE2 đã sẵn sàng hiển thị sau {time.time() - start_time:.1f}s.", flush=True) time.sleep(2.0) # Chờ 2s ổn định return True # Nếu tiến trình chạy ổn định hơn 15s và đã có HWND if time.time() - process_detected_time >= 15.0 and cls.find_client_hwnd() is not None: print(f"[ClientLauncher] Tiến trình POE2 đã chạy ổn định (>15s) và có HWND sau {time.time() - start_time:.1f}s.", flush=True) return True time.sleep(1.0) print(f"[ClientLauncher Warning] Hết thời gian chờ ({wait_timeout_sec}s) cửa sổ game xuất hiện.", flush=True) return False @classmethod def recover_client_if_crashed( cls, log_watcher: Optional[Any] = None, exe_path: Optional[str] = None, wait_timeout_sec: int = 45, ) -> bool: """ [INV-CLIENT-CRASH-RECOVER] (Doc 69 §3.5 / §5 P3) Phát hiện sự cố Client Crash / Văng game: - Nếu game đang chạy bình thường: Trả về True ngay lập tức. - Nếu game bị đóng/crash (process dies, hwnd mất): 1. Ghi log cảnh báo CRASH. 2. Gọi launch_client() khởi chạy lại. 3. Chờ cửa sổ sẵn sàng và kiểm tra chuyển cảnh Hideout/Town. """ if cls.is_client_running() and cls.is_client_window_ready(): return True print("\n[ClientLauncher Watchdog] ⚠️ PHÁT HIỆN CLIENT POE2 BỊ CRASH HOẶC ĐÃ ĐÓNG! Đang kích hoạt tự động mở lại...", flush=True) ok = cls.launch_client(exe_path=exe_path, wait_timeout_sec=wait_timeout_sec) if not ok: print("[ClientLauncher Watchdog] ❌ Khởi chạy lại game client thất bại!", flush=True) return False print("[ClientLauncher Watchdog] ✅ Khởi chạy lại game client thành công! Chờ ổn định màn hình tải...", flush=True) time.sleep(3.0) return True # Đồng bộ module alias trong sys.modules để 'client_launcher' và 'src.assistant_tool.client_launcher' trỏ cùng 1 module _current_mod = sys.modules.get(__name__) if _current_mod: if "client_launcher" not in sys.modules: sys.modules["client_launcher"] = _current_mod if "src.assistant_tool.client_launcher" not in sys.modules: sys.modules["src.assistant_tool.client_launcher"] = _current_mod