5c11ee2fde
Adds next-sentence suggestions while typing, weekly input insights and a personal phrase memory to the desktop app, and fixes custom instructions so they process the text instead of inserting the instruction's own wording. Local model requests are now bounded and individually cancellable. Bumps the product version to 1.5.0 (Android/iOS build 1050000), refreshes the landing and web download links, and records the new INPUT feature rows and the open verification gaps in the infrastructure map.
428 lines
16 KiB
Python
428 lines
16 KiB
Python
"""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
|