diff --git a/src/lazy_enter/bridge.py b/src/lazy_enter/bridge.py index d3c0a69..e011d48 100644 --- a/src/lazy_enter/bridge.py +++ b/src/lazy_enter/bridge.py @@ -7,6 +7,7 @@ import threading import time from lazy_enter.config import Config +from lazy_enter.output_filter import clean_terminal_output from lazy_enter.pty_manager import PtyManager from lazy_enter.slack_handler import SlackHandler @@ -23,6 +24,7 @@ class Bridge: self._output_thread: threading.Thread | None = None self._running = False self._channel: str = self.config.allowed_channel_id + self._last_sent_output: str = "" self.slack.on_message(self._handle_message) self.slack.on_command(self._handle_command) @@ -51,6 +53,7 @@ class Bridge: return self._channel = channel + self._last_sent_output = "" self.pty = PtyManager(self.config.default_shell) self.pty.start() self._running = True @@ -65,6 +68,7 @@ class Bridge: if self.pty: self.pty.stop() self.pty = None + self._last_sent_output = "" self.slack.send_message(channel, ":stop_sign: 세션이 종료되었습니다.") def _poll_output(self) -> None: @@ -73,14 +77,18 @@ class Bridge: while self._running and self.pty and self.pty.is_alive: output = self.pty.read_output(timeout=self.config.pty_read_timeout) if output: - buffer += output + cleaned = clean_terminal_output(output) + if cleaned: + buffer += f"{cleaned}\n" if buffer: # 메시지 길이 제한 적용 - message = buffer[: self.config.max_message_length] + message = buffer[: self.config.max_message_length].rstrip() if len(buffer) > self.config.max_message_length: message += "\n... (truncated)" - self.slack.send_message(self._channel, f"```\n{message}\n```") + if message and message != self._last_sent_output: + self.slack.send_message(self._channel, f"```\n{message}\n```") + self._last_sent_output = message buffer = "" time.sleep(self.config.output_buffer_interval) diff --git a/src/lazy_enter/output_filter.py b/src/lazy_enter/output_filter.py new file mode 100644 index 0000000..4553f81 --- /dev/null +++ b/src/lazy_enter/output_filter.py @@ -0,0 +1,32 @@ +"""PTY 출력에서 터미널 제어 시퀀스를 제거한다.""" + +from __future__ import annotations + +import re + +# ANSI CSI / OSC / DCS / 단일 ESC 시퀀스를 폭넓게 제거한다. +_ANSI_ESCAPE_RE = re.compile( + r"(?:\x1B\[[0-?]*[ -/]*[@-~])" + r"|(?:\x1B\][^\x1B\x07]*(?:\x07|\x1B\\))" + r"|(?:\x1BP[^\x1B]*(?:\x1B\\))" + r"|(?:\x1B[@-_])" +) + +# ESC 문자가 유실된 상태로 남는 CSI 토큰(예: [38;2;153;153;153m) 제거. +_BROKEN_CSI_RE = re.compile(r"\[(?:[0-9;<=>?]|(?:\d+;))*\d*[A-Za-z]") + +# C0 제어문자 중 개행/탭/캐리지리턴을 제외하고 제거. +_CONTROL_RE = re.compile(r"[\x00-\x08\x0B-\x1F\x7F]") + + +def clean_terminal_output(text: str) -> str: + """Slack 전송용 텍스트를 정리한다.""" + cleaned = _ANSI_ESCAPE_RE.sub("", text) + cleaned = _BROKEN_CSI_RE.sub("", cleaned) + cleaned = cleaned.replace("\r", "") + cleaned = _CONTROL_RE.sub("", cleaned) + + # 제어 코드 제거 후 남는 빈 줄을 축소한다. + lines = [line.rstrip() for line in cleaned.splitlines()] + compact = "\n".join(lines).strip() + return compact diff --git a/tests/test_output_filter.py b/tests/test_output_filter.py new file mode 100644 index 0000000..f6a7489 --- /dev/null +++ b/tests/test_output_filter.py @@ -0,0 +1,18 @@ +"""output_filter 모듈 테스트.""" + +from lazy_enter.output_filter import clean_terminal_output + + +def test_clean_terminal_output_removes_ansi_sequences(): + raw = "\x1b[?2004h\x1b[38;2;153;153;153mClaude Code\x1b[39m" + assert clean_terminal_output(raw) == "Claude Code" + + +def test_clean_terminal_output_removes_broken_csi_tokens(): + raw = "[?2004h[38;2;153;153;153mhello[39m" + assert clean_terminal_output(raw) == "hello" + + +def test_clean_terminal_output_preserves_plain_text(): + raw = "line1\nline2" + assert clean_terminal_output(raw) == "line1\nline2"