# -*- coding: utf-8 -*- """yjm_clipboard_sync.py 텍스트 클립보드 동기화 보조 모듈. 현재 단계(v2.9): - 자동 양방향 동기화는 하지 않는다. - viewer가 명시적으로 요청하면 host PC 클립보드 텍스트를 viewer로 보낸다. - viewer에서 보낸 텍스트를 host PC 클립보드에 넣는다. - 이미지/파일 클립보드는 아직 다루지 않는다. - 브라우저 보안 정책 때문에 viewer 쪽 자동 클립보드 쓰기는 실패할 수 있다. 실패하면 viewer.html에서 복사 버튼/텍스트 상자로 수동 복사하도록 한다. """ from __future__ import annotations import json import time from dataclasses import dataclass from typing import Callable, Optional from host.clipboard import preview_text, read_text, write_text from host.protocol import ( LEGACY_MSG_CLIPBOARD, MSG_CLIPBOARD_DATA, MSG_CLIPBOARD_GET, MSG_CLIPBOARD_SET, MSG_ERROR, MSG_LOG, make_msg, ) try: import tkinter as tk except Exception: # pragma: no cover tk = None NotifyCallback = Callable[[dict], None] StatusCallback = Callable[[str], None] EnabledCallback = Callable[[], bool] @dataclass class ClipboardResult: ok: bool message: str = "" class ClipboardSyncManager: """Tk clipboard 기반 텍스트 동기화 관리자.""" def __init__( self, root, *, is_enabled_callback: Optional[EnabledCallback] = None, notify_viewers_callback: Optional[NotifyCallback] = None, status_callback: Optional[StatusCallback] = None, logger=None, poll_ms: int = 900, max_chars: int = 200_000, auto_sync: bool = False, ): self.root = root self.is_enabled_callback = is_enabled_callback or (lambda: False) self.notify_viewers_callback = notify_viewers_callback self.status_callback = status_callback self.logger = logger self.poll_ms = max(300, int(poll_ms)) self.max_chars = max(1000, int(max_chars)) self.auto_sync = bool(auto_sync) self._last_text: str | None = None self._last_set_from_remote: str | None = None self._polling = False self._last_notice_at = 0.0 def is_enabled(self) -> bool: try: return bool(self.is_enabled_callback()) except Exception: return False def start(self) -> None: # v2.9: 자동 클립보드 감시는 기본 OFF. # 혼란을 막기 위해 명시적 요청/전송 버튼으로만 동작시킨다. if not self.auto_sync: self._polling = False return if self._polling: return self._polling = True self.root.after(self.poll_ms, self._poll_once) def stop(self) -> None: self._polling = False def notify_state(self, enabled: bool) -> None: state = "enabled" if enabled else "disabled" self._notify(make_msg( MSG_LOG, source="clipboard", action="state", enabled=bool(enabled), state=state, message="클립보드 수동 전송 허용" if enabled else "클립보드 수동 전송 차단", )) if self.status_callback: try: self.status_callback("원격: 클립보드 수동 전송\n허용됨" if enabled else "원격: 클립보드 수동 전송\n차단됨") except Exception: pass def handle_text_message(self, message: str) -> ClipboardResult: """WebSocket text JSON 중 clipboard 관련 메시지만 처리한다. v3 표준 타입: - clipboard_set: viewer -> host 텍스트 설정 - clipboard_get: viewer -> host 현재 클립보드 요청 v2 호환 타입: - {type:"clipboard", action:"viewer_to_host"} - {type:"clipboard", action:"request_host_clipboard"} """ try: data = json.loads(message) except Exception: return ClipboardResult(False, "ignored") if not isinstance(data, dict): return ClipboardResult(False, "ignored") msg_type = data.get("type") or "" action = data.get("action") or "" is_set = msg_type == MSG_CLIPBOARD_SET or (msg_type == LEGACY_MSG_CLIPBOARD and action in ("viewer_to_host", "set_host")) is_get = msg_type == MSG_CLIPBOARD_GET or (msg_type == LEGACY_MSG_CLIPBOARD and action == "request_host_clipboard") if is_set: if not self.is_enabled(): return ClipboardResult(False, "host clipboard sync disabled") text = data.get("text") if text is None: text = "" if not isinstance(text, str): text = str(text) if len(text) > self.max_chars: text = text[: self.max_chars] self.root.after(0, lambda t=text: self._set_host_clipboard(t)) return ClipboardResult(True, "clipboard set scheduled") if is_get: if not self.is_enabled(): return ClipboardResult(False, "host clipboard sync disabled") self.root.after(0, self._send_current_clipboard) return ClipboardResult(True, "clipboard request scheduled") return ClipboardResult(False, "ignored") def _read_clipboard_text(self) -> str | None: return read_text(self.root, max_chars=self.max_chars) def _set_host_clipboard(self, text: str) -> None: try: written = write_text(text, self.root, max_chars=self.max_chars) self._last_text = text self._last_set_from_remote = text self._notify(make_msg( MSG_LOG, source="clipboard", action="ack_host_set", message=f"host 클립보드 적용 완료 ({written}자)", length=written, preview=self._preview(text), )) if self.status_callback: self.status_callback(f"원격: 클립보드 수신\n{self._preview(text)}") if self.logger: self.logger.info("clipboard received from viewer: %d chars", written) except Exception as exc: self._notify(make_msg(MSG_ERROR, source="clipboard", message=str(exc))) if self.logger: self.logger.error("clipboard set error: %s", exc) def _send_current_clipboard(self) -> None: text = self._read_clipboard_text() if text is None: self._notify(make_msg(MSG_ERROR, source="clipboard", message="host clipboard has no text")) return self._last_text = text self._notify_host_text(text, reason="request") def _poll_once(self) -> None: if not self._polling: return try: if self.is_enabled(): text = self._read_clipboard_text() if text is not None and text != self._last_text: self._last_text = text self._notify_host_text(text, reason="changed") except Exception as exc: if self.logger: self.logger.debug("clipboard poll error: %s", exc) finally: if self._polling: self.root.after(self.poll_ms, self._poll_once) def _notify_host_text(self, text: str, *, reason: str) -> None: self._notify(make_msg( MSG_CLIPBOARD_DATA, reason=reason, text=text, length=len(text), preview=self._preview(text), )) now = time.time() if self.status_callback and now - self._last_notice_at > 1.2: self._last_notice_at = now try: self.status_callback(f"원격: 클립보드 전송\n{self._preview(text)}") except Exception: pass def _notify(self, payload: dict) -> None: if self.notify_viewers_callback: try: self.notify_viewers_callback(payload) except Exception: pass @staticmethod def _preview(text: str, limit: int = 42) -> str: return preview_text(text, limit=limit)