""" 交互式终端面板 - 用 paramiko invoke_shell 打开真实 shell - 用户敲的每个字符直接发到远程 - 远程输出（含 ANSI 颜色）实时显示 - 本地历史（↑↓ 调出） - Ctrl+C / Ctrl+D / Ctrl+L 等常用键 """ import os import re import time from typing import Optional, List from PyQt5.QtCore import Qt, QTimer, pyqtSignal, QObject from PyQt5.QtGui import ( QFont, QFontDatabase, QTextCursor, QColor, QTextCharFormat, QKeyEvent, ) from PyQt5.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QPlainTextEdit, QPushButton, QLabel, QComboBox, QCheckBox, QApplication, ) from core.ssh_client import SSHConnection from core.snippets import SnippetManager # ANSI 颜色映射（16 色） ANSI_FG = { "30": "#2e2e2e", "31": "#ef5350", "32": "#66bb6a", "33": "#ffa726", "34": "#42a5f5", "35": "#ab47bc", "36": "#26c6da", "37": "#cfd8dc", "90": "#78909c", "91": "#ff7043", "92": "#9ccc65", "93": "#fff176", "94": "#64b5f6", "95": "#ce93d8", "96": "#80deea", "97": "#ffffff", } ANSI_BG = { "40": "#2e2e2e", "41": "#ef5350", "42": "#66bb6a", "43": "#ffa726", "44": "#42a5f5", "45": "#ab47bc", "46": "#26c6da", "47": "#cfd8dc", "100": "#78909c", } # ANSI CSI 序列：ESC [ ... _ANSI_RE = re.compile(r"\x1b\[([\x30-\x3f]*)([\x20-\x2f]*)([\x40-\x7e])") # OSC 序列：ESC ] ... BEL(0x07) 或 ESC \ 终止（PS1 标题/工作目录等） _OSC_RE = re.compile(r"\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)") # 单字符 ESC 序列：ESC c/=/> 等（RIS、DECSC 等），一般无害；这里跳过 ESC 后跟非 [ 的 1-2 字节 _ESC_SINGLE_RE = re.compile(r"\x1b[PX^_](?:[^\x1b]*?\x1b\\)?") # DCS/SOS/PM/APC：ESC P/X/^/_ ... ST 跳过 _DCS_RE = re.compile(r"\x1b[PX^_].*?\x1b\\") def _strip_ansi(text: str) -> str: """去掉所有 ANSI/OSC/DCS 转义，得到纯文本（用于历史回放、长度计算）""" text = _OSC_RE.sub("", text) text = _DCS_RE.sub("", text) text = _ANSI_RE.sub("", text) return text class _ShellReader(QObject): """在独立线程中读 channel；通过信号把数据发给 UI""" data_received = pyqtSignal(str) closed = pyqtSignal(str) def __init__(self, chan, poll_ms: int = 30): super().__init__() self.chan = chan self.poll_ms = poll_ms self._running = True # 用 QTimer 模拟后台读取（不需要真线程，paramiko recv 是阻塞的， # 但只要 settimeout(0) 非阻塞就 OK） self._timer = QTimer() self._timer.timeout.connect(self._poll) def start(self): self._timer.start(self.poll_ms) def stop(self): self._running = False self._timer.stop() try: self.chan.close() except Exception: pass def _poll(self): if not self._running: return try: if self.chan.closed: self.closed.emit("远程 shell 已关闭") self.stop() return # 非阻塞读，缓冲 4096 data = b"" while self.chan.recv_ready(): chunk = self.chan.recv(4096) if not chunk: break data += chunk # 防止一次把 buffer 抽干 if len(data) > 65536: break if data: # 用 utf-8 解码（忽略错误字符） text = data.decode("utf-8", errors="replace") self.data_received.emit(text) except Exception as e: # EAGAIN 之类的不算错误 if "EAGAIN" in str(e) or "timed out" in str(e).lower(): return self.closed.emit(f"读取错误: {e}") self.stop() class TerminalPanel(QWidget): """交互式终端""" closed = pyqtSignal() # 终端关闭（用于通知主窗口清理状态） def __init__(self, parent=None): super().__init__(parent) self.conn: Optional[SSHConnection] = None self.chan = None self.reader: Optional[_ShellReader] = None self.history: List[str] = [] # 本地命令历史 self.history_idx: int = -1 # -1 = 当前未浏览历史 self._pending_input: str = "" # 浏览历史时保留当前输入 self._current_line_start: int = 0 # 当前命令行在文档中的起始位置 self._connected = False # ANSI 状态 self._ansi_fg = "#e0e0e0" self._ansi_bg = None self._ansi_bold = False self._ansi_underline = False self._ansi_invert = False # OSC 0/2 设置的窗口标题 self._window_title = "" # 命令片段 self.snippet_mgr = SnippetManager() self._build_ui() self._apply_style() # ============================================================ # UI # ============================================================ def _build_ui(self): v = QVBoxLayout(self) v.setContentsMargins(0, 0, 0, 0) v.setSpacing(0) # 顶部工具栏 toolbar = QHBoxLayout() toolbar.setContentsMargins(6, 4, 6, 4) self.status_label = QLabel("未连接") self.status_label.setStyleSheet("color: #888;") toolbar.addWidget(self.status_label) toolbar.addStretch(1) # 命令片段 toolbar.addWidget(QLabel("📋")) self.snippet_combo = QComboBox() self.snippet_combo.setMinimumWidth(180) self.snippet_combo.setToolTip("选择命令片段快速发送到终端") self.snippet_combo.currentIndexChanged.connect(self._on_snippet_selected) toolbar.addWidget(self.snippet_combo) self._refresh_snippets() self.btn_snippet_mgr = QPushButton("⚙") self.btn_snippet_mgr.setFixedWidth(28) self.btn_snippet_mgr.setToolTip("管理命令片段") self.btn_snippet_mgr.clicked.connect(self._open_snippet_manager) toolbar.addWidget(self.btn_snippet_mgr) toolbar.addSpacing(8) self.btn_clear = QPushButton("清屏") self.btn_clear.clicked.connect(self._clear_screen) self.btn_reset = QPushButton("重连 shell") self.btn_reset.clicked.connect(self._reopen_shell) toolbar.addWidget(self.btn_clear) toolbar.addWidget(self.btn_reset) v.addLayout(toolbar) # 终端显示 self.term = QPlainTextEdit() self.term.setReadOnly(False) # 允许键盘输入 self.term.setUndoRedoEnabled(False) self.term.setLineWrapMode(QPlainTextEdit.WidgetWidth) self.term.setMaximumBlockCount(5000) # 限制行数防止卡 # 等宽字体 font = QFontDatabase.systemFont(QFontDatabase.FixedFont) font.setPointSize(11) self.term.setFont(font) v.addWidget(self.term, 1) # 状态栏 self.bottom_label = QLabel("提示: 连接主机后这里会出现真实的 shell 提示符，可直接键入命令") self.bottom_label.setStyleSheet("color: #666; padding: 4px;") v.addWidget(self.bottom_label) # 安装事件过滤器：拦截按键 self.term.installEventFilter(self) def _apply_style(self): self.term.setStyleSheet(""" QPlainTextEdit { background: #0c0c0c; color: #e0e0e0; selection-background-color: #264f78; selection-color: #ffffff; } """) # ============================================================ # 命令片段 # ============================================================ def _refresh_snippets(self): """刷新下拉框""" self.snippet_combo.blockSignals(True) self.snippet_combo.clear() self.snippet_combo.addItem("-- 选择命令片段 --", "") for s in self.snippet_mgr.list_all(): self.snippet_combo.addItem(f"{s['name']} ({s['cmd'][:30]})", s["cmd"]) self.snippet_combo.blockSignals(False) def _on_snippet_selected(self, index: int): """选择片段后发送命令到终端""" if index <= 0: return cmd = self.snippet_combo.itemData(index) if not cmd: return # 重置下拉框选中项（让用户能重复选同一个） self.snippet_combo.blockSignals(True) self.snippet_combo.setCurrentIndex(0) self.snippet_combo.blockSignals(False) # 发送命令 self.send_command(cmd) def send_command(self, cmd: str): """发送一条命令到远程 shell（自动加换行）""" if not self.chan or not self._connected: self._set_status("未连接，无法发送命令", "#c62828") return data = cmd.encode("utf-8", errors="replace") if not cmd.endswith("\n"): data += b"\n" try: self.chan.send(data) except Exception as e: self._set_status(f"发送失败: {e}", "#c62828") def _open_snippet_manager(self): from .snippet_dialog import SnippetManagerDialog dlg = SnippetManagerDialog(self.snippet_mgr, self) dlg.exec_() self._refresh_snippets() # ============================================================ # 生命周期 # ============================================================ def attach(self, conn: SSHConnection): """绑定 SSH 连接并打开 shell""" self.conn = conn if not conn or not conn.connected: self._set_status("未连接", "#c62828") return self._open_shell() def _open_shell(self): # 估算一个 cols/rows font_metrics = self.term.fontMetrics() ch_w = max(font_metrics.horizontalAdvance("M"), 1) ch_h = max(font_metrics.height(), 1) view = self.term.viewport().size() cols = max(40, view.width() // ch_w) rows = max(10, view.height() // ch_h) chan = self.conn.open_shell("xterm-256color", cols=cols, rows=rows) if not chan: self._set_status("打开 shell 失败", "#c62828") return self.chan = chan self.reader = _ShellReader(chan, poll_ms=30) self.reader.data_received.connect(self._on_data) self.reader.closed.connect(self._on_closed) self.reader.start() self._connected = True host = self.conn.host self._set_status(f"已连接 {self.conn.username}@{host}", "#2e7d32") self.bottom_label.setText( "提示: 直接键入命令，Enter 执行，↑↓ 调出历史，Ctrl+C 中断，Ctrl+L 清屏，Ctrl+D 退出 shell" ) def _reopen_shell(self): self.close_shell() if self.conn and self.conn.connected: self.term.clear() self._open_shell() def close_shell(self): """关闭当前 shell（不断 SSH 连接）""" if self.reader: self.reader.stop() self.reader = None self.chan = None self._connected = False def _on_closed(self, msg: str): self._set_status(f"shell 已关闭: {msg}", "#ff8f00") self._connected = False self.chan = None self.closed.emit() # ============================================================ # 数据接收 + ANSI 渲染 # ============================================================ def _on_data(self, text: str): """远程 shell 发来一段字节；处理 CR/LF 渲染 + ANSI 颜色""" # 转换 \r\n / \r 为统一换行（终端协议：CR = 光标回行首，LF = 换行） # 简化：CR 单独处理（覆盖当前行），LF 真的换行 cursor = self.term.textCursor() cursor.movePosition(QTextCursor.End) i = 0 while i < len(text): ch = text[i] if ch == "\x1b" and i + 1 < len(text): nxt = text[i + 1] # CSI: ESC [ ... if nxt == "[": m = _ANSI_RE.match(text, i) if m: self._apply_ansi(m.group(1), m.group(3)) i = m.end() continue # OSC: ESC ] ... BEL | ESC \ elif nxt == "]": m = _OSC_RE.match(text, i) if m: # OSC 0/1/2 设置标题/工作目录—— GUI 终端里通常更新窗口标题； # 我们这里不维护标题栏，但 PS1 里的 OSC 也照样被吃掉。 # 提取 OSC 内容以便调试 content = m.group(0)[2:-1] # 去掉 ESC ] 和终止符 # 可选：把第一个参数后的内容存到 title if content and content[0:1] in ("0", "1", "2") and ";" in content: self._window_title = content.split(";", 1)[1] i = m.end() continue # DCS / SOS / PM / APC: ESC P/X/^/_ ... ESC \ elif nxt in ("P", "X", "^", "_"): m = _DCS_RE.match(text, i) if m: i = m.end() continue # 其他 ESC 序列（ESC = / > / c 等）—— 单字节，吞掉 else: i += 2 continue elif ch == "\r": # \r 单独：回到行首；\n 真的换行 if i + 1 < len(text) and text[i+1] == "\n": # \r\n 视作一个换行（远程很多 echo 行为） cursor.insertText("\n", self._cur_format()) self._current_line_start = cursor.position() i += 2 continue # 单独 \r：删除当前行已输入的内容（\b 模拟） cursor.movePosition(QTextCursor.EndOfLine, QTextCursor.KeepAnchor) cursor.removeSelectedText() self._current_line_start = cursor.position() i += 1 continue elif ch == "\n": cursor.insertText("\n", self._cur_format()) self._current_line_start = cursor.position() i += 1 continue elif ch == "\x07": # BEL QApplication.beep() i += 1 continue elif ch == "\x08": # BS: 退格 cursor.deletePreviousChar() i += 1 continue else: cursor.insertText(ch, self._cur_format()) i += 1 # 滚动到底 sb = self.term.verticalScrollBar() sb.setValue(sb.maximum()) def _cur_format(self) -> QTextCharFormat: """根据当前 ANSI 状态构造 QTextCharFormat""" f = QTextCharFormat() f.setForeground(QColor(self._ansi_fg)) if self._ansi_bold: f.setFontWeight(QFont.Bold) if self._ansi_underline: f.setFontUnderline(True) if self._ansi_invert: f.setBackground(QColor(self._ansi_fg)) f.setForeground(QColor("#000000")) return f def _apply_ansi(self, params: str, final: str): """处理 CSI 序列（只覆盖最常见的 SGR 颜色/样式）""" if final != "m": # 其他 CSI 序列（光标移动、清屏等）—— 简化忽略；复杂终端需要 curses/pty 库 return if not params: params = "0" codes = params.split(";") for c in codes: if c in ("0", ""): self._ansi_fg = "#e0e0e0" self._ansi_bg = None self._ansi_bold = False self._ansi_underline = False self._ansi_invert = False elif c in ANSI_FG: self._ansi_fg = ANSI_FG[c] elif c in ANSI_BG: self._ansi_bg = ANSI_BG[c] elif c == "1": self._ansi_bold = True elif c == "4": self._ansi_underline = True elif c == "7": self._ansi_invert = True elif c == "22": self._ansi_bold = False elif c == "24": self._ansi_underline = False elif c == "27": self._ansi_invert = False elif c == "39": self._ansi_fg = "#e0e0e0" elif c == "49": self._ansi_bg = None # 38;5;N 256 色简化忽略 def _reset_ansi(self): self._ansi_fg = "#e0e0e0" self._ansi_bg = None self._ansi_bold = False self._ansi_underline = False self._ansi_invert = False def _set_status(self, text: str, color: str = "#888"): self.status_label.setText(text) self.status_label.setStyleSheet(f"color: {color};") # ============================================================ # 键盘输入 # ============================================================ def eventFilter(self, obj, event): if obj is not self.term: return super().eventFilter(obj, event) if event.type() != QKeyEvent.KeyPress: return super().eventFilter(obj, event) key = event.key() mods = event.modifiers() # Ctrl 组合 if mods & Qt.ControlModifier: if key == Qt.Key_C: self._send_signal(b"\x03") # SIGINT return True if key == Qt.Key_D: # 行为取决于 shell；在行首按 Ctrl+D = EOF cursor = self.term.textCursor() cursor.movePosition(QTextCursor.End) if cursor.columnNumber() == 0 or self._is_at_line_start(cursor): self._send_bytes(b"\x04") return True if key == Qt.Key_L: self._clear_screen() return True if key == Qt.Key_A: self._move_cursor_to(QTextCursor.StartOfLine) return True if key == Qt.Key_E: self._move_cursor_to(QTextCursor.EndOfLine) return True if key == Qt.Key_U: self._kill_to_start_of_line() return True if key == Qt.Key_K: self._kill_to_end_of_line() return True if key == Qt.Key_W: self._kill_prev_word() return True return False # 普通键 if key == Qt.Key_Return or key == Qt.Key_Enter: self._on_enter() return True if key == Qt.Key_Backspace: self._on_backspace() return True if key == Qt.Key_Up: self._history_prev() return True if key == Qt.Key_Down: self._history_next() return True if key == Qt.Key_Left: # 只允许在当前行内左移 self._safe_move(QTextCursor.Left) return True if key == Qt.Key_Right: self._safe_move(QTextCursor.Right) return True if key == Qt.Key_Home: self._move_to_current_line_start() return True if key == Qt.Key_End: self._move_to_current_line_end() return True if key == Qt.Key_Delete: self._on_delete() return True if key == Qt.Key_Tab: # tab 补全 if self.chan and self._connected: self._send_bytes(b"\t") return True return False # 防止用户把光标移到历史输出里改字 if key in (Qt.Key_PageUp, Qt.Key_PageDown): return False # 普通字符：发到远程 text = event.text() if text and self.chan and self._connected: self._send_bytes(text.encode("utf-8", errors="replace")) return True return False def _send_signal(self, b: bytes): if self.chan and self._connected: try: self.chan.send(b) except Exception: pass def _send_bytes(self, b: bytes): if self.chan and self._connected: try: self.chan.send(b) except Exception: pass def _on_enter(self): # 取出当前行输入，发给 shell cursor = self.term.textCursor() cursor.movePosition(QTextCursor.End) end_pos = cursor.position() cursor.setPosition(self._current_line_start) cursor.setPosition(end_pos, QTextCursor.KeepAnchor) cmd = cursor.selectedText() # 记录历史（去重空行） cmd_stripped = _strip_ansi(cmd).strip() if cmd_stripped: if not self.history or self.history[-1] != cmd_stripped: self.history.append(cmd_stripped) if len(self.history) > 200: self.history = self.history[-200:] self.history_idx = -1 self._send_bytes(b"\r") def _on_backspace(self): cursor = self.term.textCursor() if cursor.position() <= self._current_line_start: return # 已到行首，不删 # 不能跨行删（保持当前行） if self._is_at_line_start(cursor): return cursor.deletePreviousChar() def _on_delete(self): cursor = self.term.textCursor() if cursor.atEnd(): return if cursor.columnNumber() == 0 and not cursor.hasSelection(): return cursor.deleteChar() def _history_prev(self): if not self.history: return if self.history_idx == -1: # 保存当前行 cursor = self.term.textCursor() cursor.setPosition(self._current_line_start) cursor.movePosition(QTextCursor.End, QTextCursor.KeepAnchor) self._pending_input = cursor.selectedText() self.history_idx = len(self.history) - 1 elif self.history_idx > 0: self.history_idx -= 1 self._replace_current_line(self.history[self.history_idx]) def _history_next(self): if self.history_idx == -1: return if self.history_idx < len(self.history) - 1: self.history_idx += 1 self._replace_current_line(self.history[self.history_idx]) else: self.history_idx = -1 self._replace_current_line(self._pending_input) def _replace_current_line(self, text: str): cursor = self.term.textCursor() cursor.setPosition(self._current_line_start) cursor.movePosition(QTextCursor.End, QTextCursor.KeepAnchor) cursor.removeSelectedText() cursor.insertText(text) # 把光标移到行尾 cursor.movePosition(QTextCursor.EndOfLine) self.term.setTextCursor(cursor) def _move_to_current_line_start(self): cursor = self.term.textCursor() cursor.setPosition(self._current_line_start) self.term.setTextCursor(cursor) def _move_to_current_line_end(self): cursor = self.term.textCursor() cursor.movePosition(QTextCursor.EndOfLine) self.term.setTextCursor(cursor) def _move_cursor_to(self, op): cursor = self.term.textCursor() cursor.movePosition(op) self.term.setTextCursor(cursor) def _kill_to_start_of_line(self): cursor = self.term.textCursor() cursor.setPosition(self._current_line_start, QTextCursor.KeepAnchor) cursor.removeSelectedText() def _kill_to_end_of_line(self): cursor = self.term.textCursor() cursor.movePosition(QTextCursor.EndOfLine, QTextCursor.KeepAnchor) cursor.removeSelectedText() def _kill_prev_word(self): cursor = self.term.textCursor() # 删到上一个空格/行首 start = cursor.position() cursor.movePosition(QTextCursor.PreviousCharacter, QTextCursor.KeepAnchor, 1) while cursor.position() > self._current_line_start and not cursor.selectedText().isspace(): cursor.movePosition(QTextCursor.PreviousCharacter, QTextCursor.KeepAnchor, 1) cursor.removeSelectedText() def _safe_move(self, op): cursor = self.term.textCursor() if op == QTextCursor.Left and cursor.position() <= self._current_line_start: return cursor.movePosition(op) self.term.setTextCursor(cursor) def _is_at_line_start(self, cursor) -> bool: return cursor.position() <= self._current_line_start def _clear_screen(self): # 真正清屏：发 Ctrl+L 给 shell if self.chan and self._connected: self._send_bytes(b"\x0c") # 顺手也清一下本地显示 self.term.clear() def resizeEvent(self, e): super().resizeEvent(e) # 通知 shell 终端大小变了 if self.chan and self._connected and self.conn: fm = self.term.fontMetrics() ch_w = max(fm.horizontalAdvance("M"), 1) ch_h = max(fm.height(), 1) view = self.term.viewport().size() cols = max(40, view.width() // ch_w) rows = max(10, view.height() // ch_h) self.conn.resize_shell(self.chan, cols, rows)