""" Bộ Quản Lý Vòng Đời Tiến Trình C++ Core Engine (Process Controller) Khởi chạy, truyền cờ cấu hình và theo dõi luồng xuất log thời gian thực. """ import os import subprocess import sys import threading from typing import Any, Callable, Dict, List, Optional try: from src.assistant_tool.client_launcher import Poe2ClientLauncher except ImportError: try: from client_launcher import Poe2ClientLauncher # type: ignore[no-redef] except ImportError: Poe2ClientLauncher = None # type: ignore[assignment,misc] try: from src.assistant_tool.optical_hp_sensor import OpticalHPSensor except ImportError: try: from optical_hp_sensor import OpticalHPSensor # type: ignore[no-redef] except ImportError: OpticalHPSensor = None # type: ignore[assignment,misc] class CoreController: def __init__(self, bin_path: Optional[str] = None): if bin_path: self.bin_path = bin_path else: # Tự động tìm kiếm file thực thi trong các thư mục bin phổ biến root_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")) cands = [ os.path.join(root_dir, "bin", "Release", "AutoPOE2_Core.exe"), os.path.join(root_dir, "bin", "AutoPOE2_Core.exe"), os.path.join(root_dir, "build", "bin", "Release", "AutoPOE2_Core.exe"), os.path.join(root_dir, "build", "Release", "AutoPOE2_Core.exe"), ] self.bin_path = next((p for p in cands if os.path.exists(p)), cands[0]) self._process: Optional[subprocess.Popen] = None self._hProcess = None self._log_listeners: List[Callable[[str], None]] = [] self._state_listeners: List[Callable[[bool], None]] = [] self._reader_thread: Optional[threading.Thread] = None self._starting: bool = False self._lock = threading.Lock() def register_log_callback(self, cb: Callable[[str], None]): """Đăng ký nhận từng dòng log thời gian thực từ C++ Core.""" self._log_listeners.append(cb) def register_state_callback(self, cb: Callable[[bool], None]): """Đăng ký nhận thay đổi trạng thái (True = Running, False = Stopped).""" self._state_listeners.append(cb) def is_running(self) -> bool: if self._process is not None: return self._process.poll() is None if self._hProcess is not None: try: import ctypes from ctypes import wintypes exit_code = wintypes.DWORD() if ctypes.windll.kernel32.GetExitCodeProcess(self._hProcess, ctypes.byref(exit_code)): return exit_code.value == 259 # STILL_ACTIVE = 259 except Exception: pass return False def get_pid(self) -> Optional[int]: if self._process is not None and self._process.poll() is None: return self._process.pid if self._hProcess is not None: try: import ctypes return ctypes.windll.kernel32.GetProcessId(self._hProcess) except Exception: pass return None def start(self, config: Dict[str, Any]) -> bool: """ Khởi động AutoPOE2_Core.exe với các tham số từ giao diện. """ with self._lock: if self.is_running() or self._starting: return True # Dọn dẹp tiến trình Core cũ nếu có trước khi đặt cờ starting self._kill_orphan_core_processes() self._starting = True try: return self._do_start(config) finally: with self._lock: self._starting = False def build_cmd(self, config: Dict[str, Any]) -> List[str]: """ Xây dựng danh sách tham số dòng lệnh cho AutoPOE2_Core.exe từ config. Safety Failsafe: Chỉ khi Poe2ClientLauncher.is_client_running() là True mới được phép truyền cờ --move! Nếu game chưa chạy, C++ Core sẽ không bao giờ được phép gửi tín hiệu chuột/phím ra desktop! """ cmd = [self.bin_path] # 1. Phản xạ if config.get("auto_dodge", True): cmd.append("--autododge") else: cmd.append("--no-dodge") if config.get("smart_flask", True): cmd.append("--autoflask") else: cmd.append("--no-flask") if config.get("no_mana_flask", False) or not config.get("mana_flask_enabled", True): cmd.append("--no-mana-flask") if config.get("no_life_flask", False) or not config.get("life_flask_enabled", True): cmd.append("--no-life-flask") if config.get("boss_rage_flask", False): cmd.append("--boss-rage-flask") delay = config.get("rage_flask_delay", 4000) cmd.extend(["--rage-flask-delay", str(delay)]) if config.get("kiting", False): cmd.append("--kiting") if not config.get("emergency_logout", True): cmd.append("--no-chicken") # Ngưỡng máu flask_hp = config.get("flask_hp", 55) cmd.extend(["--flask-hp", str(flask_hp)]) # Ngưỡng mana flask_mana = config.get("flask_mana", 25) cmd.extend(["--mana-hp", str(flask_mana)]) # Kỹ năng chính & Chế độ đánh xa (Ranged Combat RMB) primary_skill = config.get("primary_skill", "rmb") is_ranged = config.get("is_ranged", True) if primary_skill == "rmb" or is_ranged: cmd.extend(["--primary-skill", "rmb", "--ranged"]) elif primary_skill: cmd.extend(["--primary-skill", str(primary_skill)]) max_mana = config.get("max_mana", 0) if not max_mana or max_mana == 0: max_mana = config.get("detected_max_mana", 0) try: max_mana = int(max_mana or 0) except (ValueError, TypeError): max_mana = 0 max_es = config.get("max_es", 0) if not max_es or max_es == 0: max_es = config.get("detected_max_es", 0) try: max_es = int(max_es or 0) except (ValueError, TypeError): max_es = 0 max_ward = config.get("max_ward", 0) if not max_ward or max_ward == 0: max_ward = config.get("detected_max_ward", 0) try: max_ward = int(max_ward or 0) except (ValueError, TypeError): max_ward = 0 # Cấu hình Máu tối đa (Expected Max HP) — RCA 2026-09-07: # 1. Nếu người dùng chủ động gõ số cụ thể (> 0) -> truyền cờ --max-hp (Manual Override). # 2. Nếu để "Auto"/0/rỗng -> KHÔNG truyền --max-hp: Core tự nhận diện bằng bộ lọc # cấu trúc LifeComponent đã gia cố (bất biến ES cur <= 20000). Giá trị # detected/optical chỉ dùng cho lớp xác thực HUD (badge UNVERIFIED), # KHÔNG đưa vào đối số khởi động để tránh poisoned calibration khi cảm biến đọc sai. raw_max_hp = config.get("expected_max_hp") if raw_max_hp is None or str(raw_max_hp).strip().lower() in ("auto", "", "0"): raw_max_hp = config.get("max_hp", 0) raw_str = str(raw_max_hp).strip().lower() if raw_str in ("ci", "chaos inoculation"): max_hp = 1 elif raw_str in ("auto", "", "0"): max_hp = 0 else: try: max_hp = int(raw_max_hp) except (ValueError, TypeError): max_hp = 0 if max_hp == 0: detected_hp = config.get("detected_max_hp", 0) if detected_hp and int(detected_hp) > 0: max_hp = int(detected_hp) self._emit_log(f"[CoreController Auto-Detect] 🎯 Đã nhận diện Max HP = {max_hp} từ cấu hình quét màn hình -> Truyền cờ --max-hp {max_hp}") # Optical live-scan injection: Mặc định bật khi max_hp == 0 (Auto mode) if max_hp == 0 and config.get("optical_inject_max_hp", True) and OpticalHPSensor is not None: try: sensor = OpticalHPSensor() vitals = sensor.detect_vitals_from_screen() if vitals and vitals.max_hp > 0: max_hp = vitals.max_hp self._emit_log(f"[CoreController Auto-Detect] 🎯 Optical Sensor tự động quét được Max HP = {max_hp} từ màn hình game POE2 -> Truyền cờ --max-hp {max_hp}") if max_mana == 0 and vitals.max_mana > 0: max_mana = vitals.max_mana if max_es == 0 and hasattr(vitals, "max_shield") and vitals.max_shield > 0: max_es = vitals.max_shield if max_ward == 0 and hasattr(vitals, "max_ward") and vitals.max_ward > 0: max_ward = vitals.max_ward except Exception: pass if max_hp == 1: cmd.extend(["--ci", "--max-hp", "1"]) elif max_hp > 0: cmd.extend(["--max-hp", str(max_hp)]) if max_mana > 0: cmd.extend(["--max-mana", str(max_mana)]) if max_es > 0: cmd.extend(["--max-es", str(max_es)]) if max_ward > 0: cmd.extend(["--max-ward", str(max_ward)]) burst_hp = config.get("burst_hp", 22) cmd.extend(["--burst-hp", str(burst_hp)]) # 2. Nhặt đồ if config.get("auto_loot", True): cmd.append("--autoloot") loot_radius = config.get("loot_radius", 35.0) cmd.extend(["--loot-radius", str(loot_radius)]) if config.get("fast_loot", True): cmd.append("--fast-loot") else: cmd.append("--no-loot") # 3. Kỹ năng & Combo if config.get("auto_combo", True): cmd.append("--autocombo") else: cmd.append("--no-combo") # 4. Quest & Tuần tra dò map (Auto-Patrol) if config.get("auto_quest", True): cmd.append("--autoquest") else: cmd.append("--no-quest") if config.get("auto_patrol", True): cmd.append("--autopatrol") else: cmd.append("--no-patrol") # 4b. Kịch bản chơi Map Endgame (Map Scenario & Boss Rush) map_scenario = config.get("map_scenario") if map_scenario is not None: s_str = str(map_scenario).lower() cmd.extend(["--map-scenario", s_str]) if s_str in ("0", "boss_rush", "boss"): cmd.append("--boss-rush") elif s_str in ("1", "atlas_quest_rush", "atlas_quest", "atlas"): cmd.append("--atlas-quest") elif s_str in ("2", "fast_clear", "fast"): cmd.append("--fast-clear") elif s_str in ("3", "full_clear", "full"): cmd.append("--full-clear") if config.get("auto_portal_after_boss", True): cmd.append("--auto-portal-exit") else: cmd.append("--no-auto-portal") # 5. Cho phép di chuyển chuột & phím vật lý thực tế # Safety Failsafe: Chỉ khi Poe2ClientLauncher.is_client_running() là True mới được phép truyền cờ --move! # Nếu game chưa chạy, C++ Core sẽ không bao giờ được phép gửi tín hiệu chuột/phím ra desktop! is_client_running = False if Poe2ClientLauncher is not None: try: is_client_running = bool(Poe2ClientLauncher.is_client_running()) except Exception: is_client_running = False if not is_client_running: for alt_name in ("client_launcher", "src.assistant_tool.client_launcher"): alt_mod = sys.modules.get(alt_name) if alt_mod and hasattr(alt_mod, "Poe2ClientLauncher"): alt_cls = getattr(alt_mod, "Poe2ClientLauncher") if alt_cls is not Poe2ClientLauncher and hasattr(alt_cls, "is_client_running"): try: if alt_cls.is_client_running(): is_client_running = True break except Exception: pass if is_client_running: cmd.append("--move") else: self._emit_log("[CoreController Safety Failsafe] Game client POE2 chưa chạy -> TUYỆT ĐỐI KHÔNG truyền cờ --move để bảo vệ desktop!") # 6. Chế độ di chuyển: mouse (Click-to-Move) hoặc wasd move_mode = config.get("move_mode", "mouse") cmd.extend(["--move-mode", move_mode]) # 7. Tự động đăng nhập lại khi mất kết nối (Auto Reconnect & Login) if config.get("auto_login", True): cmd.append("--autologin") reconnect_delay = config.get("login_delay", 5000) cmd.extend(["--login-delay", str(reconnect_delay)]) char_slot = config.get("char_slot", 1) cmd.extend(["--char-slot", str(char_slot)]) league_name = config.get("league_name", "Forbidden Rites") cmd.extend(["--league", league_name]) login_pass = config.get("login_pass", "") if login_pass: cmd.extend(["--login-pass", login_pass]) else: cmd.append("--no-autologin") # 8. Zoom/Fog thuộc Visual Tool riêng — Core chỉ nhận khi config tường minh if config.get("zoom_enabled", False): zoom_mult = float(config.get("zoom_multiplier", 4.0)) cmd.extend(["--zoom", str(zoom_mult)]) if config.get("disable_fog", False): cmd.append("--no-fog") if config.get("camera_addr"): cmd.extend(["--camera-addr", str(config["camera_addr"])]) if config.get("fog_addr"): cmd.extend(["--fog-addr", str(config["fog_addr"])]) if config.get("ward_addr"): cmd.extend(["--ward-addr", str(config["ward_addr"])]) return cmd build_command = build_cmd def _do_start(self, config: Dict[str, Any]) -> bool: if not os.path.exists(self.bin_path): err_msg = f"[CoreController] Không tìm thấy file thực thi tại: {self.bin_path}" print(err_msg, flush=True) self._emit_log(err_msg) return False # 0. Tự động mở game nếu cấu hình auto_launch_client bật và game chưa chạy if config.get("auto_launch_client", True): try: if Poe2ClientLauncher is not None and not Poe2ClientLauncher.is_client_running(): exe_path = config.get("poe2_exe_path") self._emit_log("[CoreController] Đang tự động mở PathOfExile.exe Standalone (Non-blocking)...") threading.Thread( target=Poe2ClientLauncher.launch_client, args=(exe_path,), kwargs={"wait_timeout_sec": 35}, daemon=True ).start() except Exception as e: self._emit_log(f"[CoreController Warning] Không thể tự động mở game: {e}") # Xây dựng danh sách tham số dòng lệnh từ config cmd = self.build_cmd(config) if not os.path.exists(self.bin_path) or os.path.getsize(self.bin_path) < 10000: err_msg = f"[CoreController] File thực thi chưa sẵn sàng hoặc rỗng: {self.bin_path}. Vui lòng chạy lệnh: cmake --build build --config Release" print(err_msg, flush=True) self._emit_log(err_msg) return False # Dọn dẹp triệt để bất kỳ tiến trình Core cũ nào đang chạy ngầm để tránh xung đột Shared Memory self._kill_orphan_core_processes() # Kiểm tra xem Python hiện tại có đang chạy quyền Administrator hay không is_admin = False if os.name == "nt": try: import ctypes is_admin = bool(ctypes.windll.shell32.IsUserAnAdmin()) except Exception: is_admin = False if not is_admin: self._emit_log("[CoreController] 🔐 Hệ thống chạy Standard User -> Kích hoạt UAC để cấp quyền Administrator cho Core...") return self._start_elevated(cmd) try: self._emit_log(f"[CoreController] Đang khởi chạy: {' '.join(cmd)}") self._process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, encoding="utf-8", errors="replace", bufsize=1, creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0 ) # Khởi chạy luồng đọc stdout self._reader_thread = threading.Thread(target=self._read_output, daemon=True) self._reader_thread.start() self._emit_state(True) return True except OSError as e: # WinError 740: The requested operation requires elevation if getattr(e, "winerror", None) == 740 or "requires elevation" in str(e).lower(): self._emit_log("[CoreController] ⚠️ Tiến trình Core yêu cầu quyền Administrator để đọc bộ nhớ game.") return self._start_elevated(cmd) err = f"[CoreController Error] Lỗi khởi động tiến trình: {e}" print(err, flush=True) self._emit_log(err) self._emit_state(False) return False except Exception as e: err = f"[CoreController Error] Lỗi khởi động tiến trình: {e}" print(err, flush=True) self._emit_log(err) self._emit_state(False) return False def _start_elevated(self, cmd: List[str]) -> bool: """ Khởi chạy tiến trình với quyền Administrator qua ShellExecuteExW (verb='runas'). Tự động hiển thị hộp thoại UAC cho người dùng cấp quyền. """ try: import ctypes from ctypes import wintypes SEE_MASK_NOCLOSEPROCESS = 0x00000040 class SHELLEXECUTEINFOW(ctypes.Structure): _fields_ = [ ("cbSize", wintypes.DWORD), ("fMask", wintypes.ULONG), ("hwnd", wintypes.HWND), ("lpVerb", wintypes.LPCWSTR), ("lpFile", wintypes.LPCWSTR), ("lpParameters", wintypes.LPCWSTR), ("lpDirectory", wintypes.LPCWSTR), ("nShow", ctypes.c_int), ("hInstApp", wintypes.HINSTANCE), ("lpIDList", wintypes.LPVOID), ("lpClass", wintypes.LPCWSTR), ("hkeyClass", wintypes.HKEY), ("dwHotKey", wintypes.DWORD), ("hIconOrMonitor", wintypes.HANDLE), ("hProcess", wintypes.HANDLE) ] params = " ".join(f'"{c}"' if " " in c else c for c in cmd[1:]) bin_file = os.path.abspath(cmd[0]) work_dir = os.path.dirname(bin_file) sei = SHELLEXECUTEINFOW() sei.cbSize = ctypes.sizeof(SHELLEXECUTEINFOW) sei.fMask = SEE_MASK_NOCLOSEPROCESS sei.lpVerb = "runas" sei.lpFile = bin_file sei.lpParameters = params sei.lpDirectory = work_dir sei.nShow = 1 # SW_SHOWNORMAL self._emit_log("[CoreController] 🔐 Đang kích hoạt hộp thoại UAC Windows để cấp quyền Administrator...") ret = ctypes.windll.shell32.ShellExecuteExW(ctypes.byref(sei)) if not ret or not sei.hProcess: err_code = ctypes.GetLastError() err = f"[CoreController Error] Không thể khởi chạy với quyền Admin hoặc bạn đã bấm 'No' trên cửa sổ UAC (Mã: {err_code})." self._emit_log(err) self._emit_state(False) return False self._hProcess = sei.hProcess self._emit_log("[CoreController] ✅ Đã khởi chạy tiến trình Core Engine với quyền Administrator thành công!") # Khởi chạy luồng theo dõi tiến trình elevated và đọc log self._reader_thread = threading.Thread(target=self._monitor_elevated_process, daemon=True) self._reader_thread.start() self._emit_state(True) return True except Exception as e: err = f"[CoreController Error] Lỗi khởi động tiến trình quyền Admin: {e}" self._emit_log(err) self._emit_state(False) return False def _monitor_elevated_process(self): """Theo dõi tiến trình elevated và tail tệp core_log.txt để hiển thị realtime log.""" import ctypes import time from ctypes import wintypes log_path = os.path.join(os.path.dirname(os.path.abspath(self.bin_path)), "core_log.txt") last_size = 0 if os.path.exists(log_path): last_size = os.path.getsize(log_path) exit_code = wintypes.DWORD(259) while getattr(self, "_hProcess", None): if ctypes.windll.kernel32.GetExitCodeProcess(self._hProcess, ctypes.byref(exit_code)): if exit_code.value != 259: break # Đọc log mới từ core_log.txt nếu có if os.path.exists(log_path): try: cur_size = os.path.getsize(log_path) if cur_size > last_size: with open(log_path, "r", encoding="utf-8", errors="replace") as f: f.seek(last_size) new_lines = f.read() last_size = cur_size for line in new_lines.splitlines(): clean = line.strip() if clean: self._emit_log(clean) elif cur_size < last_size: last_size = 0 # File bị xoá hoặc reset except Exception: pass time.sleep(0.1) self._emit_log("[CoreController] Tiến trình Core Engine (Elevated) đã dừng.") self._emit_state(False) def stop(self): """Dừng tiến trình Core Engine an toàn.""" with self._lock: self._starting = False if not self.is_running(): self._emit_state(False) return self._emit_log("[CoreController] Đang gửi tín hiệu dừng tiến trình...") try: if self._process: self._process.terminate() self._process.wait(timeout=2.0) except Exception: try: if self._process: self._process.kill() except Exception: pass self._process = None if hasattr(self, "_hProcess") and self._hProcess: try: import ctypes ctypes.windll.kernel32.TerminateProcess(self._hProcess, 0) ctypes.windll.kernel32.CloseHandle(self._hProcess) except Exception: pass self._hProcess = None # Dọn dẹp bất kỳ tiến trình Core mồ côi nào để giải phóng Shared Memory hoàn toàn self._kill_orphan_core_processes() self._emit_log("[CoreController] Tiến trình đã dừng hoàn toàn.") self._emit_state(False) def _kill_orphan_core_processes(self): """Đảm bảo không có tiến trình AutoPOE2_Core.exe nào chạy ngầm xung đột Shared Memory.""" if self.is_running() or getattr(self, "_starting", False): return try: if os.name == "nt": subprocess.run( ["taskkill", "/F", "/IM", "AutoPOE2_Core.exe"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False ) except Exception: pass def recalibrate_xyz(self, shm_bridge=None) -> bool: """Kích hoạt hiệu chuẩn lại tọa độ XYZ qua IPC hoặc tiến trình riêng.""" if shm_bridge and shm_bridge.send_recalibrate_xyz(): self._emit_log("[CoreController] Đã gửi lệnh Calibrate Tọa Độ XYZ tới Core Engine thành công!") return True self._emit_log("[CoreController Cảnh Báo] Core chưa kết nối IPC, không thể gửi lệnh Calibrate!") return False def _read_output(self): """Đọc dòng log từ stdout của tiến trình con và chuyển tới GUI.""" if not self._process or not self._process.stdout: return try: for line in iter(self._process.stdout.readline, ""): if not line: break clean_line = line.strip() if clean_line: self._emit_log(clean_line) except Exception: pass finally: self._emit_state(False) def _emit_log(self, text: str): for cb in self._log_listeners: try: cb(text) except Exception: pass def _emit_state(self, is_running: bool): for cb in self._state_listeners: try: cb(is_running) except Exception: pass