#!/usr/bin/env python3 import base64 import ctypes import fcntl import json import os import platform import shutil import socket import subprocess import sys import tempfile import threading import time import urllib.error import urllib.parse import urllib.request import uuid import wave BASE_URL = "https://mcko.kittieslabs.tech" CLIENT_TOKEN = "mXo9D2J6spSUGH4tigKayPohRaovnUITEt7YaExgr2U" CLIENT_VERSION = "1779136732" CONNECT_TIMEOUT = 900 MAX_DRAFT_IMAGES = 8 MAX_AUDIO_UPLOAD_BYTES = 25165824 UPDATE_CHECK_INTERVAL = 30 MAX_ATTEMPTS = 4 PLATFORM = platform.system().lower() IS_DARWIN = PLATFORM == "darwin" IS_LINUX = PLATFORM == "linux" if IS_DARWIN: LOG_PATH = os.path.expanduser("~/Library/Logs/ScreenAnswer/client.log") APP_ROOT = os.path.expanduser("~/Library/Application Support/ScreenAnswer") APPLICATIONS_DIR = None AUTOSTART_DIR = None else: STATE_HOME = os.environ.get("XDG_STATE_HOME") or os.path.expanduser("~/.local/state") DATA_HOME = os.environ.get("XDG_DATA_HOME") or os.path.expanduser("~/.local/share") CONFIG_HOME = os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser("~/.config") LOG_PATH = os.path.join(STATE_HOME, "screen-answer", "client.log") APP_ROOT = os.path.join(DATA_HOME, "screen-answer") APPLICATIONS_DIR = os.path.join(DATA_HOME, "applications") AUTOSTART_DIR = os.path.join(CONFIG_HOME, "autostart") BIN_DIR = os.path.join(APP_ROOT, "bin") DRAFT_DIR = os.path.join(APP_ROOT, "draft") DRAFT_STATE_PATH = os.path.join(DRAFT_DIR, "state.json") AUDIO_DIR = os.path.join(APP_ROOT, "audio") DAEMON_LOG_PATH = os.path.join(os.path.dirname(LOG_PATH), "listener.log") LAST_ANSWER_PATH = os.path.join(APP_ROOT, "last-answer.txt") LAST_AUDIO_STATE_PATH = os.path.join(APP_ROOT, "last-audio.json") CLIENT_ID_PATH = os.path.join(APP_ROOT, "client-id") SESSION_ENV_PATH = os.path.join(APP_ROOT, "session-env.json") CAPTURE_MODES = {"full", "select"} LISTENER_LOCK_FILE = "/tmp/.screen_answer_listener.lock" SESSION_ENV_KEYS = [ "DISPLAY", "WAYLAND_DISPLAY", "XDG_RUNTIME_DIR", "DBUS_SESSION_BUS_ADDRESS", "XAUTHORITY", "DESKTOP_SESSION", "XDG_SESSION_TYPE", ] AUDIO_SAMPLE_RATE = 16000 AUDIO_CHANNELS = 1 AUDIO_SAMPLE_WIDTH = 2 AUDIO_CHUNK_FRAMES = 2048 AUDIO_WAV_MIME = "audio/wav" PRESETS = { "default": "", "number": "Return only the final numeric answer if the screenshot expects a number. Otherwise return the shortest exact answer.", "word": "Return one short word unless the screenshot absolutely requires more.", "comma": "If multiple fragments must be combined, return them joined with commas in the needed order.", "exact": "Preserve exact punctuation, case, and spacing when they matter.", } _CLIENT_ID = None _RUN_LOCK = threading.Lock() _DRAFT_LOCK = threading.Lock() _PENDING_LOCK = threading.Lock() _AUDIO_LOCK = threading.Lock() _DRAFT_IMAGES = [] _DRAFT_TEXTS = [] _PENDING_CAPTURE = "" _PENDING_CAPTURE_HINT = "" _PENDING_SEND = False _PENDING_SEND_HINT = "" _AUDIO_RECORDER = None _PULSE_LIB = None _PULSE_SIMPLE_LIB = None def _write_local_log(message): try: os.makedirs(os.path.dirname(LOG_PATH), exist_ok=True) with open(LOG_PATH, "a", encoding="utf-8") as handle: handle.write(time.strftime("%Y-%m-%d %H:%M:%S") + " " + message + "\n") except Exception: pass def get_client_id(): global _CLIENT_ID if _CLIENT_ID: return _CLIENT_ID try: os.makedirs(APP_ROOT, exist_ok=True) if os.path.exists(CLIENT_ID_PATH): with open(CLIENT_ID_PATH, "r", encoding="utf-8") as handle: value = handle.read().strip() if value: _CLIENT_ID = value[:120] return _CLIENT_ID _CLIENT_ID = str(uuid.uuid4()) with open(CLIENT_ID_PATH, "w", encoding="utf-8") as handle: handle.write(_CLIENT_ID + "\n") return _CLIENT_ID except Exception: _CLIENT_ID = socket.gethostname() or "unknown-client" return _CLIENT_ID def post_json(url, payload): raw = json.dumps(payload).encode("utf-8") headers = { "Content-Type": "application/json", "X-Client-Key": CLIENT_TOKEN, "User-Agent": "screen-answer-client/1.0", } return urllib.request.Request(url, data=raw, headers=headers, method="POST") def _post_log_payload(payload): try: req = post_json(BASE_URL + "/v1/client-log", payload) with urllib.request.urlopen(req, timeout=min(3, CONNECT_TIMEOUT)) as response: response.read() except Exception: pass def send_log_to_server(event, message="", **fields): payload = { "client_id": get_client_id(), "client": socket.gethostname(), "event": (event or "log")[:80], "message": (message or "")[:1000], "fields": fields or {}, } try: threading.Thread(target=_post_log_payload, args=(payload,), daemon=True).start() except Exception: pass def log_line(message, event="log", **fields): _write_local_log(message) send_log_to_server(event, message, **fields) def server_error_code(body): try: payload = json.loads(body) except Exception: return "" if not isinstance(payload, dict): return "" return str(payload.get("error") or "").strip() def remove_files(paths): for path in paths or []: try: os.remove(path) except Exception: pass class DraftRejectedError(RuntimeError): def __init__(self, message, clear_draft=False): super().__init__(message) self.clear_draft = clear_draft def queue_draft_capture(capture_mode, hint, reason): global _PENDING_CAPTURE, _PENDING_CAPTURE_HINT with _PENDING_LOCK: _PENDING_CAPTURE = capture_mode or "full" _PENDING_CAPTURE_HINT = hint or _PENDING_CAPTURE_HINT log_line( "capture queued mode=" + (capture_mode or "full") + " reason=" + reason, event="draft_add_queued", kind="image", mode=capture_mode or "full", reason=reason, ) def queue_draft_send(hint, reason): global _PENDING_SEND, _PENDING_SEND_HINT with _PENDING_LOCK: _PENDING_SEND = True if hint: _PENDING_SEND_HINT = hint log_line( "send draft queued reason=" + reason, event="draft_send_queued", reason=reason, ) def maybe_dispatch_queued_work(default_hint): global _PENDING_CAPTURE, _PENDING_CAPTURE_HINT, _PENDING_SEND, _PENDING_SEND_HINT with _PENDING_LOCK: queued_capture = _PENDING_CAPTURE capture_hint = _PENDING_CAPTURE_HINT or default_hint queued_send = _PENDING_SEND send_hint = _PENDING_SEND_HINT or default_hint _PENDING_CAPTURE = "" _PENDING_CAPTURE_HINT = "" _PENDING_SEND = False _PENDING_SEND_HINT = "" if queued_capture: log_line( "queued capture dispatch mode=" + queued_capture, event="draft_add_queue_run", kind="image", mode=queued_capture, ) threading.Thread(target=add_screenshot_to_draft, args=(queued_capture, capture_hint), daemon=True).start() if queued_send: queue_draft_send(send_hint, "after_capture") return if not queued_send: return with _DRAFT_LOCK: has_payload = bool(_DRAFT_IMAGES) or any(str(text).strip() for text in _DRAFT_TEXTS) if not has_payload: log_line("queued send dropped empty draft", event="draft_send_queue_drop") return log_line("queued send dispatch", event="draft_send_queue_run") threading.Thread(target=send_draft, args=(send_hint,), daemon=True).start() def command_exists(name): return shutil.which(name) is not None def capture_session_env(): try: os.makedirs(APP_ROOT, exist_ok=True) payload = {} for key in SESSION_ENV_KEYS: value = os.environ.get(key) if value: payload[key] = value runtime_dir = payload.get("XDG_RUNTIME_DIR") or os.environ.get("XDG_RUNTIME_DIR", "") bus_path = os.path.join(runtime_dir, "bus") if runtime_dir else "" if bus_path and os.path.exists(bus_path) and not payload.get("DBUS_SESSION_BUS_ADDRESS"): payload["DBUS_SESSION_BUS_ADDRESS"] = "unix:path=" + bus_path if payload: with open(SESSION_ENV_PATH, "w", encoding="utf-8") as handle: json.dump(payload, handle, ensure_ascii=False) except Exception: pass def load_session_env(): try: with open(SESSION_ENV_PATH, "r", encoding="utf-8") as handle: data = json.load(handle) if isinstance(data, dict): return {str(key): str(value) for key, value in data.items() if str(value)} except Exception: pass return {} def desktop_env(): env = dict(os.environ) for key, value in load_session_env().items(): env.setdefault(key, value) runtime_dir = env.get("XDG_RUNTIME_DIR", "") bus_path = os.path.join(runtime_dir, "bus") if runtime_dir else "" if bus_path and os.path.exists(bus_path) and not env.get("DBUS_SESSION_BUS_ADDRESS"): env["DBUS_SESSION_BUS_ADDRESS"] = "unix:path=" + bus_path return env def run_clipboard_command(command, timeout=5, input_bytes=None): kwargs = { "check": True, "stdout": subprocess.DEVNULL, "stderr": subprocess.DEVNULL, } if timeout: kwargs["timeout"] = timeout if input_bytes is not None: kwargs["input"] = input_bytes if IS_LINUX: kwargs["env"] = desktop_env() return subprocess.run(command, **kwargs) def read_clipboard_command(command, timeout=5): kwargs = { "stdout": subprocess.PIPE, "stderr": subprocess.PIPE, "timeout": timeout, } if IS_LINUX: kwargs["env"] = desktop_env() return subprocess.run(command, **kwargs) def _pulse_libs(): global _PULSE_LIB, _PULSE_SIMPLE_LIB if _PULSE_LIB is None: _PULSE_LIB = ctypes.CDLL("libpulse.so.0") if _PULSE_SIMPLE_LIB is None: _PULSE_SIMPLE_LIB = ctypes.CDLL("libpulse-simple.so.0") return _PULSE_LIB, _PULSE_SIMPLE_LIB class pa_sample_spec(ctypes.Structure): _fields_ = [ ("format", ctypes.c_int), ("rate", ctypes.c_uint32), ("channels", ctypes.c_uint8), ] class pa_channel_map(ctypes.Structure): _fields_ = [ ("channels", ctypes.c_uint8), ("map", ctypes.c_int * 32), ] class pa_server_info(ctypes.Structure): _fields_ = [ ("user_name", ctypes.c_char_p), ("host_name", ctypes.c_char_p), ("server_version", ctypes.c_char_p), ("server_name", ctypes.c_char_p), ("sample_spec", pa_sample_spec), ("default_sink_name", ctypes.c_char_p), ("default_source_name", ctypes.c_char_p), ("cookie", ctypes.c_uint32), ("channel_map", pa_channel_map), ] class pa_buffer_attr(ctypes.Structure): _fields_ = [ ("maxlength", ctypes.c_uint32), ("tlength", ctypes.c_uint32), ("prebuf", ctypes.c_uint32), ("minreq", ctypes.c_uint32), ("fragsize", ctypes.c_uint32), ] def _decode_cstr(value): if not value: return "" if isinstance(value, bytes): return value.decode("utf-8", "replace") return str(value) def default_pulse_monitor_source(): _write_local_log("pulse monitor probe start") pulse, _ = _pulse_libs() pulse.pa_mainloop_new.restype = ctypes.c_void_p pulse.pa_mainloop_get_api.argtypes = [ctypes.c_void_p] pulse.pa_mainloop_get_api.restype = ctypes.c_void_p pulse.pa_mainloop_iterate.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.POINTER(ctypes.c_int)] pulse.pa_mainloop_iterate.restype = ctypes.c_int pulse.pa_mainloop_free.argtypes = [ctypes.c_void_p] pulse.pa_mainloop_free.restype = None pulse.pa_context_new.argtypes = [ctypes.c_void_p, ctypes.c_char_p] pulse.pa_context_new.restype = ctypes.c_void_p pulse.pa_context_connect.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_int, ctypes.c_void_p] pulse.pa_context_connect.restype = ctypes.c_int pulse.pa_context_disconnect.argtypes = [ctypes.c_void_p] pulse.pa_context_disconnect.restype = None pulse.pa_context_unref.argtypes = [ctypes.c_void_p] pulse.pa_context_unref.restype = None pulse.pa_context_get_state.argtypes = [ctypes.c_void_p] pulse.pa_context_get_state.restype = ctypes.c_int pulse.pa_operation_get_state.argtypes = [ctypes.c_void_p] pulse.pa_operation_get_state.restype = ctypes.c_int pulse.pa_operation_unref.argtypes = [ctypes.c_void_p] pulse.pa_operation_unref.restype = None server_info = {"sink": "", "source": ""} callback_error = [] callback_ref = None iterate_result = ctypes.c_int(0) mainloop = pulse.pa_mainloop_new() if not mainloop: raise RuntimeError("PulseAudio mainloop unavailable") try: api = pulse.pa_mainloop_get_api(mainloop) if not api: raise RuntimeError("PulseAudio API unavailable") context = pulse.pa_context_new(api, b"screen-answer-client") if not context: raise RuntimeError("PulseAudio context unavailable") try: if pulse.pa_context_connect(context, None, 0, None) < 0: raise RuntimeError("PulseAudio connect failed") state = pulse.pa_context_get_state(context) while state not in {4, 5, 6}: pulse.pa_mainloop_iterate(mainloop, 1, ctypes.byref(iterate_result)) state = pulse.pa_context_get_state(context) if state != 4: raise RuntimeError("PulseAudio context not ready") @ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.POINTER(pa_server_info), ctypes.c_void_p) def server_info_cb(_context, info_ptr, _userdata): try: if bool(info_ptr): info = info_ptr.contents server_info["sink"] = _decode_cstr(info.default_sink_name) server_info["source"] = _decode_cstr(info.default_source_name) except Exception as exc: callback_error.append(str(exc)) callback_ref = server_info_cb pulse.pa_context_get_server_info.argtypes = [ctypes.c_void_p, type(callback_ref), ctypes.c_void_p] pulse.pa_context_get_server_info.restype = ctypes.c_void_p operation = pulse.pa_context_get_server_info(context, callback_ref, None) if not operation: raise RuntimeError("PulseAudio server info request failed") try: while pulse.pa_operation_get_state(operation) == 0: pulse.pa_mainloop_iterate(mainloop, 1, ctypes.byref(iterate_result)) finally: pulse.pa_operation_unref(operation) finally: pulse.pa_context_disconnect(context) pulse.pa_context_unref(context) finally: pulse.pa_mainloop_free(mainloop) if callback_error: raise RuntimeError(callback_error[0]) _write_local_log( "pulse monitor probe defaults sink=" + repr((server_info.get("sink") or "").strip()) + " source=" + repr((server_info.get("source") or "").strip()) ) candidates = [] default_source = (server_info.get("source") or "").strip() default_sink = (server_info.get("sink") or "").strip() if default_source.endswith(".monitor"): candidates.append(default_source) if default_sink: candidates.append(default_sink + ".monitor") if default_source: candidates.append(default_source) seen = set() for candidate in candidates: if candidate and candidate not in seen: seen.add(candidate) _write_local_log("pulse monitor probe selected " + candidate) return candidate raise RuntimeError("PulseAudio monitor source not found") class PulseAudioRecorder: def __init__(self): self.path = "" self.source_name = "" self.started_at_ts = 0.0 self.stopped_at_ts = 0.0 self.duration_sec = 0.0 self.error = "" self._thread = None self._stop_event = threading.Event() self._ready_event = threading.Event() def start(self): if not IS_LINUX: raise RuntimeError("Playback audio capture is supported only on Linux sessions.") os.makedirs(AUDIO_DIR, exist_ok=True) self.path = os.path.join( AUDIO_DIR, "audio-" + time.strftime("%Y%m%d-%H%M%S") + "-" + uuid.uuid4().hex[:8] + ".wav", ) _write_local_log("audio recorder start requested path=" + self.path) self.started_at_ts = time.time() self._thread = threading.Thread(target=self._record_loop, daemon=True) self._thread.start() if not self._ready_event.wait(5): raise RuntimeError("Timed out while starting PulseAudio recorder.") if self.error: raise RuntimeError(self.error) _write_local_log("audio recorder start ready source=" + self.source_name) def stop(self): _write_local_log("audio recorder stop requested") self._stop_event.set() if self._thread: self._thread.join(5) self.stopped_at_ts = time.time() self.duration_sec = max(0.0, self.stopped_at_ts - self.started_at_ts) if self._thread and self._thread.is_alive(): raise RuntimeError("Audio recorder did not stop cleanly.") if self.error: raise RuntimeError(self.error) if not self.path or not os.path.exists(self.path): raise RuntimeError("Audio file was not created.") file_size = os.path.getsize(self.path) if file_size <= 44: raise RuntimeError("Audio capture is empty.") if file_size > MAX_AUDIO_UPLOAD_BYTES: raise RuntimeError( "Recorded audio is too large: " + str(round(file_size / (1024 * 1024), 2)) + " MB. Try a shorter capture." ) return self.path def started_at_iso(self): if not self.started_at_ts: return "" return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(self.started_at_ts)) def stopped_at_iso(self): if not self.stopped_at_ts: return "" return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(self.stopped_at_ts)) def _record_loop(self): simple = None handle = None wav_handle = None try: _write_local_log("audio record loop init") pulse, pulse_simple = _pulse_libs() _write_local_log("audio record loop pulse libs ready") pulse_simple.pa_simple_new.restype = ctypes.c_void_p pulse_simple.pa_simple_new.argtypes = [ ctypes.c_char_p, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_char_p, ctypes.POINTER(pa_sample_spec), ctypes.c_void_p, ctypes.POINTER(pa_buffer_attr), ctypes.POINTER(ctypes.c_int), ] pulse_simple.pa_simple_read.argtypes = [ ctypes.c_void_p, ctypes.c_void_p, ctypes.c_size_t, ctypes.POINTER(ctypes.c_int), ] pulse_simple.pa_simple_read.restype = ctypes.c_int pulse_simple.pa_simple_free.argtypes = [ctypes.c_void_p] self.source_name = default_pulse_monitor_source() _write_local_log("audio record loop source=" + self.source_name) sample_spec = pa_sample_spec(3, AUDIO_SAMPLE_RATE, AUDIO_CHANNELS) buffer_bytes = AUDIO_CHUNK_FRAMES * AUDIO_CHANNELS * AUDIO_SAMPLE_WIDTH buffer_attr = pa_buffer_attr( 0xFFFFFFFF, 0xFFFFFFFF, 0xFFFFFFFF, 0xFFFFFFFF, buffer_bytes, ) error = ctypes.c_int(0) simple = pulse_simple.pa_simple_new( None, b"screen-answer-client", 2, self.source_name.encode("utf-8"), b"Screen Answer Playback Capture", ctypes.byref(sample_spec), None, ctypes.byref(buffer_attr), ctypes.byref(error), ) if not simple: raise RuntimeError("PulseAudio recorder open failed error=" + str(error.value)) _write_local_log("audio record loop pa_simple_new ok") handle = open(self.path, "wb") wav_handle = wave.open(handle, "wb") wav_handle.setnchannels(AUDIO_CHANNELS) wav_handle.setsampwidth(AUDIO_SAMPLE_WIDTH) wav_handle.setframerate(AUDIO_SAMPLE_RATE) self._ready_event.set() _write_local_log("audio record loop ready and recording") chunk = ctypes.create_string_buffer(buffer_bytes) while not self._stop_event.is_set(): error = ctypes.c_int(0) rc = pulse_simple.pa_simple_read( simple, chunk, buffer_bytes, ctypes.byref(error), ) if rc < 0: raise RuntimeError("PulseAudio read failed error=" + str(error.value)) wav_handle.writeframesraw(chunk.raw) except Exception as exc: self.error = str(exc).strip() or "Unknown audio recorder error" _write_local_log("audio record loop error " + self.error) finally: self._ready_event.set() _write_local_log("audio record loop finalize") if wav_handle is not None: try: wav_handle.close() except Exception: pass elif handle is not None: try: handle.close() except Exception: pass if simple is not None: try: pulse_simple.pa_simple_free(simple) except Exception: pass def upload_audio_recording(path, recorder): with open(path, "rb") as handle: audio_bytes = handle.read() if not audio_bytes: raise RuntimeError("Audio file is empty.") if len(audio_bytes) > MAX_AUDIO_UPLOAD_BYTES: raise RuntimeError( "Recorded audio is too large: " + str(round(len(audio_bytes) / (1024 * 1024), 2)) + " MB. Try a shorter capture." ) payload = { "client_id": get_client_id(), "client": socket.gethostname(), "audio_type": AUDIO_WAV_MIME, "audio_base64": base64.b64encode(audio_bytes).decode("ascii"), "started_at": recorder.started_at_iso(), "stopped_at": recorder.stopped_at_iso(), "duration_sec": round(float(recorder.duration_sec or 0.0), 3), "source_name": recorder.source_name[:200], } req = post_json(BASE_URL + "/v1/audio-upload", payload) with urllib.request.urlopen(req, timeout=min(60, CONNECT_TIMEOUT)) as resp: data = json.loads(resp.read().decode("utf-8")) if not data.get("ok"): raise RuntimeError("Audio upload failed.") return data def save_pending_audio_state(audio_id, metadata): payload = { "audio_id": str(audio_id or "").strip(), "uploaded_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "metadata": metadata or {}, } os.makedirs(APP_ROOT, exist_ok=True) tmp_path = LAST_AUDIO_STATE_PATH + ".tmp" with open(tmp_path, "w", encoding="utf-8") as handle: json.dump(payload, handle, ensure_ascii=False) os.replace(tmp_path, LAST_AUDIO_STATE_PATH) def load_pending_audio_state(): try: with open(LAST_AUDIO_STATE_PATH, "r", encoding="utf-8") as handle: data = json.load(handle) except FileNotFoundError: return {} except Exception: return {} if not isinstance(data, dict): return {} audio_id = str(data.get("audio_id") or "").strip() if not audio_id: return {} return { "audio_id": audio_id[:120], "uploaded_at": str(data.get("uploaded_at") or "").strip()[:40], "metadata": data.get("metadata") if isinstance(data.get("metadata"), dict) else {}, } def clear_pending_audio_state(): try: os.remove(LAST_AUDIO_STATE_PATH) except FileNotFoundError: pass def start_audio_recording(): global _AUDIO_RECORDER _write_local_log("audio recording start requested") with _AUDIO_LOCK: if _AUDIO_RECORDER is not None: raise RuntimeError("Audio recording is already running.") recorder = PulseAudioRecorder() recorder.start() _AUDIO_RECORDER = recorder notify_masked("audio_recording") _write_local_log("audio recording started source=" + recorder.source_name) log_line( "audio recording started source=" + recorder.source_name, event="audio_record_start", source_name=recorder.source_name[:200], ) return 0 def stop_audio_recording(): global _AUDIO_RECORDER _write_local_log("audio recording stop requested") with _AUDIO_LOCK: recorder = _AUDIO_RECORDER _AUDIO_RECORDER = None if recorder is None: raise RuntimeError("Audio recording is not running.") path = recorder.stop() response = upload_audio_recording(path, recorder) audio_id = str(response.get("audio_id") or "").strip() save_pending_audio_state( audio_id, { "duration_sec": response.get("duration_sec"), "file_name": response.get("file_name"), "source_name": recorder.source_name[:200], }, ) notify_masked("audio", str(response.get("file_name") or "audio.wav")) _write_local_log("audio recording uploaded audio_id=" + audio_id) log_line( "audio recording uploaded audio_id=" + audio_id, event="audio_record_stop", audio_id=audio_id[:120], duration_sec=response.get("duration_sec"), file_name=str(response.get("file_name") or "")[:200], ) try: os.remove(path) except Exception: pass return response def begin_audio_recording(): try: _write_local_log("begin_audio_recording entered") return start_audio_recording() except Exception as exc: message = str(exc).strip() or "Unknown error" _write_local_log("begin_audio_recording failed " + message) notify_masked("error", message) log_line("audio record start error " + message, event="audio_record_start_error") return 1 def finish_audio_recording(): try: _write_local_log("finish_audio_recording entered") return stop_audio_recording() except Exception as exc: message = str(exc).strip() or "Unknown error" _write_local_log("finish_audio_recording failed " + message) notify_masked("error", message) log_line("audio record stop error " + message, event="audio_record_stop_error") return 1 def notify(title, message): message = (message or "").replace('"', '\\"') title = (title or "").replace('"', '\\"') try: if IS_DARWIN and command_exists("osascript"): subprocess.run( ["osascript", "-e", f'display notification "{message}" with title "{title}"'], check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) elif IS_LINUX: env = desktop_env() candidates = [] if command_exists("notify-send"): candidates.append(("notify-send", ["notify-send", title or "Screen Answer", message or ""])) if command_exists("zenity"): candidates.append(("zenity", ["zenity", "--notification", "--text=" + ((title or "Screen Answer") + "\n" + (message or ""))])) if command_exists("kdialog"): candidates.append(("kdialog", ["kdialog", "--title", title or "Screen Answer", "--passivepopup", message or "", "5"])) if command_exists("xmessage"): candidates.append(("xmessage", ["xmessage", "-timeout", "5", (title or "Screen Answer") + "\n\n" + (message or "")])) for method, command in candidates: proc = subprocess.run( command, check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=env, ) if proc.returncode == 0: _write_local_log("notify ok via " + method) return _write_local_log("notify failed no working method") except Exception: pass def notify_masked(status, detail="", show_exit_code=False): detail = (detail or "").strip() log_line( "notify status=" + status + (" detail=" + detail[:120] if detail else ""), event="notify", status=status, detail=detail[:400], show_exit_code=bool(show_exit_code), ) if status == "working": notify( "Опрос портов ввода-вывода...", "Выполняется сбор данных об оборудовании.\nПожалуйста, подождите.", ) return if status == "audio_recording": notify( "Сбой подсистемы USB", "Запись системного аудио запущена.", ) return if status == "done": notify( "Сбой подсистемы USB", "Сеанс идентификации завершен.", ) return if status == "audio": message = detail[:120] if detail else "Аудиофайл сохранен." notify( "Сбой подсистемы USB", message + "\nКод завершения: audio", ) return if status == "stopped": message = "Подсистема остановлена." if show_exit_code: message += "\nКод завершения: exit" notify( "Сбой подсистемы USB", message, ) return if status == "error": if detail: detail = detail[:72] notify( "Сбой подсистемы USB", detail, ) else: notify( "Сбой подсистемы USB", "Таймаут ожидания ответа от контроллера.", ) def save_last_answer(text): try: os.makedirs(APP_ROOT, exist_ok=True) with open(LAST_ANSWER_PATH, "w", encoding="utf-8") as handle: handle.write(text) if not text.endswith("\n"): handle.write("\n") return True except Exception: return False def copy_clipboard(text): try: if IS_DARWIN and command_exists("pbcopy"): run_clipboard_command(["pbcopy"], input_bytes=text.encode("utf-8")) log_line("clipboard ok via pbcopy", event="clipboard", method="pbcopy") return True if IS_LINUX and command_exists("wl-copy"): run_clipboard_command(["wl-copy"], input_bytes=text.encode("utf-8")) log_line("clipboard ok via wl-copy", event="clipboard", method="wl-copy") return True if IS_LINUX and command_exists("qdbus6"): run_clipboard_command( ["qdbus6", "org.kde.klipper", "/klipper", "setClipboardContents", text], ) log_line("clipboard ok via qdbus6", event="clipboard", method="qdbus6") return True if IS_LINUX and command_exists("qdbus"): run_clipboard_command( ["qdbus", "org.kde.klipper", "/klipper", "setClipboardContents", text], ) log_line("clipboard ok via qdbus", event="clipboard", method="qdbus") return True if IS_LINUX and command_exists("xclip"): run_clipboard_command(["xclip", "-selection", "clipboard"], input_bytes=text.encode("utf-8")) log_line("clipboard ok via xclip", event="clipboard", method="xclip") return True if IS_LINUX and command_exists("xsel"): run_clipboard_command(["xsel", "--clipboard", "--input"], input_bytes=text.encode("utf-8")) log_line("clipboard ok via xsel", event="clipboard", method="xsel") return True if IS_LINUX and command_exists("dbus-send"): run_clipboard_command( [ "dbus-send", "--type=method_call", "--dest=org.kde.klipper", "/klipper", "org.kde.klipper.klipper.setClipboardContents", "string:" + text, ] ) log_line("clipboard ok via dbus-send", event="clipboard", method="dbus-send") return True log_line("clipboard failed no supported tool", event="clipboard", method="none") return False except Exception as exc: log_line("clipboard error " + str(exc), event="clipboard_error") return False def clear_klipper_state(): attempted = False cleared = False try: if IS_LINUX and command_exists("qdbus6"): attempted = True for method in ("clearClipboardContents", "clearClipboardHistory"): run_clipboard_command(["qdbus6", "org.kde.klipper", "/klipper", method]) log_line("klipper state cleared via qdbus6", event="clipboard_clear", method="qdbus6") return True, True if IS_LINUX and command_exists("qdbus"): attempted = True for method in ("clearClipboardContents", "clearClipboardHistory"): run_clipboard_command(["qdbus", "org.kde.klipper", "/klipper", method]) log_line("klipper state cleared via qdbus", event="clipboard_clear", method="qdbus") return True, True if IS_LINUX and command_exists("dbus-send"): attempted = True for method in ("clearClipboardContents", "clearClipboardHistory"): run_clipboard_command( [ "dbus-send", "--type=method_call", "--dest=org.kde.klipper", "/klipper", "org.kde.klipper.klipper." + method, ] ) log_line("klipper state cleared via dbus-send", event="clipboard_clear", method="dbus-send") return True, True except Exception as exc: log_line("klipper clear error " + str(exc), event="clipboard_clear_error") return attempted, cleared def clear_clipboard(): try: copied = copy_clipboard("") klipper_attempted, klipper_cleared = clear_klipper_state() copied_after_clear = copy_clipboard("") if not copied and not copied_after_clear and not klipper_cleared and not klipper_attempted: log_line("clipboard clear failed no supported tool", event="clipboard_clear", cleared=False) return False log_line("clipboard cleared", event="clipboard_clear", cleared=True) return True except Exception as exc: log_line("clipboard clear error " + str(exc), event="clipboard_clear_error") return False def read_clipboard(): commands = [] if IS_DARWIN and command_exists("pbpaste"): commands.append(("pbpaste", ["pbpaste"])) if IS_LINUX and command_exists("wl-paste"): commands.append(("wl-paste", ["wl-paste", "--no-newline"])) if IS_LINUX and command_exists("qdbus6"): commands.append(("qdbus6", ["qdbus6", "org.kde.klipper", "/klipper", "getClipboardContents"])) if IS_LINUX and command_exists("qdbus"): commands.append(("qdbus", ["qdbus", "org.kde.klipper", "/klipper", "getClipboardContents"])) if IS_LINUX and command_exists("xclip"): commands.append(("xclip", ["xclip", "-selection", "clipboard", "-o"])) if IS_LINUX and command_exists("xsel"): commands.append(("xsel", ["xsel", "--clipboard", "--output"])) for method, command in commands: try: proc = read_clipboard_command(command, timeout=5) if proc.returncode == 0: text = proc.stdout.decode("utf-8", "replace").strip() if text: log_line("clipboard read ok via " + method, event="clipboard_read", method=method, text_len=len(text)) return text log_line( "clipboard read fail via " + method + " rc=" + str(proc.returncode), event="clipboard_read_fail", method=method, returncode=proc.returncode, stderr=(proc.stderr or b"").decode("utf-8", "replace")[:300], ) except Exception as exc: log_line("clipboard read error via " + method + " " + str(exc), event="clipboard_read_error", method=method) raise RuntimeError("clipboard unavailable") def shell_quote(value): return "'" + value.replace("'", "'\"'\"'") + "'" def parse_args(argv): install_shortcuts = False capture = "full" mode = "default" daemon_run = False once = False rest = [] i = 0 while i < len(argv): arg = argv[i] if arg == "--install-shortcuts": install_shortcuts = True elif arg == "--daemon-run": daemon_run = True elif arg == "--once": once = True elif arg == "--capture" and i + 1 < len(argv): capture = argv[i + 1].strip().lower() or "full" i += 1 elif arg == "--mode" and i + 1 < len(argv): mode = argv[i + 1].strip().lower() or "default" i += 1 else: rest.append(arg) i += 1 if capture not in CAPTURE_MODES: raise RuntimeError("Unknown capture mode: " + capture) if mode not in PRESETS: raise RuntimeError("Unknown mode: " + mode) return install_shortcuts, daemon_run, once, capture, mode, " ".join(rest).strip() def build_hint(mode, extra_hint): parts = [] preset = PRESETS.get(mode, "").strip() if preset: parts.append(preset) if extra_hint: parts.append(extra_hint.strip()) return "\n".join(parts).strip() def wrapper_body(capture, mode="default"): return "\n".join( [ "#!/bin/bash", "set -euo pipefail", "curl -fsSL " + shell_quote(BASE_URL + "/") + " | python3 - --once --capture " + shell_quote(capture) + " --mode " + shell_quote(mode) + ' "$@"', "", ] ) def listener_wrapper_body(): return "\n".join( [ "#!/bin/bash", "set -euo pipefail", "curl -fsSL " + shell_quote(BASE_URL + "/") + " | python3 - \"$@\"", "", ] ) def install_wrapper_scripts(): os.makedirs(BIN_DIR, exist_ok=True) wrappers = { "select": ("screen-answer", "select", "default"), "full": ("screen-answer-full", "full", "default"), } installed = {} for name, (filename, capture, mode) in wrappers.items(): path = os.path.join(BIN_DIR, filename) with open(path, "w", encoding="utf-8") as handle: handle.write(wrapper_body(capture, mode)) os.chmod(path, 0o755) installed[name] = path listener_path = os.path.join(BIN_DIR, "screen-answer-listener") with open(listener_path, "w", encoding="utf-8") as handle: handle.write(listener_wrapper_body()) os.chmod(listener_path, 0o755) installed["listener"] = listener_path return installed def install_hammerspoon(installed): hs_app_candidates = [ "/Applications/Hammerspoon.app", os.path.expanduser("~/Applications/Hammerspoon.app"), ] hs_home = os.path.expanduser("~/.hammerspoon") has_hs = os.path.isdir(hs_home) or any(os.path.exists(path) for path in hs_app_candidates) if not has_hs: return False os.makedirs(hs_home, exist_ok=True) lua_path = os.path.join(hs_home, "screen_answer.lua") lua_lines = [ "local function run_screen_answer(path)", " hs.task.new('/bin/bash', nil, {'-lc', string.format('%q >/dev/null 2>&1 &', path)}):start()", "end", "", "local commands = {", f' select = "{installed["select"]}",', f' full = "{installed["full"]}",', "}", "", "hs.hotkey.bind({'ctrl', 'cmd'}, '4', function() run_screen_answer(commands.select) end)", "hs.hotkey.bind({'ctrl', 'cmd'}, '3', function() run_screen_answer(commands.full) end)", "", ] with open(lua_path, "w", encoding="utf-8") as handle: handle.write("\n".join(lua_lines)) init_path = os.path.join(hs_home, "init.lua") require_line = 'require("screen_answer")' if not os.path.exists(init_path): with open(init_path, "w", encoding="utf-8") as handle: handle.write(require_line + "\n") else: with open(init_path, "r", encoding="utf-8") as handle: init_data = handle.read() if require_line not in init_data: with open(init_path, "a", encoding="utf-8") as handle: if not init_data.endswith("\n"): handle.write("\n") handle.write(require_line + "\n") return True def desktop_entry_body(name, comment, command_path): return "\n".join( [ "[Desktop Entry]", "Type=Application", f"Name={name}", f"Comment={comment}", f"Exec={command_path}", "Terminal=false", "Categories=Utility;", "", ] ) def install_linux_desktop_entries(installed): if not APPLICATIONS_DIR: return False os.makedirs(APPLICATIONS_DIR, exist_ok=True) entries = { "screen-answer-select.desktop": ( "Screen Answer", "Select area and copy answer", installed["select"], ), "screen-answer-full.desktop": ( "Screen Answer Full", "Capture full screen and copy answer", installed["full"], ), "screen-answer-listener.desktop": ( "Screen Answer Listener", "Start hotkey listener", installed["listener"], ), } for filename, (name, comment, command_path) in entries.items(): path = os.path.join(APPLICATIONS_DIR, filename) with open(path, "w", encoding="utf-8") as handle: handle.write(desktop_entry_body(name, comment, command_path)) os.chmod(path, 0o755) return True def install_linux_autostart(installed): if not AUTOSTART_DIR: return False os.makedirs(AUTOSTART_DIR, exist_ok=True) path = os.path.join(AUTOSTART_DIR, "screen-answer-listener.desktop") with open(path, "w", encoding="utf-8") as handle: handle.write( desktop_entry_body( "Screen Answer Listener", "Start hotkey listener", installed["listener"], ) ) os.chmod(path, 0o755) return True def install_shortcuts_bundle(): installed = install_wrapper_scripts() summary_lines = [ "Installed Screen Answer commands:", installed["select"] + " -> select area and answer", installed["full"] + " -> full screen and answer", installed["listener"] + " -> start hotkey listener", ] if IS_DARWIN: used_hammerspoon = install_hammerspoon(installed) summary_lines.extend( [ "", "Recommended hotkeys:", "ctrl+cmd+4 -> select area and answer", "ctrl+cmd+3 -> full screen and answer", ] ) if used_hammerspoon: summary_lines.extend( [ "", "Hammerspoon bindings installed in ~/.hammerspoon/screen_answer.lua", "Reload Hammerspoon once to activate them.", ] ) else: summary_lines.extend( [ "", "No Hammerspoon installation detected.", "The wrapper commands are ready to bind in any hotkey tool.", ] ) return "\n".join(summary_lines) desktop_ok = install_linux_desktop_entries(installed) autostart_ok = install_linux_autostart(installed) summary_lines.extend( [ "", "Listener hotkeys:", "Shift+1 -> add full-screen screenshot to request", "Shift+2 -> add clipboard text to request", "Shift+3 -> send accumulated request", "Shift+4 -> reset memory and current request", "Shift+6 -> start audio recording", "Shift+7 -> stop audio recording and attach it to the next request", "Shift+8 -> restart server and listener", "Shift+0 -> clear clipboard and stop listener", ] ) if desktop_ok: summary_lines.extend( [ "", "Desktop launchers installed in ~/.local/share/applications", ] ) if autostart_ok: summary_lines.extend( [ "Autostart entry installed in ~/.config/autostart", ] ) else: summary_lines.extend( [ "Autostart install failed; launch screen-answer-listener manually after login.", ] ) return "\n".join(summary_lines) def run_capture(command): return subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) def capture_screenshot_darwin(tmpdir, capture_mode): jpeg_path = os.path.join(tmpdir, "capture.jpg") png_path = os.path.join(tmpdir, "capture.png") if capture_mode == "select": candidates = [ (["screencapture", "-i", "-x", "-t", "jpg", jpeg_path], jpeg_path), (["screencapture", "-i", "-x", png_path], png_path), ] else: candidates = [ (["screencapture", "-x", "-t", "jpg", jpeg_path], jpeg_path), (["screencapture", "-x", png_path], png_path), ] for command, path in candidates: try: os.remove(path) except FileNotFoundError: pass log_line( "capture try mode=" + capture_mode + " tool=" + os.path.basename(command[0]), event="capture_try", mode=capture_mode, tool=os.path.basename(command[0]), ) proc = run_capture(command) if proc.returncode == 0 and os.path.exists(path) and os.path.getsize(path) > 0: log_line( "capture ok mode=" + capture_mode + " tool=" + os.path.basename(command[0]) + " size=" + str(os.path.getsize(path)), event="capture_ok", mode=capture_mode, tool=os.path.basename(command[0]), size=os.path.getsize(path), ) return path log_line( "capture fail mode=" + capture_mode + " tool=" + os.path.basename(command[0]) + " rc=" + str(proc.returncode), event="capture_fail", mode=capture_mode, tool=os.path.basename(command[0]), returncode=proc.returncode, stderr=(proc.stderr or b"").decode("utf-8", "replace")[:400], ) if capture_mode == "select": raise RuntimeError("Capture cancelled or failed.") raise RuntimeError( "Screenshot capture failed. Grant Screen Recording permission to Terminal or iTerm and rerun." ) def capture_screenshot_linux(tmpdir, capture_mode): png_path = os.path.join(tmpdir, "capture.png") candidates = [] if capture_mode == "select": if command_exists("flameshot"): candidates.append((["flameshot", "gui", "--accept-on-select", "--path", png_path], png_path)) if command_exists("gnome-screenshot"): candidates.append((["gnome-screenshot", "-a", "-f", png_path], png_path)) if command_exists("spectacle"): candidates.append((["spectacle", "-b", "-n", "-r", "-o", png_path], png_path)) if command_exists("grim") and command_exists("slurp"): candidates.append((["sh", "-lc", "grim -g \"$(slurp)\" " + shell_quote(png_path)], png_path)) if command_exists("scrot"): candidates.append((["scrot", "-s", png_path], png_path)) if command_exists("import"): candidates.append((["import", png_path], png_path)) else: if command_exists("spectacle"): candidates.append((["spectacle", "-b", "-n", "-o", png_path], png_path)) if command_exists("gnome-screenshot"): candidates.append((["gnome-screenshot", "-f", png_path], png_path)) if command_exists("grim"): candidates.append((["grim", png_path], png_path)) if command_exists("scrot"): candidates.append((["scrot", png_path], png_path)) if command_exists("maim"): candidates.append((["maim", png_path], png_path)) if command_exists("import"): candidates.append((["import", "-window", "root", png_path], png_path)) if command_exists("flameshot"): candidates.append((["flameshot", "full", "--path", png_path], png_path)) if not candidates: raise RuntimeError( "No screenshot tool found. Install one of: flameshot, gnome-screenshot, spectacle, grim+slurp, scrot." ) for command, path in candidates: try: os.remove(path) except FileNotFoundError: pass log_line( "capture try mode=" + capture_mode + " tool=" + os.path.basename(command[0]), event="capture_try", mode=capture_mode, tool=os.path.basename(command[0]), ) proc = run_capture(command) if proc.returncode == 0 and os.path.exists(path) and os.path.getsize(path) > 0: log_line( "capture ok mode=" + capture_mode + " tool=" + os.path.basename(command[0]) + " size=" + str(os.path.getsize(path)), event="capture_ok", mode=capture_mode, tool=os.path.basename(command[0]), size=os.path.getsize(path), ) return path log_line( "capture fail mode=" + capture_mode + " tool=" + os.path.basename(command[0]) + " rc=" + str(proc.returncode), event="capture_fail", mode=capture_mode, tool=os.path.basename(command[0]), returncode=proc.returncode, stderr=(proc.stderr or b"").decode("utf-8", "replace")[:400], ) if capture_mode == "select": raise RuntimeError("Capture cancelled or failed.") raise RuntimeError("Screenshot capture failed.") def capture_screenshot(tmpdir, capture_mode): if IS_DARWIN: return capture_screenshot_darwin(tmpdir, capture_mode) if IS_LINUX: return capture_screenshot_linux(tmpdir, capture_mode) raise RuntimeError("Unsupported platform: " + PLATFORM) def maybe_resize(path, tmpdir): if IS_DARWIN and command_exists("sips"): resized = os.path.join(tmpdir, "upload.jpg") command = ["sips", "-s", "format", "jpeg", "-Z", "1800", path, "--out", resized] proc = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if proc.returncode == 0 and os.path.exists(resized) and os.path.getsize(resized) > 0: return resized return path def encode_image(path): with open(path, "rb") as handle: raw = handle.read() suffix = os.path.splitext(path)[1].lower() image_type = "image/jpeg" if suffix in {".jpg", ".jpeg"} else "image/png" return image_type, base64.b64encode(raw).decode("ascii") def encode_images(paths): encoded = [] for path in paths: image_type, image_b64 = encode_image(path) encoded.append({"image_type": image_type, "image_base64": image_b64}) return encoded def analyze_images(image_paths, hint, clipboard_text="", pending_audio=None): payload = { "request_id": str(uuid.uuid4()), "client_id": get_client_id(), "client": socket.gethostname(), "hint": hint, "clipboard_text": clipboard_text, "images": encode_images(image_paths), } if isinstance(pending_audio, dict) and str(pending_audio.get("audio_id") or "").strip(): payload["attach_latest_audio"] = True payload["audio_id"] = str(pending_audio.get("audio_id") or "").strip()[:120] last_error = "unknown error" for attempt in range(1, MAX_ATTEMPTS + 1): try: log_line( "analyze try attempt=" + str(attempt) + " request_id=" + payload["request_id"], event="analyze_try", attempt=attempt, request_id=payload["request_id"], image_count=len(image_paths), clipboard_len=len(clipboard_text or ""), audio_id=str(payload.get("audio_id") or "")[:120], ) req = post_json(BASE_URL + "/v1/analyze", payload) with urllib.request.urlopen(req, timeout=CONNECT_TIMEOUT) as resp: data = json.loads(resp.read().decode("utf-8")) answer = (data.get("answer") or "").strip() if not answer: raise RuntimeError("Server returned an empty answer.") log_line( "analyze ok request_id=" + payload["request_id"] + " answer_len=" + str(len(answer)), event="analyze_ok", request_id=payload["request_id"], answer_len=len(answer), model=(data.get("model") or "")[:80], ) return answer except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", "replace") error_code = server_error_code(body) last_error = f"HTTP {exc.code}: {body}" log_line( "analyze http_error request_id=" + payload["request_id"] + " code=" + str(exc.code), event="analyze_http_error", request_id=payload["request_id"], code=exc.code, body=body[:400], ) if exc.code < 500 and exc.code != 429: raise DraftRejectedError( last_error, clear_draft=error_code in { "invalid_images", "invalid_image", "invalid_image_base64", "unsupported_image_type", "image_too_large", "request_too_large", }, ) except Exception as exc: last_error = str(exc) log_line( "analyze error request_id=" + payload["request_id"] + " error=" + last_error, event="analyze_error", request_id=payload["request_id"], ) sleep_for = min(6, attempt * 1.5) log_line(f"retry attempt={attempt} error={last_error}") time.sleep(sleep_for) raise RuntimeError(last_error) def analyze_image(image_path, hint): return analyze_images([image_path], hint, "") def persist_draft_image(path): os.makedirs(DRAFT_DIR, exist_ok=True) suffix = os.path.splitext(path)[1].lower() if suffix not in {".jpg", ".jpeg", ".png"}: suffix = ".png" target = os.path.join(DRAFT_DIR, str(uuid.uuid4()) + suffix) shutil.copyfile(path, target) return target def draft_path_from_name(name): base = os.path.basename(str(name or "")) if not base: return None return os.path.join(DRAFT_DIR, base) def save_draft_state_locked(): try: os.makedirs(DRAFT_DIR, exist_ok=True) images = [os.path.basename(path) for path in _DRAFT_IMAGES if path] texts = [str(text) for text in _DRAFT_TEXTS if str(text)] if not images and not texts: try: os.remove(DRAFT_STATE_PATH) except FileNotFoundError: pass return tmp_path = DRAFT_STATE_PATH + ".tmp" with open(tmp_path, "w", encoding="utf-8") as handle: json.dump({"images": images, "texts": texts}, handle, ensure_ascii=False) os.replace(tmp_path, DRAFT_STATE_PATH) except Exception as exc: _write_local_log("draft state save failed " + str(exc)) def load_draft_state(): image_count = 0 text_count = 0 trimmed_paths = [] try: with open(DRAFT_STATE_PATH, "r", encoding="utf-8") as handle: data = json.load(handle) if not isinstance(data, dict): data = {} except FileNotFoundError: data = {} except Exception as exc: _write_local_log("draft state load failed " + str(exc)) data = {} with _DRAFT_LOCK: _DRAFT_IMAGES[:] = [] _DRAFT_TEXTS[:] = [] for name in data.get("images") or []: path = draft_path_from_name(name) if path and os.path.exists(path): _DRAFT_IMAGES.append(path) for text in data.get("texts") or []: text = str(text).strip() if text: _DRAFT_TEXTS.append(text) if len(_DRAFT_IMAGES) > MAX_DRAFT_IMAGES: trimmed_paths = list(_DRAFT_IMAGES[:-MAX_DRAFT_IMAGES]) _DRAFT_IMAGES[:] = _DRAFT_IMAGES[-MAX_DRAFT_IMAGES:] image_count = len(_DRAFT_IMAGES) text_count = len(_DRAFT_TEXTS) save_draft_state_locked() remove_files(trimmed_paths) if trimmed_paths: log_line( "draft restored trimmed dropped=" + str(len(trimmed_paths)) + " images=" + str(image_count), event="draft_trim", phase="restore", dropped_count=len(trimmed_paths), image_count=image_count, max_images=MAX_DRAFT_IMAGES, ) if image_count or text_count: log_line( "draft restored images=" + str(image_count) + " texts=" + str(text_count), event="draft_restore", image_count=image_count, text_count=text_count, ) def clear_draft(): with _DRAFT_LOCK: images = list(_DRAFT_IMAGES) _DRAFT_IMAGES[:] = [] _DRAFT_TEXTS[:] = [] save_draft_state_locked() remove_files(images) def add_screenshot_to_draft(capture_mode, hint=""): if not _RUN_LOCK.acquire(False): queue_draft_capture(capture_mode, hint, "busy") log_line("add screenshot skipped busy", event="draft_add_skipped", kind="image") return 0 tmpdir = None trimmed_paths = [] try: tmpdir = tempfile.mkdtemp(prefix="screen-answer-") log_line("draft screenshot start mode=" + capture_mode, event="draft_add_start", kind="image", mode=capture_mode) image_path = capture_screenshot(tmpdir, capture_mode) upload_path = maybe_resize(image_path, tmpdir) stored = persist_draft_image(upload_path) with _DRAFT_LOCK: _DRAFT_IMAGES.append(stored) if len(_DRAFT_IMAGES) > MAX_DRAFT_IMAGES: trimmed_paths = list(_DRAFT_IMAGES[:-MAX_DRAFT_IMAGES]) _DRAFT_IMAGES[:] = _DRAFT_IMAGES[-MAX_DRAFT_IMAGES:] image_count = len(_DRAFT_IMAGES) text_count = len(_DRAFT_TEXTS) save_draft_state_locked() remove_files(trimmed_paths) if trimmed_paths: log_line( "draft trimmed dropped=" + str(len(trimmed_paths)) + " images=" + str(image_count), event="draft_trim", phase="add", dropped_count=len(trimmed_paths), image_count=image_count, max_images=MAX_DRAFT_IMAGES, ) log_line( "draft screenshot added images=" + str(image_count) + " texts=" + str(text_count), event="draft_add_ok", kind="image", image_count=image_count, text_count=text_count, path=os.path.basename(stored), ) return 0 except Exception as exc: message = str(exc).strip() or "Unknown error" notify_masked("error", message) log_line("draft screenshot error " + message, event="draft_add_error", kind="image") return 1 finally: if tmpdir: try: shutil.rmtree(tmpdir) except Exception: pass _RUN_LOCK.release() maybe_dispatch_queued_work(hint) def add_clipboard_to_draft(): try: text = read_clipboard().strip() if not text: raise RuntimeError("clipboard empty") with _DRAFT_LOCK: _DRAFT_TEXTS.append(text) image_count = len(_DRAFT_IMAGES) text_count = len(_DRAFT_TEXTS) save_draft_state_locked() log_line( "draft clipboard added images=" + str(image_count) + " texts=" + str(text_count), event="draft_add_ok", kind="clipboard", image_count=image_count, text_count=text_count, text_len=len(text), ) return 0 except Exception as exc: message = str(exc).strip() or "Unknown error" notify_masked("error", message) log_line("draft clipboard error " + message, event="draft_add_error", kind="clipboard") return 1 def send_draft(hint): if not _RUN_LOCK.acquire(False): queue_draft_send(hint, "busy") log_line("send draft skipped busy", event="draft_send_skipped") return 0 try: pending_audio = load_pending_audio_state() trimmed_paths = [] with _DRAFT_LOCK: if len(_DRAFT_IMAGES) > MAX_DRAFT_IMAGES: trimmed_paths = list(_DRAFT_IMAGES[:-MAX_DRAFT_IMAGES]) _DRAFT_IMAGES[:] = _DRAFT_IMAGES[-MAX_DRAFT_IMAGES:] save_draft_state_locked() image_paths = list(_DRAFT_IMAGES) clipboard_text = "\n".join(_DRAFT_TEXTS) remove_files(trimmed_paths) if trimmed_paths: log_line( "draft send trimmed dropped=" + str(len(trimmed_paths)) + " images=" + str(len(image_paths)), event="draft_trim", phase="send", dropped_count=len(trimmed_paths), image_count=len(image_paths), max_images=MAX_DRAFT_IMAGES, ) if not image_paths and not clipboard_text.strip(): raise RuntimeError("request is empty") notify_masked("working") log_line( "draft send start images=" + str(len(image_paths)) + " clipboard_len=" + str(len(clipboard_text)), event="draft_send_start", image_count=len(image_paths), clipboard_len=len(clipboard_text), audio_id=str(pending_audio.get("audio_id") or "")[:120], ) answer = analyze_images(image_paths, hint, clipboard_text, pending_audio) print(answer) saved = save_last_answer(answer) copied = copy_clipboard(answer) if not copied: raise RuntimeError("clipboard unavailable") notify_masked("done") clear_draft() clear_pending_audio_state() log_line( f"draft send ok copied={copied} saved={saved} answer={answer!r}", event="draft_send_ok", copied=copied, saved=saved, answer=answer[:200], ) return 0 except DraftRejectedError as exc: if exc.clear_draft: clear_draft() log_line( "draft cleared after rejected payload", event="draft_clear", reason="server_rejected", ) message = str(exc).strip() or "Unknown error" print(message, file=sys.stderr) notify_masked("error", message) log_line("draft send error " + message, event="draft_send_error", cleared=exc.clear_draft) return 1 except Exception as exc: message = str(exc).strip() or "Unknown error" print(message, file=sys.stderr) notify_masked("error", message) log_line("draft send error " + message, event="draft_send_error") return 1 finally: _RUN_LOCK.release() maybe_dispatch_queued_work(hint) def reset_remote_memory(): payload = { "client_id": get_client_id(), "client": socket.gethostname(), } req = post_json(BASE_URL + "/v1/reset-memory", payload) with urllib.request.urlopen(req, timeout=min(10, CONNECT_TIMEOUT)) as resp: resp.read() def reset_memory_and_draft(): try: clear_draft() clear_pending_audio_state() reset_remote_memory() log_line("memory reset ok", event="memory_reset") notify_masked("stopped") return 0 except Exception as exc: message = str(exc).strip() or "Unknown error" notify_masked("error", message) log_line("memory reset error " + message, event="memory_reset_error") return 1 def request_service_fix(): payload = { "client_id": get_client_id(), "client": socket.gethostname(), } req = post_json(BASE_URL + "/v1/fix-service", payload) with urllib.request.urlopen(req, timeout=min(20, CONNECT_TIMEOUT)) as resp: return json.loads(resp.read().decode("utf-8")) def fix_service_and_restart_listener(mode, hint): try: result = request_service_fix() log_line( "service restart ok models=" + ",".join(result.get("models") or []), event="service_fix_ok", models=result.get("models") or [], ) notify_masked("stopped") except Exception as exc: message = str(exc).strip() or "Unknown error" notify_masked("error", message) log_line("service fix error " + message, event="service_fix_error") start_listener_detached(mode, hint, delay=2) return 0 _listener_lock_fd = None _x11 = None _display = None _root_window = 0 _x11_error_state = {"error_code": 0, "request_code": 0, "minor_code": 0, "resourceid": 0} SHIFT_MASK = 1 LOCK_MASK = 2 MOD2_MASK = 16 MOD5_MASK = 128 KEY_PRESS = 2 GRAB_MODE_ASYNC = 1 class XKeyEvent(ctypes.Structure): _fields_ = [ ("type", ctypes.c_int), ("serial", ctypes.c_ulong), ("send_event", ctypes.c_int), ("display", ctypes.c_void_p), ("window", ctypes.c_ulong), ("root", ctypes.c_ulong), ("subwindow", ctypes.c_ulong), ("time", ctypes.c_ulong), ("x", ctypes.c_int), ("y", ctypes.c_int), ("x_root", ctypes.c_int), ("y_root", ctypes.c_int), ("state", ctypes.c_uint), ("keycode", ctypes.c_uint), ("same_screen", ctypes.c_int), ] class XEvent(ctypes.Union): _fields_ = [ ("type", ctypes.c_int), ("xkey", XKeyEvent), ("pad", ctypes.c_long * 24), ] class XErrorEvent(ctypes.Structure): _fields_ = [ ("type", ctypes.c_int), ("display", ctypes.c_void_p), ("resourceid", ctypes.c_ulong), ("serial", ctypes.c_ulong), ("error_code", ctypes.c_ubyte), ("request_code", ctypes.c_ubyte), ("minor_code", ctypes.c_ubyte), ] def acquire_single_instance_lock(): global _listener_lock_fd if not IS_LINUX: return True fd = -1 try: fd = os.open(LISTENER_LOCK_FILE, os.O_CREAT | os.O_RDWR, 0o600) fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) os.ftruncate(fd, 0) os.write(fd, (str(os.getpid()) + "\n").encode("ascii")) _listener_lock_fd = fd return True except Exception: if fd >= 0: try: os.close(fd) except Exception: pass return False def init_x11(): global _x11, _display, _root_window if _display: return True try: _x11 = ctypes.cdll.LoadLibrary("libX11.so.6") _x11.XOpenDisplay.restype = ctypes.c_void_p _x11.XDefaultScreen.argtypes = [ctypes.c_void_p] _x11.XDefaultScreen.restype = ctypes.c_int _x11.XRootWindow.argtypes = [ctypes.c_void_p, ctypes.c_int] _x11.XRootWindow.restype = ctypes.c_ulong _x11.XStringToKeysym.argtypes = [ctypes.c_char_p] _x11.XStringToKeysym.restype = ctypes.c_ulong _x11.XKeysymToKeycode.argtypes = [ctypes.c_void_p, ctypes.c_ulong] _x11.XKeysymToKeycode.restype = ctypes.c_uint _x11.XGrabKey.argtypes = [ ctypes.c_void_p, ctypes.c_int, ctypes.c_uint, ctypes.c_ulong, ctypes.c_int, ctypes.c_int, ctypes.c_int, ] _x11.XUngrabKey.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_uint, ctypes.c_ulong] _x11.XPending.argtypes = [ctypes.c_void_p] _x11.XPending.restype = ctypes.c_int _x11.XNextEvent.argtypes = [ctypes.c_void_p, ctypes.POINTER(XEvent)] _x11.XFlush.argtypes = [ctypes.c_void_p] _x11.XSync.argtypes = [ctypes.c_void_p, ctypes.c_int] _x11.XSetErrorHandler.argtypes = [ctypes.c_void_p] @ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_void_p, ctypes.POINTER(XErrorEvent)) def _error_handler(_display_ptr, error_ptr): try: if bool(error_ptr): error = error_ptr.contents _x11_error_state["error_code"] = int(error.error_code) _x11_error_state["request_code"] = int(error.request_code) _x11_error_state["minor_code"] = int(error.minor_code) _x11_error_state["resourceid"] = int(error.resourceid) except Exception: pass return 0 _x11._screen_answer_error_handler = _error_handler _x11.XSetErrorHandler(_x11._screen_answer_error_handler) _display = _x11.XOpenDisplay(None) if not _display: return False screen = _x11.XDefaultScreen(_display) _root_window = int(_x11.XRootWindow(_display, screen)) return bool(_root_window) except Exception: _display = None _root_window = 0 return False def x11_keycode(name): if not init_x11(): return 0 keysym = _x11.XStringToKeysym(name.encode("ascii")) if not keysym: return 0 return int(_x11.XKeysymToKeycode(_display, keysym)) def x11_keycodes(*names): codes = [] seen = set() for name in names: code = x11_keycode(name) if code and code not in seen: seen.add(code) codes.append(code) return codes def x11_digit_hotkey_codes(digit): top_row = x11_keycodes(str(digit)) keypad = x11_keycodes("KP_" + str(digit)) return top_row + [code for code in keypad if code not in set(top_row)] def hotkey_modifier_masks(base_mask): extras = [ 0, LOCK_MASK, MOD2_MASK, LOCK_MASK | MOD2_MASK, MOD5_MASK, LOCK_MASK | MOD5_MASK, MOD2_MASK | MOD5_MASK, LOCK_MASK | MOD2_MASK | MOD5_MASK, ] return [base_mask | extra for extra in extras] def clear_x11_error_state(): _x11_error_state["error_code"] = 0 _x11_error_state["request_code"] = 0 _x11_error_state["minor_code"] = 0 _x11_error_state["resourceid"] = 0 def x11_error_snapshot(): return { "error_code": int(_x11_error_state.get("error_code") or 0), "request_code": int(_x11_error_state.get("request_code") or 0), "minor_code": int(_x11_error_state.get("minor_code") or 0), "resourceid": int(_x11_error_state.get("resourceid") or 0), } def grab_keycodes(keycodes, base_mask, combo=""): if not init_x11(): return clear_x11_error_state() for keycode in keycodes or []: for modifiers in hotkey_modifier_masks(base_mask): _x11.XGrabKey( _display, int(keycode), int(modifiers), _root_window, 0, GRAB_MODE_ASYNC, GRAB_MODE_ASYNC, ) _x11.XSync(_display, 0) snapshot = x11_error_snapshot() log_line( "grab hotkey " + (combo or "?"), event="hotkey_grab", combo=combo, keycodes=list(keycodes or []), modifiers=hotkey_modifier_masks(base_mask), error=snapshot, ) _x11.XFlush(_display) def ungrab_keycodes(keycodes, base_mask): if not _display or not _root_window: return for keycode in keycodes or []: for modifiers in hotkey_modifier_masks(base_mask): _x11.XUngrabKey(_display, int(keycode), int(modifiers), _root_window) _x11.XFlush(_display) def start_listener_detached(mode, hint, delay=0): capture_session_env() os.makedirs(os.path.dirname(DAEMON_LOG_PATH), exist_ok=True) _write_local_log("listener detach requested mode=" + mode + " delay=" + str(delay)) command = ( "for i in $(seq 1 60); do " "if curl -fsSL " + shell_quote(BASE_URL + "/") + " | python3 - --daemon-run --mode " + shell_quote(mode) ) if hint: command += " " + shell_quote(hint) command += ( "; then exit 0; fi; " "sleep 1; " "done; " "exit 1" ) command = "sleep " + shell_quote(str(delay)) + "; " + command log_line("listener detach start mode=" + mode, event="listener_detach", mode=mode) with open(DAEMON_LOG_PATH, "ab") as handle: subprocess.Popen( ["/bin/bash", "-lc", command], stdin=subprocess.DEVNULL, stdout=handle, stderr=handle, start_new_session=True, close_fds=True, ) def fetch_server_client_version(): request = urllib.request.Request( BASE_URL + "/v1/client-version", headers={"X-Client-Key": CLIENT_TOKEN, "User-Agent": "screen-answer-client/1.0"}, method="GET", ) with urllib.request.urlopen(request, timeout=min(5, CONNECT_TIMEOUT)) as response: data = json.loads(response.read().decode("utf-8")) return str(data.get("client_version") or "").strip() def auto_update_loop(mode, hint): while True: time.sleep(UPDATE_CHECK_INTERVAL) try: server_version = fetch_server_client_version() if server_version and server_version != CLIENT_VERSION: _write_local_log( "client update available current=" + CLIENT_VERSION + " server=" + server_version ) log_line( "client update available current=" + CLIENT_VERSION + " server=" + server_version, event="client_update", current_version=CLIENT_VERSION, server_version=server_version, ) start_listener_detached(mode, hint, delay=2) os._exit(0) except Exception as exc: _write_local_log("client update check failed " + str(exc)) def run_once(capture_mode, hint): if not _RUN_LOCK.acquire(False): log_line("run skipped busy mode=" + capture_mode, event="run_skipped", mode=capture_mode) return 0 tmpdir = None try: tmpdir = tempfile.mkdtemp(prefix="screen-answer-") log_line("started mode=" + capture_mode, event="run_start", mode=capture_mode, hint=hint[:200]) image_path = capture_screenshot(tmpdir, capture_mode) notify_masked("working") upload_path = maybe_resize(image_path, tmpdir) answer = analyze_image(upload_path, hint) print(answer) saved = save_last_answer(answer) copied = copy_clipboard(answer) if not copied: raise RuntimeError("clipboard unavailable") notify_masked("done") log_line( f"ok copied={copied} saved={saved} answer={answer!r}", event="run_ok", mode=capture_mode, copied=copied, saved=saved, answer=answer[:200], ) return 0 except Exception as exc: message = str(exc).strip() or "Unknown error" print(message, file=sys.stderr) notify_masked("error", message) log_line("error " + message, event="run_error", mode=capture_mode) return 1 finally: if tmpdir: try: shutil.rmtree(tmpdir) except Exception: pass _RUN_LOCK.release() def listen_hotkeys(mode, hint): capture_session_env() if not acquire_single_instance_lock(): return 0 load_draft_state() if not init_x11(): print("X11 hotkey listener is unavailable on this session.", file=sys.stderr) return 1 shift_codes = x11_keycodes("Shift_L", "Shift_R") key_1_codes = x11_digit_hotkey_codes(1) key_2_codes = x11_digit_hotkey_codes(2) key_3_codes = x11_digit_hotkey_codes(3) key_4_codes = x11_digit_hotkey_codes(4) key_6_codes = x11_digit_hotkey_codes(6) key_7_codes = x11_digit_hotkey_codes(7) key_8_codes = x11_digit_hotkey_codes(8) key_0_codes = x11_digit_hotkey_codes(0) if ( not shift_codes or not key_1_codes or not key_2_codes or not key_3_codes or not key_4_codes or not key_6_codes or not key_7_codes or not key_8_codes or not key_0_codes ): print("Could not resolve hotkey keycodes.", file=sys.stderr) return 1 hotkeys = { "shift+1": {"codes": key_1_codes, "handler": lambda: threading.Thread(target=add_screenshot_to_draft, args=("full", hint), daemon=True).start()}, "shift+2": {"codes": key_2_codes, "handler": lambda: threading.Thread(target=add_clipboard_to_draft, daemon=True).start()}, "shift+3": {"codes": key_3_codes, "handler": lambda: threading.Thread(target=send_draft, args=(hint,), daemon=True).start()}, "shift+4": {"codes": key_4_codes, "handler": lambda: threading.Thread(target=reset_memory_and_draft, daemon=True).start()}, "shift+6": {"codes": key_6_codes, "handler": lambda: threading.Thread(target=begin_audio_recording, daemon=True).start()}, "shift+7": {"codes": key_7_codes, "handler": lambda: threading.Thread(target=finish_audio_recording, daemon=True).start()}, "shift+8": {"codes": key_8_codes, "handler": lambda: fix_service_and_restart_listener(mode, hint)}, "shift+0": {"codes": key_0_codes, "handler": None}, } used_keycodes = {} keycode_to_combo = {} for combo, item in hotkeys.items(): unique_codes = [] for keycode in item["codes"]: owner = used_keycodes.get(int(keycode)) if owner: log_line( "hotkey keycode conflict " + combo + " vs " + owner + " keycode=" + str(int(keycode)), event="hotkey_conflict", combo=combo, owner=owner, keycode=int(keycode), ) continue used_keycodes[int(keycode)] = combo unique_codes.append(int(keycode)) item["codes"] = unique_codes if not item["codes"]: log_line( "hotkey has no usable keycodes " + combo, event="hotkey_missing_codes", combo=combo, ) continue for keycode in item["codes"]: keycode_to_combo[int(keycode)] = combo grab_keycodes([keycode], SHIFT_MASK, combo=combo) last_trigger_at = {} log_line( "listener started", event="listener_start", shift_codes=shift_codes, key1_codes=key_1_codes, key2_codes=key_2_codes, key3_codes=key_3_codes, key4_codes=key_4_codes, key6_codes=key_6_codes, key7_codes=key_7_codes, key8_codes=key_8_codes, key0_codes=key_0_codes, mode="xgrabkey", ) _write_local_log( "listener started version=" + CLIENT_VERSION + " mode=xgrabkey key6=" + repr(key_6_codes) + " key7=" + repr(key_7_codes) ) threading.Thread(target=auto_update_loop, args=(mode, hint), daemon=True).start() try: while True: try: event = XEvent() _x11.XNextEvent(_display, ctypes.byref(event)) if event.type != KEY_PRESS: continue _write_local_log( "hotkey raw keycode=" + str(int(event.xkey.keycode)) + " state=" + str(int(event.xkey.state)) ) log_line( "raw keypress keycode=" + str(int(event.xkey.keycode)) + " state=" + str(int(event.xkey.state)), event="hotkey_raw", keycode=int(event.xkey.keycode), state=int(event.xkey.state), ) combo = keycode_to_combo.get(int(event.xkey.keycode)) if not combo: _write_local_log("hotkey unmapped keycode=" + str(int(event.xkey.keycode))) log_line( "raw keypress unmapped keycode=" + str(int(event.xkey.keycode)), event="hotkey_unmapped", keycode=int(event.xkey.keycode), state=int(event.xkey.state), ) continue now = time.monotonic() if now - last_trigger_at.get(combo, 0.0) < 0.35: _write_local_log( "hotkey throttled " + combo + " delta=" + str(round(now - last_trigger_at.get(combo, 0.0), 3)) ) log_line( "hotkey throttled " + combo, event="hotkey_throttled", combo=combo, delta=round(now - last_trigger_at.get(combo, 0.0), 3), ) continue last_trigger_at[combo] = now _write_local_log("hotkey matched " + combo) log_line("hotkey " + combo, event="hotkey", combo=combo) if combo == "shift+0": clear_clipboard() notify_masked("stopped", show_exit_code=True) log_line("listener stopped by hotkey", event="listener_stop") break if combo == "shift+8": hotkeys[combo]["handler"]() break handler = hotkeys[combo]["handler"] if handler: handler() except KeyboardInterrupt: break except Exception as exc: log_line("listener error " + str(exc), event="listener_error") time.sleep(0.2) finally: for combo, item in hotkeys.items(): ungrab_keycodes(item["codes"], SHIFT_MASK) return 0 def main(): install_shortcuts, daemon_run, once, capture_mode, mode, cli_hint = parse_args(sys.argv[1:]) capture_session_env() if install_shortcuts: summary = install_shortcuts_bundle() print(summary) notify("Screen Answer", "Shortcut helpers installed") log_line("installed shortcuts bundle") return 0 hint = build_hint(mode, cli_hint or os.environ.get("SCREENANSWER_HINT", "").strip()) if IS_LINUX and not once and not daemon_run: start_listener_detached(mode, hint, delay=1) return 0 if IS_LINUX and not once: return listen_hotkeys(mode, hint) return run_once(capture_mode, hint) if __name__ == "__main__": raise SystemExit(main())