Files
sshclient/ui/terminal_panel.py
T
Hermes 9dbd500399 feat: real interactive terminal (Xshell/PuTTY style)
User feedback: 'your terminal can only send commands, not type
freely. Useless.'

Replaced the old 'single command + output' terminal with a true
interactive shell panel:

NEW: ui/terminal_panel.py
- Uses paramiko invoke_shell() to open a long-lived shell session
- Every keystroke is forwarded to the remote shell (not a textbox)
- Remote output (incl. ANSI color codes) is parsed and rendered
  inline in a QPlainTextEdit (16-color SGR, bold, underline, invert)
- Cursor kept on the current input line; user can't edit past output
- Local command history with up/down navigation (stores last 200)
- Standard terminal shortcuts: Ctrl+C (SIGINT), Ctrl+D (EOF),
  Ctrl+L (clear), Ctrl+A/E (line start/end), Ctrl+U/K/W
- Terminal resize forwarded to remote PTY (vim/top work correctly)
- Tab key forwarded to remote for completion

CHANGED: core/ssh_client.py
- Added open_shell(term_type, cols, rows) -> Channel
- Added resize_shell(chan, cols, rows)

CHANGED: ui/main_window.py
- Replaced old _build_terminal_tab (now removed) with TerminalPanel
- On host connect -> auto-open shell, switch to Terminal tab
- On host disconnect / switch -> close shell cleanly

NEW: test_terminal.py (5 tests, all pass)
- Real SSH connection, opens shell
- Sends 'ls /tmp' via simulated keypresses, verifies 162-line output
- Sends ANSI color command, verifies colored text appears
- Tests up-arrow history recall
- Verifies prompt returns after command completes

All existing tests still pass (core 6/6 + UI 3/3 + E2E 6/6 + terminal 5/5)
2026-07-28 21:36:49 +08:00

588 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
交互式终端面板
- 用 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)