CLI 视觉升级:tui.py 真彩色界面
- 配色与 GUI 同源(石墨底/青绿点缀),支持 NO_COLOR 与非 tty 自动降级纯文本 - 向导:对齐横幅、❯ 方向键选择菜单(↑↓/数字/回车,q 取消),彩色表单提示 - 隧道运行时实时面板:状态·运行时长·累计连接·滚动日志,原地刷新 - Ctrl+C 退出时面板干净收起;非交互终端自动退化为编号选择
This commit is contained in:
leefer
@@ -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`,公网访问不到。
|
||||
|
||||
@@ -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("提示: 当前平台密码以明文保存于配置文件(权限已设为仅本用户可读)。")
|
||||
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user