From 5ea19b75ae89aada0d7c0243213a0f2bd1e4922c Mon Sep 17 00:00:00 2001 From: xiaoli-cli bot Date: Tue, 23 Jun 2026 09:14:15 +0000 Subject: [PATCH] =?UTF-8?q?refactor:=20terminal=20=E6=8F=92=E4=BB=B6?= =?UTF-8?q?=E9=87=8D=E5=86=99=20v2=20-=20=E6=9E=81=E7=AE=80=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新设计: AI 的虚拟键盘和屏幕 - open: 打开终端 - type: 打字 - key: 按键 (enter/tab/ctrl+c/方向键/f1-f12等) - read: 看整个屏幕 - last [N]: 只看最近N行（默认20，避免上下文爆炸） - wait: 等待输出 从600行精简到250行，接口从12个减到8个 --- plugins/terminal.py | 707 ++++++++++++++++---------------------------- 1 file changed, 258 insertions(+), 449 deletions(-) diff --git a/plugins/terminal.py b/plugins/terminal.py index a37d72a..828eed9 100644 --- a/plugins/terminal.py +++ b/plugins/terminal.py @@ -1,245 +1,204 @@ """ -终端模拟器插件 — 纯标准库，Windows/Linux/macOS 通用 -=================================================== -持久交互式终端会话，AI 可以像人一样打开终端、敲命令、 -看输出、继续交互，支持 Python REPL、SSH、top 等。 +终端模拟器 v2 — AI 的虚拟键盘和屏幕 +===================================== +设计理念：AI 面前有一个终端，它能做的事就是： + 1. 打字（type） + 2. 按键（key） + 3. 看屏幕（read） +就这么简单。 """ import os import re import sys import time -import json import uuid -import shutil -import signal +import json import threading import subprocess import platform from collections import deque - -# ═══════════════════════════════════════════════════════════════ -# ToolResult 轻量替代（避免依赖 xcli_core） -# ═══════════════════════════════════════════════════════════════ - -class _R: - """极简返回值""" - @staticmethod - def ok(data): return str(data) - @staticmethod - def fail(msg): return f"[错误] {msg}" - - -# ═══════════════════════════════════════════════════════════════ -# 单个终端会话 -# ═══════════════════════════════════════════════════════════════ - _IS_WIN = platform.system() == "Windows" -class TerminalSession: - """ - 一个持久的交互式终端会话。 - 纯标准库实现：subprocess.Popen + threading.Thread， - 不依赖 pty、pty.openpty、select.poll 等 Unix 特有 API。 - """ - _SENTINEL = "<<nul"] - creationflags = subprocess.CREATE_NO_WINDOW - else: - env["TERM"] = "dumb" - env["NO_COLOR"] = "1" - env["PROMPT_COMMAND"] = "" - env["PYTHONUNBUFFERED"] = "1" # Python REPL 不缓冲 - creationflags = 0 - - # 只有 bash 才加 --norc --noprofile - shell_base = os.path.basename(self.shell) - if shell_base in ("bash", "bash.exe"): + env = {**os.environ, "TERM": "dumb", "NO_COLOR": "1", + "PYTHONUNBUFFERED": "1", "PROMPT_COMMAND": ""} + if not _IS_WIN: + base = os.path.basename(self.shell) + if "bash" in base: env["PS1"] = "" - shell_args = [self.shell, "--norc", "--noprofile"] - elif "python" in shell_base: - # Python 需要 -i 才能管道下交互 - shell_args = [self.shell, "-i"] + args = [self.shell, "--norc", "--noprofile"] + elif "python" in base: + args = [self.shell, "-i"] else: - shell_args = [self.shell] + args = [self.shell] + flags = 0 + else: + args = [self.shell, "/K", "chcp 65001 >nul"] + flags = subprocess.CREATE_NO_WINDOW - try: - self._proc = subprocess.Popen( - shell_args, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - cwd=self.cwd, - env=env, - bufsize=0, - creationflags=creationflags if _IS_WIN else 0, - ) - except FileNotFoundError: - raise RuntimeError(f"Shell 不存在: {self.shell}") - except Exception as e: - raise RuntimeError(f"启动终端失败: {e}") - - self._reader = threading.Thread(target=self._read_loop, daemon=True) - self._reader.start() + self._proc = subprocess.Popen( + args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, bufsize=0, cwd=self.cwd, + env=env, creationflags=flags if _IS_WIN else 0, + ) + threading.Thread(target=self._reader, daemon=True).start() time.sleep(0.3) - self.last_active = time.time() - def _read_loop(self): - """后台持续读取子进程 stdout""" - proc = self._proc - fd = proc.stdout.fileno() - while proc.poll() is None: + def _reader(self): + fd = self._proc.stdout.fileno() + while self._proc.poll() is None: try: chunk = os.read(fd, 4096) if not chunk: break text = chunk.decode("utf-8", errors="replace") with self._lock: - self._raw_buf += text + self._raw += text for line in text.split("\n"): - self._output_buf.append(line) + self._screen.append(line) except (OSError, ValueError): break - def send(self, text: str, wait: float = 0.3, timeout: float = 15) -> str: - """发送文本，返回后续输出""" + def alive(self): + return self._proc and self._proc.poll() is None + + # ── 核心操作：打字 ── + + def type(self, text: str): + """像人一样打字，每个字符发到 stdin""" if not self.alive(): - return "[错误] 终端进程已退出" - - self.last_active = time.time() - with self._lock: - self._raw_buf = "" - + return try: self._proc.stdin.write(text.encode("utf-8")) self._proc.stdin.flush() - except (OSError, BrokenPipeError) as e: - return f"[错误] 发送失败: {e}" - - return self._collect(wait, timeout) - - def send_cmd(self, command: str, timeout: float = 15) -> str: - """ - 发送命令并可靠等待输出结束。 - 用临时文件做哨兵，避免 stdout 混入提示符。 - """ - if not self.alive(): - return "[错误] 终端进程已退出" - - self.last_active = time.time() - with self._lock: - self._raw_buf = "" - - # 哨兵临时文件 - marker = f"/tmp/.xli_{uuid.uuid4().hex[:8]}" if not _IS_WIN else f"%TEMP%\\.xli_{uuid.uuid4().hex[:8]}" - - # 发送: 命令 + 哨兵文件创建 - payload = f"{command}\n{'touch' if not _IS_WIN else 'type nul >'} {marker}\n" - try: - self._proc.stdin.write(payload.encode("utf-8")) - self._proc.stdin.flush() - except (OSError, BrokenPipeError) as e: - return f"[错误] 发送失败: {e}" - - # 等待哨兵文件出现 - real_marker = marker if not _IS_WIN else os.path.expandvars(marker) - deadline = time.time() + timeout - while time.time() < deadline: - if os.path.exists(real_marker): - break - time.sleep(0.03) - - # 给 stdout 一点时间把剩余输出冲完 - time.sleep(0.1) - - # 收集输出 - with self._lock: - out = self._raw_buf - self._raw_buf = "" - - # 清理哨兵文件 - try: - os.unlink(real_marker) - except OSError: + except (OSError, BrokenPipeError): pass - # 过滤掉命令回显和哨兵行 - lines = out.split("\n") - result = [] - for line in lines: - s = line.strip() - if s == command.strip(): - continue - if ".xli_" in s: - continue - if s.startswith("touch ") or s.startswith("type nul"): - continue - result.append(line) + # ── 核心操作：按键 ── - out = self._clean("\n".join(result)) + _KEYS = { + "enter": "\n", "return": "\n", + "tab": "\t", + "backspace": "\x7f", "bs": "\x7f", + "esc": "\x1b", "escape": "\x1b", + "space": " ", + "delete": "\x1b[3~", "del": "\x1b[3~", + "insert": "\x1b[2~", + "up": "\x1b[A", "down": "\x1b[B", + "right": "\x1b[C", "left": "\x1b[D", + "home": "\x1b[H", "end": "\x1b[F", + "pageup": "\x1b[5~", "pagedown": "\x1b[6~", + "f1": "\x1bOP", "f2": "\x1bOQ", "f3": "\x1bOR", "f4": "\x1bOS", + "f5": "\x1b[15~", "f6": "\x1b[17~", "f7": "\x1b[18~", "f8": "\x1b[19~", + "f9": "\x1b[20~", "f10": "\x1b[21~", "f11": "\x1b[23~", "f12": "\x1b[24~", + } + + # Ctrl+ 字母映射 + _CTRL = {chr(i): chr(i - 64) for i in range(65, 91)} # A-Z → \x01-\x1a + + def key(self, name: str): + """ + 按一个键。 + 支持: enter, tab, esc, up, down, left, right, + ctrl+c, ctrl+d, ctrl+z, ctrl+l, ctrl+a, ctrl+e, + f1-f12, backspace, delete, home, end, space, ... + """ + if not self.alive(): + return + name = name.lower().strip() + + # Ctrl+ 组合 + if name.startswith("ctrl+"): + letter = name[5:] + seq = self._CTRL.get(letter.upper()) + if seq: + self._send_raw(seq) + return + + # 普通按键 + seq = self._KEYS.get(name) + if seq: + self._send_raw(seq) + elif len(name) == 1: + # 单个字符直接发 + self._send_raw(name) + + def combo(self, *keys): + """ + 组合键，如 combo("ctrl", "shift", "t") + 目前主要支持 ctrl+单字母 + """ + parts = [k.lower() for k in keys] + if "ctrl" in parts and len(parts) == 2: + other = [k for k in parts if k != "ctrl"][0] + self.key(f"ctrl+{other}") + elif "alt" in parts: + # Alt+key → ESC + key + other = [k for k in parts if k != "alt"][0] + self._send_raw("\x1b" + other) + elif "shift" in parts: + other = [k for k in parts if k != "shift"][0] + self._send_raw(other.upper() if len(other) == 1 else other) + + def _send_raw(self, data: str): + try: + self._proc.stdin.write(data.encode("utf-8")) + self._proc.stdin.flush() + except (OSError, BrokenPipeError): + pass + + # ── 核心操作：看屏幕 ── + + def read(self, lines: int = 0) -> str: + """读取屏幕当前内容""" with self._lock: - self._last_cmd_output = out - return out - - def _collect(self, wait: float, timeout: float) -> str: - """收集输出""" - time.sleep(wait) - deadline = time.time() + timeout - buf = "" - while time.time() < deadline: - with self._lock: - new = self._raw_buf - self._raw_buf = "" - buf += new - if new: - time.sleep(0.05) + if lines > 0: + buf = list(self._screen)[-lines:] else: - time.sleep(0.1) - with self._lock: - new = self._raw_buf - self._raw_buf = "" - buf += new - if not new: - break - return self._clean(buf) + buf = list(self._screen) + return self._clean("\n".join(buf)) + + def read_new(self, timeout: float = 1) -> str: + """等待并读取新输出""" + with self._lock: + self._raw = "" + time.sleep(timeout) + with self._lock: + out = self._raw + self._raw = "" + return self._clean(out) + + def wait(self, seconds: float = 1): + """等待，让终端输出完""" + time.sleep(seconds) def _clean(self, text: str) -> str: if not text: return "" - # 去 ANSI 转义 text = re.sub(r'\x1b\[[0-9;]*[a-zA-Z]', '', text) text = re.sub(r'\x1b\][^\x07]*\x07', '', text) text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f]', '', text) @@ -247,38 +206,10 @@ class TerminalSession: text = re.sub(r'\n{4,}', '\n\n\n', text) return text.strip() - def alive(self) -> bool: - return self._proc is not None and self._proc.poll() is None - - def recent(self, lines: int = 50) -> str: - """返回原始输出缓冲的最近几行""" - with self._lock: - buf = list(self._output_buf)[-lines:] - return self._clean("\n".join(buf)) - - def status(self) -> dict: - with self._lock: - lines = len(self._last_cmd_output.split('\n')) if self._last_cmd_output else 0 - return { - "id": self.id, - "label": self.label, - "shell": self.shell, - "cwd": self.cwd, - "alive": self.alive(), - "pid": self._proc.pid if self._proc else None, - "uptime": round(time.time() - self.created_at, 1), - "idle": round(time.time() - self.last_active, 1), - "lines": lines, - } - def close(self): try: if self._proc and self._proc.poll() is None: - try: - self._proc.stdin.write(b"exit\n") - self._proc.stdin.flush() - except Exception: - pass + self._send_raw("exit\n") time.sleep(0.2) if self._proc.poll() is None: self._proc.terminate() @@ -296,12 +227,12 @@ class TerminalSession: # ═══════════════════════════════════════════════════════════════ class Liugin: - """终端模拟器 — 纯标准库，全平台""" + """终端模拟器 v2 — AI 的虚拟键盘和屏幕""" def __init__(self): self.cli = None - self._sessions: dict = {} - self._default_id = "main" + self._terminals: dict = {} + self._current = "main" def set_cli(self, cli): self.cli = cli @@ -309,119 +240,88 @@ class Liugin: def get_tool_info(self): return { "name": "terminal", - "description": "交互式终端模拟器 — 持久 Shell 会话，纯标准库实现，" - "Windows/Linux/macOS 通用，支持 Python REPL、SSH 等交互式程序", + "description": "终端模拟器 — AI 的虚拟键盘和屏幕。" + "启动终端、打字、按键、看屏幕，像人一样操作。", "keywords": [ - "终端", "terminal", "shell", "交互", "会话", "REPL", - "持久", "cmd", "powershell", "bash", "pty", "windows", + "终端", "terminal", "shell", "键盘", "按键", "打字", + "屏幕", "交互", "REPL", "cmd", "bash", "python", ], "usage": ( "terminal <操作> [参数]\n\n" - "操作:\n" - " open [shell] [label] - 打开新终端 (默认: 系统 shell)\n" - " send <命令> [--timeout N] - 发送命令并获取输出\n" - " input <文本> - 发送纯输入 (不换行)\n" - " read [行数] - 读取最近输出\n" - " switch - 切换会话\n" - " list - 列出所有会话\n" - " close [ID] - 关闭会话\n" - " status - 当前状态\n" - " cwd [路径] - 获取/切换目录\n" - " env [VALUE] - 环境变量\n" - " keys <按键名> - 发送特殊按键\n\n" + " open [shell] - 打开终端，返回初始画面\n" + " type <文字> - 打字（发送到终端）\n" + " key <按键名> - 按一个键（enter/tab/ctrl+c/方向键...）\n" + " read [行数] - 看屏幕内容\n" + " last [行数] - 看最近N行（默认20，避免上下文爆炸）\n" + " wait [秒] - 等待输出完毕\n" + " switch - 切换终端\n" + " list - 列出所有终端\n" + " close [ID] - 关闭终端\n\n" + "按键: enter tab esc space backspace delete\n" + " up down left right home end\n" + " ctrl+c ctrl+d ctrl+z ctrl+l ctrl+a ctrl+e\n" + " f1-f12\n\n" "示例:\n" - " terminal open - 打开默认 shell\n" - " terminal open python3 \"Python\" - 打开 Python REPL\n" - " terminal send dir - Windows 列目录\n" - " terminal send ls -la - Linux 列目录\n" - " terminal send ssh user@host - SSH 连接\n" - " terminal keys Ctrl+C - 中断程序\n" - " terminal close - 关闭当前会话" + " terminal open - 打开 bash\n" + " terminal open python3 - 打开 Python\n" + " terminal type ls -la - 打字\n" + " terminal key enter - 按回车\n" + " terminal read - 看屏幕\n" + " terminal key ctrl+c - 中断" ), } def get_mcp_definition(self): return { "name": "terminal", - "description": "交互式终端模拟器 — 持久 Shell 会话，支持交互式程序", + "description": "终端模拟器 — AI 的虚拟键盘和屏幕", "inputSchema": { "type": "object", "properties": { "action": { "type": "string", - "enum": ["open", "send", "input", "read", "switch", - "list", "close", "status", "cwd", "env", "keys"], - "description": "操作类型", + "enum": ["open", "type", "key", "read", "last", "wait", + "switch", "list", "close"], }, - "command": {"type": "string", "description": "命令或文本"}, - "session_id": {"type": "string", "description": "会话 ID"}, - "shell": {"type": "string", "description": "Shell 路径"}, - "label": {"type": "string", "description": "会话标签"}, - "timeout": {"type": "number", "description": "超时秒数 (默认 15)"}, - "lines": {"type": "integer", "description": "读取行数"}, - "value": {"type": "string", "description": "环境变量值"}, - "keys": {"type": "string", "description": "特殊按键"}, + "text": {"type": "string", "description": "要打的字 / shell路径"}, + "key": {"type": "string", "description": "按键名"}, + "last": {"type": "integer", "description": "最近N行"}, + "seconds": {"type": "number", "description": "等待秒数"}, + "session": {"type": "string", "description": "终端 ID"}, }, "required": ["action"], }, } def convert_mcp_args(self, arguments): - action = arguments.get("action", "") - parts = [action] - cmd = arguments.get("command", "") - if cmd: - parts.append(cmd) - sid = arguments.get("session_id", "") - if sid: - parts.append(f"--sid {sid}") - shell = arguments.get("shell", "") - if shell: - parts.append(f"--shell {shell}") - label = arguments.get("label", "") - if label: - parts.append(f"--label {label}") - timeout = arguments.get("timeout") - if timeout is not None: - parts.append(f"--timeout {timeout}") - lines = arguments.get("lines") - if lines is not None: - parts.append(f"--lines {lines}") - value = arguments.get("value") - if value is not None: - parts.append(value) - keys = arguments.get("keys") - if keys: - parts.append(keys) + a = arguments.get("action", "") + parts = [a] + for k in ["text", "key", "lines", "seconds", "session"]: + v = arguments.get(k) + if v is not None: + parts.append(str(v)) return " ".join(parts) - _KEY_MAP = { - "ctrl+c": "\x03", "ctrl+d": "\x04", "ctrl+z": "\x1a", - "ctrl+l": "\x0c", "ctrl+a": "\x01", "ctrl+e": "\x05", - "ctrl+k": "\x0b", "ctrl+u": "\x15", "ctrl+w": "\x17", - "tab": "\t", "enter": "\n", - "up": "\x1b[A", "down": "\x1b[B", - "right": "\x1b[C", "left": "\x1b[D", - "home": "\x1b[H", "end": "\x1b[F", - "backspace": "\x7f", "esc": "\x1b", - "pageup": "\x1b[5~", "pagedown": "\x1b[6~", - } - def handle(self, args: str) -> str: parts = args.strip().split(maxsplit=1) if not parts: return self._help() action = parts[0].lower() rest = parts[1].strip() if len(parts) > 1 else "" + + # 支持 `read 20` 省写 + if rest.isdigit() and action in ('read', 'last'): + action, rest = 'last', rest + h = { - "open": self._h_open, "send": self._h_send, - "input": self._h_input, "read": self._h_read, + "open": self._h_open, "type": self._h_type, + "key": self._h_key, "read": self._h_read, + "last": self._h_last, "wait": self._h_wait, "switch": self._h_switch, "list": self._h_list, - "close": self._h_close, "status": self._h_status, - "cwd": self._h_cwd, "env": self._h_env, "keys": self._h_keys, + "close": self._h_close, }.get(action) if not h: - return f"错误: 不支持 '{action}'\n{self._help()}" + return f"不支持 '{action}'\n{self._help()}" try: return h(rest) except Exception as e: @@ -429,174 +329,83 @@ class Liugin: def _help(self): return ( - "终端模拟器:\n" - " open [shell] [label] - 打开新会话\n" - " send <命令> - 执行命令\n" - " input <文本> - 纯输入\n" - " read [行数] - 读输出\n" - " switch/list/close - 会话管理\n" - " cwd/env/keys - 目录/变量/按键" + "terminal open [shell] - 打开终端\n" + "terminal type <文字> - 打字\n" + "terminal key <按键> - 按键\n" + "terminal read [行数] - 看屏幕\n" + "terminal wait [秒] - 等待\n" + "terminal list - 列出终端\n" + "terminal close - 关闭" ) - def _get(self, sid: str = "") -> TerminalSession: - sid = sid or self._default_id - if sid not in self._sessions: - self._sessions[sid] = TerminalSession(sid) - return self._sessions[sid] - - def _opts(self, rest: str) -> dict: - opts = {} - for m in re.finditer(r'--(\w+)\s+(\S+)', rest): - opts[m.group(1)] = m.group(2) - opts["_"] = re.sub(r'--\w+\s+\S+', '', rest).strip() - return opts + def _get(self, sid: str = "") -> Terminal: + sid = sid or self._current + if sid not in self._terminals: + self._terminals[sid] = Terminal(sid) + return self._terminals[sid] def _h_open(self, rest: str) -> str: - o = self._opts(rest) - shell = o.get("shell", "") - label = o.get("label", "") - sid = o.get("sid", "") - pos = o.get("_", "") - if pos and not shell: - ps = pos.split() - shell = ps[0] - if len(ps) > 1 and not label: - label = ps[1] - if not sid: - sid = f"s{len(self._sessions) + 1}" if self._sessions else self._default_id - try: - s = TerminalSession(sid, shell=shell, label=label) - self._sessions[sid] = s - self._default_id = sid # 自动切换到新会话 - info = s.status() - return ( - f"✓ 终端已打开\n" - f" 会话: {sid} ({info['label']})\n" - f" Shell: {info['shell']}\n" - f" PID: {info['pid']}\n" - f" 输出:\n{s.recent(10) or '(空)'}" - ) - except RuntimeError as e: - return f"✗ 打开失败: {e}" + sid = f"s{len(self._terminals) + 1}" if self._terminals else "main" + t = Terminal(sid, shell=rest.strip()) + self._terminals[sid] = t + self._current = sid + screen = t.read() + return f"✓ 终端 {sid} 已打开\n{screen}" - def _h_send(self, rest: str) -> str: + def _h_type(self, rest: str) -> str: if not rest: - return "错误: 请提供命令" - o = self._opts(rest) - cmd = o.get("_", rest).strip() - timeout = float(o.get("timeout", "15")) - if not cmd: - cmd = re.sub(r'--timeout\s+\S+', '', rest).strip() - if not cmd: - return "错误: 请提供命令" - s = self._get() - out = s.send_cmd(cmd, timeout=timeout) - tag = "✓" if s.alive() else "✗" - return f"{tag} {cmd}\n{out}" + return "请提供要打的字" + t = self._get() + t.type(rest) + t.wait(0.3) + return f"← 打字: {rest}" - def _h_input(self, rest: str) -> str: + def _h_key(self, rest: str) -> str: if not rest: - return "错误: 请提供文本" - s = self._get() - # 加换行让 REPL 执行 - out = s.send(rest + "\n", wait=0.5) - with s._lock: - s._last_cmd_output = out - return f"← 输入: {rest}\n{out}" + return "请提供按键名" + t = self._get() + t.key(rest) + t.wait(0.3) + return f"← 按键: {rest}" def _h_read(self, rest: str) -> str: - """读取输出 — 优先用缓存，否则从原始缓冲读""" - s = self._get() - with s._lock: - cached = s._last_cmd_output - if cached: - return cached - # REPL 场景：从原始缓冲读 - return s.recent(30) or "(无输出)" + t = self._get() + return t.read() or "(屏幕为空)" + + def _h_last(self, rest: str) -> str: + t = self._get() + n = int(rest) if rest.strip().isdigit() else 20 + return t.read(n) or "(屏幕为空)" + + def _h_wait(self, rest: str) -> str: + secs = float(rest) if rest.strip() else 1.0 + t = self._get() + t.wait(secs) + return t.read() def _h_switch(self, rest: str) -> str: sid = rest.strip() - if not sid: - return "错误: 请提供会话 ID" - if sid not in self._sessions: - return f"错误: 不存在 '{sid}'。可用: {', '.join(self._sessions)}" - self._default_id = sid - s = self._sessions[sid] - return f"✓ 已切换到 {sid} ({s.label})" + if sid not in self._terminals: + return f"终端 '{sid}' 不存在" + self._current = sid + return f"✓ 切换到 {sid}" def _h_list(self, _: str) -> str: - if not self._sessions: - return "暂无终端。用 terminal open 打开。" - lines = [f"终端 ({len(self._sessions)} 个):", ""] - for sid, s in self._sessions.items(): - st = s.status() - cur = " ← 当前" if sid == self._default_id else "" - dot = "●" if st["alive"] else "○" - lines.append( - f" {dot} {sid}: {st['label']} " - f"[{st['shell']}] " - f"空闲{st['idle']}s " - f"{st['lines']}行{cur}" - ) + if not self._terminals: + return "无终端" + lines = [] + for sid, t in self._terminals.items(): + cur = " ← 当前" if sid == self._current else "" + dot = "●" if t.alive() else "○" + lines.append(f" {dot} {sid} [{t.shell}]{cur}") return "\n".join(lines) def _h_close(self, rest: str) -> str: - sid = rest.strip() or self._default_id - if sid not in self._sessions: - return f"错误: 不存在 '{sid}'" - s = self._sessions.pop(sid) - s.close() - if self._default_id == sid: - self._default_id = next(iter(self._sessions), "main") - return f"✓ 终端 {sid} ({s.label}) 已关闭" - - def _h_status(self, _: str) -> str: - s = self._get() - st = s.status() - return ( - f"终端状态:\n" - f" 会话: {st['id']} ({st['label']})\n" - f" Shell: {st['shell']}\n" - f" PID: {st['pid']}\n" - f" 目录: {st['cwd']}\n" - f" 状态: {'运行中' if st['alive'] else '已退出'}\n" - f" 运行: {st['uptime']}s 空闲: {st['idle']}s\n" - f" 输出: {st['lines']}行" - ) - - def _h_cwd(self, rest: str) -> str: - s = self._get() - target = rest.strip() - if not target: - return f"当前目录: {s.send_cmd('cd' if _IS_WIN else 'pwd', timeout=5)}" - cmd = f"cd /d {target} && cd" if _IS_WIN else f"cd {target} && pwd" - out = s.send_cmd(cmd, timeout=5) - if out and not out.startswith("[错误"): - s.cwd = out.strip().split("\n")[-1].strip() - return f"✓ 目录: {s.cwd}" - return f"切换失败: {out}" - - def _h_env(self, rest: str) -> str: - s = self._get() - parts = rest.strip().split(maxsplit=1) - if not parts: - return s.send_cmd("set" if _IS_WIN else "env | head -30", timeout=5) - key = parts[0] - if len(parts) == 1: - var = f"%{key}%" if _IS_WIN else f"${key}" - return s.send_cmd(f"echo {var}", timeout=5) - val = parts[1] - cmd = f"set {key}={val}" if _IS_WIN else f"export {key}={val}" - s.send_cmd(cmd, timeout=5) - return f"✓ {key}={val}" - - def _h_keys(self, rest: str) -> str: - if not rest: - return f"可用: {', '.join(self._KEY_MAP)}" - s = self._get() - name = rest.strip().lower() - seq = self._KEY_MAP.get(name) - if not seq: - return f"错误: 未知按键 '{name}'\n可用: {', '.join(self._KEY_MAP)}" - out = s.send(seq, wait=0.3) - return f"← {name}\n{out}" + sid = rest.strip() or self._current + if sid not in self._terminals: + return f"终端 '{sid}' 不存在" + self._terminals[sid].close() + del self._terminals[sid] + if self._current == sid: + self._current = next(iter(self._terminals), "main") + return f"✓ 终端 {sid} 已关闭"