diff --git a/app.py b/app.py index 72d0711..140c18c 100644 --- a/app.py +++ b/app.py @@ -2,6 +2,7 @@ import os import select import sys import time +import unicodedata from collections import deque from importlib import import_module from pathlib import Path @@ -53,7 +54,9 @@ TOOL_REGISTRY = { "module": "intelligent_tool", "class": "IntelligentCommunicationTool", "kwargs": { - "model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-small") + "model": os.environ.get( + "MISTLE_LLM_MODEL", "mistral/mistral-small" + ) }, "description": "Uses an LLM to craft a polite reply to private tells.", }, @@ -61,7 +64,9 @@ TOOL_REGISTRY = { "module": "intelligent_tool", "class": "IntelligentCommunicationTool", "kwargs": { - "model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-small") + "model": os.environ.get( + "MISTLE_LLM_MODEL", "mistral/mistral-small" + ) }, "description": "Alias of 'intelligent'. Uses an LLM to craft a polite reply to private tells.", }, @@ -74,6 +79,38 @@ TOOL_DESCRIPTIONS = { from telnetclient import TelnetClient +_UMLAUT_TRANSLATION = str.maketrans( + { + "ä": "ae", + "ö": "oe", + "ü": "ue", + "Ä": "Ae", + "Ö": "Oe", + "Ü": "Ue", + "ß": "ss", + } +) + + +def sanitize_for_mud(text: str) -> str: + """Return ASCII-only text suitable for Silberland's input parser.""" + if not text: + return "" + + replaced = text.translate(_UMLAUT_TRANSLATION) + normalized = unicodedata.normalize("NFKD", replaced) + cleaned: list[str] = [] + for ch in normalized: + if unicodedata.combining(ch): + continue + code = ord(ch) + if 32 <= code <= 126: + cleaned.append(ch) + else: + cleaned.append("?") + return "".join(cleaned) + + class SessionState: """Share Telnet session state safely across threads.""" @@ -113,8 +150,9 @@ class SessionState: self._listeners_lock = Lock() def send(self, client: TelnetClient, message: str) -> None: + sanitized = sanitize_for_mud(message) with self._send_lock: - client.send(message) + client.send(sanitized) def tool_send( self, @@ -125,12 +163,14 @@ class SessionState: stop_event: Event, ) -> bool: """Send on behalf of the tool while respecting a minimum cadence.""" + sanitized = sanitize_for_mud(message) + while not stop_event.is_set(): with self._send_lock: now = time.time() elapsed = now - self._last_tool_send if elapsed >= min_interval: - client.send(message) + client.send(sanitized) self._last_tool_send = now return True wait_time = min_interval - elapsed @@ -340,10 +380,10 @@ def login( state.update_output(banner) if user: - client.send(user) + client.send(sanitize_for_mud(user)) time.sleep(0.2) if password: - client.send(password) + client.send(sanitize_for_mud(password)) response = client.receive(timeout=response_timeout) if response: @@ -414,7 +454,7 @@ def graceful_shutdown( if state: state.send(client, exit_command) else: - client.send(exit_command) + client.send(sanitize_for_mud(exit_command)) farewell = client.receive(timeout=2.0) if farewell: print(farewell, end="" if farewell.endswith("\n") else "\n")