wow/hardware_control.py

import importlib.util
import json
import os
import random
import sys
import time
from ctypes import c_long, c_longlong, windll

import pythoncom
import win32com.client

try:
    import pydirectinput
except Exception as exc:
    pydirectinput = None
    _PYDIRECTINPUT_IMPORT_ERROR = exc
else:
    _PYDIRECTINPUT_IMPORT_ERROR = None
    pydirectinput.FAILSAFE = False
    pydirectinput.PAUSE = 0


MAIN_CONFIG_FILE = "wow_multikey_qt.json"
BACKEND_TIANYA = "tianya"
BACKEND_GHOST = "ghost"
BACKEND_DIRECTINPUT = "directinput"
TIANYA_DLL_PATH = "ddl/wyhkm.dll"
GHOSTBOX_MODULE_PATH = "Ghostbox_ddl/ghostbox.py"


class _TianyaBackend:
    name = BACKEND_TIANYA

    def __init__(self, controller, dll_path=TIANYA_DLL_PATH):
        self._controller = controller
        self._dll_path_setting = dll_path
        self._wyhkm = None
        self.dll_path = None

    def configure(self, dll_path=None):
        if dll_path:
            self._dll_path_setting = dll_path

    def label(self):
        return self.name if self.is_available() else f"{self.name}_unavailable"

    def is_available(self):
        if self._wyhkm:
            try:
                if self._wyhkm.IsOpen(0):
                    return True
            except Exception:
                self._wyhkm = None

        self.dll_path = self._controller._resolve_resource_path(self._dll_path_setting)
        if not self.dll_path:
            self._controller._log(f">>> [tianya_control] DLL not found: {self._dll_path_setting}")
            self._wyhkm = None
            return False

        try:
            hkmdll = windll.LoadLibrary(self.dll_path)
            hkmdll.DllInstall.argtypes = (c_long, c_longlong)

            if hkmdll.DllInstall(1, 2) < 0:
                self._controller._log(">>> [tianya_control] DllInstall failed")
                self._wyhkm = None
                return False

            pythoncom.CoInitialize()
            self._wyhkm = win32com.client.Dispatch("wyp.hkm")

            dev_id = self._wyhkm.SearchDevice(0x2612, 0x1701, 0)
            if dev_id == -1:
                self._controller._log(">>> [tianya_control] Device not found (Index 0)")
                self._wyhkm = None
                return False

            if not self._wyhkm.Open(dev_id, 0):
                self._controller._log(">>> [tianya_control] Open device failed")
                self._wyhkm = None
                return False

            self._wyhkm.SetMode(1, 1)
            self._wyhkm.SetMode(2, 1)
            self._wyhkm.SetKeyInterval(30, 50)
            self._wyhkm.SetMouseInterval(30, 50)

            self._controller._log(f">>> [tianya_control] Device opened and initialized: {dev_id}")
            return True
        except Exception as exc:
            self._controller._log(f">>> [tianya_control] Init failed: {exc}")
            self._wyhkm = None
            return False

    def _normalize_key(self, key_str):
        return str(key_str).strip().upper()

    def delay_rnd(self, min_ms, max_ms):
        if not self.is_available():
            return False
        try:
            self._wyhkm.DelayRnd(int(min_ms), int(max_ms))
            return True
        except Exception:
            return False

    def key_down(self, key_str):
        if not self.is_available():
            return False
        try:
            self._wyhkm.KeyDown(self._normalize_key(key_str))
            return True
        except Exception as exc:
            self._controller._log(f">>> [tianya_control] KeyDown failed ({key_str}): {exc}")
            return False

    def key_up(self, key_str):
        if not self.is_available():
            return False
        try:
            self._wyhkm.KeyUp(self._normalize_key(key_str))
            return True
        except Exception as exc:
            self._controller._log(f">>> [tianya_control] KeyUp failed ({key_str}): {exc}")
            return False

    def key_press(self, key_str):
        if not self.is_available():
            return False
        try:
            self._wyhkm.KeyPress(self._normalize_key(key_str))
            return True
        except Exception as exc:
            self._controller._log(f">>> [tianya_control] KeyPress failed ({key_str}): {exc}")
            return False

    def move_to(self, x, y):
        if not self.is_available():
            return False
        try:
            return bool(self._wyhkm.MoveTo(int(x), int(y)))
        except Exception as exc:
            self._controller._log(f">>> [tianya_control] MoveTo failed ({x}, {y}): {exc}")
            return False

    def move_r(self, dx, dy):
        if not self.is_available():
            return False
        try:
            return bool(self._wyhkm.MoveR(int(dx), int(dy)))
        except Exception as exc:
            self._controller._log(f">>> [tianya_control] MoveR failed ({dx}, {dy}): {exc}")
            return False

    def left_click(self):
        if not self.is_available():
            return False
        try:
            return bool(self._wyhkm.LeftClick())
        except Exception as exc:
            self._controller._log(f">>> [tianya_control] LeftClick failed: {exc}")
            return False

    def right_click(self):
        if not self.is_available():
            return False
        try:
            return bool(self._wyhkm.RightClick())
        except Exception as exc:
            self._controller._log(f">>> [tianya_control] RightClick failed: {exc}")
            return False

    def left_down(self):
        if not self.is_available():
            return False
        try:
            return bool(self._wyhkm.LeftDown())
        except Exception as exc:
            self._controller._log(f">>> [tianya_control] LeftDown failed: {exc}")
            return False

    def left_up(self):
        if not self.is_available():
            return False
        try:
            return bool(self._wyhkm.LeftUp())
        except Exception as exc:
            self._controller._log(f">>> [tianya_control] LeftUp failed: {exc}")
            return False


class _GhostboxBackend:
    name = BACKEND_GHOST
    LEFT_BUTTON = 1
    RIGHT_BUTTON = 2

    def __init__(self, controller, module_path=GHOSTBOX_MODULE_PATH):
        self._controller = controller
        self._module_path_setting = module_path
        self._module = None
        self.module_path = None

    def configure(self, module_path=None):
        if module_path and module_path != self._module_path_setting:
            self._module_path_setting = module_path
            self._module = None

    def label(self):
        return self.name if self.is_available() else f"{self.name}_unavailable"

    def _load_module(self):
        if self._module is not None:
            return self._module

        self.module_path = self._controller._resolve_resource_path(self._module_path_setting)
        if not self.module_path:
            self._controller._log(f">>> [ghost_control] module not found: {self._module_path_setting}")
            return None

        try:
            spec = importlib.util.spec_from_file_location("ghostbox_runtime", self.module_path)
            if spec is None or spec.loader is None:
                self._controller._log(f">>> [ghost_control] load spec failed: {self.module_path}")
                return None
            module = importlib.util.module_from_spec(spec)
            sys.modules[spec.name] = module
            spec.loader.exec_module(module)
            self._module = module
            return module
        except Exception as exc:
            self._controller._log(f">>> [ghost_control] load failed: {exc}")
            self._module = None
            return None

    def is_available(self):
        gb = self._load_module()
        if gb is None:
            return False

        try:
            if gb.isconnected():
                return True
        except Exception:
            pass

        try:
            result = gb.opendevice()
            connected = bool(gb.isconnected())
            if not connected:
                self._controller._log(f">>> [ghost_control] Open device failed: result={result}")
                return False

            try:
                gb.setpresskeydelay(30, 50)
                gb.setpressmousebuttondelay(30, 50)
                gb.setmousemovementdelay(3, 5)
            except Exception:
                pass

            self._controller._log(">>> [ghost_control] Device opened and initialized")
            return True
        except Exception as exc:
            self._controller._log(f">>> [ghost_control] Init failed: {exc}")
            return False

    def _normalize_key(self, key_str):
        return str(key_str).strip().lower()

    def _call(self, action, func, *args):
        if not self.is_available():
            return False
        try:
            func(*args)
            return True
        except Exception as exc:
            self._controller._log(f">>> [ghost_control] {action} failed: {exc}")
            return False

    def delay_rnd(self, min_ms, max_ms):
        low = min(int(min_ms), int(max_ms))
        high = max(int(min_ms), int(max_ms))
        time.sleep(random.uniform(low, high) / 1000.0)
        return True

    def key_down(self, key_str):
        gb = self._load_module()
        if gb is None:
            return False
        return self._call(f"KeyDown({key_str})", gb.presskeybyname, self._normalize_key(key_str))

    def key_up(self, key_str):
        gb = self._load_module()
        if gb is None:
            return False
        return self._call(f"KeyUp({key_str})", gb.releasekeybyname, self._normalize_key(key_str))

    def key_press(self, key_str):
        gb = self._load_module()
        if gb is None:
            return False
        return self._call(f"KeyPress({key_str})", gb.pressandreleasekeybyname, self._normalize_key(key_str))

    def move_to(self, x, y):
        gb = self._load_module()
        if gb is None:
            return False
        return self._call(f"MoveTo({x}, {y})", gb.movemouseto, int(x), int(y))

    def move_r(self, dx, dy):
        gb = self._load_module()
        if gb is None:
            return False
        return self._call(f"MoveR({dx}, {dy})", gb.movemouserelative, int(dx), int(dy))

    def left_click(self):
        gb = self._load_module()
        if gb is None:
            return False
        return self._call("LeftClick", gb.pressandreleasemousebutton, self.LEFT_BUTTON)

    def right_click(self):
        gb = self._load_module()
        if gb is None:
            return False
        return self._call("RightClick", gb.pressandreleasemousebutton, self.RIGHT_BUTTON)

    def left_down(self):
        gb = self._load_module()
        if gb is None:
            return False
        return self._call("LeftDown", gb.pressmousebutton, self.LEFT_BUTTON)

    def left_up(self):
        gb = self._load_module()
        if gb is None:
            return False
        return self._call("LeftUp", gb.releasemousebutton, self.LEFT_BUTTON)


class _DirectInputBackend:
    name = BACKEND_DIRECTINPUT

    def __init__(self, controller):
        self._controller = controller
        self._last_error = None

    def label(self):
        return self.name if self.is_available() else f"{self.name}_unavailable"

    def is_available(self):
        if pydirectinput is not None:
            return True
        if self._last_error != _PYDIRECTINPUT_IMPORT_ERROR:
            self._last_error = _PYDIRECTINPUT_IMPORT_ERROR
            self._controller._log(f">>> [input_control] pydirectinput unavailable: {_PYDIRECTINPUT_IMPORT_ERROR}")
        return False

    def _normalize_key(self, key_str):
        return str(key_str).strip().lower()

    def delay_rnd(self, min_ms, max_ms):
        if not self.is_available():
            return False
        low = min(int(min_ms), int(max_ms))
        high = max(int(min_ms), int(max_ms))
        time.sleep(random.uniform(low, high) / 1000.0)
        return True

    def key_down(self, key_str):
        if not self.is_available():
            return False
        try:
            pydirectinput.keyDown(self._normalize_key(key_str))
            return True
        except Exception as exc:
            self._controller._log(f">>> [directinput] keyDown failed ({key_str}): {exc}")
            return False

    def key_up(self, key_str):
        if not self.is_available():
            return False
        try:
            pydirectinput.keyUp(self._normalize_key(key_str))
            return True
        except Exception as exc:
            self._controller._log(f">>> [directinput] keyUp failed ({key_str}): {exc}")
            return False

    def key_press(self, key_str):
        if not self.is_available():
            return False
        try:
            pydirectinput.press(self._normalize_key(key_str))
            return True
        except Exception as exc:
            self._controller._log(f">>> [directinput] press failed ({key_str}): {exc}")
            return False

    def move_to(self, x, y):
        if not self.is_available():
            return False
        try:
            pydirectinput.moveTo(int(x), int(y))
            return True
        except Exception as exc:
            self._controller._log(f">>> [directinput] moveTo failed ({x}, {y}): {exc}")
            return False

    def move_r(self, dx, dy):
        if not self.is_available():
            return False
        try:
            pydirectinput.moveRel(int(dx), int(dy))
            return True
        except Exception as exc:
            self._controller._log(f">>> [directinput] moveRel failed ({dx}, {dy}): {exc}")
            return False

    def left_click(self):
        if not self.is_available():
            return False
        try:
            pydirectinput.click(button="left")
            return True
        except Exception as exc:
            self._controller._log(f">>> [directinput] left click failed: {exc}")
            return False

    def right_click(self):
        if not self.is_available():
            return False
        try:
            pydirectinput.click(button="right")
            return True
        except Exception as exc:
            self._controller._log(f">>> [directinput] right click failed: {exc}")
            return False

    def left_down(self):
        if not self.is_available():
            return False
        try:
            pydirectinput.mouseDown(button="left")
            return True
        except Exception as exc:
            self._controller._log(f">>> [directinput] left mouseDown failed: {exc}")
            return False

    def left_up(self):
        if not self.is_available():
            return False
        try:
            pydirectinput.mouseUp(button="left")
            return True
        except Exception as exc:
            self._controller._log(f">>> [directinput] left mouseUp failed: {exc}")
            return False


class HardwareController:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            cls._instance = super(HardwareController, cls).__new__(cls)
        return cls._instance

    def __init__(
        self,
        dll_path=TIANYA_DLL_PATH,
        use_hardware_input=None,
        use_tianya_box=None,
        use_ghost_box=None,
        ghostbox_module_path=GHOSTBOX_MODULE_PATH,
        backend=None,
    ):
        if getattr(self, "_bootstrapped", False):
            self._tianya_backend.configure(dll_path=dll_path)
            self._ghost_backend.configure(module_path=ghostbox_module_path)
            if backend is not None or use_hardware_input is not None or use_tianya_box is not None or use_ghost_box is not None:
                self.configure(
                    use_hardware_input=use_hardware_input,
                    use_tianya_box=use_tianya_box,
                    use_ghost_box=use_ghost_box,
                    dll_path=dll_path,
                    ghostbox_module_path=ghostbox_module_path,
                    backend=backend,
                )
            return

        self._bootstrapped = True
        self._backend = BACKEND_TIANYA
        self._last_backend_log = None
        self._tianya_backend = _TianyaBackend(self, dll_path)
        self._ghost_backend = _GhostboxBackend(self, ghostbox_module_path)
        self._directinput_backend = _DirectInputBackend(self)

        if backend is None and use_hardware_input is None and use_tianya_box is None and use_ghost_box is None:
            use_tianya_box, use_ghost_box = self._load_backend_preference(default_tianya=True)
        self.configure(
            use_hardware_input=use_hardware_input,
            use_tianya_box=use_tianya_box,
            use_ghost_box=use_ghost_box,
            backend=backend,
        )

    def _runtime_base_dir(self):
        if getattr(sys, "frozen", False):
            return os.path.dirname(sys.executable)
        return os.path.dirname(os.path.abspath(__file__))

    def _resolve_resource_path(self, relative_path):
        if os.path.isabs(relative_path):
            return relative_path if os.path.exists(relative_path) else None

        candidates = []
        if getattr(sys, "frozen", False):
            exe_dir = os.path.dirname(sys.executable)
            if exe_dir:
                candidates.append(os.path.join(exe_dir, relative_path))
            meipass = getattr(sys, "_MEIPASS", "")
            if meipass:
                candidates.append(os.path.join(meipass, relative_path))

        module_dir = os.path.dirname(os.path.abspath(__file__))
        candidates.append(os.path.join(module_dir, relative_path))
        candidates.append(os.path.abspath(relative_path))

        seen = set()
        for candidate in candidates:
            candidate = os.path.abspath(candidate)
            if candidate in seen:
                continue
            seen.add(candidate)
            if os.path.exists(candidate):
                return candidate
        return None

    def _config_path(self):
        candidates = [os.path.join(self._runtime_base_dir(), MAIN_CONFIG_FILE)]
        if getattr(sys, "frozen", False):
            meipass = getattr(sys, "_MEIPASS", "")
            if meipass:
                candidates.append(os.path.join(meipass, MAIN_CONFIG_FILE))
        candidates.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), MAIN_CONFIG_FILE))

        seen = set()
        for candidate in candidates:
            candidate = os.path.abspath(candidate)
            if candidate in seen:
                continue
            seen.add(candidate)
            if os.path.exists(candidate):
                return candidate
        return os.path.join(self._runtime_base_dir(), MAIN_CONFIG_FILE)

    def _load_backend_preference(self, default_tianya=True):
        config_path = self._config_path()
        if not os.path.exists(config_path):
            return default_tianya, False
        try:
            with open(config_path, "r", encoding="utf-8") as f:
                cfg = json.load(f)
            bot_cfg = (cfg or {}).get("bot") or {}
            has_new_flags = "use_tianya_box" in bot_cfg or "use_ghost_box" in bot_cfg
            if has_new_flags:
                return bool(bot_cfg.get("use_tianya_box", False)), bool(bot_cfg.get("use_ghost_box", False))
            return bool(bot_cfg.get("use_hardware_input", default_tianya)), False
        except Exception as exc:
            self._log(f">>> [input_control] load config failed: {exc}")
            return default_tianya, False

    def _log(self, message):
        print(message)
        try:
            log_path = os.path.join(self._runtime_base_dir(), "hardware_control.log")
            with open(log_path, "a", encoding="utf-8") as f:
                f.write(message + "\n")
        except Exception:
            pass

    def _normalize_backend(self, backend):
        if backend in (BACKEND_TIANYA, "hardware"):
            return BACKEND_TIANYA
        if backend == BACKEND_GHOST:
            return BACKEND_GHOST
        if backend == BACKEND_DIRECTINPUT:
            return BACKEND_DIRECTINPUT
        return BACKEND_DIRECTINPUT

    def _select_backend(self, use_tianya_box, use_ghost_box):
        if bool(use_tianya_box):
            return BACKEND_TIANYA
        if bool(use_ghost_box):
            return BACKEND_GHOST
        return BACKEND_DIRECTINPUT

    def _current_backend(self):
        if self._backend == BACKEND_TIANYA:
            return self._tianya_backend
        if self._backend == BACKEND_GHOST:
            return self._ghost_backend
        return self._directinput_backend

    def _log_backend(self):
        label = self.backend_label()
        if label != self._last_backend_log:
            self._last_backend_log = label
            self._log(f">>> [input_control] backend={label}")

    def configure(
        self,
        use_hardware_input=None,
        use_tianya_box=None,
        use_ghost_box=None,
        dll_path=None,
        ghostbox_module_path=None,
        backend=None,
    ):
        self._tianya_backend.configure(dll_path=dll_path)
        self._ghost_backend.configure(module_path=ghostbox_module_path)

        if backend is not None:
            self._backend = self._normalize_backend(backend)
        else:
            if use_tianya_box is None and use_ghost_box is None:
                if use_hardware_input is None:
                    use_tianya_box, use_ghost_box = self._load_backend_preference(default_tianya=True)
                else:
                    use_tianya_box, use_ghost_box = bool(use_hardware_input), False
            else:
                use_tianya_box = bool(use_tianya_box)
                use_ghost_box = bool(use_ghost_box)
                if use_hardware_input is not None and not bool(use_hardware_input):
                    use_tianya_box, use_ghost_box = False, False
            self._backend = self._select_backend(use_tianya_box, use_ghost_box)

        self._log_backend()
        return self._backend

    def uses_hardware_input(self):
        return self._backend in (BACKEND_TIANYA, BACKEND_GHOST)

    def uses_tianya_box(self):
        return self._backend == BACKEND_TIANYA

    def uses_ghost_box(self):
        return self._backend == BACKEND_GHOST

    def backend_label(self):
        return self._current_backend().label()

    def is_available(self):
        return self._current_backend().is_available()

    def delay_rnd(self, min_ms, max_ms):
        return self._current_backend().delay_rnd(min_ms, max_ms)

    def key_down(self, key_str):
        return self._current_backend().key_down(key_str)

    def key_up(self, key_str):
        return self._current_backend().key_up(key_str)

    def key_press(self, key_str):
        return self._current_backend().key_press(key_str)

    def keyDown(self, key_str):
        return self.key_down(key_str)

    def keyUp(self, key_str):
        return self.key_up(key_str)

    def press(self, key_str):
        return self.key_press(key_str)

    def move_to(self, x, y):
        return self._current_backend().move_to(x, y)

    def move_r(self, dx, dy):
        return self._current_backend().move_r(dx, dy)

    def MoveR(self, dx, dy):
        return self.move_r(dx, dy)

    def left_click(self):
        return self._current_backend().left_click()

    def right_click(self):
        return self._current_backend().right_click()

    def left_down(self):
        return self._current_backend().left_down()

    def left_up(self):
        return self._current_backend().left_up()


hw_ctrl = HardwareController()