1106 lines
34 KiB
Python
Executable File
1106 lines
34 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""User-level GUI bridge for the adaptive WeChat automation service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import secrets
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
|
|
|
|
HOME = Path.home()
|
|
STATE_DIR = HOME / ".local" / "state" / "wechat-adaptive-gui-bridge"
|
|
PENDING_TEXT = STATE_DIR / "pending.txt"
|
|
PENDING_TOKEN = STATE_DIR / "pending.token"
|
|
PENDING_TTL_SECONDS = 120
|
|
|
|
DISPLAY = ":1"
|
|
XAUTHORITY = HOME / ".Xauthority"
|
|
XDG_RUNTIME_DIR = Path(f"/run/user/{os.getuid()}")
|
|
DBUS_ADDRESS = f"unix:path={XDG_RUNTIME_DIR}/bus"
|
|
|
|
BASE_WIDTH = 1440
|
|
BASE_HEIGHT = 873
|
|
BASE_DRAFT_POINT = (441, 786)
|
|
BASE_SEND_POINT = (1390, 801)
|
|
BASE_INPUT_BOX = (300, 732, 1439, 872)
|
|
BASE_INPUT_TEXT_ROI = (320, 748, 960, 824)
|
|
INPUT_TEXT_ROI_NAME = "input_text_core_v2"
|
|
PIXEL_CHANGE_THRESHOLD = 32
|
|
INK_PIXEL_THRESHOLD = 160
|
|
MAX_EMPTY_ROI_INK_PIXELS = 40
|
|
MIN_DRAFT_CHANGED_PIXELS = 80
|
|
MIN_SEND_CHANGED_PIXELS = 80
|
|
CLEAR_CONFIRM_TIMEOUT_SECONDS = 0.5
|
|
DRAFT_CONFIRM_TIMEOUT_SECONDS = 0.8
|
|
SEND_CONFIRM_TIMEOUT_SECONDS = 0.9
|
|
CONFIRM_POLL_SECONDS = 0.05
|
|
VNC_SETTLE_SECONDS = 0.05
|
|
VNC_HOST = "127.0.0.1"
|
|
VNC_PORT = 5901
|
|
|
|
|
|
class GuiBridgeError(RuntimeError):
|
|
def __init__(self, message: str, timings: list[dict[str, object]]) -> None:
|
|
super().__init__(message)
|
|
self.timings = timings
|
|
|
|
|
|
def elapsed_ms(started: float) -> float:
|
|
return round(max(0.0, (time.monotonic() - started) * 1000), 3)
|
|
|
|
|
|
def append_timing(
|
|
timings: list[dict[str, object]],
|
|
phase: str,
|
|
started: float,
|
|
**fields: object,
|
|
) -> None:
|
|
value: dict[str, object] = {
|
|
"phase": phase,
|
|
"elapsed_ms": elapsed_ms(started),
|
|
}
|
|
for key, item in fields.items():
|
|
if isinstance(item, (str, int, float, bool, type(None))):
|
|
value[key] = item
|
|
timings.append(value)
|
|
|
|
|
|
def environment() -> dict[str, str]:
|
|
return os.environ | {
|
|
"DISPLAY": DISPLAY,
|
|
"XAUTHORITY": str(XAUTHORITY),
|
|
"XDG_RUNTIME_DIR": str(XDG_RUNTIME_DIR),
|
|
"DBUS_SESSION_BUS_ADDRESS": DBUS_ADDRESS,
|
|
}
|
|
|
|
|
|
def run(command: list[str], *, input_text: str | None = None) -> str:
|
|
result = subprocess.run(
|
|
command,
|
|
input=input_text,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
timeout=20,
|
|
env=environment(),
|
|
)
|
|
if result.returncode != 0:
|
|
raise RuntimeError("WeChat GUI command failed")
|
|
return result.stdout
|
|
|
|
|
|
def ensure_state_dir() -> None:
|
|
STATE_DIR.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
os.chmod(STATE_DIR, 0o700)
|
|
|
|
|
|
def write_private(path: Path, value: str) -> None:
|
|
path.write_text(value, encoding="utf-8")
|
|
os.chmod(path, 0o600)
|
|
|
|
|
|
def window_id() -> str:
|
|
matches = run(
|
|
["/usr/bin/xdotool", "search", "--onlyvisible", "--class", "wechat"]
|
|
)
|
|
for candidate in matches.splitlines():
|
|
candidate = candidate.strip()
|
|
if not candidate:
|
|
continue
|
|
title = run(["/usr/bin/xdotool", "getwindowname", candidate]).strip()
|
|
if title == "Weixin":
|
|
return candidate
|
|
raise RuntimeError("visible main WeChat window not found")
|
|
|
|
|
|
def geometry(identifier: str) -> tuple[int, int]:
|
|
output = run(
|
|
["/usr/bin/xdotool", "getwindowgeometry", "--shell", identifier]
|
|
)
|
|
values = dict(
|
|
line.split("=", 1)
|
|
for line in output.splitlines()
|
|
if "=" in line
|
|
)
|
|
try:
|
|
return int(values["WIDTH"]), int(values["HEIGHT"])
|
|
except (KeyError, ValueError) as error:
|
|
raise RuntimeError("WeChat window geometry is unavailable") from error
|
|
|
|
|
|
def position(identifier: str) -> tuple[int, int]:
|
|
output = run(
|
|
["/usr/bin/xdotool", "getwindowgeometry", "--shell", identifier]
|
|
)
|
|
values = dict(
|
|
line.split("=", 1)
|
|
for line in output.splitlines()
|
|
if "=" in line
|
|
)
|
|
try:
|
|
return int(values["X"]), int(values["Y"])
|
|
except (KeyError, ValueError) as error:
|
|
raise RuntimeError("WeChat window position is unavailable") from error
|
|
|
|
|
|
def scaled_point(
|
|
point: tuple[int, int],
|
|
*,
|
|
width: int,
|
|
height: int,
|
|
) -> tuple[int, int]:
|
|
return (
|
|
round(point[0] * width / BASE_WIDTH),
|
|
round(point[1] * height / BASE_HEIGHT),
|
|
)
|
|
|
|
|
|
def scaled_box(
|
|
box: tuple[int, int, int, int],
|
|
*,
|
|
width: int,
|
|
height: int,
|
|
) -> tuple[int, int, int, int]:
|
|
left, top, right, bottom = box
|
|
return (
|
|
max(0, min(width - 1, round(left * width / BASE_WIDTH))),
|
|
max(0, min(height - 1, round(top * height / BASE_HEIGHT))),
|
|
max(1, min(width, round(right * width / BASE_WIDTH))),
|
|
max(1, min(height, round(bottom * height / BASE_HEIGHT))),
|
|
)
|
|
|
|
|
|
def receive(connection: socket.socket, size: int) -> bytes:
|
|
value = b""
|
|
while len(value) < size:
|
|
chunk = connection.recv(size - len(value))
|
|
if not chunk:
|
|
raise RuntimeError("VNC closed during keyboard handshake")
|
|
value += chunk
|
|
return value
|
|
|
|
|
|
def vnc_key_event(connection: socket.socket, *, down: bool, keysym: int) -> None:
|
|
connection.sendall(struct.pack(">BBHI", 4, int(down), 0, keysym))
|
|
|
|
|
|
def vnc_pointer_event(
|
|
connection: socket.socket,
|
|
*,
|
|
buttons: int,
|
|
x: int,
|
|
y: int,
|
|
) -> None:
|
|
connection.sendall(struct.pack(">BBHH", 5, buttons, x, y))
|
|
|
|
|
|
def vnc_key_press(connection: socket.socket, keysym: int) -> None:
|
|
vnc_key_event(connection, down=True, keysym=keysym)
|
|
vnc_key_event(connection, down=False, keysym=keysym)
|
|
|
|
|
|
def vnc_select_all_and_clear(connection: socket.socket) -> None:
|
|
vnc_key_event(connection, down=True, keysym=0xFFE3)
|
|
vnc_key_press(connection, ord("a"))
|
|
vnc_key_event(connection, down=False, keysym=0xFFE3)
|
|
vnc_key_press(connection, 0xFF08)
|
|
|
|
|
|
def vnc_connection() -> socket.socket:
|
|
connection = socket.create_connection((VNC_HOST, VNC_PORT), timeout=5)
|
|
connection.settimeout(5)
|
|
version = receive(connection, 12)
|
|
if not version.startswith(b"RFB "):
|
|
connection.close()
|
|
raise RuntimeError("VNC protocol is unavailable")
|
|
connection.sendall(b"RFB 003.008\n")
|
|
count = receive(connection, 1)[0]
|
|
security_types = receive(connection, count)
|
|
if 1 not in security_types:
|
|
connection.close()
|
|
raise RuntimeError("VNC no-auth input path is unavailable")
|
|
connection.sendall(b"\x01")
|
|
if struct.unpack(">I", receive(connection, 4))[0] != 0:
|
|
connection.close()
|
|
raise RuntimeError("VNC input authentication failed")
|
|
connection.sendall(b"\x01")
|
|
receive(connection, 4)
|
|
receive(connection, 16)
|
|
receive(connection, struct.unpack(">I", receive(connection, 4))[0])
|
|
return connection
|
|
|
|
|
|
def set_x_clipboard(text: str) -> None:
|
|
process = subprocess.Popen(
|
|
["/usr/bin/xclip", "-selection", "clipboard", "-in"],
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
text=True,
|
|
start_new_session=True,
|
|
env=environment(),
|
|
)
|
|
try:
|
|
process.communicate(text, timeout=5)
|
|
except subprocess.TimeoutExpired as error:
|
|
process.kill()
|
|
process.communicate()
|
|
raise RuntimeError("X11 clipboard operation timed out") from error
|
|
if process.returncode != 0:
|
|
raise RuntimeError("X11 clipboard operation failed")
|
|
|
|
deadline = time.monotonic() + 2
|
|
while time.monotonic() < deadline:
|
|
if run(["/usr/bin/xclip", "-selection", "clipboard", "-out"]) == text:
|
|
return
|
|
time.sleep(0.05)
|
|
raise RuntimeError("X11 clipboard owner did not become ready")
|
|
|
|
|
|
def vnc_focus_input(
|
|
identifier: str,
|
|
point: tuple[int, int],
|
|
*,
|
|
origin: tuple[int, int] | None = None,
|
|
) -> socket.socket:
|
|
origin_x, origin_y = origin or position(identifier)
|
|
connection = vnc_connection()
|
|
vnc_pointer_event(
|
|
connection,
|
|
buttons=1,
|
|
x=origin_x + point[0],
|
|
y=origin_y + point[1],
|
|
)
|
|
vnc_pointer_event(
|
|
connection,
|
|
buttons=0,
|
|
x=origin_x + point[0],
|
|
y=origin_y + point[1],
|
|
)
|
|
time.sleep(VNC_SETTLE_SECONDS)
|
|
return connection
|
|
|
|
|
|
def scaled_region(
|
|
box: tuple[int, int, int, int],
|
|
*,
|
|
width: int,
|
|
height: int,
|
|
) -> tuple[int, int, int, int]:
|
|
left, top, right, bottom = scaled_box(box, width=width, height=height)
|
|
return left, top, max(1, right - left), max(1, bottom - top)
|
|
|
|
|
|
def vnc_paste(connection: socket.socket, text: str) -> None:
|
|
if len(text.encode("utf-8")) > 65536:
|
|
raise RuntimeError("draft text exceeds VNC clipboard limit")
|
|
set_x_clipboard(text)
|
|
vnc_key_event(connection, down=True, keysym=0xFFE3)
|
|
vnc_key_press(connection, ord("v"))
|
|
vnc_key_event(connection, down=False, keysym=0xFFE3)
|
|
time.sleep(VNC_SETTLE_SECONDS)
|
|
|
|
|
|
def vnc_click(*, x: int, y: int) -> None:
|
|
if x < 0 or y < 0 or x > 0xFFFF or y > 0xFFFF:
|
|
raise RuntimeError("VNC click coordinates are out of range")
|
|
connection = vnc_connection()
|
|
try:
|
|
vnc_pointer_event(connection, buttons=1, x=x, y=y)
|
|
vnc_pointer_event(connection, buttons=0, x=x, y=y)
|
|
time.sleep(VNC_SETTLE_SECONDS)
|
|
finally:
|
|
connection.close()
|
|
|
|
|
|
def input_image(
|
|
identifier: str,
|
|
*,
|
|
width: int,
|
|
height: int,
|
|
box: tuple[int, int, int, int] = BASE_INPUT_BOX,
|
|
origin: tuple[int, int] | None = None,
|
|
):
|
|
from PIL import Image
|
|
|
|
descriptor, raw_path = tempfile.mkstemp(prefix="wechat-input-", suffix=".png")
|
|
os.close(descriptor)
|
|
path = Path(raw_path)
|
|
os.chmod(path, 0o600)
|
|
left, top, region_width, region_height = scaled_region(
|
|
box,
|
|
width=width,
|
|
height=height,
|
|
)
|
|
try:
|
|
if os.environ.get("JWEBOT_GUI_REGION_SCROT", "1").strip() != "0":
|
|
try:
|
|
origin_x, origin_y = origin or position(identifier)
|
|
run(
|
|
[
|
|
"/usr/bin/scrot",
|
|
"--silent",
|
|
"--overwrite",
|
|
"--autoselect",
|
|
(
|
|
f"{origin_x + left},{origin_y + top},"
|
|
f"{region_width},{region_height}"
|
|
),
|
|
"--format",
|
|
"png",
|
|
"--file",
|
|
str(path),
|
|
]
|
|
)
|
|
with Image.open(path) as source:
|
|
image = source.convert("RGB")
|
|
image.load()
|
|
return image
|
|
except (OSError, RuntimeError):
|
|
pass
|
|
run(
|
|
[
|
|
"/usr/bin/scrot",
|
|
"--silent",
|
|
"--overwrite",
|
|
"--window",
|
|
identifier,
|
|
"--format",
|
|
"png",
|
|
"--file",
|
|
str(path),
|
|
]
|
|
)
|
|
with Image.open(path) as source:
|
|
image = source.convert("RGB")
|
|
image.load()
|
|
return image.crop((left, top, left + region_width, top + region_height))
|
|
finally:
|
|
path.unlink(missing_ok=True)
|
|
|
|
|
|
def changed_pixels(before, after) -> int:
|
|
from PIL import ImageChops
|
|
|
|
if before.size != after.size:
|
|
raise RuntimeError("input geometry changed during draft")
|
|
difference = ImageChops.difference(before, after)
|
|
return sum(
|
|
int(max(pixel) > PIXEL_CHANGE_THRESHOLD)
|
|
for pixel in difference.getdata()
|
|
)
|
|
|
|
|
|
def ink_pixels(image) -> int:
|
|
return sum(
|
|
int(min(pixel) < INK_PIXEL_THRESHOLD)
|
|
for pixel in image.getdata()
|
|
)
|
|
|
|
|
|
def wait_for_input_empty(
|
|
identifier: str,
|
|
*,
|
|
width: int,
|
|
height: int,
|
|
timeout_seconds: float = CLEAR_CONFIRM_TIMEOUT_SECONDS,
|
|
box: tuple[int, int, int, int] = BASE_INPUT_TEXT_ROI,
|
|
origin: tuple[int, int] | None = None,
|
|
) -> tuple[object, int, int, float]:
|
|
started = time.monotonic()
|
|
deadline = time.monotonic() + timeout_seconds
|
|
last = input_image(
|
|
identifier,
|
|
width=width,
|
|
height=height,
|
|
box=box,
|
|
origin=origin,
|
|
)
|
|
ink = ink_pixels(last)
|
|
polls = 1
|
|
while ink > MAX_EMPTY_ROI_INK_PIXELS and time.monotonic() < deadline:
|
|
time.sleep(CONFIRM_POLL_SECONDS)
|
|
polls += 1
|
|
last = input_image(
|
|
identifier,
|
|
width=width,
|
|
height=height,
|
|
box=box,
|
|
origin=origin,
|
|
)
|
|
ink = ink_pixels(last)
|
|
return last, ink, polls, elapsed_ms(started)
|
|
|
|
|
|
def wait_for_input_change(
|
|
identifier: str,
|
|
before,
|
|
*,
|
|
width: int,
|
|
height: int,
|
|
threshold: int,
|
|
timeout_seconds: float,
|
|
box: tuple[int, int, int, int] = BASE_INPUT_TEXT_ROI,
|
|
origin: tuple[int, int] | None = None,
|
|
) -> tuple[object, int, int, float]:
|
|
started = time.monotonic()
|
|
deadline = time.monotonic() + timeout_seconds
|
|
last = before
|
|
changed = 0
|
|
polls = 0
|
|
while True:
|
|
polls += 1
|
|
last = input_image(
|
|
identifier,
|
|
width=width,
|
|
height=height,
|
|
box=box,
|
|
origin=origin,
|
|
)
|
|
changed = changed_pixels(before, last)
|
|
if changed >= threshold or time.monotonic() >= deadline:
|
|
return last, changed, polls, elapsed_ms(started)
|
|
time.sleep(CONFIRM_POLL_SECONDS)
|
|
|
|
|
|
def roi_timing_fields(
|
|
*,
|
|
width: int,
|
|
height: int,
|
|
box: tuple[int, int, int, int] = BASE_INPUT_TEXT_ROI,
|
|
) -> dict[str, object]:
|
|
x, y, roi_width, roi_height = scaled_region(box, width=width, height=height)
|
|
return {
|
|
"method": "scrot-roi",
|
|
"roi": INPUT_TEXT_ROI_NAME,
|
|
"x": x,
|
|
"y": y,
|
|
"width": roi_width,
|
|
"height": roi_height,
|
|
}
|
|
|
|
|
|
def clear_pending() -> None:
|
|
PENDING_TOKEN.unlink(missing_ok=True)
|
|
PENDING_TEXT.unlink(missing_ok=True)
|
|
|
|
|
|
def click_existing_connection(
|
|
connection: socket.socket,
|
|
*,
|
|
origin: tuple[int, int],
|
|
point: tuple[int, int],
|
|
) -> None:
|
|
x = origin[0] + point[0]
|
|
y = origin[1] + point[1]
|
|
if x < 0 or y < 0 or x > 0xFFFF or y > 0xFFFF:
|
|
raise RuntimeError("VNC click coordinates are out of range")
|
|
vnc_pointer_event(connection, buttons=1, x=x, y=y)
|
|
vnc_pointer_event(connection, buttons=0, x=x, y=y)
|
|
time.sleep(VNC_SETTLE_SECONDS)
|
|
|
|
|
|
def deliver_text(
|
|
text: str,
|
|
*,
|
|
before_send: Callable[[], None] | None = None,
|
|
) -> dict[str, object]:
|
|
timings: list[dict[str, object]] = []
|
|
total_started = time.monotonic()
|
|
if not text:
|
|
raise RuntimeError("delivery text is empty")
|
|
if len(text) > 900:
|
|
raise RuntimeError("delivery text exceeds the allowed length")
|
|
|
|
step_started = time.monotonic()
|
|
identifier = window_id()
|
|
width, height = geometry(identifier)
|
|
origin = position(identifier)
|
|
input_point = scaled_point(BASE_DRAFT_POINT, width=width, height=height)
|
|
send_point = scaled_point(BASE_SEND_POINT, width=width, height=height)
|
|
append_timing(
|
|
timings,
|
|
"deliver.window",
|
|
step_started,
|
|
width=width,
|
|
height=height,
|
|
x=input_point[0],
|
|
y=input_point[1],
|
|
)
|
|
|
|
step_started = time.monotonic()
|
|
connection = vnc_focus_input(identifier, input_point, origin=origin)
|
|
append_timing(
|
|
timings,
|
|
"deliver.focus_input",
|
|
step_started,
|
|
method="vnc-click",
|
|
x=input_point[0],
|
|
y=input_point[1],
|
|
settle_ms=round(VNC_SETTLE_SECONDS * 1000),
|
|
)
|
|
try:
|
|
step_started = time.monotonic()
|
|
vnc_select_all_and_clear(connection)
|
|
time.sleep(0.05)
|
|
append_timing(
|
|
timings,
|
|
"deliver.clear_existing",
|
|
step_started,
|
|
method="ctrl-a-backspace",
|
|
settle_ms=50,
|
|
)
|
|
|
|
step_started = time.monotonic()
|
|
before, ink, polls, poll_ms = wait_for_input_empty(
|
|
identifier,
|
|
width=width,
|
|
height=height,
|
|
origin=origin,
|
|
)
|
|
clear_confirmed = ink <= MAX_EMPTY_ROI_INK_PIXELS
|
|
timings.append(
|
|
{
|
|
"phase": "deliver.clear_confirm_poll",
|
|
"elapsed_ms": poll_ms,
|
|
"polls": polls,
|
|
"pixels": ink,
|
|
"threshold": MAX_EMPTY_ROI_INK_PIXELS,
|
|
"timeout_ms": round(CLEAR_CONFIRM_TIMEOUT_SECONDS * 1000),
|
|
"outcome": "confirmed" if clear_confirmed else "dirty",
|
|
"flag": clear_confirmed,
|
|
"flag_wait_ms": poll_ms,
|
|
"confirmed_by": "input_roi_empty" if clear_confirmed else "dirty",
|
|
**roi_timing_fields(width=width, height=height),
|
|
}
|
|
)
|
|
if not clear_confirmed:
|
|
raise GuiBridgeError("input box could not be cleared", timings)
|
|
|
|
step_started = time.monotonic()
|
|
vnc_paste(connection, text)
|
|
append_timing(
|
|
timings,
|
|
"deliver.clipboard_paste",
|
|
step_started,
|
|
method="xclip-ctrl-v",
|
|
settle_ms=round(VNC_SETTLE_SECONDS * 1000),
|
|
)
|
|
|
|
_, changed, polls, poll_ms = wait_for_input_change(
|
|
identifier,
|
|
before,
|
|
width=width,
|
|
height=height,
|
|
threshold=MIN_DRAFT_CHANGED_PIXELS,
|
|
timeout_seconds=DRAFT_CONFIRM_TIMEOUT_SECONDS,
|
|
box=BASE_INPUT_TEXT_ROI,
|
|
origin=origin,
|
|
)
|
|
draft_confirmed = changed >= MIN_DRAFT_CHANGED_PIXELS
|
|
timings.append(
|
|
{
|
|
"phase": "deliver.draft_confirm_poll",
|
|
"elapsed_ms": poll_ms,
|
|
"polls": polls,
|
|
"pixels": changed,
|
|
"threshold": MIN_DRAFT_CHANGED_PIXELS,
|
|
"timeout_ms": round(DRAFT_CONFIRM_TIMEOUT_SECONDS * 1000),
|
|
"outcome": "confirmed" if draft_confirmed else "timeout",
|
|
"flag": draft_confirmed,
|
|
"flag_wait_ms": poll_ms,
|
|
"confirmed_by": (
|
|
"input_roi_changed" if draft_confirmed else "timeout"
|
|
),
|
|
**roi_timing_fields(width=width, height=height),
|
|
}
|
|
)
|
|
if not draft_confirmed:
|
|
step_started = time.monotonic()
|
|
vnc_select_all_and_clear(connection)
|
|
append_timing(
|
|
timings,
|
|
"deliver.failed_cleanup",
|
|
step_started,
|
|
method="ctrl-a-backspace",
|
|
)
|
|
raise GuiBridgeError("draft paste could not be confirmed", timings)
|
|
|
|
if before_send is not None:
|
|
step_started = time.monotonic()
|
|
try:
|
|
before_send()
|
|
except Exception:
|
|
cleanup_started = time.monotonic()
|
|
vnc_select_all_and_clear(connection)
|
|
append_timing(
|
|
timings,
|
|
"deliver.callback_cleanup",
|
|
cleanup_started,
|
|
method="ctrl-a-backspace",
|
|
)
|
|
raise GuiBridgeError("before-send callback failed", timings)
|
|
finally:
|
|
append_timing(timings, "deliver.before_send_callback", step_started)
|
|
|
|
step_started = time.monotonic()
|
|
before_send_image = input_image(
|
|
identifier,
|
|
width=width,
|
|
height=height,
|
|
box=BASE_INPUT_TEXT_ROI,
|
|
origin=origin,
|
|
)
|
|
append_timing(
|
|
timings,
|
|
"deliver.snapshot_before_return",
|
|
step_started,
|
|
**roi_timing_fields(width=width, height=height),
|
|
)
|
|
|
|
step_started = time.monotonic()
|
|
vnc_key_press(connection, 0xFF0D)
|
|
append_timing(
|
|
timings,
|
|
"deliver.return_key",
|
|
step_started,
|
|
method="vnc-enter",
|
|
x=input_point[0],
|
|
y=input_point[1],
|
|
settle_ms=round(VNC_SETTLE_SECONDS * 1000),
|
|
)
|
|
_, changed, polls, poll_ms = wait_for_input_change(
|
|
identifier,
|
|
before_send_image,
|
|
width=width,
|
|
height=height,
|
|
threshold=MIN_SEND_CHANGED_PIXELS,
|
|
timeout_seconds=SEND_CONFIRM_TIMEOUT_SECONDS,
|
|
box=BASE_INPUT_TEXT_ROI,
|
|
origin=origin,
|
|
)
|
|
return_confirmed = changed >= MIN_SEND_CHANGED_PIXELS
|
|
timings.append(
|
|
{
|
|
"phase": "deliver.return_confirm_poll",
|
|
"elapsed_ms": poll_ms,
|
|
"polls": polls,
|
|
"pixels": changed,
|
|
"threshold": MIN_SEND_CHANGED_PIXELS,
|
|
"timeout_ms": round(SEND_CONFIRM_TIMEOUT_SECONDS * 1000),
|
|
"outcome": "confirmed" if return_confirmed else "fallback",
|
|
"flag": return_confirmed,
|
|
"flag_wait_ms": poll_ms,
|
|
"confirmed_by": (
|
|
"input_roi_cleared" if return_confirmed else "fallback"
|
|
),
|
|
**roi_timing_fields(width=width, height=height),
|
|
}
|
|
)
|
|
|
|
if not return_confirmed:
|
|
step_started = time.monotonic()
|
|
before_button = input_image(
|
|
identifier,
|
|
width=width,
|
|
height=height,
|
|
box=BASE_INPUT_TEXT_ROI,
|
|
origin=origin,
|
|
)
|
|
append_timing(
|
|
timings,
|
|
"deliver.snapshot_before_button",
|
|
step_started,
|
|
**roi_timing_fields(width=width, height=height),
|
|
)
|
|
|
|
step_started = time.monotonic()
|
|
click_existing_connection(
|
|
connection,
|
|
origin=origin,
|
|
point=send_point,
|
|
)
|
|
append_timing(
|
|
timings,
|
|
"deliver.button_click",
|
|
step_started,
|
|
method="vnc-click",
|
|
x=send_point[0],
|
|
y=send_point[1],
|
|
settle_ms=round(VNC_SETTLE_SECONDS * 1000),
|
|
)
|
|
_, changed, polls, poll_ms = wait_for_input_change(
|
|
identifier,
|
|
before_button,
|
|
width=width,
|
|
height=height,
|
|
threshold=MIN_SEND_CHANGED_PIXELS,
|
|
timeout_seconds=SEND_CONFIRM_TIMEOUT_SECONDS,
|
|
box=BASE_INPUT_TEXT_ROI,
|
|
origin=origin,
|
|
)
|
|
button_confirmed = changed >= MIN_SEND_CHANGED_PIXELS
|
|
timings.append(
|
|
{
|
|
"phase": "deliver.button_confirm_poll",
|
|
"elapsed_ms": poll_ms,
|
|
"polls": polls,
|
|
"pixels": changed,
|
|
"threshold": MIN_SEND_CHANGED_PIXELS,
|
|
"timeout_ms": round(SEND_CONFIRM_TIMEOUT_SECONDS * 1000),
|
|
"outcome": "confirmed" if button_confirmed else "timeout",
|
|
"flag": button_confirmed,
|
|
"flag_wait_ms": poll_ms,
|
|
"confirmed_by": (
|
|
"input_roi_cleared" if button_confirmed else "timeout"
|
|
),
|
|
**roi_timing_fields(width=width, height=height),
|
|
}
|
|
)
|
|
if changed < MIN_SEND_CHANGED_PIXELS:
|
|
raise GuiBridgeError("send action could not be visually confirmed", timings)
|
|
finally:
|
|
connection.close()
|
|
|
|
append_timing(timings, "deliver.total", total_started)
|
|
return {"ok": True, "state": "sent", "timings": timings}
|
|
|
|
|
|
def draft(text: str) -> None:
|
|
timings: list[dict[str, object]] = []
|
|
total_started = time.monotonic()
|
|
if not text:
|
|
raise RuntimeError("draft text is empty")
|
|
if len(text) > 900:
|
|
raise RuntimeError("draft text exceeds the allowed length")
|
|
|
|
step_started = time.monotonic()
|
|
ensure_state_dir()
|
|
append_timing(timings, "draft.state_dir", step_started)
|
|
step_started = time.monotonic()
|
|
identifier = window_id()
|
|
width, height = geometry(identifier)
|
|
origin = position(identifier)
|
|
point = scaled_point(BASE_DRAFT_POINT, width=width, height=height)
|
|
append_timing(
|
|
timings,
|
|
"draft.window",
|
|
step_started,
|
|
width=width,
|
|
height=height,
|
|
x=point[0],
|
|
y=point[1],
|
|
)
|
|
step_started = time.monotonic()
|
|
connection = vnc_focus_input(identifier, point, origin=origin)
|
|
append_timing(
|
|
timings,
|
|
"draft.focus_input",
|
|
step_started,
|
|
method="vnc-click",
|
|
x=point[0],
|
|
y=point[1],
|
|
settle_ms=round(VNC_SETTLE_SECONDS * 1000),
|
|
)
|
|
try:
|
|
step_started = time.monotonic()
|
|
vnc_select_all_and_clear(connection)
|
|
time.sleep(0.05)
|
|
append_timing(
|
|
timings,
|
|
"draft.clear_existing",
|
|
step_started,
|
|
method="ctrl-a-backspace",
|
|
settle_ms=50,
|
|
)
|
|
step_started = time.monotonic()
|
|
before, ink, polls, poll_ms = wait_for_input_empty(
|
|
identifier,
|
|
width=width,
|
|
height=height,
|
|
origin=origin,
|
|
)
|
|
clear_confirmed = ink <= MAX_EMPTY_ROI_INK_PIXELS
|
|
timings.append(
|
|
{
|
|
"phase": "draft.clear_confirm_poll",
|
|
"elapsed_ms": poll_ms,
|
|
"polls": polls,
|
|
"pixels": ink,
|
|
"threshold": MAX_EMPTY_ROI_INK_PIXELS,
|
|
"timeout_ms": round(CLEAR_CONFIRM_TIMEOUT_SECONDS * 1000),
|
|
"outcome": "confirmed" if clear_confirmed else "dirty",
|
|
"flag": clear_confirmed,
|
|
"flag_wait_ms": poll_ms,
|
|
"confirmed_by": "input_roi_empty" if clear_confirmed else "dirty",
|
|
**roi_timing_fields(width=width, height=height),
|
|
}
|
|
)
|
|
if not clear_confirmed:
|
|
raise RuntimeError("input box could not be cleared")
|
|
step_started = time.monotonic()
|
|
vnc_paste(connection, text)
|
|
append_timing(
|
|
timings,
|
|
"draft.clipboard_paste",
|
|
step_started,
|
|
method="xclip-ctrl-v",
|
|
settle_ms=round(VNC_SETTLE_SECONDS * 1000),
|
|
)
|
|
finally:
|
|
connection.close()
|
|
_, changed, polls, poll_ms = wait_for_input_change(
|
|
identifier,
|
|
before,
|
|
width=width,
|
|
height=height,
|
|
threshold=MIN_DRAFT_CHANGED_PIXELS,
|
|
timeout_seconds=DRAFT_CONFIRM_TIMEOUT_SECONDS,
|
|
box=BASE_INPUT_TEXT_ROI,
|
|
origin=origin,
|
|
)
|
|
draft_confirmed = changed >= MIN_DRAFT_CHANGED_PIXELS
|
|
timings.append(
|
|
{
|
|
"phase": "draft.confirm_poll",
|
|
"elapsed_ms": poll_ms,
|
|
"polls": polls,
|
|
"pixels": changed,
|
|
"threshold": MIN_DRAFT_CHANGED_PIXELS,
|
|
"timeout_ms": round(DRAFT_CONFIRM_TIMEOUT_SECONDS * 1000),
|
|
"outcome": "confirmed" if draft_confirmed else "timeout",
|
|
"flag": draft_confirmed,
|
|
"flag_wait_ms": poll_ms,
|
|
"confirmed_by": "input_roi_changed" if draft_confirmed else "timeout",
|
|
**roi_timing_fields(width=width, height=height),
|
|
}
|
|
)
|
|
if changed < MIN_DRAFT_CHANGED_PIXELS:
|
|
step_started = time.monotonic()
|
|
failed_connection = vnc_focus_input(identifier, point, origin=origin)
|
|
try:
|
|
vnc_select_all_and_clear(failed_connection)
|
|
finally:
|
|
failed_connection.close()
|
|
append_timing(
|
|
timings,
|
|
"draft.failed_cleanup",
|
|
step_started,
|
|
method="ctrl-a-backspace",
|
|
)
|
|
raise RuntimeError("draft paste could not be confirmed")
|
|
|
|
step_started = time.monotonic()
|
|
write_private(PENDING_TEXT, text)
|
|
token = secrets.token_hex(16)
|
|
write_private(PENDING_TOKEN, token + "\n")
|
|
append_timing(timings, "draft.pending_write", step_started)
|
|
append_timing(timings, "draft.total", total_started)
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"ok": True,
|
|
"state": "drafted",
|
|
"confirmation_token": token,
|
|
"expires_in_seconds": PENDING_TTL_SECONDS,
|
|
"timings": timings,
|
|
}
|
|
)
|
|
)
|
|
|
|
|
|
def send(supplied_token: str) -> None:
|
|
timings: list[dict[str, object]] = []
|
|
total_started = time.monotonic()
|
|
step_started = time.monotonic()
|
|
if not supplied_token or not PENDING_TOKEN.is_file() or not PENDING_TEXT.is_file():
|
|
raise RuntimeError("no armed draft")
|
|
expected = PENDING_TOKEN.read_text(encoding="utf-8").strip()
|
|
if not secrets.compare_digest(supplied_token, expected):
|
|
raise RuntimeError("confirmation token mismatch")
|
|
if time.time() - PENDING_TOKEN.stat().st_mtime > PENDING_TTL_SECONDS:
|
|
clear_pending()
|
|
raise RuntimeError("confirmation token expired")
|
|
append_timing(timings, "send.token_check", step_started)
|
|
|
|
step_started = time.monotonic()
|
|
identifier = window_id()
|
|
width, height = geometry(identifier)
|
|
origin = position(identifier)
|
|
append_timing(
|
|
timings,
|
|
"send.window",
|
|
step_started,
|
|
width=width,
|
|
height=height,
|
|
)
|
|
step_started = time.monotonic()
|
|
before = input_image(
|
|
identifier,
|
|
width=width,
|
|
height=height,
|
|
box=BASE_INPUT_TEXT_ROI,
|
|
origin=origin,
|
|
)
|
|
append_timing(
|
|
timings,
|
|
"send.snapshot_before_return",
|
|
step_started,
|
|
**roi_timing_fields(width=width, height=height),
|
|
)
|
|
send_point = scaled_point(BASE_SEND_POINT, width=width, height=height)
|
|
input_point = scaled_point(BASE_DRAFT_POINT, width=width, height=height)
|
|
step_started = time.monotonic()
|
|
connection = vnc_focus_input(identifier, input_point, origin=origin)
|
|
try:
|
|
vnc_key_press(connection, 0xFF0D)
|
|
finally:
|
|
connection.close()
|
|
append_timing(
|
|
timings,
|
|
"send.return_key",
|
|
step_started,
|
|
method="vnc-enter",
|
|
x=input_point[0],
|
|
y=input_point[1],
|
|
settle_ms=round(VNC_SETTLE_SECONDS * 1000),
|
|
)
|
|
_, changed, polls, poll_ms = wait_for_input_change(
|
|
identifier,
|
|
before,
|
|
width=width,
|
|
height=height,
|
|
threshold=MIN_SEND_CHANGED_PIXELS,
|
|
timeout_seconds=SEND_CONFIRM_TIMEOUT_SECONDS,
|
|
box=BASE_INPUT_TEXT_ROI,
|
|
origin=origin,
|
|
)
|
|
return_confirmed = changed >= MIN_SEND_CHANGED_PIXELS
|
|
timings.append(
|
|
{
|
|
"phase": "send.return_confirm_poll",
|
|
"elapsed_ms": poll_ms,
|
|
"polls": polls,
|
|
"pixels": changed,
|
|
"threshold": MIN_SEND_CHANGED_PIXELS,
|
|
"timeout_ms": round(SEND_CONFIRM_TIMEOUT_SECONDS * 1000),
|
|
"outcome": "confirmed" if return_confirmed else "fallback",
|
|
"flag": return_confirmed,
|
|
"flag_wait_ms": poll_ms,
|
|
"confirmed_by": "input_roi_cleared" if return_confirmed else "fallback",
|
|
**roi_timing_fields(width=width, height=height),
|
|
}
|
|
)
|
|
if changed < MIN_SEND_CHANGED_PIXELS:
|
|
step_started = time.monotonic()
|
|
before = input_image(
|
|
identifier,
|
|
width=width,
|
|
height=height,
|
|
box=BASE_INPUT_TEXT_ROI,
|
|
origin=origin,
|
|
)
|
|
append_timing(
|
|
timings,
|
|
"send.snapshot_before_button",
|
|
step_started,
|
|
**roi_timing_fields(width=width, height=height),
|
|
)
|
|
step_started = time.monotonic()
|
|
connection = vnc_focus_input(identifier, send_point, origin=origin)
|
|
try:
|
|
pass
|
|
finally:
|
|
connection.close()
|
|
append_timing(
|
|
timings,
|
|
"send.button_click",
|
|
step_started,
|
|
method="vnc-click",
|
|
x=send_point[0],
|
|
y=send_point[1],
|
|
settle_ms=round(VNC_SETTLE_SECONDS * 1000),
|
|
)
|
|
_, changed, polls, poll_ms = wait_for_input_change(
|
|
identifier,
|
|
before,
|
|
width=width,
|
|
height=height,
|
|
threshold=MIN_SEND_CHANGED_PIXELS,
|
|
timeout_seconds=SEND_CONFIRM_TIMEOUT_SECONDS,
|
|
box=BASE_INPUT_TEXT_ROI,
|
|
origin=origin,
|
|
)
|
|
button_confirmed = changed >= MIN_SEND_CHANGED_PIXELS
|
|
timings.append(
|
|
{
|
|
"phase": "send.button_confirm_poll",
|
|
"elapsed_ms": poll_ms,
|
|
"polls": polls,
|
|
"pixels": changed,
|
|
"threshold": MIN_SEND_CHANGED_PIXELS,
|
|
"timeout_ms": round(SEND_CONFIRM_TIMEOUT_SECONDS * 1000),
|
|
"outcome": "confirmed" if button_confirmed else "timeout",
|
|
"flag": button_confirmed,
|
|
"flag_wait_ms": poll_ms,
|
|
"confirmed_by": "input_roi_cleared"
|
|
if button_confirmed
|
|
else "timeout",
|
|
**roi_timing_fields(width=width, height=height),
|
|
}
|
|
)
|
|
if changed < MIN_SEND_CHANGED_PIXELS:
|
|
raise RuntimeError("send action could not be visually confirmed")
|
|
step_started = time.monotonic()
|
|
clear_pending()
|
|
append_timing(timings, "send.clear_pending", step_started)
|
|
append_timing(timings, "send.total", total_started)
|
|
print(json.dumps({"ok": True, "state": "sent", "timings": timings}))
|
|
|
|
|
|
def clear() -> None:
|
|
identifier = window_id()
|
|
width, height = geometry(identifier)
|
|
connection = vnc_focus_input(
|
|
identifier,
|
|
scaled_point(BASE_DRAFT_POINT, width=width, height=height),
|
|
)
|
|
try:
|
|
vnc_select_all_and_clear(connection)
|
|
finally:
|
|
connection.close()
|
|
clear_pending()
|
|
print(json.dumps({"ok": True, "state": "cleared"}))
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
commands = parser.add_subparsers(dest="command", required=True)
|
|
draft_parser = commands.add_parser("draft")
|
|
draft_parser.add_argument("text")
|
|
send_parser = commands.add_parser("send")
|
|
send_parser.add_argument("token")
|
|
commands.add_parser("clear")
|
|
args = parser.parse_args()
|
|
|
|
if args.command == "draft":
|
|
draft(args.text)
|
|
elif args.command == "send":
|
|
send(args.token)
|
|
else:
|
|
clear()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|