# -*- coding: utf-8 -*- """ yjm_remote_control.py viewer.html에서 들어오는 원격 제어 이벤트를 실제 Windows 입력으로 실행한다. v2.7 추가: - 마우스 down/up 기반 드래그 지원 - Ctrl/Shift/Alt + 마우스 클릭/드래그 지원 - Ctrl+A, Ctrl+C, Ctrl+V, Ctrl+X 같은 조합키 안정화 - 중간 버튼, 휠 + modifier 지원 - viewer에서 OSK(Windows 화상 키보드, osk.exe) 실행 요청 지원 보안 원칙: - 기본값은 차단. - Python UI의 "원격 제어 허용" 체크가 켜져야만 실행. - viewer.html에서도 "제어: ON"을 눌러야 이벤트를 보냄. - 좌표는 현재 캡처 대상 rect 기준으로 변환한다. """ from __future__ import annotations import ctypes import json import logging import os import subprocess import time from dataclasses import dataclass from typing import Callable, Optional import win32api import win32con @dataclass class ControlResult: ok: bool message: str data: Optional[dict] = None # Win32 SendInput: Unicode 문자 입력용 ULONG_PTR = ctypes.c_ulonglong if ctypes.sizeof(ctypes.c_void_p) == 8 else ctypes.c_ulong class KEYBDINPUT(ctypes.Structure): _fields_ = [ ("wVk", ctypes.c_ushort), ("wScan", ctypes.c_ushort), ("dwFlags", ctypes.c_ulong), ("time", ctypes.c_ulong), ("dwExtraInfo", ULONG_PTR), ] class MOUSEINPUT(ctypes.Structure): _fields_ = [ ("dx", ctypes.c_long), ("dy", ctypes.c_long), ("mouseData", ctypes.c_ulong), ("dwFlags", ctypes.c_ulong), ("time", ctypes.c_ulong), ("dwExtraInfo", ULONG_PTR), ] class HARDWAREINPUT(ctypes.Structure): _fields_ = [("uMsg", ctypes.c_ulong), ("wParamL", ctypes.c_ushort), ("wParamH", ctypes.c_ushort)] class INPUTUNION(ctypes.Union): _fields_ = [("ki", KEYBDINPUT), ("mi", MOUSEINPUT), ("hi", HARDWAREINPUT)] class INPUT(ctypes.Structure): _fields_ = [("type", ctypes.c_ulong), ("union", INPUTUNION)] INPUT_KEYBOARD = 1 KEYEVENTF_KEYUP = 0x0002 KEYEVENTF_UNICODE = 0x0004 VK_MAP = { "Backspace": win32con.VK_BACK, "Tab": win32con.VK_TAB, "Enter": win32con.VK_RETURN, "Return": win32con.VK_RETURN, "Escape": win32con.VK_ESCAPE, "Esc": win32con.VK_ESCAPE, "Space": win32con.VK_SPACE, " ": win32con.VK_SPACE, "Delete": win32con.VK_DELETE, "Del": win32con.VK_DELETE, "Insert": win32con.VK_INSERT, "Ins": win32con.VK_INSERT, "Home": win32con.VK_HOME, "End": win32con.VK_END, "PageUp": win32con.VK_PRIOR, "PageDown": win32con.VK_NEXT, "ArrowLeft": win32con.VK_LEFT, "ArrowRight": win32con.VK_RIGHT, "ArrowUp": win32con.VK_UP, "ArrowDown": win32con.VK_DOWN, "Control": win32con.VK_CONTROL, "Ctrl": win32con.VK_CONTROL, "Shift": win32con.VK_SHIFT, "Alt": win32con.VK_MENU, "Meta": win32con.VK_LWIN, "Win": win32con.VK_LWIN, "ContextMenu": win32con.VK_APPS, "PrintScreen": win32con.VK_SNAPSHOT, "Pause": win32con.VK_PAUSE, "CapsLock": win32con.VK_CAPITAL, "NumLock": win32con.VK_NUMLOCK, "ScrollLock": win32con.VK_SCROLL, } for i in range(1, 25): # pywin32 win32con은 VK_F1~VK_F24를 보통 제공한다. VK_MAP[f"F{i}"] = getattr(win32con, f"VK_F{i}", 0x70 + i - 1) MOD_VK = { "ctrl": win32con.VK_CONTROL, "alt": win32con.VK_MENU, "shift": win32con.VK_SHIFT, "meta": win32con.VK_LWIN, } BUTTON_FLAGS = { "left": (win32con.MOUSEEVENTF_LEFTDOWN, win32con.MOUSEEVENTF_LEFTUP), "middle": (win32con.MOUSEEVENTF_MIDDLEDOWN, win32con.MOUSEEVENTF_MIDDLEUP), "right": (win32con.MOUSEEVENTF_RIGHTDOWN, win32con.MOUSEEVENTF_RIGHTUP), } BUTTON_BY_CODE = { 0: "left", 1: "middle", 2: "right", "0": "left", "1": "middle", "2": "right", "left": "left", "middle": "middle", "right": "right", } def _safe_text(s: object, max_len: int = 200) -> str: if s is None: return "" text = str(s) if len(text) > max_len: text = text[:max_len] return text def _norm_mods(mods: object) -> dict[str, bool]: if not isinstance(mods, dict): return {"ctrl": False, "alt": False, "shift": False, "meta": False} return {name: bool(mods.get(name)) for name in ("ctrl", "alt", "shift", "meta")} class RemoteInputController: def __init__( self, source_manager, is_enabled_callback: Callable[[], bool], logger: Optional[logging.Logger] = None, status_callback: Optional[Callable[[str], None]] = None, ): self.source_manager = source_manager self.is_enabled_callback = is_enabled_callback self.logger = logger or logging.getLogger(__name__) self.status_callback = status_callback or (lambda _text: None) self.last_action_at = 0.0 self.last_move_status_at = 0.0 self.last_coord_ack_at = 0.0 self.min_interval_sec = 0.003 self._mouse_is_down = False self._mouse_button = "left" self._mouse_mod_vks: list[int] = [] def is_enabled(self) -> bool: try: return bool(self.is_enabled_callback()) except Exception: return False def _status(self, text: str) -> None: try: self.status_callback(text) except Exception: pass def _coord_ack(self, action: str, msg: dict, screen_x: int, screen_y: int, *, force: bool = False) -> Optional[dict]: # move 이벤트는 매우 자주 오므로 viewer 표시용 ack는 과도하게 보내지 않는다. now = time.perf_counter() if not force and action in ("move", "drag_move") and now - self.last_coord_ack_at < 0.08: return None self.last_coord_ack_at = now return { "type": "control_ack", "action": action, "screen_x": int(screen_x), "screen_y": int(screen_y), "canvas_x": float(msg.get("x", 0)), "canvas_y": float(msg.get("y", 0)), "frame_w": int(float(msg.get("frame_w", 0) or 0)), "frame_h": int(float(msg.get("frame_h", 0) or 0)), } def handle_text_message(self, text: str) -> ControlResult: try: msg = json.loads(text) except Exception: return ControlResult(False, "invalid json") if not isinstance(msg, dict) or msg.get("type") != "control": return ControlResult(False, "ignored") if not self.is_enabled(): return ControlResult(False, "remote control disabled") now = time.perf_counter() action = str(msg.get("action", "")) # move/drag는 빈도가 높으므로 별도 rate-limit. down/up/key는 너무 막으면 체감이 나빠진다. if action in ("move", "drag_move"): if now - self.last_action_at < self.min_interval_sec: return ControlResult(False, "rate limited") self.last_action_at = now try: if action in ("move", "drag_move"): x, y = self._map_canvas_to_screen(msg) self._move(x, y) if now - self.last_move_status_at > 0.25: self.last_move_status_at = now self._status("원격: 마우스 드래그중..." if self._mouse_is_down else "원격: 마우스 이동중...") return ControlResult(True, f"{action} {x},{y}", self._coord_ack(action, msg, x, y)) if action == "mouse_down": x, y = self._map_canvas_to_screen(msg) button = self._button_name(msg.get("button", 0)) mods = _norm_mods(msg.get("mods")) self._status(self._mod_label(mods, f"원격: {button} 버튼 누름")) self._mouse_down(x, y, button, mods) return ControlResult(True, f"mouse_down {button} {x},{y}", self._coord_ack(action, msg, x, y, force=True)) if action == "mouse_up": x, y = self._map_canvas_to_screen(msg) button = self._button_name(msg.get("button", self._mouse_button)) self._status(f"원격: {button} 버튼 놓음") self._mouse_up(x, y, button) return ControlResult(True, f"mouse_up {button} {x},{y}", self._coord_ack(action, msg, x, y, force=True)) if action in ("click", "dblclick", "rightclick"): x, y = self._map_canvas_to_screen(msg) mods = _norm_mods(msg.get("mods")) if action == "rightclick": button = "right" count = 1 label = "우클릭" else: button = self._button_name(msg.get("button", 0)) count = 2 if action == "dblclick" else 1 label = "더블클릭" if count == 2 else "클릭" self._status(self._mod_label(mods, f"원격: {label}")) self._click(x, y, button, count=count, mods=mods) return ControlResult(True, f"{action} {button} {x},{y}", self._coord_ack(action, msg, x, y, force=True)) if action == "wheel": x, y = self._map_canvas_to_screen(msg) mods = _norm_mods(msg.get("mods")) delta = int(float(msg.get("delta", 0))) self._status(self._mod_label(mods, "원격: 휠 스크롤")) self._wheel(x, y, delta, mods=mods) return ControlResult(True, f"wheel {delta} at {x},{y}", self._coord_ack(action, msg, x, y, force=True)) if action == "type_text": text_value = _safe_text(msg.get("text"), max_len=1000) if not text_value: return ControlResult(False, "empty text") self._status("원격: 타이핑중...") self._type_unicode(text_value) return ControlResult(True, f"type {len(text_value)} chars") if action == "key": key = _safe_text(msg.get("key"), max_len=40) mods = _norm_mods(msg.get("mods")) self._status(self._mod_label(mods, f"원격: 키 입력 {key}")) self._press_key(key, mods) return ControlResult(True, f"key {key}") if action == "key_down": key = _safe_text(msg.get("key"), max_len=40) vk = self._key_to_vk(key) if vk is not None: self._keybd(vk, True) return ControlResult(True, f"key_down {key}") return ControlResult(False, f"unknown key_down {key}") if action == "key_up": key = _safe_text(msg.get("key"), max_len=40) vk = self._key_to_vk(key) if vk is not None: self._keybd(vk, False) return ControlResult(True, f"key_up {key}") return ControlResult(False, f"unknown key_up {key}") if action in ("launch_osk", "osk"): self._launch_osk() self._status("원격: 화상 키보드 실행\nosk.exe") return ControlResult(True, "osk launched") if action == "release_all": self._release_mouse_mods() self._mouse_is_down = False return ControlResult(True, "released") return ControlResult(False, f"unknown action: {action}") except Exception as exc: # 예외가 나면 stuck modifier 방지를 위해 일단 놓는다. self._release_mouse_mods() self.logger.error("remote control error: %s", exc) return ControlResult(False, str(exc)) def _map_canvas_to_screen(self, msg: dict) -> tuple[int, int]: target = self.source_manager.get_target() if not target: raise RuntimeError("capture target not available") left, top, right, bottom = target.rect target_w = max(1, int(right - left)) target_h = max(1, int(bottom - top)) frame_w = max(1, int(float(msg.get("frame_w", target_w)))) frame_h = max(1, int(float(msg.get("frame_h", target_h)))) x = float(msg.get("x", 0)) y = float(msg.get("y", 0)) sx = int(round(int(left) + max(0.0, min(frame_w - 1, x)) * target_w / frame_w)) sy = int(round(int(top) + max(0.0, min(frame_h - 1, y)) * target_h / frame_h)) return sx, sy @staticmethod def _move(x: int, y: int) -> None: win32api.SetCursorPos((int(x), int(y))) @staticmethod def _button_name(button: object) -> str: return BUTTON_BY_CODE.get(button, "left") @staticmethod def _mod_label(mods: dict[str, bool], base: str) -> str: names = [] if mods.get("ctrl"): names.append("Ctrl") if mods.get("shift"): names.append("Shift") if mods.get("alt"): names.append("Alt") if mods.get("meta"): names.append("Win") if not names: return base return f"원격: {'+'.join(names)}\n{base.replace('원격: ', '')}" def _press_mouse_mods(self, mods: dict[str, bool]) -> None: self._release_mouse_mods() self._mouse_mod_vks = [] for name in ("ctrl", "alt", "shift", "meta"): if bool(mods.get(name)): vk = MOD_VK[name] self._mouse_mod_vks.append(vk) self._keybd(vk, True) time.sleep(0.005) def _release_mouse_mods(self) -> None: for vk in reversed(self._mouse_mod_vks): try: self._keybd(vk, False) time.sleep(0.003) except Exception: pass self._mouse_mod_vks = [] def _mouse_down(self, x: int, y: int, button: str = "left", mods: dict[str, bool] | None = None) -> None: self._move(x, y) self._press_mouse_mods(mods or {}) down, _up = BUTTON_FLAGS.get(button, BUTTON_FLAGS["left"]) win32api.mouse_event(down, 0, 0, 0, 0) self._mouse_is_down = True self._mouse_button = button def _mouse_up(self, x: int, y: int, button: str = "left") -> None: self._move(x, y) _down, up = BUTTON_FLAGS.get(button, BUTTON_FLAGS.get(self._mouse_button, BUTTON_FLAGS["left"])) win32api.mouse_event(up, 0, 0, 0, 0) self._mouse_is_down = False self._release_mouse_mods() def _click(self, x: int, y: int, button: str = "left", count: int = 1, mods: dict[str, bool] | None = None) -> None: self._move(x, y) self._press_mouse_mods(mods or {}) down, up = BUTTON_FLAGS.get(button, BUTTON_FLAGS["left"]) try: for _ in range(max(1, int(count))): win32api.mouse_event(down, 0, 0, 0, 0) time.sleep(0.035) win32api.mouse_event(up, 0, 0, 0, 0) time.sleep(0.055) finally: self._release_mouse_mods() def _wheel(self, x: int, y: int, delta: int, mods: dict[str, bool] | None = None) -> None: self._move(x, y) if delta == 0: return self._press_mouse_mods(mods or {}) try: wheel = -120 if delta > 0 else 120 win32api.mouse_event(win32con.MOUSEEVENTF_WHEEL, 0, 0, wheel, 0) finally: self._release_mouse_mods() @staticmethod def _keybd(vk: int, down: bool) -> None: flags = 0 if down else win32con.KEYEVENTF_KEYUP win32api.keybd_event(int(vk), 0, flags, 0) def _press_vk(self, vk: int) -> None: self._keybd(vk, True) time.sleep(0.015) self._keybd(vk, False) def _key_to_vk(self, key: str) -> Optional[int]: if key in VK_MAP: return VK_MAP[key] if len(key) == 1: ch = key.upper() if "A" <= ch <= "Z" or "0" <= ch <= "9": return ord(ch) return None def _press_key(self, key: str, mods: dict | None = None) -> None: mods = _norm_mods(mods) vk = self._key_to_vk(key) if vk is None: # 처리 못 하는 특수키는 무시한다. 일반 문자는 type_text로 받아야 한다. return down_mods: list[int] = [] for name in ("ctrl", "alt", "shift", "meta"): if bool(mods.get(name)): mvk = MOD_VK[name] down_mods.append(mvk) self._keybd(mvk, True) time.sleep(0.005) try: self._press_vk(vk) finally: for mvk in reversed(down_mods): self._keybd(mvk, False) time.sleep(0.005) @staticmethod def _send_unicode_unit(code_unit: int, keyup: bool = False) -> None: flags = KEYEVENTF_UNICODE | (KEYEVENTF_KEYUP if keyup else 0) inp = INPUT(type=INPUT_KEYBOARD, union=INPUTUNION(ki=KEYBDINPUT(0, code_unit, flags, 0, 0))) ctypes.windll.user32.SendInput(1, ctypes.byref(inp), ctypes.sizeof(INPUT)) def _type_unicode(self, text: str) -> None: # SendInput Unicode는 IME 상태와 별개로 대부분의 텍스트 입력창에 문자 입력 가능. # 일부 관리자 권한 창/UAC/게임/보안 프로그램에서는 차단될 수 있다. encoded = text.encode("utf-16-le", errors="surrogatepass") for i in range(0, len(encoded), 2): unit = encoded[i] | (encoded[i + 1] << 8) self._send_unicode_unit(unit, keyup=False) self._send_unicode_unit(unit, keyup=True) time.sleep(0.003) def _launch_osk(self) -> None: # Windows 화상 키보드. shell=True 없이 우선 실행하고, 실패하면 start 명령으로 재시도. try: subprocess.Popen(["osk.exe"], close_fds=True) return except Exception as first_exc: self.logger.debug("osk direct launch failed: %s", first_exc) try: subprocess.Popen(["cmd", "/c", "start", "", "osk.exe"], close_fds=True) return except Exception as second_exc: self.logger.error("osk launch failed: %s", second_exc) raise RuntimeError(f"osk 실행 실패: {second_exc}")