# -*- coding: utf-8 -*- # FILE: _St/RTC8790_8791/yjm_win2rtc_capture.py | ROLE: RTC8790/8791 원격 화면·Tray UI 서비스 """ yjm_win2rtc_capture.py 현재 단계: WebSocket JPEG preview + LAN/WAN 공유 준비 + 캡처 대상 선택 확장 지원 캡처 대상: 1) 선택 창 HWND 2) 디스플레이 0, 1, ... 3) 전체 바탕화면 주의: 아직 WebRTC가 아니다. WebRTC/aiortc로 넘어가기 전, 캡처·프레임·포트·공유 URL을 검증하는 단계다. """ from __future__ import annotations import argparse import asyncio import io import json import logging import os import secrets import socket import subprocess import sys import threading import time import webbrowser from functools import partial from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, urlparse import urllib.request import urllib.error ST_ROOT = Path(__file__).resolve().parent.parent if str(ST_ROOT) not in sys.path: sys.path.insert(0, str(ST_ROOT)) TRAY_ICON_PATH = ST_ROOT / "Shared" / "assets" / "ig_styellow_tray.ico" try: from Shared.windows_tray import WindowsTrayIcon except Exception: WindowsTrayIcon = None import tkinter as tk from tkinter import ttk import mss from PIL import Image, ImageDraw import win32gui import win32api from yjm_capture_sources import CaptureSourceManager from yjm_monitor_probe import set_dpi_awareness from yjm_net_upnp import UpnpPortMapper, make_viewer_url from yjm_remote_control import RemoteInputController from yjm_remote_status import RemoteActivityOverlay from yjm_file_transfer import make_file_transfer_handler from yjm_clipboard_sync import ClipboardSyncManager from host.protocol import MSG_CLIPBOARD_GET, MSG_HOST_OPTION_SET, MSG_HOST_OPTION_ACK, make_msg from yjm_settings import AppSettings, SnapshotSettings, UiSettings, load_settings, save_settings from yjm_ai_snapshot_ring import SnapshotRingWorker try: from websockets.asyncio.server import serve from websockets.datastructures import Headers from websockets.http11 import Response except Exception as exc: # pragma: no cover - 사용 환경 진단용 serve = None Headers = None Response = None _WEBSOCKETS_IMPORT_ERROR = exc else: _WEBSOCKETS_IMPORT_ERROR = None selected_hwnd: int | None = None hwnd_lock = threading.Lock() BUILD_VERSION = "2.3.8.324" BUILD_TAG = "v324" def st_token_file_path(base_dir: Path) -> Path: """RTC/Auto8700 공통 인증 토큰 파일. 배포 ZIP에는 포함하지 않는다.""" return base_dir.resolve().parent / "_St.token" def load_st_token_entries(path: Path, create: bool = True, logger=None) -> list[dict]: """SSH authorized_keys처럼 한 줄에 토큰 1개 + 선택 이름/comment를 허용한다.""" entries: list[dict] = [] try: if path.exists(): for line in path.read_text(encoding="utf-8", errors="ignore").splitlines(): raw = line.strip() if not raw or raw.startswith("#"): continue parts = raw.split(None, 1) token_value = (parts[0] if parts else "").strip() comment = (parts[1] if len(parts) > 1 else "").strip() if len(token_value) >= 16: entries.append({"token": token_value, "comment": comment}) except Exception as exc: if logger: logger.warning("_St.token 읽기 실패: %s", exc) if not entries and create: try: path.parent.mkdir(parents=True, exist_ok=True) token_value = secrets.token_urlsafe(32) text = "# _St shared tokens: one token per line, optional name/comment after whitespace\n" text += f"{token_value} rtc-auto8700-default\n" path.write_text(text, encoding="utf-8") entries.append({"token": token_value, "comment": "rtc-auto8700-default"}) if logger: logger.info("_St.token 자동 생성: %s", path) except Exception as exc: if logger: logger.warning("_St.token 생성 실패: %s", exc) return entries def st_token_values(entries: list[dict]) -> list[str]: vals: list[str] = [] for ent in entries or []: tok = str((ent or {}).get("token") or "").strip() if tok: vals.append(tok) return vals def st_token_allowed(got: str, allowed_tokens: list[str]) -> bool: got = str(got or "").strip() if not got: return False for tok in allowed_tokens or []: if tok and secrets.compare_digest(got, tok): return True return False def find_free_port(host: str, preferred_port: int, max_try: int = 60) -> int: """preferred_port가 사용 중이면 다음 포트를 찾는다.""" bind_host = host or "0.0.0.0" for port in range(int(preferred_port), int(preferred_port) + int(max_try)): try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind((bind_host, port)) return port except OSError: continue raise OSError(f"사용 가능한 포트를 찾지 못했습니다: {preferred_port}~{preferred_port + max_try - 1}") class WindowOverlay: """마우스가 멈춘 foreground HWND를 선택하고 빨간 테두리를 표시한다.""" def __init__(self, root, logger, on_hwnd_change_callback, is_locked_callback=None, is_window_mode_callback=None): self.root = root self.logger = logger self.on_hwnd_change_callback = on_hwnd_change_callback self.is_locked_callback = is_locked_callback or (lambda: False) self.is_window_mode_callback = is_window_mode_callback or (lambda: True) self.overlay = None self.last_rect = None self.running = True self.thread = None self.last_mouse_pos = None self.last_mouse_move_time = 0 self.mouse_stationary_threshold = 2 def start_monitoring(self): if not self.thread or not self.thread.is_alive(): self.thread = threading.Thread(target=self.monitor_foreground_window, daemon=True) self.thread.start() self.logger.debug("Started HWND monitoring thread") def _is_locked(self) -> bool: try: return bool(self.is_locked_callback()) except Exception: return False def _is_window_mode(self) -> bool: try: return bool(self.is_window_mode_callback()) except Exception: return True def _is_own_window(self, hwnd: int) -> bool: if hwnd == 0: return True try: if hwnd == self.root.winfo_id(): return True title = win32gui.GetWindowText(hwnd) or "" if title.startswith("yjm_win2rtc"): return True if "viewer.html" in title and "yjm_win2rtc" in title: return True except Exception: pass return False def monitor_foreground_window(self): global selected_hwnd while self.running: try: if not self._is_window_mode(): self.hide_overlay() time.sleep(0.2) continue current_pos = win32gui.GetCursorPos() current_time = time.time() if self.last_mouse_pos and self.last_mouse_pos != current_pos: self.last_mouse_move_time = current_time self.last_mouse_pos = current_pos if current_time - self.last_mouse_move_time >= self.mouse_stationary_threshold: hwnd = win32gui.GetForegroundWindow() with hwnd_lock: current_selected = selected_hwnd if self._is_locked() and current_selected: if win32gui.IsWindow(current_selected): rect = win32gui.GetWindowRect(current_selected) self.create_overlay(rect) else: selected_hwnd = None self.on_hwnd_change_callback(None, None) time.sleep(0.1) continue if hwnd != selected_hwnd and not self._is_own_window(hwnd): selected_hwnd = hwnd rect = win32gui.GetWindowRect(hwnd) title = win32gui.GetWindowText(hwnd) self.create_overlay(rect) self.on_hwnd_change_callback(hwnd, title) self.logger.debug("Selected HWND=%s title=%s rect=%s", hwnd, title, rect) elif selected_hwnd and not win32gui.IsWindow(selected_hwnd): selected_hwnd = None self.hide_overlay() self.on_hwnd_change_callback(None, None) time.sleep(0.1) except Exception as exc: self.logger.error("HWND monitoring error: %s", exc) time.sleep(1) def create_overlay(self, rect): if not self._is_window_mode(): self.hide_overlay() return x, y, x1, y1 = rect width = x1 - x height = y1 - y if width <= 0 or height <= 0: return try: if not self.overlay: self.overlay = tk.Toplevel(self.root) self.overlay.overrideredirect(True) self.overlay.attributes("-topmost", True) self.overlay.attributes("-transparentcolor", "white") self.canvas = tk.Canvas(self.overlay, bg="white", highlightthickness=0) self.canvas.pack(fill=tk.BOTH, expand=True) if rect != self.last_rect: self.overlay.geometry(f"{width}x{height}+{x}+{y}") self.canvas.config(width=width, height=height) self.canvas.delete("all") self.canvas.create_rectangle(2, 2, width - 2, height - 2, outline="red", width=4) self.last_rect = rect except Exception as exc: self.logger.error("Overlay creation/update error: %s", exc) def hide_overlay(self): if self.overlay: try: self.overlay.destroy() except Exception: pass self.overlay = None self.last_rect = None def stop(self): self.running = False self.hide_overlay() class WsFrameServer: """캡처 스레드에서 들어온 JPEG bytes를 모든 viewer.html 클라이언트에 보낸다. viewer.html에서 들어오는 텍스트 JSON은 원격 제어 이벤트로 처리할 수 있다. """ def __init__(self, host="0.0.0.0", port=8765, token="", logger=None, auto_port=True, control_handler=None): self.host = host self.requested_port = int(port) self.port = int(port) self.token = token self.logger = logger or logging.getLogger(__name__) self.auto_port = auto_port self.loop = None self.thread = None self.clients = set() self.running = False self.ready_event = threading.Event() self.control_handler = control_handler async def handler(self, websocket): queue = asyncio.Queue(maxsize=1) self.clients.add(queue) self.logger.info("viewer connected: clients=%d", len(self.clients)) async def send_frames(): while True: frame = await queue.get() await websocket.send(frame) async def receive_control(): async for message in websocket: if isinstance(message, str) and self.control_handler: result = self.control_handler(message) if result and getattr(result, "ok", False): self.logger.debug("control ok: %s", getattr(result, "message", "")) elif result and getattr(result, "message", "") != "ignored": self.logger.debug("control blocked: %s", getattr(result, "message", "")) send_task = asyncio.create_task(send_frames()) recv_task = asyncio.create_task(receive_control()) try: done, pending = await asyncio.wait( {send_task, recv_task}, return_when=asyncio.FIRST_EXCEPTION, ) for task in done: exc = task.exception() if exc: raise exc except Exception as exc: self.logger.debug("viewer disconnected: %s", exc) finally: for task in (send_task, recv_task): task.cancel() try: if self.control_handler: self.control_handler(json.dumps({"type": "control", "action": "release_all"})) except Exception as exc: self.logger.debug("viewer disconnect release_all failed: %s", exc) self.clients.discard(queue) self.logger.info("viewer removed: clients=%d", len(self.clients)) def _check_token_process_request(self, connection, request): if not self.token: return None try: parsed = urlparse(request.path) token = parse_qs(parsed.query).get("token", [""])[0] if secrets.compare_digest(token, self.token): return None except Exception: pass if Response is None or Headers is None: return None body = b"Forbidden: invalid token\n" return Response( 403, "Forbidden", Headers([("Content-Type", "text/plain"), ("Content-Length", str(len(body)))]), body, ) async def _run(self): if serve is None: raise RuntimeError(f"websockets import 실패: {_WEBSOCKETS_IMPORT_ERROR}") if self.auto_port: self.port = find_free_port(self.host, self.requested_port) async with serve( self.handler, self.host, self.port, max_size=None, compression=None, ping_interval=20, ping_timeout=20, process_request=self._check_token_process_request, ): self.logger.info("WebSocket server ws://%s:%s", self.host, self.port) self.ready_event.set() await asyncio.Future() def start(self): if self.running: return self.running = True self.ready_event.clear() self.thread = threading.Thread(target=self._thread_main, daemon=True) self.thread.start() def _thread_main(self): self.loop = asyncio.new_event_loop() asyncio.set_event_loop(self.loop) try: self.loop.run_until_complete(self._run()) except Exception as exc: self.logger.error("WebSocket server error: %s", exc) finally: self.running = False self.ready_event.set() def push_frame(self, frame_bytes: bytes): if not self.loop or not self.running: return self.loop.call_soon_threadsafe(self._push_message_in_loop, frame_bytes) def push_text(self, text: str): if not self.loop or not self.running: return self.loop.call_soon_threadsafe(self._push_message_in_loop, str(text)) def _push_message_in_loop(self, message): for queue in list(self.clients): try: if queue.full(): try: queue.get_nowait() except asyncio.QueueEmpty: pass queue.put_nowait(message) except Exception: self.clients.discard(queue) class StaticViewerServer: def __init__(self, directory, host="0.0.0.0", port=8000, logger=None, auto_port=True, handler_factory=None): self.directory = str(Path(directory).resolve()) self.host = host self.requested_port = int(port) self.port = int(port) self.logger = logger or logging.getLogger(__name__) self.auto_port = auto_port self.handler_factory = handler_factory self.httpd = None self.thread = None def start(self): if self.httpd: return if self.auto_port: self.port = find_free_port(self.host, self.requested_port) if self.handler_factory: handler = self.handler_factory(self.directory) else: handler = partial(SimpleHTTPRequestHandler, directory=self.directory) self.httpd = ThreadingHTTPServer((self.host, self.port), handler) self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True) self.thread.start() self.logger.info("HTTP viewer http://%s:%s/viewer.html", self.host, self.port) def stop(self): if self.httpd: self.httpd.shutdown() self.httpd.server_close() self.httpd = None class CaptureWorker: def __init__(self, frame_server, source_manager, logger, fps=12, jpeg_quality=70, max_width=1280, draw_border=True): self.frame_server = frame_server self.source_manager = source_manager self.logger = logger self.fps = int(fps) self.jpeg_quality = int(jpeg_quality) self.max_width = int(max_width) self.draw_border = bool(draw_border) self.running = False self.thread = None self.frame_count = 0 self.last_target_title = "" self.last_frame_size = {"width": 0, "height": 0} self.last_capture_rect = {} self.last_capture_mode = "" self.last_capture_saved_path = "" def update_options(self, fps=None, jpeg_quality=None, max_width=None, draw_border=None): if fps is not None: self.fps = max(1, int(fps)) if jpeg_quality is not None: self.jpeg_quality = max(1, min(95, int(jpeg_quality))) if max_width is not None: self.max_width = max(0, int(max_width)) if draw_border is not None: self.draw_border = bool(draw_border) def start(self): if self.running: self.logger.info("capture already running") return self.running = True self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() self.logger.info( "capture started: fps=%s quality=%s max_width=%s border=%s", self.fps, self.jpeg_quality, self.max_width, self.draw_border, ) def stop(self): self.running = False def _run(self): with mss.mss() as sct: while self.running: started = time.perf_counter() try: target = self.source_manager.get_target() if not target: time.sleep(0.2) continue monitor = target.monitor_dict self.last_capture_rect = dict(monitor) self.last_capture_mode = str(target.mode) shot = sct.grab(monitor) pil = Image.frombytes("RGB", shot.size, shot.bgra, "raw", "BGRX") if self.draw_border: draw = ImageDraw.Draw(pil) w, h = pil.size for i in range(4): draw.rectangle((i, i, w - 1 - i, h - 1 - i), outline=(255, 0, 0)) if self.max_width and pil.width > self.max_width: new_height = int(pil.height * (self.max_width / pil.width)) pil = pil.resize((self.max_width, max(1, new_height)), Image.Resampling.BILINEAR) self.last_frame_size = {"width": int(pil.width), "height": int(pil.height)} buf = io.BytesIO() pil.save(buf, format="JPEG", quality=self.jpeg_quality, optimize=False) self.frame_server.push_frame(buf.getvalue()) self.frame_count += 1 self.last_target_title = target.title except Exception as exc: self.logger.error("capture error: %s", exc) time.sleep(0.5) interval = 1.0 / max(1, int(self.fps)) elapsed = time.perf_counter() - started time.sleep(max(0.001, interval - elapsed)) class Win2RtcApp: def __init__(self, root, args): set_dpi_awareness() self.root = root self.args = args self.logger = logging.getLogger("yjm_win2rtc") self.logger.setLevel(logging.DEBUG if args.verbose else logging.INFO) handler = logging.StreamHandler() handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s")) self.logger.addHandler(handler) self.root.title("RTC8790_8791 - Windows Remote Desktop Bridge") base_dir = Path(__file__).resolve().parent self.base_dir = base_dir self.tray = WindowsTrayIcon("RTC8790_8791", "RTC8790_8791 - 클릭하여 화면 복원", icon_path=TRAY_ICON_PATH) if WindowsTrayIcon is not None else None self.tray_hidden = False self.config_path = self.resolve_same_name_config_path(base_dir) self.legacy_config_path = base_dir / "rtc8790_8791_config.json" self.app_settings = self.load_app_settings_with_legacy_fallback() self._settings_save_after_id = None self.restore_main_window_geometry() self._st_token_entries = load_st_token_entries(st_token_file_path(base_dir), create=bool(args.token), logger=self.logger) self._st_tokens = st_token_values(self._st_token_entries) self.token = (self._st_tokens[0] if (args.token and self._st_tokens) else "") self.upnp = UpnpPortMapper(self.logger) self.lan_ip = UpnpPortMapper.get_lan_ip() self.viewer_dir = base_dir / "web" self.state_file = Path(args.state_file).resolve() if getattr(args, "state_file", "") else None self.state_log_lines: list[dict] = [] ui = self.app_settings.ui self.source_manager = CaptureSourceManager(self.get_selected_hwnd) self.control_var = tk.BooleanVar(value=bool(ui.remote_control)) self.file_transfer_var = tk.BooleanVar(value=bool(ui.file_transfer)) self.clipboard_var = tk.BooleanVar(value=bool(ui.clipboard)) self.remote_activity_overlay = RemoteActivityOverlay(self.root, hold_ms=1300) self.recv_dir = base_dir / "recv_files" self.remote_control = RemoteInputController( self.source_manager, is_enabled_callback=lambda: self.control_var.get(), logger=self.logger, status_callback=self.remote_activity_overlay.show, ) self.clipboard_sync = ClipboardSyncManager( self.root, is_enabled_callback=lambda: self.clipboard_var.get(), notify_viewers_callback=self.notify_viewers, status_callback=self.remote_activity_overlay.show, logger=self.logger, poll_ms=900, max_chars=200_000, ) self.clipboard_sync.start() self.ws_server = WsFrameServer( args.host, args.ws_port, self.token, self.logger, auto_port=not args.no_auto_port, control_handler=self.handle_ws_text_message, ) def _http_handler_factory(directory): base_handler = make_file_transfer_handler( web_dir=directory, recv_dir=self.recv_dir, token=self.token, is_enabled_callback=lambda: self.file_transfer_var.get(), status_callback=self.remote_activity_overlay.show, web_notify_callback=self.notify_viewers, logger=self.logger, ) cdp_proxy_tokens = list(self._st_tokens) class RtcHttpHandler(base_handler): """RTC viewer에서 Auto8700 CDP API를 안전하게 호출하기 위한 host-side 프록시. 원격 모바일 브라우저의 127.0.0.1은 host PC가 아니므로, viewer JS가 Auto8700 8707을 직접 호출하지 않고 RTC HTTP 서버를 경유한다. """ auto8700_base = "http://127.0.0.1:8707" cdp_get_map = { "/api/auto8700/cdp/status": "/api/cdp/status", "/api/auto8700/cdp/targets": "/api/cdp/targets", "/api/auto8700/cdp/ig/status": "/api/cdp/ig/status", "/api/auto8700/cdp/ig/schedule_storage": "/api/cdp/ig/schedule_storage", } cdp_post_map = { "/api/auto8700/cdp/evaluate": "/api/cdp/evaluate", "/api/auto8700/cdp/bring_to_front": "/api/cdp/bring_to_front", "/api/auto8700/cdp/ig/bring_to_front": "/api/cdp/ig/bring_to_front", "/api/auto8700/cdp/ig/send_event": "/api/cdp/ig/send_event", "/api/auto8700/cdp/ig/side_open": "/api/cdp/ig/side_open", } def _proxy_json(self, method: str, target_path: str, body: bytes = b"") -> None: try: url = self.auto8700_base + target_path headers = {"User-Agent": "rtc8790-auto8700-cdp-proxy"} data = None if method.upper() == "POST": data = body or b"{}" headers["Content-Type"] = "application/json; charset=utf-8" req = urllib.request.Request(url, data=data, method=method.upper(), headers=headers) with urllib.request.urlopen(req, timeout=5.0) as resp: raw = resp.read() code = int(getattr(resp, "status", 200) or 200) ctype = resp.headers.get("Content-Type") or "application/json; charset=utf-8" self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(raw))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(raw) except Exception as exc: payload = json.dumps({"ok": False, "error": "auto8700_proxy_failed", "message": str(exc)}, ensure_ascii=False).encode("utf-8") self.send_response(502) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(payload))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(payload) def _check_cdp_proxy_token(self) -> bool: try: if not cdp_proxy_tokens: return True parsed = urlparse(self.path) got_qs = parse_qs(parsed.query).get("token", [""])[0] got_hdr = self.headers.get("X-RTC-Token") or self.headers.get("X-Rtc-Token") or "" return st_token_allowed(got_qs, cdp_proxy_tokens) or st_token_allowed(got_hdr, cdp_proxy_tokens) except Exception: return False def do_GET(self): # noqa: N802 parsed = urlparse(self.path) if parsed.path in self.cdp_get_map: if not self._check_cdp_proxy_token(): return self._reject(403, "invalid token") return self._proxy_json("GET", self.cdp_get_map[parsed.path]) return super().do_GET() def do_POST(self): # noqa: N802 parsed = urlparse(self.path) if parsed.path in self.cdp_post_map: if not self._check_cdp_proxy_token(): return self._reject(403, "invalid token") try: length = int(self.headers.get("Content-Length") or 0) except Exception: length = 0 body = self.rfile.read(length) if length else b"{}" return self._proxy_json("POST", self.cdp_post_map[parsed.path], body) return super().do_POST() return RtcHttpHandler self.http_server = StaticViewerServer( self.viewer_dir, args.host, args.http_port, self.logger, auto_port=not args.no_auto_port, handler_factory=_http_handler_factory, ) self.capture_worker = CaptureWorker( self.ws_server, self.source_manager, self.logger, fps=max(1, int(ui.fps or args.fps)), jpeg_quality=max(20, min(95, int(ui.jpeg_quality or args.quality))), max_width=max(0, int(ui.max_width if ui.max_width is not None else args.max_width)), draw_border=bool(ui.draw_border), ) self.snapshot_ring = SnapshotRingWorker( self.source_manager, self.logger, status_callback=lambda msg: self.root.after(0, lambda: self.status_var.set(msg)) if hasattr(self, "status_var") else None, ) ui = self.app_settings.ui self.hwnd_var = tk.StringVar(value="선택된 창: 없음") self.source_var = tk.StringVar(value=str(ui.source or "desktop")) self.display_var = tk.IntVar(value=max(0, int(ui.display_index or 0))) self.lock_var = tk.BooleanVar(value=bool(ui.lock_window)) self.border_var = tk.BooleanVar(value=bool(ui.draw_border)) self.fps_var = tk.IntVar(value=max(1, int(ui.fps or args.fps))) self.quality_var = tk.IntVar(value=max(20, min(95, int(ui.jpeg_quality or args.quality)))) self.max_width_var = tk.IntVar(value=max(0, int(ui.max_width if ui.max_width is not None else args.max_width))) self.control_var = tk.BooleanVar(value=bool(ui.remote_control)) self.file_transfer_var = tk.BooleanVar(value=bool(ui.file_transfer)) self.clipboard_var = tk.BooleanVar(value=bool(ui.clipboard)) self.status_var = tk.StringVar(value="대기 중") self.local_url_var = tk.StringVar(value="서버 시작 전") self.share_url_var = tk.StringVar(value="UPnP 실행 전") snap = self.app_settings.snapshot self.ai_snapshot_var = tk.BooleanVar(value=bool(snap.enabled)) self.ai_snapshot_folder_var = tk.StringVar(value=str(snap.folder)) self.ai_snapshot_keep_var = tk.IntVar(value=int(snap.keep_count)) self.ai_snapshot_interval_var = tk.DoubleVar(value=float(snap.interval_sec)) self.ai_snapshot_min_interval_var = tk.DoubleVar(value=float(snap.min_save_interval_sec)) self.ai_snapshot_change_var = tk.DoubleVar(value=float(snap.change_threshold_percent)) self.ai_snapshot_max_width_var = tk.IntVar(value=int(snap.max_width)) self.ai_snapshot_quality_var = tk.IntVar(value=int(snap.jpeg_quality)) self.overlay = WindowOverlay( self.root, self.logger, self.on_hwnd_change, is_locked_callback=lambda: self.lock_var.get(), is_window_mode_callback=lambda: self.source_var.get() == "window", ) self.overlay.start_monitoring() self.setup_ui() self.source_manager.set_desktop() self.snapshot_ring.update_settings(self.app_settings.snapshot) self.root.protocol("WM_DELETE_WINDOW", self.on_close) self.root.bind("", self.on_root_configure, add="+") self.install_settings_traces() self.refresh_local_url_label() self.write_state("init") self.root.after(80, self.hide_to_tray_on_startup) self.root.after(200, self.poll_tray_restore) self.root.after(300, self.apply_startup_options) self.root.after(1000, self.periodic_state_write) @staticmethod def resolve_same_name_config_path(base_dir: Path) -> Path: """실행한 PY 파일과 동일한 파일명.json을 설정 파일로 사용한다.""" try: argv0 = Path(sys.argv[0]).resolve() if argv0.exists() and argv0.suffix.lower() in (".py", ".pyw", ".exe"): return argv0.with_suffix(".json") except Exception: pass return base_dir / "rtc8790_8791_main.json" def load_app_settings_with_legacy_fallback(self) -> AppSettings: try: if self.config_path.exists(): return load_settings(self.config_path) if self.legacy_config_path.exists(): return load_settings(self.legacy_config_path) except Exception: pass return AppSettings() def restore_main_window_geometry(self) -> None: ui = getattr(self.app_settings, "ui", UiSettings()) geometry = str(getattr(ui, "window_geometry", "") or "").strip() if not self.is_safe_tk_geometry(geometry): geometry = "940x850" try: self.root.geometry(geometry) except Exception: self.root.geometry("940x850") state = str(getattr(ui, "window_state", "normal") or "normal").lower() if state == "zoomed": self.root.after(100, lambda: self.safe_set_root_state("zoomed")) @staticmethod def is_safe_tk_geometry(value: str) -> bool: import re return bool(re.fullmatch(r"\d{3,5}x\d{3,5}([+-]\d{1,6}[+-]\d{1,6})?", str(value or ""))) def safe_set_root_state(self, state: str) -> None: try: self.root.state(state) except Exception: pass def build_ui_settings(self) -> UiSettings: try: state = str(self.root.state() or "normal") except Exception: state = "normal" if state == "iconic": state = "normal" try: geometry = self.root.geometry() except Exception: geometry = "940x850" if not self.is_safe_tk_geometry(geometry): geometry = "940x850" return UiSettings( window_geometry=geometry, window_state=state, source=str(self.source_var.get() or "desktop"), display_index=max(0, int(self.display_var.get() or 0)), lock_window=bool(self.lock_var.get()), remote_control=bool(self.control_var.get()), file_transfer=bool(self.file_transfer_var.get()), clipboard=bool(self.clipboard_var.get()), fps=max(1, int(self.fps_var.get() or 12)), jpeg_quality=max(20, min(95, int(self.quality_var.get() or 70))), max_width=max(0, int(self.max_width_var.get() or 0)), draw_border=bool(self.border_var.get()), ) def save_all_settings(self) -> None: self.app_settings.ui = self.build_ui_settings() self.app_settings.snapshot = self.build_ai_snapshot_settings() save_settings(self.config_path, self.app_settings) def save_all_settings_quietly(self) -> None: try: self.save_all_settings() except Exception as exc: try: self.logger.debug("settings save failed: %s", exc) except Exception: pass def schedule_settings_save(self, delay_ms: int = 700) -> None: try: if self._settings_save_after_id is not None: self.root.after_cancel(self._settings_save_after_id) except Exception: pass try: self._settings_save_after_id = self.root.after(int(delay_ms), self._finish_scheduled_settings_save) except Exception: self.save_all_settings_quietly() def _finish_scheduled_settings_save(self) -> None: self._settings_save_after_id = None self.save_all_settings_quietly() def on_root_configure(self, event=None) -> None: try: if event is not None and event.widget is not self.root: return self.schedule_settings_save(1000) except Exception: pass def install_settings_traces(self) -> None: names = ( "source_var", "display_var", "lock_var", "border_var", "fps_var", "quality_var", "max_width_var", "control_var", "file_transfer_var", "clipboard_var", "ai_snapshot_var", "ai_snapshot_folder_var", "ai_snapshot_keep_var", "ai_snapshot_interval_var", "ai_snapshot_min_interval_var", "ai_snapshot_change_var", "ai_snapshot_max_width_var", "ai_snapshot_quality_var", ) for name in names: try: getattr(self, name).trace_add("write", lambda *_: self.schedule_settings_save()) except Exception: pass def append_state_log(self, src: str, message: str, level: str = "info") -> None: try: item = {"ts": time.strftime("%H:%M:%S"), "src": str(src), "message": str(message), "level": str(level)} self.state_log_lines.append(item) self.state_log_lines = self.state_log_lines[-200:] except Exception: pass def build_state(self) -> dict: try: target = self.source_manager.get_target() except Exception: target = None rect = {} mode = str(getattr(self.source_manager, "mode", "") or "") title = "" if target: left, top, right, bottom = target.rect rect = {"left": int(left), "top": int(top), "right": int(right), "bottom": int(bottom), "width": int(right-left), "height": int(bottom-top)} mode = str(target.mode or mode) title = str(target.title or "") else: try: rect = dict(getattr(self.capture_worker, "last_capture_rect", {}) or {}) mode = str(getattr(self.capture_worker, "last_capture_mode", mode) or mode) except Exception: pass local_url = "" lan_url = "" try: local_url = self.get_local_viewer_url() lan_url = self.get_lan_viewer_url() except Exception: pass return { "build_tag": BUILD_TAG, "build_version": BUILD_VERSION, "pid": os.getpid(), "config_path": str(self.config_path), "local_url": local_url, "lan_url": lan_url, "clients": len(getattr(self.ws_server, "clients", []) or []), "capture": { "mode": mode, "rect": rect, "title": title or getattr(self.capture_worker, "last_target_title", ""), "frame_size": getattr(self.capture_worker, "last_frame_size", {"width": 0, "height": 0}), "frame_count": int(getattr(self.capture_worker, "frame_count", 0) or 0), "running": bool(getattr(self.capture_worker, "running", False)), }, "control": {"enabled": bool(self.control_var.get())}, "http": {"host": self.args.host, "port": int(getattr(self.http_server, "port", self.args.http_port))}, "ws": {"host": self.args.host, "port": int(getattr(self.ws_server, "port", self.args.ws_port))}, "marker": {"found": 0, "reason": "main_py_rtc_no_marker_scan", "markers": {}}, "cursor": self.get_cursor_state(rect), "last_capture_saved_path": str(getattr(self.capture_worker, "last_capture_saved_path", "") or ""), "log_lines": self.state_log_lines[-120:], } def get_cursor_state(self, rect: dict | None = None) -> dict: try: x, y = win32api.GetCursorPos() rect = rect if isinstance(rect, dict) else dict(getattr(self.capture_worker, "last_capture_rect", {}) or {}) left = int(rect.get("left", 0) or 0) top = int(rect.get("top", 0) or 0) width = int(rect.get("width", 0) or 0) height = int(rect.get("height", 0) or 0) frame = getattr(self.capture_worker, "last_frame_size", {}) or {} fw = int(frame.get("width", 0) or 0) fh = int(frame.get("height", 0) or 0) frame_x = None frame_y = None inside_capture = False if width > 0 and height > 0 and fw > 0 and fh > 0: inside_capture = left <= int(x) < left + width and top <= int(y) < top + height frame_x = round((int(x) - left) * fw / width, 1) frame_y = round((int(y) - top) * fh / height, 1) return { "ok": True, "x": int(x), "y": int(y), "frame_x": frame_x, "frame_y": frame_y, "inside_capture": bool(inside_capture), "capture_rect": {"left": left, "top": top, "width": width, "height": height}, "frame_size": {"width": fw, "height": fh}, } except Exception as exc: return {"ok": False, "error": str(exc)} def save_current_capture(self) -> dict: try: target = self.source_manager.get_target() if not target: return {"ok": False, "error": "capture_target_missing", "message": "캡처 대상이 없습니다"} out_dir = self.base_dir / "captures" out_dir.mkdir(parents=True, exist_ok=True) stamp = time.strftime("%Y%m%d_%H%M%S") path = out_dir / f"rtc_capture_{stamp}.jpg" with mss.mss() as sct: shot = sct.grab(target.monitor_dict) pil = Image.frombytes("RGB", shot.size, shot.bgra, "raw", "BGRX") pil.save(path, format="JPEG", quality=int(self.capture_worker.jpeg_quality), optimize=False) self.capture_worker.last_capture_saved_path = str(path) self.append_state_log("RTC", f"캡처 저장 {path}") self.write_state("capture_saved") return {"ok": True, "path": str(path), "message": f"캡처 저장: {path}", "mode": target.mode, "rect": dict(target.monitor_dict)} except Exception as exc: self.append_state_log("RTC", f"캡처 저장 실패: {exc}", "error") return {"ok": False, "error": str(exc), "message": f"캡처 저장 실패: {exc}"} def write_state(self, reason: str = "") -> None: if not getattr(self, "state_file", None): return try: data = self.build_state() if reason: data["state_reason"] = str(reason) self.state_file.parent.mkdir(parents=True, exist_ok=True) self.state_file.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") except Exception as exc: try: self.logger.debug("state write failed: %s", exc) except Exception: pass def periodic_state_write(self) -> None: # v126: no per-second disk writes. Viewer state is pushed through WebSocket; # state_file is written only on lifecycle/URL-changing events for WS bootstrap compatibility. try: self.notify_viewers({"type": "rtc_state", "state": self.build_state()}) finally: try: self.root.after(1000, self.periodic_state_write) except Exception: pass def apply_startup_options(self) -> None: """WS에서 실행했을 때 원격 데몬과 viewer를 자동으로 연결한다.""" try: source = str(getattr(self.args, "source", "") or "").lower() if source == "display": self.display_var.set(max(0, int(getattr(self.args, "display_index", self.display_var.get()) or 0))) self.on_display_selected() elif source == "window": self.source_var.set("window") self.on_source_changed() elif source == "desktop": self.source_var.set("desktop") self.on_source_changed() else: self.on_source_changed() control_arg = str(getattr(self.args, "control", "") or "").lower() if control_arg in ("1", "true", "yes", "on", "켜기"): self.control_var.set(True) self.on_control_changed() elif control_arg in ("0", "false", "no", "off", "끄기"): self.control_var.set(False) self.on_control_changed() if bool(getattr(self.args, "auto_start", False)): self.start_preview() self.append_state_log("RTC", "host/main.py auto-start preview") self.write_state("auto_start") if str(getattr(self.args, "viewer", "no") or "no").lower() == "open": webbrowser.open(self.get_local_viewer_url()) self.append_state_log("RTC", "viewer opened from host/main.py") self.write_state("viewer_open") except Exception as exc: try: self.status_var.set(f"WS 자동 RTC 시작 실패: {exc}") self.append_state_log("RTC", f"auto-start failed: {exc}", "error") self.write_state("auto_start_failed") except Exception: pass @staticmethod def get_selected_hwnd() -> int | None: with hwnd_lock: return selected_hwnd def is_bound_all(self) -> bool: return self.args.host in ("0.0.0.0", "", None) def get_http_url_host_for_local_open(self) -> str: return "127.0.0.1" if self.is_bound_all() else self.args.host def get_lan_viewer_url(self) -> str: http_host = self.lan_ip if self.is_bound_all() else self.args.host ws_host = self.lan_ip if self.is_bound_all() else self.args.host return make_viewer_url(http_host, self.http_server.port, ws_host, self.ws_server.port, self.token) def get_local_viewer_url(self) -> str: host = self.get_http_url_host_for_local_open() ws_host = host return make_viewer_url(host, self.http_server.port, ws_host, self.ws_server.port, self.token) def notify_viewers(self, payload: dict): try: self.ws_server.push_text(json.dumps(payload, ensure_ascii=False)) except Exception as exc: self.logger.debug("viewer notify failed: %s", exc) def handle_ws_text_message(self, message: str): """viewer.html에서 들어오는 JSON 메시지를 기능별로 라우팅한다.""" # host option 변경은 별도 처리한다. 공유 URL 사용자에게 설정 변경 권한을 주는 기능이므로 # Host 쪽 [원격 제어 허용]이 켜져 있을 때만 허용한다. try: data = json.loads(message) except Exception: data = None if isinstance(data, dict) and data.get("type") == MSG_HOST_OPTION_SET: return self.handle_host_option_message(data) if isinstance(data, dict) and data.get("type") == "rtc_chat": return self.handle_rtc_chat_message(data) result = self.clipboard_sync.handle_text_message(message) if result and getattr(result, "message", "") != "ignored": return result return self.remote_control.handle_text_message(message) def handle_rtc_chat_message(self, data: dict): from yjm_remote_control import ControlResult text = str(data.get("text") or data.get("cmd") or "").strip() try: if text in ("?", "/help", "help"): msg = "명령: /status 현재상태, /capture 캡처정보, /cursor 포인터 위치, /save_capture 수동저장" self.notify_viewers({"type": "rtc_chat_result", "cmd": text, "ok": True, "message": msg}) return ControlResult(True, msg) if text == "/status": st = self.build_state() cap = st.get("capture", {}) if isinstance(st, dict) else {} cur = st.get("cursor", {}) if isinstance(st, dict) else {} msg = f"상태: capture={cap.get('mode')} running={cap.get('running')} frame={cap.get('frame_size')} cursor={cur}" self.notify_viewers({"type": "rtc_chat_result", "cmd": text, "ok": True, "message": msg, "state": st}) return ControlResult(True, msg) if text == "/cursor": cur = self.get_cursor_state() msg = f"포인터: screen=({cur.get('x')},{cur.get('y')}) frame=({cur.get('frame_x')},{cur.get('frame_y')})" if cur.get("ok") else f"포인터 실패: {cur.get('error')}" self.notify_viewers({"type": "rtc_chat_result", "cmd": text, "ok": bool(cur.get("ok")), "message": msg, "cursor": cur}) return ControlResult(bool(cur.get("ok")), msg) if text == "/capture": st = self.build_state() cap = st.get("capture", {}) if isinstance(st, dict) else {} msg = f"캡처정보: mode={cap.get('mode')} rect={cap.get('rect')} frame={cap.get('frame_size')}" self.notify_viewers({"type": "rtc_chat_result", "cmd": text, "ok": True, "message": msg, "capture": cap}) return ControlResult(True, msg) if text == "/save_capture": res = self.save_current_capture() self.notify_viewers({"type": "rtc_chat_result", "cmd": text, "ok": bool(res.get("ok")), "message": res.get("message") or "", "capture": res}) return ControlResult(bool(res.get("ok")), str(res.get("message") or res.get("error") or "capture")) msg = f"알 수 없는 RTC 명령: {text}" self.notify_viewers({"type": "rtc_chat_result", "cmd": text, "ok": False, "message": msg}) return ControlResult(False, msg) except Exception as exc: msg = f"RTC 명령 처리 실패: {exc}" self.notify_viewers({"type": "rtc_chat_result", "cmd": text, "ok": False, "message": msg}) return ControlResult(False, msg) def handle_host_option_message(self, data: dict): from yjm_remote_control import ControlResult if not self.control_var.get(): return ControlResult(False, "host option blocked: remote control disabled") name = str(data.get("name", "")) value = data.get("value") ok = False message = "unknown option" if name == "draw_border": flag = bool(value) self.border_var.set(flag) self.capture_worker.update_options(draw_border=flag) ok = True message = f"빨간 테두리 {'ON' if flag else 'OFF'}" self.status_var.set(message) payload = make_msg(MSG_HOST_OPTION_ACK, name=name, value=value, ok=ok, message=message) self.notify_viewers(payload) return ControlResult(ok, message) @staticmethod def format_display_label(idx: int, m: dict) -> str: primary = " / 주모니터" if m.get("primary") else "" source = f" / {m.get('source')}" if m.get("source") else "" device = f" / {m.get('device')}" if m.get("device") else "" real_index = m.get("index", idx + 1) return ( f"디스플레이 {idx} / mss[{real_index}] - {int(m['width'])}x{int(m['height'])} " f"@ {int(m['left'])},{int(m['top'])}{primary}{source}{device}" ) def refresh_display_combo(self): displays = self.source_manager.refresh_displays() display_names = [self.format_display_label(idx, m) for idx, m in enumerate(displays)] if not display_names: display_names = ["디스플레이 없음"] self.display_combo["values"] = display_names current = self.display_var.get() if displays: current = min(max(0, current), len(displays) - 1) self.display_var.set(current) self.display_combo.current(current) self.source_manager.set_display(current) self.status_var.set(f"디스플레이 {len(displays)}개 감지됨") for idx, m in enumerate(displays): self.logger.info("display %s: %s", idx, self.format_display_label(idx, m)) else: self.display_var.set(0) self.display_combo.current(0) self.status_var.set("디스플레이를 감지하지 못했습니다") def open_monitor_lab(self): """별도 모니터 진단 도구를 실행한다.""" try: script = Path(__file__).resolve().parent / "yjm_monitor_lab.py" subprocess.Popen([sys.executable, str(script)], cwd=str(script.parent)) self.status_var.set("모니터 진단 도구를 실행했습니다") except Exception as exc: self.status_var.set(f"모니터 진단 실행 실패: {exc}") self.logger.error("monitor lab open error: %s", exc) def setup_ui(self): frame = ttk.Frame(self.root, padding=12) frame.pack(fill=tk.BOTH, expand=True) ttk.Label(frame, text="원격 화면 대상", font=("Segoe UI", 10, "bold")).pack(anchor="w") src_box = ttk.LabelFrame(frame, text="Remote Desktop Source") src_box.pack(fill=tk.X, pady=(4, 10)) ttk.Radiobutton(src_box, text="1. 전체 바탕화면", value="desktop", variable=self.source_var, command=self.on_source_changed).grid(row=0, column=0, sticky="w", padx=8, pady=5) self.display_combo = ttk.Combobox(src_box, state="readonly", width=66) self.display_combo.bind("<>", self.on_display_selected) ttk.Radiobutton(src_box, text="2. 디스플레이", value="display", variable=self.source_var, command=self.on_source_changed).grid(row=1, column=0, sticky="w", padx=8, pady=5) self.display_combo.grid(row=1, column=1, sticky="we", padx=8, pady=5) diag_btns = ttk.Frame(src_box) diag_btns.grid(row=1, column=2, sticky="w", padx=8, pady=5) ttk.Button(diag_btns, text="디스플레이 새로고침", command=self.refresh_display_combo).pack(side=tk.LEFT, padx=(0, 6)) ttk.Button(diag_btns, text="모니터 진단", command=self.open_monitor_lab).pack(side=tk.LEFT) advanced = ttk.Frame(src_box) advanced.grid(row=2, column=0, columnspan=3, sticky="we", padx=8, pady=(2, 6)) ttk.Radiobutton( advanced, text="고급/보기용: 선택 창(HWND) 캡처 - 원격제어 메인 용도는 아님", value="window", variable=self.source_var, command=self.on_source_changed, ).pack(side=tk.LEFT) src_box.columnconfigure(1, weight=1) self.refresh_display_combo() ttk.Label(frame, textvariable=self.hwnd_var).pack(anchor="w", pady=(0, 6)) ttk.Checkbutton( frame, text="선택 창 고정(고급 창 캡처용) - 원격 데스크톱 사용 중에는 전체/디스플레이 권장", variable=self.lock_var, command=self.on_lock_changed, ).pack(anchor="w", pady=(0, 4)) ttk.Checkbutton( frame, text="원격 제어 허용 - viewer에서 클릭/우클릭/더블클릭/휠을 실제 PC 좌표로 실행", variable=self.control_var, command=self.on_control_changed, ).pack(anchor="w", pady=(0, 4)) file_row = ttk.Frame(frame) file_row.pack(fill=tk.X, pady=(0, 4)) ttk.Checkbutton( file_row, text="파일전송 허용 - viewer에서 보낸 파일을 recv_files 폴더에 저장", variable=self.file_transfer_var, command=self.on_file_transfer_changed, ).pack(side=tk.LEFT) ttk.Button(file_row, text="수신 폴더 열기", command=self.open_recv_dir).pack(side=tk.LEFT, padx=(10, 0)) clip_row = ttk.Frame(frame) clip_row.pack(fill=tk.X, pady=(0, 10)) ttk.Checkbutton( clip_row, text="클립보드 수동 전송 허용 - 자동 동기화 없음", variable=self.clipboard_var, command=self.on_clipboard_changed, ).pack(side=tk.LEFT) ttk.Button(clip_row, text="현재 host 클립보드 viewer로 전송", command=self.send_host_clipboard_now).pack(side=tk.LEFT, padx=(10, 0)) opt_box = ttk.LabelFrame(frame, text="전송 옵션") opt_box.pack(fill=tk.X, pady=(0, 10)) ttk.Label(opt_box, text="FPS").grid(row=0, column=0, sticky="w", padx=8, pady=6) ttk.Spinbox(opt_box, from_=1, to=30, textvariable=self.fps_var, width=6).grid(row=0, column=1, sticky="w", padx=6, pady=6) ttk.Label(opt_box, text="JPEG 품질").grid(row=0, column=2, sticky="w", padx=8, pady=6) ttk.Spinbox(opt_box, from_=20, to=95, textvariable=self.quality_var, width=6).grid(row=0, column=3, sticky="w", padx=6, pady=6) ttk.Label(opt_box, text="max_width").grid(row=0, column=4, sticky="w", padx=8, pady=6) ttk.Spinbox(opt_box, from_=0, to=3840, increment=160, textvariable=self.max_width_var, width=8).grid(row=0, column=5, sticky="w", padx=6, pady=6) ttk.Checkbutton(opt_box, text="원격 화면 빨간 테두리", variable=self.border_var).grid(row=0, column=6, sticky="w", padx=10, pady=6) ttk.Button(opt_box, text="옵션 적용", command=self.apply_capture_options).grid(row=0, column=7, sticky="w", padx=8, pady=6) ai_box = ttk.LabelFrame(frame, text="AI 스냅샷 / Google Drive 동기화 폴더") ai_box.pack(fill=tk.X, pady=(0, 10)) ttk.Checkbutton(ai_box, text="변화 있을 때 스냅샷 저장", variable=self.ai_snapshot_var).grid(row=0, column=0, sticky="w", padx=8, pady=5) ttk.Label(ai_box, text="폴더").grid(row=0, column=1, sticky="e", padx=4, pady=5) ttk.Entry(ai_box, textvariable=self.ai_snapshot_folder_var, width=38).grid(row=0, column=2, columnspan=3, sticky="we", padx=4, pady=5) ttk.Button(ai_box, text="폴더 열기", command=self.open_ai_snapshot_folder).grid(row=0, column=5, sticky="w", padx=6, pady=5) ttk.Label(ai_box, text="최근개수").grid(row=1, column=0, sticky="e", padx=4, pady=5) ttk.Spinbox(ai_box, from_=1, to=200, textvariable=self.ai_snapshot_keep_var, width=6).grid(row=1, column=1, sticky="w", padx=4, pady=5) ttk.Label(ai_box, text="확인초").grid(row=1, column=2, sticky="e", padx=4, pady=5) ttk.Spinbox(ai_box, from_=0.2, to=60.0, increment=0.5, textvariable=self.ai_snapshot_interval_var, width=7).grid(row=1, column=3, sticky="w", padx=4, pady=5) ttk.Label(ai_box, text="저장간격초").grid(row=1, column=4, sticky="e", padx=4, pady=5) ttk.Spinbox(ai_box, from_=0.5, to=300.0, increment=0.5, textvariable=self.ai_snapshot_min_interval_var, width=7).grid(row=1, column=5, sticky="w", padx=4, pady=5) ttk.Label(ai_box, text="변화율%").grid(row=2, column=0, sticky="e", padx=4, pady=5) ttk.Spinbox(ai_box, from_=0.1, to=100.0, increment=0.5, textvariable=self.ai_snapshot_change_var, width=7).grid(row=2, column=1, sticky="w", padx=4, pady=5) ttk.Label(ai_box, text="max_width").grid(row=2, column=2, sticky="e", padx=4, pady=5) ttk.Spinbox(ai_box, from_=0, to=3840, increment=160, textvariable=self.ai_snapshot_max_width_var, width=8).grid(row=2, column=3, sticky="w", padx=4, pady=5) ttk.Label(ai_box, text="품질").grid(row=2, column=4, sticky="e", padx=4, pady=5) ttk.Spinbox(ai_box, from_=20, to=95, textvariable=self.ai_snapshot_quality_var, width=7).grid(row=2, column=5, sticky="w", padx=4, pady=5) ttk.Button(ai_box, text="스냅샷 설정 저장+적용", command=self.save_apply_ai_snapshot_settings).grid(row=3, column=0, columnspan=2, sticky="w", padx=8, pady=(2, 8)) ttk.Label(ai_box, text="Google Drive for Desktop 동기화 폴더를 지정하면 Drive 앱이 업로드를 담당합니다. API 업로드는 별도 단계입니다.", foreground="#555555", wraplength=650).grid(row=3, column=2, columnspan=4, sticky="w", padx=4, pady=(2, 8)) ai_box.columnconfigure(2, weight=1) btns = ttk.Frame(frame) btns.pack(fill=tk.X, pady=(0, 10)) ttk.Button(btns, text="서버+캡처 시작", command=self.start_preview).pack(side=tk.LEFT, padx=(0, 6)) ttk.Button(btns, text="캡처 중지", command=self.stop_capture).pack(side=tk.LEFT, padx=(0, 6)) ttk.Button(btns, text="로컬 viewer 열기", command=self.open_local_viewer).pack(side=tk.LEFT, padx=(0, 6)) ttk.Button(btns, text="LAN URL 복사", command=self.copy_lan_url).pack(side=tk.LEFT, padx=(0, 6)) ttk.Button(btns, text=".", width=3, command=self.hide_to_tray).pack(side=tk.RIGHT, padx=(6, 0)) url_box = ttk.LabelFrame(frame, text="주소") url_box.pack(fill=tk.X, pady=(0, 10)) ttk.Label(url_box, text="LAN:").grid(row=0, column=0, sticky="nw", padx=8, pady=5) ttk.Label(url_box, textvariable=self.local_url_var, foreground="#555555", wraplength=640).grid(row=0, column=1, sticky="w", padx=8, pady=5) ttk.Label(url_box, text="공유:").grid(row=1, column=0, sticky="nw", padx=8, pady=5) ttk.Label(url_box, textvariable=self.share_url_var, foreground="#555555", wraplength=640).grid(row=1, column=1, sticky="w", padx=8, pady=5) url_box.columnconfigure(1, weight=1) upnp_btns = ttk.Frame(frame) upnp_btns.pack(fill=tk.X, pady=(0, 10)) ttk.Button(upnp_btns, text="UPnP 열고 공유 URL 복사", command=self.start_upnp_and_copy_share_url).pack(side=tk.LEFT, padx=(0, 6)) ttk.Button(upnp_btns, text="UPnP 닫기", command=self.close_upnp).pack(side=tk.LEFT, padx=(0, 6)) ttk.Label(frame, textvariable=self.status_var).pack(anchor="w", pady=(4, 0)) ttk.Label( frame, text="주의: UPnP 공유는 외부 노출입니다. 테스트가 끝나면 원격 제어 OFF + 파일전송 OFF + 클립보드 OFF + UPnP 닫기를 누르세요.", foreground="#aa0000", wraplength=720, ).pack(anchor="w", pady=(10, 0)) def build_ai_snapshot_settings(self) -> SnapshotSettings: return SnapshotSettings( enabled=bool(self.ai_snapshot_var.get()), folder=str(self.ai_snapshot_folder_var.get() or "snapshots_drive"), keep_count=max(1, int(self.ai_snapshot_keep_var.get())), interval_sec=max(0.2, float(self.ai_snapshot_interval_var.get())), min_save_interval_sec=max(0.5, float(self.ai_snapshot_min_interval_var.get())), change_threshold_percent=max(0.1, float(self.ai_snapshot_change_var.get())), max_width=max(0, int(self.ai_snapshot_max_width_var.get())), jpeg_quality=max(20, min(95, int(self.ai_snapshot_quality_var.get()))), ) def save_apply_ai_snapshot_settings(self): st = self.build_ai_snapshot_settings() self.app_settings.snapshot = st self.snapshot_ring.update_settings(st) self.save_all_settings_quietly() self.status_var.set(f"AI 스냅샷 설정 저장+적용: {'ON' if st.enabled else 'OFF'} / {st.folder} / 최근 {st.keep_count}개") def open_ai_snapshot_folder(self): st = self.build_ai_snapshot_settings() folder = Path(st.folder).expanduser() if not folder.is_absolute(): folder = Path.cwd() / folder try: folder.mkdir(parents=True, exist_ok=True) os.startfile(str(folder)) # type: ignore[attr-defined] except Exception as exc: self.status_var.set(f"AI 스냅샷 폴더 열기 실패: {exc}") def on_source_changed(self): mode = self.source_var.get() if mode == "window": self.source_manager.set_window() self.status_var.set("고급 선택 창 캡처 모드: 창 위에 마우스를 2초 멈추세요") self.schedule_settings_save() elif mode == "desktop": self.source_manager.set_desktop() self.overlay.hide_overlay() self.status_var.set("전체 바탕화면 모드") self.schedule_settings_save() elif mode == "display": self.on_display_selected() self.overlay.hide_overlay() self.status_var.set("디스플레이 모드") self.schedule_settings_save() def on_display_selected(self, event=None): idx = max(0, self.display_combo.current()) self.display_var.set(idx) self.source_manager.set_display(idx) self.source_var.set("display") self.status_var.set(f"디스플레이 {idx} 선택") self.schedule_settings_save() def on_lock_changed(self): hwnd = self.get_selected_hwnd() if self.lock_var.get(): if hwnd and win32gui.IsWindow(hwnd): self.status_var.set("선택 창 고정됨. 다른 창으로 자동 변경되지 않음") else: self.status_var.set("고정할 선택 창이 아직 없음") else: self.status_var.set("선택 창 고정 해제됨") self.schedule_settings_save() def on_control_changed(self): if self.control_var.get(): self.status_var.set("원격 제어 허용됨. 공유 URL 사용자는 viewer에서 제어 ON 후 실제 클릭할 수 있습니다.") self.schedule_settings_save() else: self.status_var.set("원격 제어 차단됨") self.schedule_settings_save() def on_file_transfer_changed(self): if self.file_transfer_var.get(): self.recv_dir.mkdir(parents=True, exist_ok=True) self.status_var.set(f"파일전송 허용됨. 수신 폴더: {self.recv_dir}") self.notify_viewers({"type": "notify", "kind": "file", "state": "enabled", "filename": "파일전송 허용"}) self.schedule_settings_save() else: self.status_var.set("파일전송 차단됨") self.notify_viewers({"type": "notify", "kind": "file", "state": "disabled", "filename": "파일전송 차단"}) self.schedule_settings_save() def on_clipboard_changed(self): enabled = bool(self.clipboard_var.get()) self.clipboard_sync.notify_state(enabled) if enabled: self.status_var.set("클립보드 수동 전송 허용됨. 자동 동기화는 하지 않습니다.") self.schedule_settings_save() else: self.status_var.set("클립보드 수동 전송 차단됨") self.schedule_settings_save() def send_host_clipboard_now(self): if not self.clipboard_var.get(): self.status_var.set("먼저 클립보드 수동 전송 허용을 체크하세요") return self.clipboard_sync.handle_text_message(json.dumps(make_msg(MSG_CLIPBOARD_GET), ensure_ascii=False)) self.status_var.set("현재 host 클립보드를 viewer로 전송 요청했습니다") def open_recv_dir(self): try: self.recv_dir.mkdir(parents=True, exist_ok=True) os.startfile(str(self.recv_dir)) # type: ignore[attr-defined] except Exception as exc: self.status_var.set(f"수신 폴더 열기 실패: {exc}") def on_hwnd_change(self, hwnd, title): if hwnd and win32gui.IsWindow(hwnd): self.hwnd_var.set(f"선택된 창: {hwnd} / {title[:80]}") else: self.hwnd_var.set("선택된 창: 없음") def apply_capture_options(self): self.capture_worker.update_options( fps=self.fps_var.get(), jpeg_quality=self.quality_var.get(), max_width=self.max_width_var.get(), draw_border=self.border_var.get(), ) self.status_var.set( f"옵션 적용: fps={self.capture_worker.fps}, quality={self.capture_worker.jpeg_quality}, max_width={self.capture_worker.max_width}, border={self.capture_worker.draw_border}" ) self.schedule_settings_save() def refresh_local_url_label(self): self.local_url_var.set(self.get_lan_viewer_url()) def start_preview(self): self.apply_capture_options() self.http_server.start() self.ws_server.start() self.ws_server.ready_event.wait(timeout=1.5) self.capture_worker.start() self.refresh_local_url_label() self.status_var.set("실행 중: 원격 화면 viewer에서 확인") self.append_state_log("RTC", "preview started") self.write_state("preview_started") def stop_capture(self): self.capture_worker.stop() self.status_var.set("캡처 중지됨. HTTP/WS 서버는 유지됨") self.append_state_log("RTC", "capture stopped") self.write_state("capture_stopped") def open_local_viewer(self): self.start_preview() webbrowser.open(self.get_local_viewer_url()) self.append_state_log("RTC", "local viewer opened") self.write_state("viewer_opened") def copy_text(self, text: str): self.root.clipboard_clear() self.root.clipboard_append(text) self.root.update() def copy_lan_url(self): self.start_preview() url = self.get_lan_viewer_url() self.copy_text(url) self.local_url_var.set(url) self.status_var.set("LAN viewer URL을 클립보드에 복사했습니다") self.append_state_log("RTC", "LAN viewer URL copied") self.write_state("url_copied") def start_upnp_and_copy_share_url(self): self.start_preview() self.status_var.set("UPnP 실행 중...") threading.Thread(target=self._upnp_worker, daemon=True).start() def _upnp_worker(self): result = self.upnp.open_ports( lan_ip=self.lan_ip, http_port=self.http_server.port, ws_port=self.ws_server.port, public_ip_url=self.args.public_ip_url, ) if result.ok: public_host = result.echo_public_ip or result.router_external_ip if not public_host: public_host = self.lan_ip share_url = make_viewer_url( public_host, result.http_external_port, public_host, result.ws_external_port, self.token, ) self.root.after(0, lambda: self._finish_upnp_ok(result.message, share_url)) else: self.root.after(0, lambda: self._finish_upnp_fail(result.message)) def _finish_upnp_ok(self, message: str, share_url: str): self.share_url_var.set(share_url) self.copy_text(share_url) self.status_var.set(f"{message} / 공유 URL 복사 완료") def _finish_upnp_fail(self, message: str): self.share_url_var.set("UPnP 실패") self.status_var.set(message) def close_upnp(self): msg = self.upnp.close_ports() self.status_var.set(msg) def hide_to_tray_on_startup(self): """실행 직후 RTC 메인창을 Tray에 숨긴다. Tray 생성 실패 시 화면을 복구한다.""" try: if self.tray is None or not self.tray.show(): self.tray_hidden = False self.root.deiconify() self.root.lift() self.append_state_log("RTC", "[TRAY-01][체크실패] 시작 Tray 생성 실패 / 메인창 표시", "error") return self.tray_hidden = True self.root.withdraw() self.append_state_log("RTC", "[TRAY-01][체크완료] 시작 즉시 Tray 숨김") except Exception as exc: self.tray_hidden = False try: self.root.deiconify() self.root.lift() except Exception: pass self.append_state_log("RTC", f"[TRAY-01][체크실패] 시작 Tray 숨김 실패: {exc}", "error") def hide_to_tray(self): """RTC 메인창을 시스템 Tray로 숨긴다.""" try: if self.tray is None or not self.tray.show(): self.status_var.set("Tray 아이콘 생성 실패 - 화면을 유지합니다") self.append_state_log("RTC", "[TRAY-02][체크실패] Tray 아이콘 생성 실패 / 화면 유지", "error") return self.save_all_settings_quietly() self.tray_hidden = True self.root.withdraw() self.append_state_log("RTC", "[TRAY-02][체크완료] 메인창 숨김 / Tray 이동") except Exception as exc: self.append_state_log("RTC", f"[TRAY-02][체크실패] Tray 이동 실패: {exc}", "error") def restore_from_tray(self): """Tray 아이콘 클릭 시 RTC 메인창을 복원한다.""" try: if self.tray is not None: self.tray.hide() self.tray_hidden = False self.root.deiconify() self.root.lift() try: self.root.attributes("-topmost", True) self.root.after(250, lambda: self.root.attributes("-topmost", False)) except Exception: pass self.root.update_idletasks() self.append_state_log("RTC", "[TRAY-03][체크완료] Tray에서 메인창 복원") except Exception as exc: self.append_state_log("RTC", f"[TRAY-03][체크실패] Tray 복원 실패: {exc}", "error") def poll_tray_restore(self): try: if self.tray is not None and self.tray.poll_restore(): self.restore_from_tray() finally: try: if self.root.winfo_exists(): self.root.after(200, self.poll_tray_restore) except Exception: pass def on_close(self): self.save_all_settings_quietly() try: if self.tray is not None: self.tray.stop() except Exception: pass self.capture_worker.stop() try: self.snapshot_ring.stop() except Exception: pass try: self.clipboard_sync.stop() self.remote_activity_overlay.destroy() except Exception: pass self.overlay.stop() self.upnp.close_ports() self.http_server.stop() self.append_state_log("RTC", "main window closed") self.write_state("closed") self.root.destroy() sys.exit(0) def parse_args(argv=None): parser = argparse.ArgumentParser(description="yjm_win2rtc Windows Remote Desktop Bridge") parser.add_argument("-v", "--verbose", action="store_true", help="상세 로그") parser.add_argument("--host", default="0.0.0.0", help="서버 바인딩 주소. 외부/LAN 공유는 0.0.0.0 권장") parser.add_argument("--http-port", type=int, default=8790, help="viewer.html HTTP 포트") parser.add_argument("--ws-port", type=int, default=8791, help="JPEG frame WebSocket 포트") parser.add_argument("--fps", type=int, default=12, help="캡처 FPS") parser.add_argument("--quality", type=int, default=70, help="JPEG 품질 1~95") parser.add_argument("--max-width", type=int, default=1280, help="전송 전 최대 가로폭. 0이면 원본 크기") parser.add_argument("--no-auto-port", action="store_true", help="포트 사용 중일 때 자동 다음 포트 탐색을 끔") parser.add_argument("--no-token", dest="token", action="store_false", help="viewer URL 토큰 사용 안 함") parser.add_argument("--border", action=argparse.BooleanOptionalAction, default=True, help="전송 프레임 자체에 빨간 테두리 표시") parser.add_argument("--public-ip-url", default="http://i.yjm.kr/ip/", help="공인 IP 확인 URL") parser.add_argument("--state-file", default="", help="WS가 읽을 RTC 상태 JSON 파일 경로") parser.add_argument("--auto-start", action="store_true", help="실행 직후 HTTP/WS/capture preview를 자동 시작") parser.add_argument("--viewer", choices=["manual", "open", "no"], default="manual", help="auto-start 후 viewer 자동 열기 여부") parser.add_argument("--control", choices=["on", "off", ""], default="", help="원격 제어 허용 초기값. 생략하면 저장값 사용") parser.add_argument("--source", choices=["desktop", "display", "window", ""], default="", help="자동 시작 캡처 대상. 생략하면 저장값 사용") parser.add_argument("--display-index", type=int, default=0, help="source=display일 때 UI 기준 디스플레이 번호") parser.add_argument("--dump-displays", action="store_true", help="디스플레이 감지 결과만 출력하고 종료") return parser.parse_args(argv) def main(argv=None): args = parse_args(argv) if args.dump_displays: for idx, m in enumerate(CaptureSourceManager.list_displays()): print(Win2RtcApp.format_display_label(idx, m)) return 0 if serve is None: print("[ERROR] websockets 패키지를 확인하세요:", _WEBSOCKETS_IMPORT_ERROR) print("설치 예: python -m pip install -r requirements.txt") return 2 root = tk.Tk() root.withdraw() Win2RtcApp(root, args) root.mainloop() return 0 if __name__ == "__main__": raise SystemExit(main())