# -*- coding: utf-8 -*- """ 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 tkinter as tk from tkinter import ttk import mss from PIL import Image, ImageDraw import win32gui 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, 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() 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", "")) data = getattr(result, "data", None) if data: try: await websocket.send(json.dumps(data, ensure_ascii=False)) except Exception as exc: self.logger.debug("control ack send failed: %s", exc) 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() 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 = "" 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 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) 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("yjm_win2rtc - Windows Remote Desktop Bridge") self.root.geometry("940x850") self.token = secrets.token_urlsafe(12) if args.token else "" self.upnp = UpnpPortMapper(self.logger) self.lan_ip = UpnpPortMapper.get_lan_ip() base_dir = Path(__file__).resolve().parent self.base_dir = base_dir self.config_path = base_dir / "yjm_win2rtc_config.json" self.app_settings = load_settings(self.config_path) self.viewer_dir = base_dir / "web" self.source_manager = CaptureSourceManager(self.get_selected_hwnd) self.control_var = tk.BooleanVar(value=False) self.remote_activity_overlay = RemoteActivityOverlay(self.root, hold_ms=1300) self.file_transfer_var = tk.BooleanVar(value=False) self.clipboard_var = tk.BooleanVar(value=False) 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): return 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, ) 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=args.fps, jpeg_quality=args.quality, max_width=args.max_width, draw_border=args.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, ) self.hwnd_var = tk.StringVar(value="선택된 창: 없음") self.source_var = tk.StringVar(value="desktop") self.display_var = tk.IntVar(value=0) self.lock_var = tk.BooleanVar(value=False) self.border_var = tk.BooleanVar(value=args.border) self.fps_var = tk.IntVar(value=args.fps) self.quality_var = tk.IntVar(value=args.quality) self.max_width_var = tk.IntVar(value=args.max_width) 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.refresh_local_url_label() @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) 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_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)) 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 save_settings(self.config_path, self.app_settings) self.snapshot_ring.update_settings(st) 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초 멈추세요") elif mode == "desktop": self.source_manager.set_desktop() self.overlay.hide_overlay() self.status_var.set("전체 바탕화면 모드") elif mode == "display": self.on_display_selected() self.overlay.hide_overlay() self.status_var.set("디스플레이 모드") 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} 선택") 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("선택 창 고정 해제됨") def on_control_changed(self): if self.control_var.get(): self.status_var.set("원격 제어 허용됨. 공유 URL 사용자는 viewer에서 제어 ON 후 실제 클릭할 수 있습니다.") else: self.status_var.set("원격 제어 차단됨") 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": "파일전송 허용"}) else: self.status_var.set("파일전송 차단됨") self.notify_viewers({"type": "notify", "kind": "file", "state": "disabled", "filename": "파일전송 차단"}) def on_clipboard_changed(self): enabled = bool(self.clipboard_var.get()) self.clipboard_sync.notify_state(enabled) if enabled: self.status_var.set("클립보드 수동 전송 허용됨. 자동 동기화는 하지 않습니다.") else: self.status_var.set("클립보드 수동 전송 차단됨") 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}" ) 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에서 확인") def stop_capture(self): self.capture_worker.stop() self.status_var.set("캡처 중지됨. HTTP/WS 서버는 유지됨") def open_local_viewer(self): self.start_preview() webbrowser.open(self.get_local_viewer_url()) 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을 클립보드에 복사했습니다") 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 on_close(self): 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.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=8000, help="viewer.html HTTP 포트") parser.add_argument("--ws-port", type=int, default=8765, 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("--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() Win2RtcApp(root, args) root.mainloop() return 0 if __name__ == "__main__": raise SystemExit(main())