diff --git a/README.md b/README.md index 5a0c81c..2fd2641 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,9 @@ curl -fsSL https://gitea.xbay.cc/leefer/ssh-e2e/raw/branch/main/install.sh | bas 安装完成会**自动进入中文配置向导**:按提示输入 SSH 地址、密码、远程端口等 (直接回车采用方括号里的默认值),填完隧道立即建立,Ctrl+C 断开。 +命令行界面支持真彩色(自动适配终端,`NO_COLOR` 关闭):方向键 `❯` 选择菜单、 +隧道运行时的实时状态面板(连接状态 · 运行时长 · 累计连接 · 滚动日志)。 + ## 典型场景 远程主机上有个服务只监听它自己的 `127.0.0.1:3000`,公网访问不到。 diff --git a/cli.py b/cli.py index 27aac6e..7eea991 100644 --- a/cli.py +++ b/cli.py @@ -23,6 +23,7 @@ sys.path.insert(0, str(Path(__file__).parent)) from engine import TunnelManager, probe_host # noqa: E402 from storage import Store, data_root # noqa: E402 +import tui # noqa: E402 def _print(msg: str) -> None: @@ -30,26 +31,32 @@ def _print(msg: str) -> None: class _QuietEmit: - """CLI 模式下把引擎事件打到终端。""" + """引擎事件 → 运行面板(无面板时打印到终端)。""" - def __init__(self): + def __init__(self, dash: "tui.Dashboard | None" = None): self._last: dict[str, str] = {} + self._dash = dash def __call__(self, ev: dict) -> None: if ev.get("type") == "tunnel": if self._last.get(ev["id"]) == ev["status"]: - return # 状态未变(如连接计数刷新),不重复打印 + return # 状态未变(如连接计数刷新) self._last[ev["id"]] = ev["status"] - t = ev["status"] - if t == "connected": - _print("[隧道] 已连接") - elif t == "error": - _print(f"[隧道] 连接失败: {ev.get('error', '')}") + if ev["status"] == "error" and self._dash: + self._dash.log(tui.err(f"✘ {ev.get('error', '')}")) + elif not self._dash: + if ev["status"] == "connected": + _print(tui.ok("[隧道] 已连接")) + elif ev["status"] == "error": + _print(tui.err(f"[隧道] 连接失败: {ev.get('error', '')}")) elif ev.get("type") == "log": e = ev["entry"] - lv = e["level"] - mark = {"info": "·", "success": "✔", "warn": "!", "error": "×"}.get(lv, "·") - _print(f"[{e['t']}] {mark} {e['msg']}") + mark = {"info": "·", "success": "✔", "warn": "!", "error": "×"}.get(e["level"], "·") + line = f"{e['t']} {mark} {e['msg']}" + if self._dash: + self._dash.log(line) + else: + _print(line) def find_host(store: Store, ref: str): @@ -113,19 +120,22 @@ def cmd_tunnel_add(store, args) -> int: def cmd_list(store, _args) -> int: if store.warn_plaintext: _print("提示: 当前平台密码以明文保存于配置文件(权限已设为仅本用户可读)。") - _print("主机:") + print(tui.bold("主机")) if not store.data["hosts"]: - _print(" (无) 添加: ssh-tunnel host add <名称> user@host") + print(tui.faint(" (无) ssh-tunnel 进入向导添加")) for h in store.data["hosts"]: - _print(f" {h['name']} {h['username']}@{h['hostname']}:{h['port']} " - f"{'私钥' if h.get('auth') == 'key' else '密码'}") - _print("隧道:") + auth = "私钥" if h.get("auth") == "key" else "密码" + print(f" {tui.accent('▣')} {h['name']} " + f"{tui.dim(f'{h["username"]}@{h["hostname"]}:{h["port"]}')} {tui.faint(auth)}") + print(tui.bold("隧道")) if not store.data["tunnels"]: - _print(" (无) 添加: ssh-tunnel tunnel add <名称> --host <主机> --lp 本地端口 --rp 远程端口") + print(tui.faint(" (无) ssh-tunnel 进入向导添加")) for t in store.data["tunnels"]: h = next((x for x in store.data["hosts"] if x["id"] == t["host_id"]), None) hn = h["name"] if h else "(主机缺失)" - _print(f" {t['name']} 127.0.0.1:{t['local_port']} → {t['remote_host']}:{t['remote_port']} 经 {hn}") + print(f" {tui.accent('⇄')} {t['name']} " + f"{tui.dim(f'127.0.0.1:{t["local_port"]} → {t["remote_host"]}:{t["remote_port"]}')} " + f"{tui.faint(f'经 {hn}')}") return 0 @@ -165,23 +175,36 @@ def cmd_remove(store, args) -> int: def run_foreground(store: Store, chosen: list) -> int: - """前台运行选中的隧道,Ctrl+C 退出。""" + """前台运行选中的隧道,Ctrl+C 退出。tty 下显示实时状态面板。""" hosts_by_id = {h["id"]: h for h in store.data["hosts"]} - mgr = TunnelManager(_QuietEmit(), data_root()) + dash = tui.Dashboard() + mgr = TunnelManager(_QuietEmit(dash), data_root()) mgr.apply_config(chosen, hosts_by_id) if store.warn_plaintext: _print("提示: 当前平台密码以明文保存于配置文件(权限已设为仅本用户可读)。") - names = ", ".join(t["name"] for t in chosen) - _print(f"启动隧道: {names} — Ctrl+C 退出") + print() + print(tui.faint(" " + ", ".join(t["name"] for t in chosen) + " — Ctrl+C 退出")) stopping = False + def _stats(): + return [{ + "name": t.cfg.get("name", t.cfg["id"]), + "local_port": t.cfg["local_port"], + "remote": f"{t.cfg['remote_host']}:{t.cfg['remote_port']}", + "status": t.status, "started_at": t.started_at, + "conns": t.conn_count, "error": t.error, + } for t in mgr.tunnels.values()] + def _stop(*_a): nonlocal stopping if not stopping: stopping = True - _print("\n正在断开所有隧道 ...") + print() # 光标移到面板下方行首 + dash.close() + _print("正在断开所有隧道 ...") + sys.stdout.flush() mgr.stop_all() signal.signal(signal.SIGINT, _stop) @@ -190,13 +213,15 @@ def run_foreground(store: Store, chosen: list) -> int: t.start() try: while not stopping: - time.sleep(0.3) - alive = [t for t in mgr.tunnels.values() if t.status != "stopped"] - if not alive: - _print("所有隧道已停止。") + time.sleep(0.5) + dash.update(_stats()) + if not [t for t in mgr.tunnels.values() if t.status != "stopped"]: + dash.close() + _print(tui.err("所有隧道已停止。")) return 1 except KeyboardInterrupt: _stop() + dash.close() return 0 @@ -269,34 +294,31 @@ def _ask_secret(prompt: str) -> str: def _wizard_pick_host(store: Store): """选择已有主机或新建;返回主机 dict 或 None(取消)。""" hosts = store.data["hosts"] - _print("\n── 第 1 步:选择 SSH 主机 ──") - for i, h in enumerate(hosts, 1): - _print(f" {i}. {h['name']} ({h['username']}@{h['hostname']}:{h['port']})") - _print(f" {len(hosts) + 1}. 新增主机") - - while True: - sel = _ask(f"请选择 [1-{len(hosts) + 1}]", str(len(hosts) + 1 if not hosts else "1")) - if not sel.isdigit() or not (1 <= int(sel) <= len(hosts) + 1): - _print(" 无效选择,请重新输入") - continue - n = int(sel) - if n <= len(hosts): - return hosts[n - 1] - return _wizard_new_host(store) + print() + print(tui.bold("◇ SSH 主机")) + opts = [f"{h['name']} ({h['username']}@{h['hostname']}:{h['port']})" for h in hosts] + opts.append(tui.dim("＋ 新增主机") if len(hosts) else "＋ 新增主机") + n = tui.select(" 选择要使用的主机:", opts, default=len(hosts) if not hosts else 0) + if n is None: + return None + if n < len(hosts): + return hosts[n] + return _wizard_new_host(store) def _wizard_new_host(store: Store): - _print("\n新增主机(带 * 的为必填):") + print() + print(tui.bold("◇ 新增主机") + tui.faint(" (带 * 为必填)")) while True: - target = _ask(" * SSH 登录目标 (格式 用户名@主机,如 root@1.2.3.4): ") + target = _ask(" " + tui.accent("SSH 登录目标") + tui.faint(" (用户名@主机,如 root@1.2.3.4)") + ": ") if "@" in target and len(target.split("@")) == 2 and target.split("@")[1]: break - _print(" 格式不对,应类似 root@1.2.3.4") + print(tui.warn(" 格式不对,应类似 root@1.2.3.4")) user, hostaddr = target.rsplit("@", 1) - port = _ask_port(" SSH 端口 [22]: ", 22) - auth = _ask(" 认证方式 1) 密码 2) 私钥 [1]: ", "1") + port = _ask_port(" " + tui.accent("SSH 端口") + tui.faint(" [22]"), 22) + auth = _ask(" " + tui.accent("认证方式") + tui.faint(" 1)密码 2)私钥 [1]"), "1") while auth not in ("1", "2"): - auth = _ask(" 请输入 1 或 2: ", "1") + auth = _ask(" 请输入 1 或 2: ", "1") rec = { "id": Store.new_id(), "name": "", "hostname": hostaddr, "port": port, @@ -304,61 +326,59 @@ def _wizard_new_host(store: Store): "password": "", "key_path": "", "key_passphrase": "", "note": "", } if rec["auth"] == "key": - rec["key_path"] = _ask(" * 私钥文件路径 (如 ~/.ssh/id_rsa): ") + rec["key_path"] = _ask(" " + tui.accent("* 私钥文件路径") + tui.faint(" (如 ~/.ssh/id_rsa)") + ": ") rec["key_passphrase"] = _ask_secret(" 私钥口令 (没有则直接回车): ") else: - rec["password"] = _ask_secret(" * 登录密码 (输入时不显示): ") - rec["name"] = _ask(" 主机备注名 [默认取地址]: ", hostaddr) + rec["password"] = _ask_secret(" " + tui.accent("* 登录密码") + tui.faint(" (输入时不显示)") + ": ") + rec["name"] = _ask(" " + tui.accent("主机备注名") + tui.faint(f" [{hostaddr}]") + ": ", hostaddr) # 连通测试,失败可重试输入 while True: - _print(f"正在测试 {user}@{hostaddr}:{port} ...") + print(tui.faint(f" 正在测试 {user}@{hostaddr}:{port} ...")) ok, msg = probe_host(rec, data_root() / "known_hosts") - _print((" ✔ " if ok else " ✘ ") + msg) + print((" " + tui.ok("✔ ") + tui.ok(msg)) if ok else (" " + tui.err("✘ ") + tui.err(msg))) if ok: break - retry = _ask(" 连接失败。重试请回车,修改密码输入 p,放弃输入 q [回车重试]: ", "") + retry = _ask(" " + tui.warn("连接失败") + tui.faint(" — 回车重试 / p 改密码 / q 放弃") + ": ", "") if retry.lower() == "q": return None if retry.lower() == "p" and rec["auth"] == "password": - rec["password"] = _ask_secret(" * 重新输入登录密码: ") + rec["password"] = _ask_secret(" 重新输入登录密码: ") store.data["hosts"].append(rec) store.save() - _print(f"已保存主机「{rec['name']}」") + print(" " + tui.ok(f"已保存主机「{rec['name']}」")) return rec def _wizard_pick_tunnel(store: Store, host: dict): """选择该主机的已有隧道或新建;返回隧道 dict 或 None。""" ts = [t for t in store.data["tunnels"] if t["host_id"] == host["id"]] - _print("\n── 第 2 步:配置隧道 ──") - for i, t in enumerate(ts, 1): - _print(f" {i}. {t['name']} (127.0.0.1:{t['local_port']} → {t['remote_host']}:{t['remote_port']})") - _print(f" {len(ts) + 1}. 新建隧道") - - while True: - sel = _ask(f"请选择 [1-{len(ts) + 1}]", str(len(ts) + 1 if not ts else "1")) - if not sel.isdigit() or not (1 <= int(sel) <= len(ts) + 1): - _print(" 无效选择,请重新输入") - continue - n = int(sel) - if n <= len(ts): - return ts[n - 1] - return _wizard_new_tunnel(store, host) + print() + print(tui.bold("◇ 隧道")) + opts = [f"{t['name']} (127.0.0.1:{t['local_port']} → {t['remote_host']}:{t['remote_port']})" + for t in ts] + opts.append("＋ 新建隧道") + n = tui.select(" 选择要启动的隧道:", opts, default=len(ts) if not ts else 0) + if n is None: + return None + if n < len(ts): + return ts[n] + return _wizard_new_tunnel(store, host) def _wizard_new_tunnel(store: Store, host: dict): - _print("\n新建隧道:把远程主机上的服务映射到本机端口") - _print(" (远程服务只监听 127.0.0.1 时,远程地址保持默认即可)") - rh = _ask(" 远程服务地址 [127.0.0.1]: ", "127.0.0.1") - rp = _ask_port(" * 远程服务端口 (如 3000): ") - lp = _ask_port(f" 本地端口 [{rp}]: ", rp) + print() + print(tui.bold("◇ 新建隧道") + tui.faint(" 把远程主机上的服务映射到本机端口")) + print(tui.faint(" (远程服务只监听 127.0.0.1 时,远程地址保持默认即可)")) + rh = _ask(" " + tui.accent("远程服务地址") + tui.faint(" [127.0.0.1]") + ": ", "127.0.0.1") + rp = _ask_port(" " + tui.accent("* 远程服务端口") + tui.faint(" (如 3000)") + ": ") + lp = _ask_port(" " + tui.accent("本地端口") + tui.faint(f" [{rp}]") + ": ", rp) for t in store.data["tunnels"]: if t["local_port"] == lp: - _print(f" 本地端口 {lp} 已被隧道「{t['name']}」占用") + print(tui.warn(f" 本地端口 {lp} 已被隧道「{t['name']}」占用")) lp = _ask_port(" 换一个本地端口: ") - name = _ask(f" 隧道备注名 [{rh}:{rp}]: ", f"{rh}:{rp}") + name = _ask(" " + tui.accent("隧道备注名") + tui.faint(f" [{rh}:{rp}]") + ": ", f"{rh}:{rp}") rec = { "id": Store.new_id(), "name": name, "host_id": host["id"], "local_port": lp, "remote_host": rh, "remote_port": rp, @@ -366,17 +386,20 @@ def _wizard_new_tunnel(store: Store, host: dict): } store.data["tunnels"].append(rec) store.save() - _print(f"已保存隧道「{name}」") + print(" " + tui.ok(f"已保存隧道「{name}」")) return rec def wizard(store: Store) -> int: _ensure_tty_stdin() - _print("╭─────────────────────────────╮") - _print("│ SSH 隧道配置向导 │") - _print("╰─────────────────────────────╯") - _print("按中文提示输入,直接回车采用 [方括号] 里的默认值;Ctrl+C 随时退出。") - _print(f"配置文件: {store.path}") + title = "SSH 隧道配置向导" + sub = "回车采用 [方括号] 默认值 · Ctrl+C 随时退出" + w = max(tui.disp_len(title) + 4, tui.disp_len(sub) + 4, 40) + print() + print(tui.accent("╭─ ") + tui.bold(title) + " " + tui.accent("─" * (w - tui.disp_len(title) - 3) + "╮")) + print(tui.accent("│") + " " + tui.faint(sub) + " " * (w - tui.disp_len(sub) - 2) + tui.accent("│")) + print(tui.accent("╰" + "─" * w + "╯")) + print(tui.faint(f" 配置文件: {store.path}")) if store.warn_plaintext: _print("提示: 当前平台密码以明文保存于配置文件(权限已设为仅本用户可读)。") diff --git a/tui.py b/tui.py new file mode 100644 index 0000000..c5375be --- /dev/null +++ b/tui.py @@ -0,0 +1,282 @@ +"""终端界面组件:真彩色样式、方向键选择菜单、隧道运行面板。 +非 tty 或设置 NO_COLOR 时自动降级为纯文本,行为不变。 +""" +from __future__ import annotations + +import os +import sys +import threading +import time +from collections import deque + +# ---------- 终端能力 ---------- + +_COLOR = None + + +def _use_color() -> bool: + global _COLOR + if _COLOR is not None: + return _COLOR + if os.environ.get("NO_COLOR"): + _COLOR = False + elif os.environ.get("FORCE_COLOR"): + _COLOR = True + else: + _COLOR = sys.stdout.isatty() + if _COLOR and sys.platform == "win32": + _enable_vt() + return _COLOR + + +def _enable_vt() -> None: + try: + import ctypes + k32 = ctypes.windll.kernel32 + h = k32.GetStdHandle(-11) # STD_OUTPUT_HANDLE + mode = ctypes.c_uint32() + if k32.GetConsoleMode(h, ctypes.byref(mode)): + k32.SetConsoleMode(h, mode.value | 0x4) # ENABLE_VIRTUAL_TERMINAL_PROCESSING + except Exception: + pass + + +# ---------- 调色板(与 GUI 同源:石墨底 + 青绿点缀) ---------- + +ACCENT = (47, 191, 161) # 青绿 +DIM = (151, 161, 173) # 次级文字 +FAINT = (110, 120, 131) # 弱化文字 +GREEN = (70, 196, 110) +RED = (226, 86, 77) +AMBER = (217, 162, 60) +TEXT = (226, 231, 238) + + +def _c(s: str, rgb: tuple, bold: bool = False) -> str: + if not _use_color(): + return s + r, g, b = rgb + prefix = f"\x1b[1m" if bold else "" + return f"{prefix}\x1b[38;2;{r};{g};{b}m{s}\x1b[0m" + + +def accent(s: str) -> str: return _c(s, ACCENT) +def dim(s: str) -> str: return _c(s, DIM) +def faint(s: str) -> str: return _c(s, FAINT) +def ok(s: str) -> str: return _c(s, GREEN) +def err(s: str) -> str: return _c(s, RED) +def warn(s: str) -> str: return _c(s, AMBER) +def bold(s: str) -> str: return _c(s, TEXT, bold=True) + + +# ---------- 显示宽度(CJK 全角按 2 列) ---------- + +_WIDE = ((0x1100, 0x115F), (0x2E80, 0x303E), (0x3041, 0x33FF), (0x3400, 0x4DBF), + (0x4E00, 0x9FFF), (0xA000, 0xA4CF), (0xAC00, 0xD7A3), (0xF900, 0xFAFF), + (0xFE30, 0xFE4F), (0xFF00, 0xFF60), (0xFFE0, 0xFFE6)) + + +def disp_len(s: str) -> int: + n = 0 + for ch in s: + o = ord(ch) + n += 2 if any(a <= o <= b for a, b in _WIDE) else 1 + return n + + +# ---------- 方向键选择菜单 ---------- + +_UP, _DOWN, _ENTER, _ESC, _OTHER = object(), object(), object(), object(), object() + + +def _read_key() -> object: + """读一个键:方向上/下、回车、Esc、其他(返回原字符)。跨平台。""" + try: + if sys.platform == "win32": + import msvcrt + ch = msvcrt.getwch() + if ch in ("\x00", "\xe0"): + ext = msvcrt.getwch() + if ext == "H": + return _UP + if ext == "P": + return _DOWN + return _OTHER + return {"\r": _ENTER, "\n": _ENTER, "\x1b": _ESC}.get(ch, ch) + import termios + import tty + fd = os.open("/dev/tty", os.O_RDONLY) + try: + old = termios.tcgetattr(fd) + tty.setraw(fd) + try: + ch = os.read(fd, 1).decode("utf-8", "ignore") + if ch == "\x1b": + rest = os.read(fd, 2).decode("utf-8", "ignore") + if rest == "[A": + return _UP + if rest == "[B": + return _DOWN + return _ESC + return {"\r": _ENTER, "\n": _ENTER}.get(ch, ch) + finally: + termios.tcsetattr(fd, termios.TCSADRAIN, old) + finally: + os.close(fd) + except Exception: + return _OTHER # 无原始终端:退化为编号输入 + + +def select(title: str, options: list, default: int = 0, cancel: bool = True): + """交互选择:tty 时 ↑↓ 移动、回车确认、数字直达、q/Esc 取消; + 非 tty(管道/自动化)退化为编号输入。返回索引;取消返回 None。""" + if len(options) == 1: + # 只有一个选项:直接展示并采用,不打扰 + print(title) + print(f" {accent('❯')} {options[0]}") + return 0 + if sys.stdin.isatty(): + return _select_keys(title, options, default, cancel) + return _select_numbered(title, options, default) + + +def _select_numbered(title: str, options: list, default: int) -> int | None: + print(title) + for i, opt in enumerate(options, 1): + mark = bold(opt) if i - 1 == default else dim(opt) + print(f" {faint(str(i) + '.')} {mark}") + while True: + try: + raw = input(faint(f" 选择 [1-{len(options)}] (回车 = {default + 1}): ")).strip() + except EOFError: + return None + if not raw: + return default + if raw.isdigit() and 1 <= int(raw) <= len(options): + return int(raw) - 1 + print(warn(f" 无效编号,请输入 1-{len(options)}")) + + +def _select_keys(title: str, options: list, default: int, cancel: bool): + hint = faint(" ↑↓ 选择 · 回车确认" + (" · q 取消" if cancel else "")) + + def _paint(cur: int, chosen: bool = False): + lines = [] + for i, opt in enumerate(options): + if i == cur: + mark = ok("❯") if chosen else accent("❯") + lines.append(f" {mark} {bold(opt)}") + else: + lines.append(f" {dim(opt)}") + return lines + + print(title) + for ln in _paint(default): + print(ln) + print(hint) + sys.stdout.flush() + drawn = len(options) + 1 # 选项 + 提示行 + + cur = default + while True: + k = _read_key() + if k is _OTHER or k is None: + sys.stdout.write(f"\x1b[{drawn}A\r" if _use_color() else "") + return _select_numbered(" (此终端不支持方向键,请输入编号)", options, cur) + if k == _UP: + cur = (cur - 1) % len(options) + elif k == _DOWN: + cur = (cur + 1) % len(options) + elif k == _ENTER: + # 定格所选,清掉菜单只留一行 + sys.stdout.write(f"\x1b[{drawn + 1}A\r\x1b[J") + print(f" {ok('❯')} {bold(options[cur])}") + return cur + elif k in (_ESC, "q", "Q"): + sys.stdout.write(f"\x1b[{drawn + 1}A\r\x1b[J") + print(f" {faint('已取消')}") + return None + elif isinstance(k, str) and k.isdigit() and 1 <= int(k) <= len(options): + cur = int(k) - 1 + sys.stdout.write(f"\x1b[{drawn + 1}A\r\x1b[J") + print(f" {ok('❯')} {bold(options[cur])}") + return cur + # 重绘选项+提示 + out = [f"\x1b[{drawn}A\r"] + for ln in _paint(cur): + out.append(ln + "\x1b[K\n") + out.append(hint + "\x1b[K") + sys.stdout.write("".join(out)) + sys.stdout.flush() + + +# ---------- 隧道运行面板 ---------- + +def fmt_up(sec: float) -> str: + sec = max(0, int(sec)) + return f"{sec // 3600:02d}:{sec % 3600 // 60:02d}:{sec % 60:02d}" + + +_STATUS_STYLE = { + "connected": (GREEN, "● 已连接"), + "connecting": (AMBER, "◐ 连接中"), + "error": (RED, "✘ 失败"), + "stopped": (FAINT, "○ 已停止"), +} + + +class Dashboard: + """运行时面板:隧道状态 + 滚动日志,原地刷新。非 tty 时为空操作。""" + + def __init__(self, logs: int = 4): + self._rows = 0 + self._logs: deque = deque(maxlen=logs) + self._stats: list = [] + self._lock = threading.Lock() + self._on = sys.stdout.isatty() + + def log(self, msg: str) -> None: + with self._lock: + self._logs.append(msg) + self._render() + + def update(self, stats: list) -> None: + with self._lock: + self._stats = stats + self._render() + + def close(self) -> None: + with self._lock: + if self._on and self._rows: + sys.stdout.write(f"\x1b[{self._rows}A\r\x1b[J") + sys.stdout.flush() + self._rows = 0 + + def _render(self) -> None: + if not self._on: + return + lines = [] + head = " ssh-e2e · 隧道运行中 " + lines.append(accent("─" + head) + faint("─" * max(0, 46 - disp_len(head)))) + for st in self._stats: + rgb, label = _STATUS_STYLE.get(st["status"], (FAINT, "○ 未知")) + dot = _c(label.split()[0], rgb) + text = " ".join(label.split()[1:]) + lines.append(f" {dot} {bold(st['name'])}") + m = f"{dim('127.0.0.1:' + str(st['local_port']))} {accent('⇄')} {dim(st['remote'])}" + uptime = fmt_up(time.time() - st["started_at"]) if st.get("started_at") else "--:--:--" + lines.append(f" {m} {_c(text, rgb)} · {faint(uptime)} · {faint('连接 ' + str(st.get('conns', 0)))}") + if st.get("error"): + lines.append(f" {err(st['error'])}") + if self._logs: + lines.append(faint("─" * 46)) + for e in self._logs: + lines.append(f" {faint(e)}") + lines.append(faint("─ Ctrl+C 断开 " + "─" * 33)) + out = [] + if self._rows: + out.append(f"\x1b[{self._rows}A\r\x1b[J") + out.append("\n".join(lines) + "\n") + sys.stdout.write("".join(out)) + sys.stdout.flush() + self._rows = len(lines)