LunaTranslator websocket server with 0 effort. put in translation optimization custom python code

myprocess.py

import base64
import ctypes
import ctypes.wintypes
import hashlib
import json
import socket
import struct
import sys
import threading

import gobject

MAGIC = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"


class WSClient:
    def __init__(self, conn, addr):
        self.conn = conn
        self.addr = addr

    def send(self, message: str):
        data = message.encode("utf-8")
        length = len(data)
        if length <= 125:
            header = struct.pack("!BB", 0x81, length)
        elif length <= 65535:
            header = struct.pack("!BBH", 0x81, 126, length)
        else:
            header = struct.pack("!BBQ", 0x81, 127, length)
        try:
            self.conn.sendall(header + data)
        except Exception:
            pass


class WSServer:
    def __init__(self, host="0.0.0.0", port=9001):
        self.host = host
        self.port = port
        self.clients: list[WSClient] = []
        self._lock = threading.Lock()
        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self._sock.bind((host, port))
        self._sock.listen(5)

    def _handshake(self, conn) -> bool:
        data = b""
        while b"\r\n\r\n" not in data:
            chunk = conn.recv(1024)
            if not chunk:
                return False
            data += chunk

        headers = {}
        lines = data.decode("utf-8", errors="ignore").split("\r\n")
        for line in lines[1:]:
            if ": " in line:
                k, v = line.split(": ", 1)
                headers[k.lower()] = v

        key = headers.get("sec-websocket-key", "")
        accept = base64.b64encode(
            hashlib.sha1((key + MAGIC).encode()).digest()
        ).decode()

        response = (
            "HTTP/1.1 101 Switching Protocols\r\n"
            "Upgrade: websocket\r\n"
            "Connection: Upgrade\r\n"
            f"Sec-WebSocket-Accept: {accept}\r\n\r\n"
        )
        conn.sendall(response.encode())
        return True

    def _handle_client(self, conn, addr):
        if not self._handshake(conn):
            conn.close()
            return

        client = WSClient(conn, addr)
        with self._lock:
            self.clients.append(client)
        print(f"WSServer: client connected {addr}")

        try:
            while True:
                conn.recv(1024)
        except Exception:
            pass
        finally:
            with self._lock:
                self.clients.remove(client)
            print(f"WSServer: client disconnected {addr}")
            conn.close()

    def broadcast(self, message: str):
        with self._lock:
            targets = list(self.clients)
        for client in targets:
            client.send(message)

    def run_forever(self):
        print(f"WSServer: listening on {self.host}:{self.port}")
        while True:
            try:
                conn, addr = self._sock.accept()
                t = threading.Thread(
                    target=self._handle_client, args=(conn, addr), daemon=True
                )
                t.start()
            except Exception:
                break


# Keep server alive across module reloads by stashing in sys.modules
if "_ws_server_instance" not in sys.modules:
    _srv = WSServer(host="0.0.0.0", port=9001)
    t = threading.Thread(target=_srv.run_forever, daemon=True)
    t.start()
    sys.modules["_ws_server_instance"] = _srv

_server: WSServer = sys.modules["_ws_server_instance"]


def _pid_to_path(pid) -> str:
    PROCESS_QUERY_INFORMATION = 0x0400
    PROCESS_VM_READ = 0x0010
    handle = ctypes.windll.kernel32.OpenProcess(
        PROCESS_QUERY_INFORMATION | PROCESS_VM_READ, False, pid
    )
    if not handle:
        return ""
    try:
        buf = ctypes.create_unicode_buffer(512)
        size = ctypes.wintypes.DWORD(512)
        ctypes.windll.kernel32.QueryFullProcessImageNameW(
            handle, 0, buf, ctypes.byref(size)
        )
        return buf.value
    finally:
        ctypes.windll.kernel32.CloseHandle(handle)


class Process:
    def process_before(self, text: str):
        hwnd = gobject.base.hwnd
        pid = None
        if hwnd:
            _pid = ctypes.wintypes.DWORD()
            ctypes.windll.user32.GetWindowThreadProcessId(hwnd, ctypes.byref(_pid))
            pid = _pid.value

        process_path = _pid_to_path(pid) if pid else ""

        payload = json.dumps({"process_path": process_path, "sentence": text})
        _server.broadcast(payload)

        return text, {}

    def process_after(self, res: str, context):
        return res