clickable_detector.py

import threading
import time
import ctypes
import sys
import os
import json
import win32con
import win32gui
import win32api
import comtypes
from comtypes import client
from ctypes import wintypes

OVERLAY_COLORKEY = win32api.RGB(255, 0, 255)

WM_UPDATE_RECTS = win32con.WM_APP + 1

# WinEvent constants (subset)
EVENT_SYSTEM_FOREGROUND = 0x0003
EVENT_OBJECT_CREATE = 0x8000
EVENT_OBJECT_DESTROY = 0x8001
EVENT_OBJECT_SHOW = 0x8002
EVENT_OBJECT_HIDE = 0x8003
EVENT_OBJECT_REORDER = 0x8004
EVENT_OBJECT_FOCUS = 0x8005
EVENT_OBJECT_SELECTION = 0x8006
WINEVENT_OUTOFCONTEXT = 0x0000

WINEVENTPROC = ctypes.WINFUNCTYPE(
    None,
    wintypes.HANDLE,  # HWINEVENTHOOK
    wintypes.DWORD,   # event
    wintypes.HWND,    # hwnd
    wintypes.LONG,    # idObject
    wintypes.LONG,    # idChild
    wintypes.DWORD,   # dwEventThread
    wintypes.DWORD,   # dwmsEventTime
)

def enable_dpi_awareness():
    try:
        shcore = ctypes.WinDLL("shcore")
        shcore.SetProcessDpiAwareness(2)
    except Exception:
        try:
            ctypes.windll.user32.SetProcessDPIAware()
        except Exception:
            pass

def virtual_screen():
    x = win32api.GetSystemMetrics(76)
    y = win32api.GetSystemMetrics(77)
    w = win32api.GetSystemMetrics(78)
    h = win32api.GetSystemMetrics(79)
    return x, y, w, h

def get_idle_ms():
    class LASTINPUTINFO(ctypes.Structure):
        _fields_ = [("cbSize", ctypes.c_uint), ("dwTime", ctypes.c_uint)]
    lii = LASTINPUTINFO()
    lii.cbSize = ctypes.sizeof(LASTINPUTINFO)
    try:
        if ctypes.windll.user32.GetLastInputInfo(ctypes.byref(lii)):
            tick = win32api.GetTickCount()
            v = int(tick - lii.dwTime)
            return max(0, v)
    except Exception:
        pass
    return 0

def _parse_hex_color(s: str, default=(255, 0, 0)):
    try:
        s = s.strip().lstrip('#')
        if len(s) == 6:
            r = int(s[0:2], 16)
            g = int(s[2:4], 16)
            b = int(s[4:6], 16)
            return r, g, b
    except Exception:
        pass
    return default

def load_config():
    cfg_path = os.path.join(os.path.dirname(__file__), 'uia_highlight_config.json')
    default = {
        "mode": "polling",
        "polling": {
            "interval_ms": 30,
            "dynamic_idle_thresholds_ms": [400, 1500],
            "dynamic_sleep_ms": [20, 50, 120]
        },
        "event": {
            "fallback_poll_ms": 250
        },
        "near_cursor": {
            "interval_ms": 30,
            "radius_px": 220,
            "ancestors": 6
        },
        "overlay": {
            "thickness": 2,
            "color": "#ff0000"
        }
    }
    try:
        with open(cfg_path, 'r', encoding='utf-8') as f:
            raw = f.read()
            # loại bỏ các dòng comment kiểu // để tránh JSON lỗi
            lines = []
            for ln in raw.splitlines():
                s = ln.strip()
                if s.startswith('//'):
                    continue
                lines.append(ln)
            text = "\n".join(lines)
            data = json.loads(text)
            # merge shallowly
            default.update({k: v for k, v in data.items() if v is not None})
    except Exception:
        pass
    return default

class ClickableFinder:
    def __init__(self):
        client.GetModule("UIAutomationCore.dll")
        from comtypes.gen import UIAutomationClient as UIA  # type: ignore
        try:
            self.automation = client.CreateObject(UIA.CUIAutomation8, interface=UIA.IUIAutomation)
        except Exception:
            self.automation = client.CreateObject(UIA.CUIAutomation, interface=UIA.IUIAutomation)
        self.UIA = UIA
        self._prepare()

    def _prepare(self):
        UIA = self.UIA
        a = self.automation
        def _or_from_seq(seq):
            try:
                return a.CreateOrConditionFromArray(tuple(seq))
            except Exception:
                c = None
                for s in seq:
                    c = s if c is None else a.CreateOrCondition(c, s)
                return c
        def _and_from_seq(seq):
            try:
                return a.CreateAndConditionFromArray(tuple(seq))
            except Exception:
                c = None
                for s in seq:
                    c = s if c is None else a.CreateAndCondition(c, s)
                return c
        p_invoke = a.CreatePropertyCondition(UIA.UIA_IsInvokePatternAvailablePropertyId, True)
        p_toggle = a.CreatePropertyCondition(UIA.UIA_IsTogglePatternAvailablePropertyId, True)
        p_expand = a.CreatePropertyCondition(UIA.UIA_IsExpandCollapsePatternAvailablePropertyId, True)
        p_select = a.CreatePropertyCondition(UIA.UIA_IsSelectionItemPatternAvailablePropertyId, True)
        p_legacy = a.CreatePropertyCondition(UIA.UIA_IsLegacyIAccessiblePatternAvailablePropertyId, True)
        p_focus = a.CreatePropertyCondition(UIA.UIA_IsKeyboardFocusablePropertyId, True)
        ct = []
        for cid in (
            UIA.UIA_ButtonControlTypeId,
            UIA.UIA_HyperlinkControlTypeId,
            UIA.UIA_MenuItemControlTypeId,
            UIA.UIA_TabItemControlTypeId,
            UIA.UIA_ListItemControlTypeId,
            UIA.UIA_TreeItemControlTypeId,
            UIA.UIA_RadioButtonControlTypeId,
            UIA.UIA_CheckBoxControlTypeId,
            UIA.UIA_SplitButtonControlTypeId,
            UIA.UIA_ComboBoxControlTypeId,
        ):
            ct.append(a.CreatePropertyCondition(UIA.UIA_ControlTypePropertyId, cid))
        clickable_types = _or_from_seq(ct)
        clickable_pats = _or_from_seq((p_invoke, p_toggle, p_expand, p_select, p_legacy, p_focus))
        clickable = a.CreateOrCondition(clickable_pats, clickable_types)
        enabled = a.CreatePropertyCondition(UIA.UIA_IsEnabledPropertyId, True)
        onscreen = a.CreatePropertyCondition(UIA.UIA_IsOffscreenPropertyId, False)
        iscontrol = a.CreatePropertyCondition(UIA.UIA_IsControlElementPropertyId, True)
        base_cond = _and_from_seq((clickable, enabled, onscreen, iscontrol))
        # Ưu tiên Control View để tăng tốc (nếu khả dụng)
        try:
            self.condition = a.CreateAndCondition(base_cond, a.ControlViewCondition)
        except Exception:
            self.condition = base_cond
        cache = a.CreateCacheRequest()
        cache.AutomationElementMode = UIA.AutomationElementMode_None
        try:
            cache.TreeScope = UIA.TreeScope_Element
            cache.TreeFilter = a.ControlViewCondition
        except Exception:
            pass
        cache.AddProperty(UIA.UIA_BoundingRectanglePropertyId)
        cache.AddProperty(UIA.UIA_NativeWindowHandlePropertyId)
        self.cache = cache

    def _rect_to_ltrb(self, r):
        if r is None:
            return None
        if hasattr(r, "width"):
            left = int(r.left)
            top = int(r.top)
            right = int(r.left + r.width)
            bottom = int(r.top + r.height)
        elif hasattr(r, "right"):
            left = int(r.left)
            top = int(r.top)
            right = int(r.right)
            bottom = int(r.bottom)
        else:
            return None
        if right - left <= 2 or bottom - top <= 2:
            return None
        return left, top, right, bottom

    def collect(self, exclude_hwnd=None, screen_bounds=None, target_hwnd=None, cursor_region=None):
        UIA = self.UIA
        a = self.automation
        try:
            hwnd = int(target_hwnd) if target_hwnd else win32gui.GetForegroundWindow()
        except Exception:
            hwnd = 0
        el_root = None
        if hwnd:
            try:
                el_root = a.ElementFromHandle(hwnd)
            except Exception:
                el_root = None
        if not el_root:
            el_root = a.GetRootElement()
        # Try fast path: BuildCache. Fallback to FindAll(Current) on provider errors.
        arr = None
        try:
            arr = el_root.FindAllBuildCache(UIA.TreeScope_Descendants, self.condition, self.cache)
        except Exception:
            try:
                arr = el_root.FindAll(UIA.TreeScope_Descendants, self.condition)
            except Exception:
                return []
            # Mark as non-cached iteration
            use_cached = False
        else:
            use_cached = True
        if not arr:
            return []
        if target_hwnd:
            try:
                lft, top, rgt, btm = win32gui.GetWindowRect(int(target_hwnd))
                vx, vy, vw, vh = lft, top, (rgt - lft), (btm - top)
            except Exception:
                vx, vy, vw, vh = screen_bounds if screen_bounds else virtual_screen()
        else:
            vx, vy, vw, vh = screen_bounds if screen_bounds else virtual_screen()
        try:
            n = arr.Length
        except Exception:
            return []
        rx = []
        seen = set()
        for i in range(n):
            try:
                el = arr.GetElement(i)
            except Exception:
                continue
            try:
                hwnd_el = (el.CachedNativeWindowHandle if use_cached else el.CurrentNativeWindowHandle)
            except Exception:
                hwnd_el = None
            if exclude_hwnd and hwnd_el:
                try:
                    if int(hwnd_el) == int(exclude_hwnd):
                        continue
                except Exception:
                    pass
            try:
                rc0 = self._rect_to_ltrb(el.CachedBoundingRectangle if use_cached else el.CurrentBoundingRectangle)
            except Exception:
                continue
            if not rc0:
                continue
            l, t, r, b = rc0
            if r <= vx or b <= vy or l >= vx + vw or t >= vy + vh:
                continue
            l = max(l, vx)
            t = max(t, vy)
            r = min(r, vx + vw)
            b = min(b, vy + vh)
            if cursor_region:
                cl, ct, cr, cb = cursor_region
                if r <= cl or b <= ct or l >= cr or t >= cb:
                    continue
                l = max(l, cl)
                t = max(t, ct)
                r = min(r, cr)
                b = min(b, cb)
            k = (l, t, r, b)
            if k in seen:
                continue
            seen.add(k)
            rx.append(k)
        return rx

class OverlayWindow:
    _instances = {}

    def __init__(self, thickness=2, colorref=None):
        self.thickness = int(thickness)
        self.rects = []
        self.hwnd = None
        self.hinst = win32api.GetModuleHandle(None)
        self.vx, self.vy, self.vw, self.vh = virtual_screen()
        self.pen_colorref = colorref if colorref is not None else win32api.RGB(255, 0, 0)

    @classmethod
    def _wndproc(cls, hwnd, msg, wparam, lparam):
        self = cls._instances.get(hwnd)
        if msg == win32con.WM_NCHITTEST:
            return win32con.HTTRANSPARENT
        if msg == win32con.WM_HOTKEY:
            try:
                win32gui.DestroyWindow(hwnd)
            except Exception:
                pass
            return 0
        if msg == win32con.WM_ERASEBKGND:
            try:
                hdc = wparam
                rc = win32gui.GetClientRect(hwnd)
                b = win32gui.CreateSolidBrush(OVERLAY_COLORKEY)
                win32gui.FillRect(hdc, rc, b)
                win32gui.DeleteObject(b)
            except Exception:
                pass
            return 1
        if msg == WM_UPDATE_RECTS:
            win32gui.InvalidateRect(hwnd, None, False)
            return 0
        if msg == win32con.WM_PAINT:
            hdc, ps = win32gui.BeginPaint(hwnd)
            try:
                # tô nền bằng colorkey để phần nền trong suốt
                rc = win32gui.GetClientRect(hwnd)
                bk = win32gui.CreateSolidBrush(OVERLAY_COLORKEY)
                win32gui.FillRect(hdc, rc, bk)
                win32gui.DeleteObject(bk)
                brush = win32gui.GetStockObject(win32con.NULL_BRUSH)
                win32gui.SelectObject(hdc, brush)
                pen = win32gui.CreatePen(win32con.PS_SOLID, max(1, self.thickness), self.pen_colorref)
                old_pen = win32gui.SelectObject(hdc, pen)
                ox, oy = -self.vx, -self.vy
                for l, t, r, b in self.rects:
                    win32gui.Rectangle(hdc, l + ox, t + oy, r + ox, b + oy)
                win32gui.SelectObject(hdc, old_pen)
                win32gui.DeleteObject(pen)
            finally:
                win32gui.EndPaint(hwnd, ps)
            return 0
        if msg == win32con.WM_DESTROY:
            win32gui.PostQuitMessage(0)
            return 0
        return win32gui.DefWindowProc(hwnd, msg, wparam, lparam)

    def create(self):
        wc = win32gui.WNDCLASS()
        wc.hInstance = self.hinst
        wc.lpszClassName = "ClickableOverlayWindow"
        wc.lpfnWndProc = OverlayWindow._wndproc
        wc.hCursor = win32gui.LoadCursor(0, win32con.IDC_ARROW)
        wc.hbrBackground = win32gui.CreateSolidBrush(OVERLAY_COLORKEY)
        atom = win32gui.RegisterClass(wc)
        ex = (
            win32con.WS_EX_LAYERED
            | win32con.WS_EX_TRANSPARENT
            | win32con.WS_EX_TOPMOST
            | win32con.WS_EX_TOOLWINDOW
            | 0x08000000
        )
        style = win32con.WS_POPUP
        self.hwnd = win32gui.CreateWindowEx(
            ex,
            atom,
            "",
            style,
            self.vx,
            self.vy,
            self.vw,
            self.vh,
            0,
            0,
            self.hinst,
            None,
        )
        OverlayWindow._instances[self.hwnd] = self
        win32gui.SetLayeredWindowAttributes(
            self.hwnd,
            OVERLAY_COLORKEY,
            255,
            win32con.LWA_COLORKEY | win32con.LWA_ALPHA,
        )
        # Hotkey thoát nhanh: Ctrl+Shift+Q
        try:
            win32api.RegisterHotKey(self.hwnd, 1, win32con.MOD_CONTROL | win32con.MOD_SHIFT, ord('Q'))
        except Exception:
            pass
        win32gui.ShowWindow(self.hwnd, win32con.SW_SHOW)
        win32gui.UpdateWindow(self.hwnd)

    def update_rects(self, rects):
        try:
            new_rects = sorted(rects)
        except Exception:
            new_rects = rects
        if getattr(self, "_sig", None) == new_rects:
            return
        self._sig = new_rects
        self.rects = new_rects
        if self.hwnd:
            win32gui.PostMessage(self.hwnd, WM_UPDATE_RECTS, 0, 0)

    def loop(self):
        win32gui.PumpMessages()

class ScannerThread(threading.Thread):
    def __init__(self, overlay: OverlayWindow, config: dict):
        super().__init__(daemon=True)
        self.overlay = overlay
        self.config = config or {}
        self.mode = (self.config.get('mode') or 'polling').lower()
        # Poll intervals
        self.poll_interval = max(0.01, float(self.config.get('polling', {}).get('interval_ms', 30)) / 1000.0)
        self.near_interval = max(0.01, float(self.config.get('near_cursor', {}).get('interval_ms', 30)) / 1000.0)
        self.event_fallback = max(0.02, float(self.config.get('event', {}).get('fallback_poll_ms', 250)) / 1000.0)
        th = self.config.get('polling', {}).get('dynamic_idle_thresholds_ms', [400, 1500])
        sl = self.config.get('polling', {}).get('dynamic_sleep_ms', [20, 50, 120])
        self.idle_thresholds = (int(th[0]), int(th[1])) if isinstance(th, (list, tuple)) and len(th) >= 2 else (400, 1500)
        self.sleep_choices = (
            float(sl[0]) / 1000.0,
            float(sl[1]) / 1000.0,
            float(sl[2]) / 1000.0,
        ) if isinstance(sl, (list, tuple)) and len(sl) >= 3 else (0.02, 0.05, 0.12)
        # Event mode
        self._evt = threading.Event()
        self._hooks = []
        self._cb = None
        self._stop = threading.Event()

    def stop(self):
        self._stop.set()
        self._evt.set()

    def _compute_sleep(self, elapsed: float) -> float:
        idle = get_idle_ms()
        lo, hi = self.idle_thresholds
        a, b, c = self.sleep_choices
        dyn = a if idle < lo else (b if idle < hi else c)
        base = self.near_interval if self.mode == 'near_cursor' else self.poll_interval
        return max(base, dyn) - max(0.0, elapsed)

    def _cursor_region(self):
        radius = int(self.config.get('near_cursor', {}).get('radius_px', 220))
        try:
            x, y = win32gui.GetCursorPos()
        except Exception:
            return None
        return (x - radius, y - radius, x + radius, y + radius)

    def _scan_once(self, finder, fg):
        t0 = time.perf_counter()
        cursor_region = self._cursor_region() if self.mode == 'near_cursor' else None
        rects = finder.collect(
            exclude_hwnd=self.overlay.hwnd,
            screen_bounds=(self.overlay.vx, self.overlay.vy, self.overlay.vw, self.overlay.vh),
            target_hwnd=fg,
            cursor_region=cursor_region,
        )
        self.overlay.update_rects(rects)
        return time.perf_counter() - t0

    def _event_callback(self, hWinEventHook, event, hwnd, idObject, idChild, dwEventThread, dwmsEventTime):
        self._evt.set()

    def _setup_hooks(self):
        user32 = ctypes.windll.user32
        try:
            user32.SetWinEventHook.argtypes = [wintypes.DWORD, wintypes.DWORD, wintypes.HMODULE, WINEVENTPROC, wintypes.DWORD, wintypes.DWORD, wintypes.DWORD]
            user32.SetWinEventHook.restype = wintypes.HANDLE
            user32.UnhookWinEvent.argtypes = [wintypes.HANDLE]
            user32.UnhookWinEvent.restype = wintypes.BOOL
        except Exception:
            pass
        self._cb = WINEVENTPROC(self._event_callback)
        events = [
            EVENT_SYSTEM_FOREGROUND,
            EVENT_OBJECT_CREATE,
            EVENT_OBJECT_DESTROY,
            EVENT_OBJECT_SHOW,
            EVENT_OBJECT_HIDE,
            EVENT_OBJECT_REORDER,
            EVENT_OBJECT_FOCUS,
            EVENT_OBJECT_SELECTION,
        ]
        for ev in events:
            try:
                h = ctypes.windll.user32.SetWinEventHook(ev, ev, 0, self._cb, 0, 0, WINEVENT_OUTOFCONTEXT)
                if h:
                    self._hooks.append(h)
            except Exception:
                pass

    def _teardown_hooks(self):
        for h in self._hooks:
            try:
                ctypes.windll.user32.UnhookWinEvent(h)
            except Exception:
                pass
        self._hooks.clear()
        self._cb = None

    def run(self):
        comtypes.CoInitialize()
        try:
            finder = ClickableFinder()
            last_fg = None
            last_reinit = time.time()
            if self.mode == 'event':
                self._setup_hooks()
            try:
                while not self._stop.is_set():
                    fg = win32gui.GetForegroundWindow()
                    if not fg or not win32gui.IsWindowVisible(fg) or win32gui.IsIconic(fg):
                        self.overlay.update_rects([])
                        if self.mode == 'event':
                            self._evt.wait(timeout=self.event_fallback)
                            self._evt.clear()
                            continue
                        time.sleep(max(self.poll_interval, 0.05))
                        continue
                    if self.mode == 'event':
                        # Wait for events, but also coalesce bursts
                        self._evt.wait(timeout=self.event_fallback)
                        self._evt.clear()
                        try:
                            self._scan_once(finder, fg)
                        except Exception:
                            # Recreate UIA on provider/COM failure
                            try:
                                finder = ClickableFinder()
                                last_reinit = time.time()
                            except Exception:
                                time.sleep(0.2)
                        continue
                    # polling / near_cursor
                    if fg != last_fg:
                        last_fg = fg
                        try:
                            elapsed = self._scan_once(finder, fg)
                        except Exception:
                            try:
                                finder = ClickableFinder()
                                last_reinit = time.time()
                            except Exception:
                                time.sleep(0.2)
                                continue
                            else:
                                time.sleep(0.05)
                                continue
                        if time.time() - last_reinit > 60:
                            try:
                                finder = ClickableFinder()
                                last_reinit = time.time()
                            except Exception:
                                pass
                        sleep_t = self._compute_sleep(elapsed)
                        if sleep_t > 0:
                            time.sleep(sleep_t)
                        continue
                    try:
                        elapsed = self._scan_once(finder, fg)
                    except Exception:
                        try:
                            finder = ClickableFinder()
                            last_reinit = time.time()
                        except Exception:
                            time.sleep(0.2)
                            continue
                        else:
                            time.sleep(0.05)
                            continue
                    if time.time() - last_reinit > 60:
                        try:
                            finder = ClickableFinder()
                            last_reinit = time.time()
                        except Exception:
                            pass
                    sleep_t = self._compute_sleep(elapsed)
                    if sleep_t > 0:
                        time.sleep(sleep_t)
            finally:
                if self.mode == 'event':
                    self._teardown_hooks()
        finally:
            try:
                comtypes.CoUninitialize()
            except Exception:
                pass

def main():
    enable_dpi_awareness()
    cfg = load_config()
    # overlay config
    try:
        th = int(cfg.get('overlay', {}).get('thickness', 2))
    except Exception:
        th = 2
    r, g, b = _parse_hex_color(str(cfg.get('overlay', {}).get('color', '#ff0000')))
    overlay = OverlayWindow(thickness=th, colorref=win32api.RGB(r, g, b))
    overlay.create()
    scanner = ScannerThread(overlay, cfg)
    scanner.start()
    try:
        overlay.loop()
    except KeyboardInterrupt:
        pass
    finally:
        scanner.stop()

if __name__ == "__main__":
    sys.exit(main())