leefer
7229642c62
- 向导开场:像素字符画 Logo(❯_ 提示符,青绿+光标绿)+ 版本/平台/配置信息 + 横线 - 运行面板头部加 ❯_ 标识;连接中状态改为 ◐◓◑◒ 旋转动画 - SSH banner 错误改为可读诊断文案;握手超时 12→25s;重连日志带尝试次数
316 lines
10 KiB
Python
316 lines
10 KiB
Python
"""终端界面组件:真彩色样式、方向键选择菜单、隧道运行面板。
|
||
非 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)
|
||
|
||
|
||
# ---------- 品牌标识:像素字符画 "❯_"(终端提示符) ----------
|
||
|
||
_LOGO_ROWS = [
|
||
"█.........",
|
||
"██........",
|
||
"█.█.......",
|
||
"██........",
|
||
"█...██████",
|
||
]
|
||
_CURSOR_FROM = 4 # 第 4 列起是"光标下划线",用绿色
|
||
|
||
|
||
def logo_lines() -> list:
|
||
"""5 行着色 logo;无色模式输出单色块画。"""
|
||
out = []
|
||
for row in _LOGO_ROWS:
|
||
line = ""
|
||
for i, ch in enumerate(row):
|
||
if ch == "█":
|
||
line += _c("█", GREEN if i >= _CURSOR_FROM else ACCENT)
|
||
else:
|
||
line += " "
|
||
out.append(line)
|
||
return out
|
||
|
||
|
||
# ---------- 显示宽度(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 or k == "k":
|
||
cur = (cur - 1) % len(options)
|
||
elif k == _DOWN or k == "j":
|
||
cur = (cur + 1) % len(options)
|
||
elif k == _ENTER:
|
||
# 定格所选,清掉整个菜单块(选项+提示),只留一行
|
||
sys.stdout.write(f"\x1b[{drawn}A\r\x1b[J")
|
||
print(f" {ok('❯')} {bold(options[cur])}")
|
||
return cur
|
||
elif k in (_ESC, "q", "Q", "\x03"):
|
||
sys.stdout.write(f"\x1b[{drawn}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}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\n")
|
||
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, "○ 已停止"),
|
||
}
|
||
|
||
|
||
_SPIN = "◐◓◑◒" # 连接中旋转帧,按时间取模
|
||
|
||
|
||
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:
|
||
if st["status"] == "connecting":
|
||
glyph = _SPIN[int(time.time() * 2) % len(_SPIN)]
|
||
dot, text = _c(glyph, AMBER), "连接中"
|
||
else:
|
||
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, AMBER if st['status'] == 'connecting' else _STATUS_STYLE[st['status']][0])} · {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)
|