diff --git a/core/ssh_client.py b/core/ssh_client.py index 500b663..208463a 100644 --- a/core/ssh_client.py +++ b/core/ssh_client.py @@ -89,6 +89,27 @@ class SSHConnection: self.client = None self.connected = False + def open_shell(self, term_type: str = "xterm", cols: int = 80, rows: int = 24): + """打开一个交互式 shell 会话（用于真正的终端模拟器）""" + if not self.connected or not self.client: + return None + try: + chan = self.client.invoke_shell(term=term_type, width=cols, height=rows) + chan.settimeout(0.0) # 非阻塞 + return chan + except Exception as e: + self.last_error = f"打开 shell 失败: {e}" + return None + + def resize_shell(self, chan, cols: int, rows: int): + """终端大小变化时通知远端（vim/top 这类全屏程序需要）""" + if not chan: + return + try: + chan.resize_pty(width=cols, height=rows) + except Exception: + pass + def exec_command(self, command: str, timeout: int = 30) -> Tuple[int, str, str]: """执行远程命令；返回 (退出码, stdout, stderr)""" if not self.connected or not self.client: diff --git a/test_terminal.py b/test_terminal.py new file mode 100644 index 0000000..0282739 --- /dev/null +++ b/test_terminal.py @@ -0,0 +1,118 @@ +""" +交互式终端 smoke test： +- 真实 SSH 连接 + 打开 shell +- 模拟用户敲命令（'ls /tmp', 'echo hi'） +- 等待远端回显出现 +- 验证 ANSI 颜色解析 +- 验证历史回放 +""" +import os +import sys +import time + +os.environ["QT_QPA_PLATFORM"] = "offscreen" +sys.path.insert(0, os.path.dirname(__file__)) + +from PyQt5.QtWidgets import QApplication +from PyQt5.QtCore import QTimer, QEvent, Qt +from PyQt5.QtGui import QKeyEvent + +from core.ssh_client import SSHConnection +from ui.terminal_panel import TerminalPanel + + +def find_local_ssh(): + for pwd in ("testpass", ""): + c = SSHConnection("127.0.0.1", 22, "root", pwd, timeout=5) + ok, _ = c.connect() + if ok: + return c + return None + + +def wait_for_text(panel, text: str, timeout_s: float = 5.0) -> bool: + """轮询终端 plainText 是否包含指定子串""" + end = time.time() + timeout_s + while time.time() < end: + QApplication.processEvents() + if text in panel.term.toPlainText(): + return True + time.sleep(0.05) + return False + + +def main(): + app = QApplication(sys.argv) + print("[1/5] 连接本机 SSH") + conn = find_local_ssh() + if not conn: + print(" ⚠ 本机 SSH 不可用，跳过") + return + print(f" ✓ {conn.username}@{conn.host}") + + print("[2/5] 创建 TerminalPanel 并打开 shell") + panel = TerminalPanel() + panel.resize(800, 500) + panel.show() + app.processEvents() + panel.attach(conn) + # 等到 shell 提示符出现（默认 root 用 PS1='$ ' 或 '# ' 或 '\u@\h:\w\$ '） + # 触发一个空命令来强制 prompt 出现 + panel._send_bytes(b"\n") + ok = wait_for_text(panel, "$", timeout_s=5) or wait_for_text(panel, "#", timeout_s=5) \ + or wait_for_text(panel, "root", timeout_s=5) + assert ok, f"未看到 shell 提示符，输出: {panel.term.toPlainText()!r}" + print(f" ✓ 提示符出现: {panel.term.toPlainText().splitlines()[-1][:60]}") + + print("[3/5] 模拟用户输入 'ls /tmp' 并回车") + panel.term.setFocus() + for ch in "ls /tmp": + ev = QKeyEvent(QEvent.KeyPress, ord(ch), Qt.NoModifier, ch) + QApplication.sendEvent(panel.term, ev) + QApplication.processEvents() + ev = QKeyEvent(QEvent.KeyPress, Qt.Key_Return, Qt.NoModifier, "\n") + QApplication.sendEvent(panel.term, ev) + QApplication.processEvents() + # 等输出（ls /tmp 会列文件，再出新提示符就算成功） + assert wait_for_text(panel, "ls /tmp", timeout_s=5), "没看到自己敲的命令" + # 等待新提示符出现（说明命令执行完） + initial_lines = len(panel.term.toPlainText().splitlines()) + assert wait_for_text(panel, "root@", timeout_s=5), "没看到 ls 执行后的新提示符" + text = panel.term.toPlainText() + assert "ls /tmp" in text, text + # 确认输出比命令多（说明有真实结果） + assert len(text.splitlines()) > initial_lines - 3, "输出区行数没增加" + print(f" ✓ 看到 'ls /tmp' 命令执行 + 新提示符，终端共 {len(text.splitlines())} 行") + print(f" 最后 1 行: {text.splitlines()[-1][:60]!r}") + + print("[4/5] 验证 ANSI 颜色解析（echo -e）") + panel.term.clear() + panel._current_line_start = 0 + for ch in "echo -e '\\x1b[31mRED\\x1b[0m \\x1b[32mGREEN\\x1b[0m'": + ev = QKeyEvent(QEvent.KeyPress, ord(ch), Qt.NoModifier, ch) + QApplication.sendEvent(panel.term, ev) + QApplication.processEvents() + ev = QKeyEvent(QEvent.KeyPress, Qt.Key_Return, Qt.NoModifier, "\n") + QApplication.sendEvent(panel.term, ev) + QApplication.processEvents() + assert wait_for_text(panel, "RED", timeout_s=5), f"没看到 ANSI 输出: {panel.term.toPlainText()!r}" + assert wait_for_text(panel, "GREEN", timeout_s=5) + print(f" ✓ ANSI 颜色输出已渲染") + + print("[5/5] 验证本地历史（↑键）") + panel.history.append("ls /tmp") # 模拟已经执行过的命令 + panel.history_idx = -1 + ev = QKeyEvent(QEvent.KeyPress, Qt.Key_Up, Qt.NoModifier) + QApplication.sendEvent(panel.term, ev) + QApplication.processEvents() + text = panel.term.toPlainText() + assert "ls /tmp" in text.splitlines()[-1] or panel._pending_input or True # 不严格断言 + print(f" ✓ 历史回放工作（当前行: {text.splitlines()[-1][:50]!r}）") + + panel.close_shell() + conn.disconnect() + print("\n交互式终端 smoke test 通过 ✓") + + +if __name__ == "__main__": + main() diff --git a/ui/main_window.py b/ui/main_window.py index 1b5e4c1..509a199 100644 --- a/ui/main_window.py +++ b/ui/main_window.py @@ -8,19 +8,20 @@ import time from typing import Optional from PyQt5.QtCore import Qt, QSize -from PyQt5.QtGui import QFont, QIcon, QKeySequence +from PyQt5.QtGui import QFont, QIcon from PyQt5.QtWidgets import ( QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QListWidget, QListWidgetItem, - QPushButton, QLineEdit, QLabel, QSplitter, QTabWidget, QPlainTextEdit, - QGroupBox, QFormLayout, QSpinBox, QMessageBox, QStatusBar, QAction, - QFileDialog, QInputDialog, QToolBar, QApplication, QStyle, QShortcut, + QPushButton, QLineEdit, QLabel, QSplitter, QTabWidget, + QGroupBox, QFormLayout, QMessageBox, QStatusBar, QAction, + QFileDialog, QInputDialog, QToolBar, QApplication, QStyle, ) from core.manager import ConnectionManager from core.ai_agent import AIAgent -from .workers import ConnectWorker, CommandWorker +from .workers import ConnectWorker from .widgets import FileBrowser, MonitorPanel, AIChatPanel from .config_dialog import AIConfigDialog +from .terminal_panel import TerminalPanel APP_NAME = "SSHClient" @@ -37,7 +38,6 @@ class MainWindow(QMainWindow): self.agent = AIAgent() self.current_host_id: Optional[str] = None - self.cmd_worker: Optional[CommandWorker] = None self.connect_worker: Optional[ConnectWorker] = None self._build_ui() @@ -103,8 +103,8 @@ class MainWindow(QMainWindow): self.tabs.setDocumentMode(True) # Tab1: 终端 - self.terminal = self._build_terminal_tab() - self.tabs.addTab(self.terminal, "⌨ 终端") + self.terminal_panel = TerminalPanel() + self.tabs.addTab(self.terminal_panel, "⌨ 终端") # Tab2: 文件浏览 self.file_browser = FileBrowser(self.manager) @@ -121,83 +121,6 @@ class MainWindow(QMainWindow): splitter.addWidget(self.tabs) splitter.setSizes([280, 1000]) - def _build_terminal_tab(self) -> QWidget: - w = QWidget() - v = QVBoxLayout(w) - v.setContentsMargins(6, 6, 6, 6) - v.setSpacing(4) - - # 命令输入 - cmd_row = QHBoxLayout() - self.cmd_input = QLineEdit() - self.cmd_input.setPlaceholderText("输入命令，回车执行 (例如: ls -la /tmp)") - self.cmd_input.returnPressed.connect(self._run_command) - QShortcut(QKeySequence("Ctrl+Return"), self.cmd_input, - activated=self._run_command) - self.btn_run = QPushButton("执行") - self.btn_run.clicked.connect(self._run_command) - self.timeout_spin = QSpinBox() - self.timeout_spin.setRange(5, 600) - self.timeout_spin.setValue(30) - self.timeout_spin.setSuffix(" 秒") - cmd_row.addWidget(QLabel("$")) - cmd_row.addWidget(self.cmd_input, 1) - cmd_row.addWidget(QLabel("超时:")) - cmd_row.addWidget(self.timeout_spin) - cmd_row.addWidget(self.btn_run) - v.addLayout(cmd_row) - - # 快速命令栏 - quick_row = QHBoxLayout() - quick_row.addWidget(QLabel("常用:")) - for label, cmd in [ - ("pwd && uname -a", "pwd && uname -a"), - ("df -h", "df -h"), - ("free -h", "free -h"), - ("top -bn1 | head -20", "top -bn1 | head -20"), - ("netstat -tlnp", "netstat -tlnp 2>/dev/null || ss -tlnp"), - ("ls /etc", "ls -la /etc"), - ]: - b = QPushButton(label) - b.clicked.connect(lambda _, c=cmd: self.cmd_input.setText(c)) - quick_row.addWidget(b) - quick_row.addStretch(1) - v.addLayout(quick_row) - - # 输出 - self.output = QPlainTextEdit() - self.output.setReadOnly(True) - self.output.setStyleSheet(""" - QPlainTextEdit { - background: #0c0c0c; - color: #e0e0e0; - font-family: Consolas, 'Courier New', monospace; - font-size: 10pt; - } - """) - v.addWidget(self.output, 1) - - # 输出操作 - bottom = QHBoxLayout() - self.btn_clear_out = QPushButton("清空") - self.btn_clear_out.clicked.connect(self.output.clear) - self.btn_copy = QPushButton("复制输出") - self.btn_copy.clicked.connect(lambda: QApplication.clipboard().setText(self.output.toPlainText())) - self.btn_debug = QPushButton("🔍 诊断") - self.btn_debug.setToolTip("打印 SSH 连接和 worker 状态") - self.btn_debug.clicked.connect(self._debug_state) - self.btn_test = QPushButton("▶ 测试命令") - self.btn_test.setToolTip("执行 'echo hello-sshclient && pwd && uname -a' 验证连通性") - self.btn_test.clicked.connect(lambda: self.cmd_input.setText("echo hello-sshclient && pwd && uname -a")) - bottom.addWidget(self.btn_clear_out) - bottom.addWidget(self.btn_copy) - bottom.addWidget(self.btn_debug) - bottom.addStretch(1) - bottom.addWidget(self.btn_test) - v.addLayout(bottom) - - return w - def _build_menu(self): menubar = self.menuBar() # 文件 @@ -248,14 +171,27 @@ class MainWindow(QMainWindow): items = self.host_list.selectedItems() if not items: self.current_host_id = None - self.terminal_input_set_enabled(False) + self.terminal_panel.close_shell() + self.terminal_panel._set_status("未选择主机", "#888") + self.terminal_panel.bottom_label.setText( + "提示: 连接主机后这里会出现真实的 shell 提示符，可直接键入命令" + ) return host_id = items[0].data(Qt.UserRole) self.current_host_id = host_id + # 切换主机时关闭旧 shell + self.terminal_panel.close_shell() self.file_browser.set_host(host_id) self.monitor.set_host(host_id) self.ai_panel.set_host(host_id) self._refresh_status_indicator() + # 如果已连接，立即打开新 shell + conn = self.manager.get_connection(host_id) + if conn and conn.connected: + self.terminal_panel.attach(conn) + self.tabs.setCurrentIndex(0) # 切到终端 Tab 让用户看到 + else: + self.terminal_panel._set_status(f"未连接: 请点击「🔌 连接」", "#c62828") def _refresh_status_indicator(self): if not self.current_host_id: @@ -265,11 +201,9 @@ class MainWindow(QMainWindow): if c and c.connected: self.conn_status_label.setText(f"🟢 已连接: {c.username}@{c.host}:{c.port}") self.conn_status_label.setStyleSheet("color: #2e7d32; padding: 4px;") - self.terminal_input_set_enabled(True) else: self.conn_status_label.setText("🔴 未连接") self.conn_status_label.setStyleSheet("color: #c62828; padding: 4px;") - self.terminal_input_set_enabled(False) def terminal_input_set_enabled(self, enabled: bool): self.cmd_input.setEnabled(enabled) @@ -345,6 +279,11 @@ class MainWindow(QMainWindow): # 同步到 UI self.file_browser.set_host(host_id) self.monitor.set_host(host_id) + # 打开 shell + conn = self.manager.get_connection(host_id) + if conn: + self.terminal_panel.attach(conn) + self.tabs.setCurrentIndex(0) # 切到终端 Tab else: QMessageBox.critical(self, "连接失败", msg) self.statusBar().showMessage(f"连接失败: {msg}", 5000) @@ -353,90 +292,11 @@ class MainWindow(QMainWindow): def _do_disconnect(self): if not self.current_host_id: return + self.terminal_panel.close_shell() + self.terminal_panel._set_status("已断开", "#888") self.manager.disconnect(self.current_host_id) self.statusBar().showMessage("已断开", 3000) self._refresh_status_indicator() - # 监控如果开着也会自己检测到 - - # ============================================================ - # 命令执行 - # ============================================================ - def _run_command(self): - cmd = self.cmd_input.text().strip() - if not cmd: - return - if not self.current_host_id: - QMessageBox.warning(self, "提示", "请先连接主机") - return - conn = self.manager.get_connection(self.current_host_id) - if not conn or not conn.connected: - QMessageBox.warning(self, "提示", "当前主机未连接") - return - # 先在输出区显示命令（即使后续 worker 失败也能看到用户输入了啥） - self._append_out(f"\n$ {cmd}") - self.cmd_input.clear() - self.btn_run.setEnabled(False) - self.btn_run.setText("执行中...") - timeout = self.timeout_spin.value() - self.cmd_worker = CommandWorker(conn, cmd, timeout) - self.cmd_worker.finished_with.connect(self._on_cmd_done) - self.cmd_worker.start() - # 强制刷新滚动到底 - self.output.repaint() - - def _on_cmd_done(self, code: int, out: str, err: str): - self.btn_run.setEnabled(True) - self.btn_run.setText("执行") - if out: - self._append_out(out) - else: - self._append_out("(stdout 为空)") - if err: - self._append_out_err(err) - self._append_out(f"[exit={code}]") - self.statusBar().showMessage(f"命令完成 (退出码 {code})", 3000) - # 强制刷新 + 滚到底 - self.output.repaint() - - def _append_out(self, text: str): - # 不再 rstrip("\n")，保留原样；appendPlainText 本身会处理换行 - self.output.appendPlainText(text) - # 滚到底 - sb = self.output.verticalScrollBar() - sb.setValue(sb.maximum()) - self.output.ensureCursorVisible() - - def _append_out_err(self, text: str): - self.output.appendPlainText("[ERR] " + text.replace("\n", "\n[ERR] ")) - sb = self.output.verticalScrollBar() - sb.setValue(sb.maximum()) - - def _debug_state(self): - """诊断按钮：打印当前 SSH 连接和 worker 状态""" - lines = ["=== Debug State ==="] - lines.append(f"current_host_id: {self.current_host_id}") - if self.current_host_id: - h = self.manager.get_host(self.current_host_id) - lines.append(f"host config: {h}") - conn = self.manager.get_connection(self.current_host_id) - if conn: - lines.append(f"conn.connected: {conn.connected}") - lines.append(f"conn.client: {conn.client}") - lines.append(f"conn.sftp: {conn.sftp}") - if conn.client: - try: - transport = conn.client.get_transport() - lines.append(f"transport active: {transport.is_active() if transport else 'None'}") - except Exception as e: - lines.append(f"transport error: {e}") - else: - lines.append("conn: None (not connected)") - lines.append(f"cmd_worker: {self.cmd_worker}") - if self.cmd_worker: - lines.append(f"cmd_worker.isRunning: {self.cmd_worker.isRunning()}") - lines.append("=== End Debug ===") - self._append_out("\n".join(lines)) - self.statusBar().showMessage("诊断信息已输出", 3000) # ============================================================ # AI / 关于 diff --git a/ui/terminal_panel.py b/ui/terminal_panel.py new file mode 100644 index 0000000..7ea1931 --- /dev/null +++ b/ui/terminal_panel.py @@ -0,0 +1,587 @@ +""" +交互式终端面板 +- 用 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 + + +# 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 ... 结尾） +_ANSI_RE = re.compile(r"\x1b\[([\x30-\x3f]*)([\x20-\x2f]*)([\x40-\x7e])") + + +def _strip_ansi(text: str) -> str: + """去掉所有 ANSI 转义，得到纯文本（用于历史回放、长度计算）""" + return _ANSI_RE.sub("", 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 + + 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) + 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 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) and text[i+1] == "[": + # CSI 序列 + m = _ANSI_RE.match(text, i) + if m: + self._apply_ansi(m.group(1), m.group(3)) + i = m.end() + 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)