diff --git a/cli.py b/cli.py index 7eea991..1b6bdac 100644 --- a/cli.py +++ b/cli.py @@ -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("提示: 当前平台密码以明文保存于配置文件(权限已设为仅本用户可读)。") diff --git a/engine.py b/engine.py index edc9256..2253aab 100644 --- a/engine.py +++ b/engine.py @@ -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: diff --git a/tui.py b/tui.py index 4686ff1..2923dcf 100644 --- a/tui.py +++ b/tui.py @@ -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: