"""UIA(UI Automation) 포커스 스냅샷 브리지. 목적: 사용자가 지금 어느 입력창에 무엇을 치고 있는지, 어렛이 어디 있는지를 읽어 "다음 문장 제안" 과 "타이핑 학습" 에 넘긴다. 설계 근거 (조사 기반, 뇌피셜 아님): - Win32 `GetGUIThreadInfo` 의 `rcCaret` 은 Chromium/Electron/VS Code 처럼 커서를 직접 그리는 앱에서 아무것도 돌려주지 않는다 (AutoHotkey CaretGetPos 문서, MS Learn "GetGUIThreadInfo" 의 rcCaret 주의 문구). 그래서 UIA 를 정본으로 쓴다. - 케어렛 rect 는 `TextPattern.GetSelection()` (축소된 caret range) → `GetBoundingRectangles()` 가 이식성 있는 경로다. Chromium 은 `ITextProvider` 는 구현하지만 `ITextPattern2::GetCaretRange` 는 구현하지 않는다 (Chromium 소스). - 비밀번호 필드는 `UIA_IsPasswordPropertyId(30019)` 로 반드시 먼저 검사한다. (MS Learn "IUIAutomationElement::get_CurrentIsPassword") - 트리 전체 순회는 금지한다. Chrome/VS Code 에서 UIA 트리 크가 10~30초 걸린다는 실측 보고가 있다 (PowerToys #46385). 포커스 요소 + 패턴 2~3회 호출만 한다. - IME(한/일) 조합 중에는 스트가 커밋되지 않고 케어렛도 부정확하다. 조합 중이면 호출자는 제안을 억제해야 하므로 `is_composing` 을 함께 돌려준다. 모든 호출은 전용 단일 스레드에서 실행한다 (comtypes COM 아파트먼트 친화성). """ from __future__ import annotations import logging import threading import time from concurrent.futures import Future, ThreadPoolExecutor, TimeoutError as FutureTimeout from typing import Any, Optional logger = logging.getLogger("sidecar.uia") # 텍스트를 그대로 들고 오는 상한. 입력창 전체 복사는 위험하므로 잘라 쓴다. MAX_TEXT_CHARS = 4000 # UIA 호출 예산. 넘기면 이번 주기는 실패로 처리하고 다음 기회를 기다린다. DEFAULT_TIMEOUT_MS = 1500 _uia_available: Optional[bool] = None _executor: Optional[ThreadPoolExecutor] = None _executor_lock = threading.Lock() _inflight = threading.Lock() def _uia_module() -> Any: """uiautomation 모듈을 지연 임포트한다 (미설치 환경에서 임포트 자체를 막지 않기 위해).""" global _uia_available try: import uiautomation # noqa: PLC0415 _uia_available = True return uiautomation except Exception as exc: # pragma: no cover - 설치 환경 의존 _uia_available = False logger.warning("uiautomation 미설치 — UIA 컨텍스트 비활성 (%s)", exc) raise def is_available() -> bool: """uiautomation 사용 가능 여부 (실제 임포트 시도 결과).""" global _uia_available if _uia_available is not None: return _uia_available try: _uia_module() except Exception: return False return bool(_uia_available) def _worker_init() -> None: """UIA 전용 스레드의 COM 초기화.""" try: import comtypes # noqa: PLC0415 comtypes.CoInitializeEx(comtypes.COINIT_MULTITHREADED) except Exception: # 이미 초기화됐거나 comtypes 가 없더라도 uiautomation 자체 초기화에 맡긴다. try: import comtypes # noqa: PLC0415 comtypes.CoInitialize() except Exception: pass def _get_executor() -> ThreadPoolExecutor: global _executor with _executor_lock: if _executor is None: _executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="uia") _executor.submit(_worker_init).result(timeout=5) return _executor def _rect_to_dict(rect: Any) -> Optional[dict]: if rect is None: return None try: left = int(rect.left) top = int(rect.top) right = int(rect.right) bottom = int(rect.bottom) except Exception: return None width = right - left height = bottom - top if width < 0 or height < 0: return None return {"x": left, "y": top, "width": width, "height": height} def _first_rect(rects: Any) -> Optional[dict]: """GetBoundingRectangles 결과에서 케어 rect 를 는다 (첫 사각형 우선).""" try: for rect in rects or []: converted = _rect_to_dict(rect) if converted is not None: return converted except Exception: return None return None def _read_text(auto: Any, control: Any) -> tuple[str, str, Optional[Any]]: """(텍스트, 출처, value_pattern) 을 돌려준다. 지원 패턴이 없으면 ('', 'none', None).""" value_pattern = None try: value_pattern = control.GetValuePattern() except Exception: value_pattern = None if value_pattern is not None: try: text = value_pattern.Value or "" return (str(text)[:MAX_TEXT_CHARS], "value", value_pattern) except Exception: pass try: text_pattern = control.GetTextPattern() except Exception: text_pattern = None if text_pattern is not None: try: text = text_pattern.DocumentRange.GetText(-1) or "" return (str(text)[:MAX_TEXT_CHARS], "text", value_pattern) except Exception: # Chromium/CEF 입력창에서 COMError 가 나는 사례가 보고돼 있다 (issue #135/#138). pass try: legacy = control.GetLegacyIAccessiblePattern() if legacy is not None: text = legacy.Value or "" return (str(text)[:MAX_TEXT_CHARS], "legacy", value_pattern) except Exception: pass return ("", "none", value_pattern) def _caret_rect_and_offset(auto: Any, control: Any) -> tuple[Optional[dict], Optional[int], bool, bool]: """케어렛 rect / 문서 시작 기준 오프셋 / 조합 진행 여부 / 비축소 선택 여부.""" caret_rect: Optional[dict] = None caret_offset: Optional[int] = None composing = False has_selection = False text_pattern = None try: text_pattern = control.GetTextPattern() except Exception: text_pattern = None if text_pattern is not None: try: selection = text_pattern.GetSelection() except Exception: selection = None if selection: for selected_range in selection: try: has_selection = ( selected_range.CompareEndpoints( auto.TextPatternRangeEndpoint.Start, selected_range, auto.TextPatternRangeEndpoint.End, ) != 0 ) except Exception: has_selection = False if has_selection: break caret_range = selection[0] try: caret_rect = _first_rect(caret_range.GetBoundingRectangles()) except Exception: caret_rect = None if caret_rect is None: # 축소된 caret range 는 빈 사각형 배열을 돌려주는 경우가 많다. # 문자 단위로 확장한 뒤 다시 시도한다 (Windows Terminal 레시피). try: expanded = caret_range.Clone() expanded.ExpandToEnclosingUnit(auto.TextUnit.Character) caret_rect = _first_rect(expanded.GetBoundingRectangles()) except Exception: caret_rect = None try: caret_offset = int( caret_range.CompareEndpoints( auto.TextPatternRangeEndpoint.Start, text_pattern.DocumentRange, auto.TextPatternRangeEndpoint.Start, ) ) except Exception: caret_offset = None # IME 조합 여부 — TextEditPattern 이 지원되는 제공자에서만 확인 가능하다. try: pattern_id = getattr(auto.PatternId, "TextEditPattern", None) if pattern_id is not None: text_edit = control.GetPattern(pattern_id) if text_edit is not None: composition = text_edit.GetActiveComposition() if composition is not None: composing_text = composition.GetText(-1) or "" composing = len(str(composing_text)) > 0 except Exception: composing = False return (caret_rect, caret_offset, composing, has_selection) def _wake_chromium_accessibility() -> bool: """Chromium 계열에 "보조기술이 있다" 고 알려 접근성 트리를 열게 한다. Chromium 문서: 브라우저가 EVENT_SYSTEM_ALERT + custom object id 1 을 알리고, 그 id 에 대해 WM_GETOBJECT 를 받으면 보조기술이 실행 중이라고 판단해 AX 트리를 만든다. NVDA 등이 쓰는 방식이며, 이걸 하지 않으면 Chrome/Edge 의 웹 입력창은 LegacyIAccessible 만 노출돼 텍스트를 읽을 수 없다(실측: edit=false src=legacy len=0). SendMessage 대신 SendMessageTimeout 을 쓰는 이유: 대상이 바쁘면 무한 대기한다. """ try: import ctypes # noqa: PLC0415 from ctypes import wintypes # noqa: PLC0415 user32 = ctypes.windll.user32 WM_GETOBJECT = 0x003D OBJID_CLIENT = 0xFFFFFFFC hwnd = user32.GetForegroundWindow() if not hwnd: return False result = ctypes.c_void_p() # 1 = 보조기술 감지용 custom object id user32.SendMessageTimeoutW( wintypes.HWND(hwnd), WM_GETOBJECT, wintypes.WPARAM(0), wintypes.LPARAM(1), 0x0002, # SMTO_ABORTIFHUNG 250, ctypes.byref(result), ) # 클라이언트 영역 객체도 함께 요청해 AX 트리를 확실히 깨운다. user32.SendMessageTimeoutW( wintypes.HWND(hwnd), WM_GETOBJECT, wintypes.WPARAM(0), wintypes.LPARAM(OBJID_CLIENT & 0xFFFFFFFF), 0x0002, 250, ctypes.byref(result), ) return True except Exception as exc: # pragma: no cover - 플랫폼 의존 logger.debug("Chromium 접근성 깨우기 실패: %s", exc) return False def _snapshot() -> dict: """전용 스레드에서 실행되는 실제 UIA 수집.""" auto = _uia_module() # Chromium/Electron 계열은 접근성 트리를 요청받아야 연다 — 먼저 깨운다. _wake_chromium_accessibility() started = time.monotonic() control = auto.GetFocusedControl() if control is None: return {"available": False, "reason": "no-focused-control"} result: dict[str, Any] = { "available": True, "isPassword": False, "isEditable": False, "isComposing": False, "hasSelection": False, "controlType": None, "controlName": None, "className": None, "textSource": "none", "text": "", "caretOffset": None, "caretRect": None, "elementRect": None, "windowTitle": None, "processId": None, } try: result["controlType"] = str(control.ControlTypeName or "") result["controlName"] = str(control.Name or "")[:200] result["className"] = str(control.ClassName or "")[:120] except Exception: pass # 1) 비밀번호 필드는 어떤 읽기도 하지 않는다 (fail-closed). try: is_password = bool(control.IsPassword) except Exception: # 판별 실패를 "안전" 으로 넘기면 안 된다 — 읽기 불가로 처리한다. return {"available": False, "reason": "is-password-unavailable"} result["isPassword"] = is_password if is_password: result["reason"] = "password-field" return result try: result["processId"] = int(control.ProcessId) except Exception: result["processId"] = None try: top = control.GetTopLevelControl() if top is not None: result["windowTitle"] = str(top.Name or "")[:300] result["elementRect"] = _rect_to_dict(top.BoundingRectangle) or result["elementRect"] except Exception: pass try: result["elementRect"] = _rect_to_dict(control.BoundingRectangle) or result["elementRect"] except Exception: pass text, source, value_pattern = _read_text(auto, control) result["textSource"] = source result["text"] = text is_readonly = False try: if value_pattern is not None: is_readonly = bool(value_pattern.IsReadOnly) except Exception: is_readonly = False control_type = result["controlType"] or "" # 터미널(Windows Terminal 등)은 ValuePattern 이 없고 TextPattern 만 있다. # 값 패턴 부재만 보고 "편집 불가" 로 판정하면 파워셸/터미널처럼 실제로 가장 # 많이 타이핑하는 표면에서 제안이 통째로 죽는다(실측: src=text, edit=false). # 텍스트 패턴이 선택/삽입 지점을 제공하면 입력 가능으로 본다. # # TextPattern 이 있으면 입력 가능으로 본다. 처음에는 GetSelection() 이 비어 # 있지 않은지까지 봤는데, Windows Terminal 은 선택이 없으면 0개를 돌려주는 # 탓에 터미널이 계속 "편집 불가" 로 판정됐다(실측 프로브: TextPattern=yes, # GetSelection=0 ranges). 값 패턴조차 없는 표면이라 다른 단서가 없다. text_readable = False try: text_readable = control.GetTextPattern() is not None except Exception: text_readable = False result["isEditable"] = bool( (value_pattern is not None and not is_readonly) or control_type in ("EditControl", "DocumentControl") or text_readable ) caret_rect, caret_offset, composing, has_selection = _caret_rect_and_offset(auto, control) result["caretRect"] = caret_rect result["caretOffset"] = caret_offset result["isComposing"] = composing result["hasSelection"] = has_selection result["capturedAt"] = int(time.time() * 1000) result["elapsedMs"] = int((time.monotonic() - started) * 1000) return result def snapshot_focus(timeout_ms: int = DEFAULT_TIMEOUT_MS) -> dict: """포커스 스냅샷. 실패/타임아웃은 예외 대신 available=False 로 돌려준다.""" if not is_available(): return {"available": False, "reason": "uiautomation-missing"} # 이전 호출이 아직 춰 있으면 새 작업을 쌓지 않고 즉시 포기한다. if not _inflight.acquire(blocking=False): return {"available": False, "reason": "busy"} try: executor = _get_executor() future: Future = executor.submit(_snapshot) try: return future.result(timeout=max(0.2, timeout_ms / 1000)) except FutureTimeout: logger.warning("UIA 스냅샷 타임아웃 (%sms) — 이번 주기 건너뜀", timeout_ms) return {"available": False, "reason": "timeout"} except Exception as exc: # pragma: no cover - 환경 의존 logger.warning("UIA 스냅샷 실패: %s", exc) return {"available": False, "reason": f"error:{type(exc).__name__}"} finally: _inflight.release() def shutdown() -> None: global _executor with _executor_lock: if _executor is not None: _executor.shutdown(wait=False, cancel_futures=True) _executor = None