text expander

#!/usr/bin/env python3
import subprocess, time, threading, signal, sys
from Xlib import X, display
from Xlib.ext import record
from Xlib.protocol import rq

TRIGGERS = {
    ":email": "email.saya@domain.co",
    ":omw": "On my way!",
}

buf = ""
is_typing = False
lock = threading.Lock()
MAX_BUF = 15

FLUSH_KEYSYMS = {
    65293, 65289, 65307, # Enter, Tab, Esc
    65507, 65508, 65513, 65514, 65515, 65516,
    65362, 65364, 65361, 65363, # Arrows
}
SEPARATORS = set(" \n\r.,;()[]{}")

local_dpy = display.Display()
record_dpy = display.Display()

def type_text(text, backspaces):
    try:
        time.sleep(0.12)
        subprocess.run(["xdotool", "key", "--delay", "30"] + ["BackSpace"] * backspaces, check=False)
        time.sleep(0.12)
        subprocess.run(["xdotool", "type", "--clearmodifiers", "--delay", "25", "--", text], check=False)
        time.sleep(0.5)
    finally:
        global is_typing
        with lock:
            is_typing = False

def handler(reply):
    global buf, is_typing
    with lock:
        if is_typing:
            return
    if reply.category!= record.FromServer or reply.client_swapped:
        return
    data = reply.data
    if not len(data) or data[0] < 2:
        return
    while len(data):
        event, data = rq.EventField(None).parse_binary_value(data, record_dpy.display, None, None)
        if event.type!= X.KeyPress:
            continue
        if event.state & (X.ControlMask | X.Mod1Mask | X.Mod4Mask):
            with lock: buf = ""
            continue

        keysym = local_dpy.keycode_to_keysym(event.detail, 1 if (event.state & X.ShiftMask) else 0)

        if keysym in FLUSH_KEYSYMS:
            with lock: buf = ""
            continue
        if keysym in (65288, 65535): # BackSpace/Delete
            with lock: buf = buf[:-1]
            continue

        if 32 <= keysym <= 126:
            char = chr(keysym)
            with lock:
                buf = (buf + char)[-MAX_BUF:]

                for trig, repl in TRIGGERS.items():
                    if buf.endswith(trig):
                        is_typing = True
                        buf = ""
                        threading.Thread(target=type_text, args=(repl, len(trig)), daemon=True).start()
                        break
                else:
                    # Clear on separator - hanya jika buffer saat ini bukan prefix dari trigger manapun
                    if char in SEPARATORS:
                        if not any(trig.startswith(buf) for trig in TRIGGERS):
                            buf = ""

def record_loop(ctx):
    try:
        record_dpy.record_enable_context(ctx, handler)
    except Exception:
        pass

ctx = record_dpy.record_create_context(0, [record.AllClients], [{
    'core_requests': (0,0),'core_replies': (0,0),'ext_requests': (0,0,0,0),
    'ext_replies': (0,0,0,0),'delivered_events': (0,0),
    'device_events': (X.KeyPress, X.KeyPress),'errors': (0,0),
    'client_started': False,'client_died': False
}])

t = threading.Thread(target=record_loop, args=(ctx,), daemon=True)
t.start()

print("fast-expander final clean jalan...")

def clean_exit(signum, frame):
    try:
        record_dpy.record_free_context(ctx)
    except Exception:
        pass
    sys.exit(0)

signal.signal(signal.SIGINT, clean_exit)
signal.signal(signal.SIGTERM, clean_exit)

try:
    while t.is_alive():
        time.sleep(0.5)
except KeyboardInterrupt:
    clean_exit(None, None)