品牌标识与握手诊断
- 向导开场:像素字符画 Logo(❯_ 提示符,青绿+光标绿)+ 版本/平台/配置信息 + 横线 - 运行面板头部加 ❯_ 标识;连接中状态改为 ◐◓◑◒ 旋转动画 - SSH banner 错误改为可读诊断文案;握手超时 12→25s;重连日志带尝试次数
This commit is contained in:
leefer
@@ -25,6 +25,25 @@ from engine import TunnelManager, probe_host # noqa: E402
|
||||
from storage import Store, data_root # noqa: E402
|
||||
import tui # noqa: E402
|
||||
|
||||
VERSION = "1.4"
|
||||
|
||||
|
||||
def _print_banner(store: Store) -> None:
|
||||
"""开场:像素 Logo + 产品信息(类 Claude Code 风格)。"""
|
||||
from platform import python_version, system
|
||||
logo = tui.logo_lines()
|
||||
info = [
|
||||
tui.bold("ssh-e2e · SSH 隧道管理器"),
|
||||
tui.dim(f"v{VERSION} · 中文交互向导"),
|
||||
tui.faint(f"{system()} · Python {python_version()}"),
|
||||
tui.faint(f"配置 {store.path}"),
|
||||
tui.faint("回车采用 [默认值] · Ctrl+C 退出"),
|
||||
]
|
||||
print()
|
||||
for row, line in zip(logo, info):
|
||||
print(f" {row} {line}")
|
||||
print(" " + tui.accent("─" * 46))
|
||||
|
||||
|
||||
def _print(msg: str) -> None:
|
||||
print(msg, flush=True)
|
||||
@@ -392,14 +411,7 @@ def _wizard_new_tunnel(store: Store, host: dict):
|
||||
|
||||
def wizard(store: Store) -> int:
|
||||
_ensure_tty_stdin()
|
||||
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}"))
|
||||
_print_banner(store)
|
||||
|
||||
if store.warn_plaintext:
|
||||
_print("提示: 当前平台密码以明文保存于配置文件(权限已设为仅本用户可读)。")
|
||||
|
||||
@@ -20,6 +20,9 @@ def friendly_error(e: Exception) -> str:
|
||||
return "SSH 认证失败:请检查用户名、密码或私钥。"
|
||||
if isinstance(e, paramiko.SSHException):
|
||||
msg = str(e)
|
||||
if "banner" in msg.lower():
|
||||
return ("SSH 握手失败:已连上端口但对方未回应 SSH 协议"
|
||||
"(远程主机宕机/过载,或端口映射网关异常)")
|
||||
if "negotiation" in msg or "timed out" in msg or "Timeout" in msg:
|
||||
return "SSH 握手超时:远程主机无响应。"
|
||||
if "not a valid" in msg.lower() or ("key" in msg.lower() and "invalid" in msg.lower()):
|
||||
@@ -59,6 +62,7 @@ class Tunnel:
|
||||
self._client: paramiko.SSHClient | None = None
|
||||
self._listener: socket.socket | None = None
|
||||
self._wakeup = threading.Event()
|
||||
self.attempts = 0 # 本次断线后的重试次数
|
||||
|
||||
# ---------- 状态 ----------
|
||||
|
||||
@@ -168,7 +172,7 @@ class Tunnel:
|
||||
port = int(host.get("port") or 22)
|
||||
kw = dict(
|
||||
hostname=host["hostname"], port=port, username=host.get("username"),
|
||||
timeout=12, banner_timeout=12, auth_timeout=12,
|
||||
timeout=15, banner_timeout=25, auth_timeout=15,
|
||||
allow_agent=False, look_for_keys=False,
|
||||
)
|
||||
if host.get("auth") == "key":
|
||||
@@ -177,8 +181,11 @@ class Tunnel:
|
||||
kw["passphrase"] = host["key_passphrase"]
|
||||
else:
|
||||
kw["password"] = host.get("password")
|
||||
self._log("info", f"正在连接 {host.get('username')}@{host['hostname']}:{port} ...")
|
||||
self.attempts += 1
|
||||
n = f"(第 {self.attempts} 次尝试)" if self.attempts > 1 else ""
|
||||
self._log("info", f"正在连接 {host.get('username')}@{host['hostname']}:{port} ... {n}")
|
||||
client.connect(**kw)
|
||||
self.attempts = 0
|
||||
try:
|
||||
client.save_host_keys(str(self.known_hosts_path))
|
||||
except Exception:
|
||||
|
||||
@@ -69,6 +69,32 @@ 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),
|
||||
@@ -225,6 +251,9 @@ _STATUS_STYLE = {
|
||||
}
|
||||
|
||||
|
||||
_SPIN = "◐◓◑◒" # 连接中旋转帧,按时间取模
|
||||
|
||||
|
||||
class Dashboard:
|
||||
"""运行时面板:隧道状态 + 滚动日志,原地刷新。非 tty 时为空操作。"""
|
||||
|
||||
@@ -256,16 +285,20 @@ class Dashboard:
|
||||
if not self._on:
|
||||
return
|
||||
lines = []
|
||||
head = " ssh-e2e · 隧道运行中 "
|
||||
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:])
|
||||
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, rgb)} · {faint(uptime)} · {faint('连接 ' + str(st.get('conns', 0)))}")
|
||||
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:
|
||||
|
||||
Reference in New Issue
Block a user