Your Name
ac29515d9f
Root cause: _current_line_start was set after \r\n but BEFORE the prompt was rendered, so Backspace could erase prompt characters after reaching the input line start. Fix: send \x7f (DEL) to shell on Backspace and \x1b[3~ on Delete instead of locally deleting characters. The shell's readline handles backspace correctly - it ignores DEL at the input line start, so the prompt is never touched. The existing \b handler in _on_data already correctly processes the shell's \b-space-\b echo sequence. Verified: typing 'abc' + 3 Backspace deletes input, 5 more Backspace leaves prompt 'root@host:~# ' intact.
702 lines
26 KiB
Python
702 lines
26 KiB
Python
"""
|
||
交互式终端面板
|
||
- 用 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 [ ... <final 0x40-0x7e>
|
||
_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 [ ... <final>
|
||
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):
|
||
# 发 DEL(\x7f) 给 shell，由 shell 的 readline 处理退格 + 回显(\b \b)。
|
||
# 这样 shell 在输入行首会忽略退格，不会删除提示符 root@host:~#。
|
||
# 旧方案是本地 deletePreviousChar()，但 _current_line_start 在提示符
|
||
# 之前，导致退到输入行首后还能继续删提示符。
|
||
if self.chan and self._connected:
|
||
self._send_bytes(b"\x7f")
|
||
return
|
||
# 未连接时 fallback：本地删除
|
||
cursor = self.term.textCursor()
|
||
if cursor.position() <= self._current_line_start:
|
||
return
|
||
cursor.deletePreviousChar()
|
||
|
||
def _on_delete(self):
|
||
# 发 Delete 转义序列给 shell，由 shell 处理
|
||
if self.chan and self._connected:
|
||
self._send_bytes(b"\x1b[3~")
|
||
return
|
||
# 未连接时 fallback：本地删除
|
||
cursor = self.term.textCursor()
|
||
if cursor.atEnd():
|
||
return
|
||
if cursor.position() < self._current_line_start:
|
||
cursor.setPosition(self._current_line_start)
|
||
self.term.setTextCursor(cursor)
|
||
return
|
||
if cursor.hasSelection():
|
||
sel_start = cursor.selectionStart()
|
||
if sel_start < self._current_line_start:
|
||
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)
|