""" AutoPOE2 Item Hover Watcher (Passive Win32 Mouse & Clipboard Listener) Theo dõi vị trí con trỏ chuột thụ động ở tần số 40Hz (25ms) qua Win32 API. Focus Interlock: Chỉ kích hoạt khi cửa sổ game PathOfExile.exe đang Active. Hover Dwell Tracker: Khi chuột đứng yên <= 5px trong >= 180ms: - Gửi xung Ctrl+C an toàn qua keybd_event. - Đọc clipboard thụ động qua win32clipboard. - Phân tích và gọi callback on_item_inspected(report, cursor_x, cursor_y). - Debounce bằng MD5 hash tránh spam phím lặp lại vô nghĩa. Mouse Move Trigger: Khi chuột di chuyển > 8px, gọi callback on_cursor_moved() để tắt tooltip tức thì (0ms). 100% Zero-Disk, Zero-Memory-Write, Zero-Ban risk. """ from __future__ import annotations import hashlib import math import os import threading import time from typing import Callable, Optional, Tuple import psutil import win32api import win32clipboard import win32con import win32gui import win32process from src.common.win32_window import attach_to_default_desktop as _attach_desktop def attach_to_default_desktop() -> bool: """Đảm bảo luồng chạy gắn vào interactive desktop 'default' của người dùng.""" return _attach_desktop() is not None from src.assistant_tool.item_inspector import ItemInspectionReport, inspect_item class ItemHoverWatcher: """ Bộ giám sát di chuột và bóc tách dữ liệu vật phẩm thụ động trên Windows. Đảm bảo an toàn tài khoản tuyệt đối với Focus Interlock và Debounce. """ DEFAULT_ALLOWED_PROCESSES = ( "PathOfExile.exe", "PathOfExileSteam.exe", "PathOfExile_x64.exe", ) def __init__( self, on_item_inspected: Optional[Callable[[ItemInspectionReport, int, int], None]] = None, on_cursor_moved: Optional[Callable[[], None]] = None, dwell_time_ms: float = 180.0, poll_interval_s: float = 0.025, allowed_processes: Optional[Tuple[str, ...]] = None, enable_focus_interlock: bool = True, simulate_keys: bool = False, ): attach_to_default_desktop() self.on_item_inspected = on_item_inspected self.on_cursor_moved = on_cursor_moved self.dwell_time_ms = dwell_time_ms self.poll_interval_s = poll_interval_s self.allowed_processes = allowed_processes or self.DEFAULT_ALLOWED_PROCESSES self.enable_focus_interlock = enable_focus_interlock self.simulate_keys = simulate_keys self._running = False self._thread: Optional[threading.Thread] = None self._lock = threading.Lock() # Trạng thái theo dõi chuột self.tooltip_active = False self.last_inspected_hash: str = "" self._mock_cursor_pos: Optional[Tuple[int, int]] = None self._mock_clipboard_text: Optional[str] = None self._mock_is_focused: Optional[bool] = None def start(self) -> None: """Khởi động luồng giám sát thụ động.""" attach_to_default_desktop() with self._lock: if self._running: return self._running = True self._thread = threading.Thread(target=self._worker_loop, daemon=True, name="ItemHoverWatcherWorker") self._thread.start() def stop(self) -> None: """Dừng luồng giám sát.""" with self._lock: self._running = False if self._thread and self._thread.is_alive(): self._thread.join(timeout=1.0) self._thread = None @property def is_running(self) -> bool: return self._running def get_cursor_pos(self) -> Tuple[int, int]: """Lấy tọa độ chuột hiện tại (hỗ trợ mock khi chạy test).""" if self._mock_cursor_pos is not None: return self._mock_cursor_pos try: return win32gui.GetCursorPos() except Exception: return (0, 0) def is_game_window_focused(self) -> bool: """Focus Interlock: Kiểm tra cửa sổ foreground có phải client POE2 không.""" if self._mock_is_focused is not None: return self._mock_is_focused if not self.enable_focus_interlock: return True try: hwnd = win32gui.GetForegroundWindow() if not hwnd: return False # Kiểm tra class name và tiêu đề cửa sổ trước title = win32gui.GetWindowText(hwnd) or "" cls_name = win32gui.GetClassName(hwnd) or "" if "poe" in cls_name.lower(): return True if "path of exile" in title.lower(): return True for p in self.allowed_processes: name_no_ext = p.replace(".exe", "") if name_no_ext.lower() in title.lower(): return True # Kiểm tra tên tiến trình qua PID _, pid = win32process.GetWindowThreadProcessId(hwnd) if pid <= 0: return False if pid == os.getpid(): return True proc = psutil.Process(pid) proc_name = proc.name().lower() return any(p.lower() in proc_name for p in self.allowed_processes) except Exception: return False def send_ctrl_c(self) -> None: """Gửi tổ hợp phím Ctrl+C an toàn qua Win32 API.""" if self.simulate_keys: return try: win32api.keybd_event(win32con.VK_CONTROL, 0, 0, 0) win32api.keybd_event(ord("C"), 0, 0, 0) time.sleep(0.015) win32api.keybd_event(ord("C"), 0, win32con.KEYEVENTF_KEYUP, 0) win32api.keybd_event(win32con.VK_CONTROL, 0, win32con.KEYEVENTF_KEYUP, 0) except Exception: pass def read_clipboard_text(self, timeout_s: float = 0.05) -> str: """Đọc nội dung văn bản từ Windows Clipboard an toàn.""" if self._mock_clipboard_text is not None: return self._mock_clipboard_text start_t = time.perf_counter() while time.perf_counter() - start_t < timeout_s: try: win32clipboard.OpenClipboard() try: if win32clipboard.IsClipboardFormatAvailable(win32con.CF_UNICODETEXT): data = win32clipboard.GetClipboardData(win32con.CF_UNICODETEXT) return data or "" elif win32clipboard.IsClipboardFormatAvailable(win32con.CF_TEXT): data = win32clipboard.GetClipboardData(win32con.CF_TEXT) if isinstance(data, bytes): return data.decode("utf-8", errors="ignore") return data or "" finally: win32clipboard.CloseClipboard() return "" except Exception: time.sleep(0.005) return "" def process_tick( self, cur_x: int, cur_y: int, state: dict, ) -> None: """ Xử lý 1 nhịp thời gian của Hover Dwell Tracker. Tách biệt thành hàm riêng để có thể test đơn vị xác định (deterministic test). """ last_pos = state.get("last_pos") anchor_pos = state.get("anchor_pos") dwell_accum_ms = state.get("dwell_accum_ms", 0.0) has_triggered = state.get("has_triggered", False) if last_pos is None or anchor_pos is None: state["last_pos"] = (cur_x, cur_y) state["anchor_pos"] = (cur_x, cur_y) state["dwell_accum_ms"] = 0.0 state["has_triggered"] = False return step_dist = math.hypot(cur_x - last_pos[0], cur_y - last_pos[1]) anchor_dist = math.hypot(cur_x - anchor_pos[0], cur_y - anchor_pos[1]) # Khi chuột di chuyển cách điểm neo > 8px -> Tắt tooltip tức thì (0ms) if anchor_dist > 8.0: if self.tooltip_active: self.tooltip_active = False if self.on_cursor_moved: try: self.on_cursor_moved() except Exception: pass state["dwell_accum_ms"] = 0.0 state["has_triggered"] = False state["anchor_pos"] = (cur_x, cur_y) elif step_dist <= 5.0: # Chuột đang đứng yên (Dwell) dwell_accum_ms += self.poll_interval_s * 1000.0 state["dwell_accum_ms"] = dwell_accum_ms if dwell_accum_ms >= self.dwell_time_ms and not has_triggered: state["has_triggered"] = True # Focus Interlock: Chỉ gửi phím khi game đang Active if self.is_game_window_focused(): self.send_ctrl_c() time.sleep(0.02) text = self.read_clipboard_text() if text and ("Item Class:" in text or "Rarity:" in text): text_hash = hashlib.md5(text.encode("utf-8", errors="ignore")).hexdigest() # Debounce: Bỏ qua nếu text giống hệt món đồ vừa đọc if text_hash != self.last_inspected_hash: self.last_inspected_hash = text_hash report = inspect_item(text) if report: self.tooltip_active = True if self.on_item_inspected: try: self.on_item_inspected(report, cur_x, cur_y) except Exception: pass else: # Chuột rung chuyển nhẹ giữa 5px và 8px: reset dwell state["dwell_accum_ms"] = 0.0 state["has_triggered"] = False state["anchor_pos"] = (cur_x, cur_y) state["last_pos"] = (cur_x, cur_y) def _worker_loop(self) -> None: """Vòng lặp nền polling chuột mỗi 25ms.""" attach_to_default_desktop() state = { "last_pos": None, "anchor_pos": None, "dwell_accum_ms": 0.0, "has_triggered": False, "last_clip_seq": 0, } try: state["last_clip_seq"] = win32clipboard.GetClipboardSequenceNumber() except Exception: pass while self._running: cur_x, cur_y = self.get_cursor_pos() self.process_tick(cur_x, cur_y, state) # Kiểm tra thay đổi clipboard tức thì (hotkey Ctrl+C / Ctrl+Alt+C) try: cur_seq = win32clipboard.GetClipboardSequenceNumber() if cur_seq != state["last_clip_seq"]: state["last_clip_seq"] = cur_seq if self.is_game_window_focused(): text = self.read_clipboard_text() if text and ("Item Class:" in text or "Rarity:" in text): text_hash = hashlib.md5(text.encode("utf-8", errors="ignore")).hexdigest() if text_hash != self.last_inspected_hash: self.last_inspected_hash = text_hash report = inspect_item(text) if report: self.tooltip_active = True if self.on_item_inspected: try: self.on_item_inspected(report, cur_x, cur_y) except Exception: pass except Exception: pass time.sleep(self.poll_interval_s)