diff --git a/README.md b/README.md index 45b1ceb..1265816 100644 --- a/README.md +++ b/README.md @@ -132,15 +132,15 @@ sbp cert renew sbp tui ``` -TUI 提供这些页面: +界面分左右两栏. 左栏固定显示栏目, 右栏显示当前栏目内容, 不需要一层层进菜单. - Dashboard: 状态总览, 启动, 停止, 重启. -- Protocols: 启用, 停用, 改端口, 改选项. -- Users: 增加, 删除, 查看分享链接, Clash 配置和客户端配置. +- Protocols: 选中协议后, 右栏显示端口, 传输, 参数, 以及启用, 停用, 改端口, 改选项. +- Users: 选中用户后, 可查看分享链接, Clash 配置, 客户端配置, 或删除用户. - Settings: 修改 host 和 SNI, 更换证书, 升级, 重装, 卸载. - Logs: 查看 journal. -操作键: `↑`/`↓` 或 `j`/`k` 移动, `Enter` 确认, `q` 返回. +操作键: `Tab` 切换左右栏, `↑`/`↓` 或 `j`/`k` 移动, `Enter` 执行, `1`-`5` 直接跳到栏目, `q` 退出. ## 证书 @@ -195,7 +195,7 @@ src/sbp/ clash.py Clash (mihomo) YAML 输出 protocols/ 每个协议一个模块 ops/ 业务操作, CLI 和 TUI 共用 - tui/ curses 界面 + tui/ curses 界面: frame 画两栏, views 是各栏内容 ``` 协议模块只做四件事: 生成 inbound, 分享链接, 客户端 outbound, Clash 节点. diff --git a/src/sbp/tui/app.py b/src/sbp/tui/app.py index ec76814..8ce242d 100644 --- a/src/sbp/tui/app.py +++ b/src/sbp/tui/app.py @@ -1,90 +1,150 @@ -"""TUI application loop and shared helpers.""" +"""TUI application: sidebar navigation and the right-hand content pane.""" from __future__ import annotations import curses import sys -from .. import constants, protocols +from .. import constants from ..errors import SbpError -from ..ops import install as install_op from ..ops import status as status_op from ..paths import Paths +from .frame import Row, draw from .term import Term -from .screens import dashboard, logs, protocols as protocols_screen, settings, users +from .views import dashboard, logs, protocols, settings, users + +FOOTER = "Tab pane ↑/↓ move Enter select 1-5 section q quit" class App: def __init__(self, paths: Paths) -> None: self.paths = paths self.term: Term + self.views = [ + dashboard.DashboardView(), + protocols.ProtocolsView(), + users.UsersView(), + settings.SettingsView(), + logs.LogsView(), + ] + self.view_index = 0 + self.focus = "content" + self.indices = [0] * len(self.views) + + @property + def view(self): + return self.views[self.view_index] def status(self) -> status_op.Status: return status_op.status(self.paths) + def rows(self) -> list[Row]: + try: + return self.view.rows(self) + except SbpError as exc: + return [Row(exc.message, kind="error", selectable=False)] + + def reset(self) -> None: + self.indices[self.view_index] = 0 + + def show_view(self, name: str) -> None: + for position, view in enumerate(self.views): + if view.name == name: + self.view_index = position + self.focus = "content" + self.reset() + return + + def external(self, title: str, action) -> bool: + ok = self.term.run_external(title, action) + self.reset() + return ok + def main(self, screen) -> None: self.term = Term(screen) while True: - status = self.status() - if not status.installed: - choice = self.term.menu("sbp", ["Install", "Quit"], subtitle="not installed") - if choice == 0: - self.install_wizard() - continue - return - subtitle = f"{status.host} {'active' if status.active else 'inactive'}" - choice = self.term.menu( - f"sbp {constants.VERSION}", - ["Dashboard", "Protocols", "Users", "Settings", "Logs", "Quit"], - subtitle=subtitle, + rows = self.rows() + index = self._clamp(rows, self.indices[self.view_index]) + self.indices[self.view_index] = index + draw( + self.term, + title=f"sbp {constants.VERSION}", + subtitle=self._subtitle(), + sidebar=[view.title for view in self.views], + sidebar_index=self.view_index, + sidebar_focused=self.focus == "sidebar", + content_title=self.view.title, + rows=rows, + index=index, + content_focused=self.focus == "content", + footer=FOOTER, ) - if choice is None or choice == 5: + key = self.term.getch() + if key == -1 or key == ord("q"): return - handler = ( - dashboard.show, - protocols_screen.show, - users.show, - settings.show, - logs.show, - )[choice] - try: - handler(self) - except SbpError as exc: - self.term.message("Error", exc.message, "error") + if key == curses.KEY_RESIZE: + continue + if key in (9, curses.KEY_BTAB): + self.focus = "content" if self.focus == "sidebar" else "sidebar" + elif key == curses.KEY_LEFT: + self.focus = "sidebar" + elif key == curses.KEY_RIGHT: + self.focus = "content" + elif key in (curses.KEY_UP, ord("k")): + self._move(rows, -1) + elif key in (curses.KEY_DOWN, ord("j")): + self._move(rows, 1) + elif key in (10, 13, curses.KEY_ENTER): + if self.focus == "sidebar": + self.focus = "content" + self.reset() + else: + self._activate(rows, index) + elif ord("1") <= key <= ord("5"): + self.view_index = key - ord("1") + self.focus = "content" + self.reset() - def install_wizard(self) -> None: - host = self.term.prompt("Install", "Client-facing host", "") - sni = self.term.prompt("Install", "TLS server name", constants.DEFAULT_SNI) - if sni is None: - return - keys = ", ".join(protocols.keys()) - selection = self.term.prompt("Install", f"Protocols ({keys})", "all") - if selection is None: - return - port_text = self.term.prompt("Install", "First port", str(constants.BASE_PORT)) - if port_text is None: - return - try: - base_port = int(port_text) - except ValueError: - self.term.message("Error", f"invalid port: {port_text}", "error") - return - selected = None if selection.strip().lower() in ("", "all") else [ - item.strip() for item in selection.split(",") if item.strip() - ] - options = install_op.InstallOptions( - host=host or None, - sni=sni or None, - protocols=selected, - base_port=base_port, - ) + def _subtitle(self) -> str: + status = self.status() + if not status.installed: + return "not installed" + state = "active" if status.active else "inactive" + return f"{status.host} {state}" - def action(progress): - install_op.install(self.paths, options, progress=progress) - for warning in options.warnings: - print(f"warning: {warning}") + def _clamp(self, rows: list[Row], index: int) -> int: + if not rows: + return 0 + index = max(0, min(index, len(rows) - 1)) + if rows[index].selectable: + return index + for offset in range(1, len(rows) + 1): + for candidate in (index + offset, index - offset): + if 0 <= candidate < len(rows) and rows[candidate].selectable: + return candidate + return index - self.term.run_external("Install sbp", action) + def _move(self, rows: list[Row], delta: int) -> None: + if self.focus == "sidebar": + self.view_index = max(0, min(len(self.views) - 1, self.view_index + delta)) + self.reset() + return + position = self.indices[self.view_index] + candidate = position + while True: + candidate += delta + if not 0 <= candidate < len(rows): + return + if rows[candidate].selectable: + self.indices[self.view_index] = candidate + return + + def _activate(self, rows: list[Row], index: int) -> None: + if not rows or not 0 <= index < len(rows): + return + action = rows[index].action + if action is not None: + action(self) def run(paths: Paths) -> int: diff --git a/src/sbp/tui/frame.py b/src/sbp/tui/frame.py new file mode 100644 index 0000000..bdadd2b --- /dev/null +++ b/src/sbp/tui/frame.py @@ -0,0 +1,109 @@ +"""Two-pane layout: a fixed sidebar on the left, the current view on the right.""" + +from __future__ import annotations + +import curses +from dataclasses import dataclass +from typing import Callable, Optional + +from .term import ACCENT, DIM, Term, display_width, pad, truncate + +SIDEBAR_MIN = 16 +SIDEBAR_MAX = 28 +MIN_WIDTH = 46 +MIN_HEIGHT = 10 + + +@dataclass(frozen=True) +class Row: + text: str + action: Optional[Callable] = None + kind: str = "" + selectable: bool = True + + +def sidebar_width(titles: list[str]) -> int: + widest = max((display_width(title) for title in titles), default=0) + return max(SIDEBAR_MIN, min(widest + 4, SIDEBAR_MAX)) + + +def window_offset(index: int, count: int, visible: int) -> int: + """Scroll just enough to keep the selected row on screen.""" + if visible <= 0 or count <= visible: + return 0 + return max(0, min(index - visible + 1, count - visible)) + + +def draw( + term: Term, + *, + title: str, + subtitle: str, + sidebar: list[str], + sidebar_index: int, + sidebar_focused: bool, + content_title: str, + rows: list[Row], + index: int, + content_focused: bool, + footer: str, +) -> None: + height, width = term.size() + term.clear() + term.put(0, 1, title, term.attr(ACCENT, curses.A_BOLD)) + if subtitle: + term.put(0, max(2, width - display_width(subtitle) - 2), subtitle, term.attr(DIM)) + term.hline(1, term.attr(DIM)) + + if width < MIN_WIDTH or height < MIN_HEIGHT: + term.put(height // 2, 2, f"terminal too small ({width}x{height})", term.attr(ACCENT)) + term.refresh() + return + + term.hline(height - 2, term.attr(DIM)) + term.put(height - 1, 1, truncate(footer, width - 2), term.attr(DIM)) + + side = sidebar_width(sidebar) + body_top = 3 + body_bottom = height - 4 + + for y in range(2, height - 2): + term.put(y, side, "│", term.attr(DIM)) + + for position, item in enumerate(sidebar): + y = body_top + position + if y > body_bottom: + break + selected = position == sidebar_index + if selected and sidebar_focused: + term.fill(y, 0, side, term.attr(ACCENT, curses.A_REVERSE)) + term.put(y, 1, f"▸ {item}", term.attr(ACCENT, curses.A_REVERSE | curses.A_BOLD)) + elif selected: + term.put(y, 1, f"▸ {item}", term.attr(ACCENT, curses.A_BOLD)) + else: + term.put(y, 1, f" {item}") + + x = side + 2 + inner = max(1, width - x - 1) + term.put(2, x, truncate(content_title, inner), term.attr(ACCENT, curses.A_BOLD)) + + visible = max(1, body_bottom - body_top + 1) + offset = window_offset(index, len(rows), visible) + for line in range(visible): + position = offset + line + if position >= len(rows): + break + row = rows[position] + selected = position == index and row.selectable + attr = term.kind_attr(row.kind) + if selected and content_focused: + term.fill(body_top + line, x, inner, term.attr(ACCENT, curses.A_REVERSE)) + term.put(body_top + line, x, pad(truncate(row.text, inner), inner), + term.attr(ACCENT, curses.A_REVERSE | curses.A_BOLD)) + elif selected: + term.put(body_top + line, x, pad(truncate(row.text, inner), inner), + term.attr(ACCENT, curses.A_BOLD)) + else: + term.put(body_top + line, x, truncate(row.text, inner), attr) + + term.refresh() diff --git a/src/sbp/tui/screens/__init__.py b/src/sbp/tui/screens/__init__.py deleted file mode 100644 index 09ea513..0000000 --- a/src/sbp/tui/screens/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""TUI screens.""" diff --git a/src/sbp/tui/screens/dashboard.py b/src/sbp/tui/screens/dashboard.py deleted file mode 100644 index c5bc597..0000000 --- a/src/sbp/tui/screens/dashboard.py +++ /dev/null @@ -1,50 +0,0 @@ -"""Dashboard: status summary and service control.""" - -from __future__ import annotations - -from typing import TYPE_CHECKING - -from ...ops import service as service_op - -if TYPE_CHECKING: - from ..app import App - - -def _summary(app: "App") -> str: - status = app.status() - lines = [ - f"Service {'active' if status.active else 'inactive'}", - f"sing-box {status.singbox_version}", - f"Address {status.host}", - f"SNI {status.sni}", - f"TLS {status.tls_mode}", - f"Cert until {status.cert_until}", - f"Users {status.user_count}", - "", - "Protocols", - ] - for row in status.protocols: - mark = "on " if row.enabled else "off" - details = ", ".join(f"{key}={value}" for key, value in row.details) - lines.append(f" [{mark}] {row.title:<18} port {row.port:<6} {row.transport} {details}") - return "\n".join(lines) - - -def show(app: "App") -> None: - while True: - status = app.status() - choice = app.term.menu( - "Dashboard", - ["Show full status", "Start service", "Stop service", "Restart service", "Back"], - subtitle=f"{status.host} {'active' if status.active else 'inactive'}", - ) - if choice is None or choice == 4: - return - if choice == 0: - app.term.scroll_text("Status", _summary(app)) - elif choice == 1: - app.term.run_external("Start service", lambda progress: service_op.start(app.paths)) - elif choice == 2: - app.term.run_external("Stop service", lambda progress: service_op.stop(app.paths)) - elif choice == 3: - app.term.run_external("Restart service", lambda progress: service_op.restart(app.paths)) diff --git a/src/sbp/tui/screens/logs.py b/src/sbp/tui/screens/logs.py deleted file mode 100644 index 2516170..0000000 --- a/src/sbp/tui/screens/logs.py +++ /dev/null @@ -1,22 +0,0 @@ -"""Journal viewer.""" - -from __future__ import annotations - -from typing import TYPE_CHECKING - -from ...ops import service as service_op - -if TYPE_CHECKING: - from ..app import App - - -def show(app: "App") -> None: - while True: - choice = app.term.menu("Logs", ["Recent 100 lines", "Recent 500 lines", "Back"]) - if choice is None or choice == 2: - return - lines = 100 if choice == 0 else 500 - app.term.scroll_text( - "Logs", - "\n".join(service_op.logs(app.paths, lines=lines)) or "no log lines", - ) diff --git a/src/sbp/tui/screens/protocols.py b/src/sbp/tui/screens/protocols.py deleted file mode 100644 index 4a01fd5..0000000 --- a/src/sbp/tui/screens/protocols.py +++ /dev/null @@ -1,99 +0,0 @@ -"""Protocol list and per-protocol actions.""" - -from __future__ import annotations - -from typing import TYPE_CHECKING - -from ... import protocols -from ...ops import protocols as protocols_op - -if TYPE_CHECKING: - from ..app import App - - -def _show(value) -> str: - if isinstance(value, bool): - return "true" if value else "false" - return str(value) - - -def _label(row, protocol) -> str: - state = "enabled" if row[1] else "off" - details = ", ".join(f"{key}={value}" for key, value in row[3]) - suffix = f" {details}" if details else "" - return f"{protocol.info.title:<18} {state:<8} {row[2]:<6} {protocol.info.transport}{suffix}" - - -def show(app: "App") -> None: - while True: - rows = protocols_op.summary(app.paths) - labels = [_label(row, protocols.get(row[0])) for row in rows] - labels.append("Back") - choice = app.term.menu("Protocols", labels, subtitle="select a protocol to manage") - if choice is None or choice == len(rows): - return - _detail(app, rows[choice][0]) - - -def _detail(app: "App", key: str) -> None: - while True: - row = next(item for item in protocols_op.summary(app.paths) if item[0] == key) - protocol = protocols.get(key) - enabled = row[1] - actions = [ - "Disable" if enabled else "Enable", - "Change port", - "Edit options", - "Show settings", - "Back", - ] - choice = app.term.menu( - protocol.info.title, - actions, - subtitle=f"{'enabled' if enabled else 'off'} port {row[2]} {protocol.info.transport}", - ) - if choice is None or choice == 4: - return - if choice == 0: - if enabled: - app.term.run_external("Disable", lambda progress: protocols_op.disable(app.paths, key)) - else: - port_text = app.term.prompt("Enable", "Port (blank = auto)", "") - port = int(port_text) if port_text and port_text.isdigit() else None - app.term.run_external( - "Enable", lambda progress: protocols_op.enable(app.paths, key, port=port) - ) - elif choice == 1: - current = str(row[2]) - value = app.term.prompt("Change port", "Port", current) - if value and value.isdigit(): - app.term.run_external( - "Change port", lambda progress: protocols_op.set_port(app.paths, key, int(value)) - ) - elif choice == 2: - _edit_options(app, key) - elif choice == 3: - lines = [f"{name}: {value}" for name, value in row[3]] or ["no editable options"] - app.term.scroll_text(protocol.info.title, "\n".join(lines)) - - -def _edit_options(app: "App", key: str) -> None: - protocol = protocols.get(key) - editable = protocol.editable() - if not editable: - app.term.message("Options", "this protocol has no editable options") - return - current = protocols_op.options(app.paths, key) - labels = [f"{label}: {_show(current.get(name, ''))}" for name, label, _ in editable] - labels.append("Back") - choice = app.term.menu("Edit options", labels) - if choice is None or choice == len(editable): - return - name, label, _kind = editable[choice] - value = app.term.prompt("Edit option", label, _show(current.get(name, ""))) - if value is None: - return - app.term.run_external( - "Apply option", - lambda progress: protocols_op.set_option(app.paths, key, name, value), - ) diff --git a/src/sbp/tui/screens/settings.py b/src/sbp/tui/screens/settings.py deleted file mode 100644 index 471422d..0000000 --- a/src/sbp/tui/screens/settings.py +++ /dev/null @@ -1,98 +0,0 @@ -"""Server settings, upgrade, and uninstall.""" - -from __future__ import annotations - -from typing import TYPE_CHECKING - -from ...ops import install as install_op -from ...ops import settings as settings_op -from ...ops import uninstall as uninstall_op -from ...ops import upgrade as upgrade_op - -if TYPE_CHECKING: - from ..app import App - -ACTIONS = ( - "Show settings", - "Set host", - "Set SNI", - "Use custom certificate", - "Renew certificate", - "Upgrade sing-box", - "Reinstall", - "Uninstall", - "Back", -) - - -def show(app: "App") -> None: - while True: - status = app.status() - choice = app.term.menu( - "Settings", - list(ACTIONS), - subtitle=f"{status.host} sni {status.sni}", - ) - if choice is None or choice == 8: - return - if choice == 0: - app.term.scroll_text( - "Settings", - "\n".join([ - f"Host {status.host}", - f"SNI {status.sni}", - f"TLS {status.tls_mode}", - f"Cert until {status.cert_until}", - f"Config {status.config_path}", - f"Created {status.created}", - ]), - ) - elif choice == 1: - value = app.term.prompt("Set host", "Client-facing host", status.host) - if value: - app.term.run_external( - "Set host", lambda progress: settings_op.set_host(app.paths, value) - ) - elif choice == 2: - value = app.term.prompt("Set SNI", "TLS server name", status.sni) - if value: - app.term.run_external( - "Set SNI", lambda progress: settings_op.set_sni(app.paths, value) - ) - elif choice == 3: - cert = app.term.prompt("Custom certificate", "Certificate path (PEM)", "") - if not cert: - continue - key = app.term.prompt("Custom certificate", "Private key path (PEM)", "") - if not key: - continue - sni = app.term.prompt("Custom certificate", "SNI", status.sni) - app.term.run_external( - "Use custom certificate", - lambda progress: settings_op.set_cert(app.paths, cert, key, sni or None), - ) - elif choice == 4: - app.term.run_external( - "Renew certificate", lambda progress: settings_op.regenerate_cert(app.paths) - ) - elif choice == 5: - app.term.run_external( - "Upgrade sing-box", lambda progress: upgrade_op.upgrade(app.paths, progress=progress) - ) - elif choice == 6: - if app.term.confirm("Reinstall", "Reinstall sing-box and regenerate the service?"): - options = install_op.InstallOptions(force=True) - app.term.run_external( - "Reinstall", - lambda progress: install_op.install(app.paths, options, progress=progress), - ) - elif choice == 7: - if app.term.confirm("Uninstall", "Stop and remove the service and binary?"): - purge = app.term.confirm( - "Uninstall", - "Also delete configuration, users, and certificates?", - ) - app.term.run_external( - "Uninstall", lambda progress: uninstall_op.uninstall(app.paths, purge=purge) - ) - return diff --git a/src/sbp/tui/screens/users.py b/src/sbp/tui/screens/users.py deleted file mode 100644 index 4e4fed5..0000000 --- a/src/sbp/tui/screens/users.py +++ /dev/null @@ -1,62 +0,0 @@ -"""User list, share links, and client configs.""" - -from __future__ import annotations - -from typing import TYPE_CHECKING - -from ... import protocols -from ...ops import show as show_op -from ...ops import users as users_op - -if TYPE_CHECKING: - from ..app import App - - -def show(app: "App") -> None: - while True: - users = users_op.list_users(app.paths) - labels = [f"{user.name:<20} {user.created[:10]}" for user in users] - labels.append("+ Add user") - labels.append("Back") - choice = app.term.menu("Users", labels, subtitle=f"{len(users)} users") - if choice is None or choice == len(users) + 1: - return - if choice == len(users): - name = app.term.prompt("Add user", "User name", "") - if name: - app.term.run_external("Add user", lambda progress: users_op.add(app.paths, name)) - continue - _detail(app, users[choice].name) - - -def _detail(app: "App", name: str) -> None: - while True: - choice = app.term.menu( - f"User {name}", - ["Show share URIs", "Show Clash YAML", "Show client config", "Delete user", "Back"], - ) - if choice is None or choice == 4: - return - if choice == 0: - uris = show_op.user_uris(app.paths, name) - text = "\n\n".join(f"{key}\n {uri}" for key, uri in uris) or "no enabled protocols" - app.term.scroll_text(f"Share URIs: {name}", text) - elif choice == 1: - app.term.scroll_text(f"Clash: {name}", show_op.user_clash(app.paths, name)) - elif choice == 2: - enabled = [row.key for row in app.status().protocols if row.enabled] - if not enabled: - app.term.message("Client config", "no enabled protocols") - continue - labels = [protocols.get(key).info.title for key in enabled] + ["Back"] - picked = app.term.menu("Client config", labels) - if picked is None or picked == len(enabled): - continue - app.term.scroll_text( - f"Client config: {name} / {enabled[picked]}", - show_op.client_config_json(app.paths, name, enabled[picked]), - ) - elif choice == 3: - if app.term.confirm("Delete user", f"Delete {name}? Clients using this user stop working."): - app.term.run_external("Delete user", lambda progress: users_op.remove(app.paths, name)) - return diff --git a/src/sbp/tui/term.py b/src/sbp/tui/term.py index 5d13d44..c1610be 100644 --- a/src/sbp/tui/term.py +++ b/src/sbp/tui/term.py @@ -1,7 +1,7 @@ -"""Curses widgets: menus, prompts, dialogs, and text paging. +"""Curses primitives: colors, drawing, dialogs, and text input. -Only this module touches curses. Screens call these helpers and never draw -raw escape sequences themselves. +Only this module touches curses. Views and the frame draw through these +helpers and never write escape sequences themselves. """ from __future__ import annotations @@ -17,6 +17,8 @@ WARN = 3 BAD = 4 DIM = 5 +KINDS = {"info": ACCENT, "good": GOOD, "warn": WARN, "error": BAD, "dim": DIM} + def display_width(text: str) -> int: width = 0 @@ -90,38 +92,44 @@ class Term: return extra return curses.color_pair(pair) | extra + def kind_attr(self, kind: str) -> int: + return self.attr(KINDS.get(kind, 0)) if kind in KINDS else 0 + def size(self) -> tuple[int, int]: - height, width = self.screen.getmaxyx() - return height, width + return self.screen.getmaxyx() + + def getch(self) -> int: + return self.screen.getch() def put(self, y: int, x: int, text: str, attr: int = 0) -> None: height, width = self.size() - if y < 0 or y >= height or x >= width: + if y < 0 or y >= height or x < 0 or x >= width or not text: return try: self.screen.addstr(y, x, truncate(text, max(0, width - x - 1)), attr) except curses.error: pass + def fill(self, y: int, x: int, width: int, attr: int = 0) -> None: + height, columns = self.size() + if y < 0 or y >= height or width <= 0: + return + try: + self.screen.addstr(y, x, " " * min(width, columns - x - 1), attr) + except curses.error: + pass + + def hline(self, y: int, attr: int = 0, x: int = 0) -> None: + height, width = self.size() + if 0 <= y < height: + self.fill(y, x, width - x - 1, attr) + def clear(self) -> None: self.screen.erase() def refresh(self) -> None: self.screen.refresh() - def header(self, title: str, subtitle: str = "") -> int: - height, width = self.size() - self.put(0, 1, title, self.attr(ACCENT, curses.A_BOLD)) - if subtitle: - self.put(0, max(2, width - display_width(subtitle) - 2), subtitle, self.attr(DIM)) - self.put(1, 0, "─" * max(1, width - 1), self.attr(DIM)) - return 3 - - def footer(self, text: str) -> None: - height, width = self.size() - self.put(height - 2, 0, "─" * max(1, width - 1), self.attr(DIM)) - self.put(height - 1, 1, text, self.attr(DIM)) - def message(self, title: str, text: str, kind: str = "info") -> None: pair = {"info": ACCENT, "good": GOOD, "warn": WARN, "error": BAD}.get(kind, ACCENT) self.clear() @@ -135,11 +143,19 @@ class Term: y += 1 self.footer(" press any key ") self.refresh() - self.screen.getch() + self.getch() def confirm(self, title: str, text: str) -> bool: - items = ["Yes", "No"] - return self.menu(title, items, subtitle=text, start=1) == 0 + self.clear() + height, width = self.size() + y = self.header(title) + for line in textwrap.wrap(text, max(20, width - 6))[: max(1, height - 8)]: + self.put(y, 3, line) + y += 1 + self.put(y + 1, 3, "y confirm n / Esc cancel", self.attr(ACCENT)) + self.footer(" y confirm any other key cancels ") + self.refresh() + return self.getch() in (ord("y"), ord("Y")) def prompt(self, title: str, label: str, default: str = "", secret: bool = False) -> Optional[str]: self.clear() @@ -148,7 +164,7 @@ class Term: self.put(3, 3, label) if default and not secret: self.put(3, 3 + display_width(label) + 1, f"[{default}]", self.attr(DIM)) - self.put(height - 1, 1, "Enter accept Esc cancel", self.attr(DIM)) + self.footer(" Enter accept Esc cancel ") self.refresh() try: curses.curs_set(1) @@ -192,6 +208,7 @@ class Term: footer: str = "↑/↓ select Enter open q back", start: int = 0, ) -> Optional[int]: + """Modal picker for short action lists.""" index = min(max(start, 0), max(0, len(items) - 1)) while True: self.clear() @@ -207,7 +224,7 @@ class Term: self.put(y + row, 2, marker + item, attr) self.footer(footer) self.refresh() - key = self.screen.getch() + key = self.getch() if key == -1: return None if key in (curses.KEY_UP, ord("k")): @@ -218,16 +235,13 @@ class Term: return index elif key in (ord("q"), 27): return None - elif key == curses.KEY_RESIZE: - continue elif ord("1") <= key <= ord("9"): candidate = key - ord("1") if candidate < len(items): return candidate def scroll_text(self, title: str, text: str, subtitle: str = "") -> None: - height, width = self.size() - available = max(1, width - 4) + available = max(1, self.size()[1] - 4) lines: list[str] = [] for source in text.splitlines() or [""]: lines.extend(wrap_display(source, available)) @@ -244,7 +258,7 @@ class Term: self.put(y + row, 2, lines[position]) self.footer("↑/↓ scroll PgUp/PgDn q back") self.refresh() - key = self.screen.getch() + key = self.getch() if key == -1 or key in (ord("q"), 27): return if key in (curses.KEY_UP, ord("k")): @@ -256,6 +270,19 @@ class Term: elif key == curses.KEY_NPAGE: top = min(max(0, len(lines) - body), top + body) + def header(self, title: str, subtitle: str = "") -> int: + height, width = self.size() + self.put(0, 1, title, self.attr(ACCENT, curses.A_BOLD)) + if subtitle: + self.put(0, max(2, width - display_width(subtitle) - 2), subtitle, self.attr(DIM)) + self.put(1, 0, "─" * max(1, width - 1), self.attr(DIM)) + return 3 + + def footer(self, text: str) -> None: + height, width = self.size() + self.put(height - 2, 0, "─" * max(1, width - 1), self.attr(DIM)) + self.put(height - 1, 1, text, self.attr(DIM)) + def run_external(self, title: str, action: Callable[[Callable[[str], None]], None]) -> bool: """Leave curses, run action, wait for Enter, then restore curses.""" try: diff --git a/src/sbp/tui/views/__init__.py b/src/sbp/tui/views/__init__.py new file mode 100644 index 0000000..1e27a4f --- /dev/null +++ b/src/sbp/tui/views/__init__.py @@ -0,0 +1 @@ +"""Content panes. Each view exposes a name, a title, and a list of rows.""" diff --git a/src/sbp/tui/views/dashboard.py b/src/sbp/tui/views/dashboard.py new file mode 100644 index 0000000..7d751cb --- /dev/null +++ b/src/sbp/tui/views/dashboard.py @@ -0,0 +1,82 @@ +"""Status summary, service control, and the install wizard.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from ... import constants, protocols +from ...ops import install as install_op +from ...ops import service as service_op +from ..frame import Row + +if TYPE_CHECKING: + from ..app import App + + +def _install(app: "App") -> None: + term = app.term + host = term.prompt("Install", "Client-facing host (blank = detect)", "") or "" + sni = term.prompt("Install", "TLS server name", constants.DEFAULT_SNI) + if sni is None: + return + keys = ", ".join(protocols.keys()) + selection = term.prompt("Install", f"Protocols ({keys})", "all") + if selection is None: + return + port_text = term.prompt("Install", "First port", str(constants.BASE_PORT)) + if port_text is None: + return + try: + base_port = int(port_text) + except ValueError: + term.message("Error", f"invalid port: {port_text}", "error") + return + chosen = None if selection.strip().lower() in ("", "all") else [ + item.strip() for item in selection.split(",") if item.strip() + ] + options = install_op.InstallOptions( + host=host or None, sni=sni or None, protocols=chosen, base_port=base_port + ) + + def action(progress): + install_op.install(app.paths, options, progress=progress) + for warning in options.warnings: + print(f"warning: {warning}") + + app.external("Install sbp", action) + + +def _service(app: "App", action: str) -> None: + call = {"start": service_op.start, "stop": service_op.stop, "restart": service_op.restart}[action] + app.external(action.capitalize(), lambda progress: call(app.paths)) + + +class DashboardView: + name = "dashboard" + title = "Dashboard" + + def rows(self, app: "App") -> list[Row]: + status = app.status() + if not status.installed: + return [ + Row("sbp is not installed", kind="warn", selectable=False), + Row("", selectable=False), + Row("Install sbp", action=_install), + ] + return [ + Row("Service", kind="header", selectable=False), + Row(f" state {'active' if status.active else 'inactive'}", + kind="good" if status.active else "warn", selectable=False), + Row(f" sing-box {status.singbox_version}", selectable=False), + Row(f" address {status.host}", selectable=False), + Row(f" sni {status.sni}", selectable=False), + Row(f" tls {status.tls_mode}", selectable=False), + Row(f" cert until {status.cert_until}", selectable=False), + Row(f" users {status.user_count}", selectable=False), + Row("", selectable=False), + Row("Actions", kind="header", selectable=False), + Row("Start service", action=lambda a: _service(a, "start")), + Row("Stop service", action=lambda a: _service(a, "stop")), + Row("Restart service", action=lambda a: _service(a, "restart")), + Row("View logs", action=lambda a: a.show_view("logs")), + ] diff --git a/src/sbp/tui/views/logs.py b/src/sbp/tui/views/logs.py new file mode 100644 index 0000000..1fa1748 --- /dev/null +++ b/src/sbp/tui/views/logs.py @@ -0,0 +1,34 @@ +"""Service log from the systemd user journal.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from ...errors import SbpError +from ...service import journal +from ..frame import Row + +if TYPE_CHECKING: + from ..app import App + +LINES = 300 + + +class LogsView: + name = "logs" + title = "Logs" + + def rows(self, app: "App") -> list[Row]: + if not app.status().installed: + return [Row("sbp is not installed", kind="warn", selectable=False)] + try: + lines = journal(app.paths, lines=LINES) + except SbpError as exc: + return [Row(exc.message, kind="error", selectable=False)] + if not lines: + return [Row("no log lines", kind="dim", selectable=False)] + return [ + Row(f"last {len(lines)} lines from journalctl --user -u {app.paths.unit_name}", + kind="dim", selectable=False), + Row("", selectable=False), + ] + [Row(line, selectable=True) for line in lines] diff --git a/src/sbp/tui/views/protocols.py b/src/sbp/tui/views/protocols.py new file mode 100644 index 0000000..e0de332 --- /dev/null +++ b/src/sbp/tui/views/protocols.py @@ -0,0 +1,118 @@ +"""Protocol list and per-protocol settings, shown in the content pane.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Optional + +from ... import protocols +from ...ops import protocols as protocols_op +from ..frame import Row + +if TYPE_CHECKING: + from ..app import App + + +def _detail_text(protocol, enabled: bool, port: int, details) -> str: + state = "enabled" if enabled else "off" + text = f"{protocol.info.title:<18} {state:<8} {port:<6} {protocol.info.transport}" + if details: + text += " " + ", ".join(f"{key}={value}" for key, value in details) + return text + + +class ProtocolsView: + name = "protocols" + title = "Protocols" + + def __init__(self) -> None: + self.selected: Optional[str] = None + + def rows(self, app: "App") -> list[Row]: + if not app.status().installed: + return [Row("sbp is not installed", kind="warn", selectable=False)] + if self.selected: + return self._detail(app, self.selected) + rows = [Row("select a protocol to manage", kind="dim", selectable=False), Row("", selectable=False)] + for key, enabled, port, details in protocols_op.summary(app.paths): + protocol = protocols.get(key) + rows.append( + Row( + _detail_text(protocol, enabled, port, details), + action=lambda a, k=key: self._open(a, k), + ) + ) + return rows + + def _open(self, app: "App", key: str) -> None: + self.selected = key + app.reset() + + def _close(self, app: "App") -> None: + self.selected = None + app.reset() + + def _detail(self, app: "App", key: str) -> list[Row]: + row = next(item for item in protocols_op.summary(app.paths) if item[0] == key) + _, enabled, port, details = row + protocol = protocols.get(key) + rows = [ + Row(protocol.info.title, kind="header", selectable=False), + Row(f" state {'enabled' if enabled else 'off'}", + kind="good" if enabled else "warn", selectable=False), + Row(f" port {port}", selectable=False), + Row(f" transport {protocol.info.transport}", selectable=False), + ] + rows.extend(Row(f" {name:<12} {value}", selectable=False) for name, value in details) + rows.append(Row("", selectable=False)) + rows.append(Row("Actions", kind="header", selectable=False)) + if enabled: + rows.append(Row("Disable", action=lambda a: self._disable(a, key))) + else: + rows.append(Row("Enable", action=lambda a: self._enable(a, key))) + rows.append(Row("Change port", action=lambda a: self._port(a, key, port))) + if protocol.editable(): + rows.append(Row("Edit options", action=lambda a: self._options(a, key))) + rows.append(Row("Back", action=lambda a: self._close(a))) + return rows + + def _disable(self, app: "App", key: str) -> None: + app.external(f"Disable {key}", lambda progress: protocols_op.disable(app.paths, key)) + + def _enable(self, app: "App", key: str) -> None: + text = app.term.prompt("Enable", "Port (blank = auto)", "") + if text is None: + return + port = int(text) if text.isdigit() else None + app.external(f"Enable {key}", lambda progress: protocols_op.enable(app.paths, key, port=port)) + + def _port(self, app: "App", key: str, current: int) -> None: + text = app.term.prompt("Change port", "Port", str(current)) + if not text or not text.isdigit(): + return + app.external( + f"Change {key} port", + lambda progress: protocols_op.set_port(app.paths, key, int(text)), + ) + + def _options(self, app: "App", key: str) -> None: + protocol = protocols.get(key) + editable = protocol.editable() + current = protocols_op.options(app.paths, key) + labels = [f"{label}: {_show(current.get(name, ''))}" for name, label, _ in editable] + choice = app.term.menu(f"{protocol.info.title} options", labels) + if choice is None: + return + name, label, _kind = editable[choice] + value = app.term.prompt("Edit option", label, _show(current.get(name, ""))) + if value is None: + return + app.external( + f"Set {key}.{name}", + lambda progress: protocols_op.set_option(app.paths, key, name, value), + ) + + +def _show(value) -> str: + if isinstance(value, bool): + return "true" if value else "false" + return str(value) diff --git a/src/sbp/tui/views/settings.py b/src/sbp/tui/views/settings.py new file mode 100644 index 0000000..2ed9582 --- /dev/null +++ b/src/sbp/tui/views/settings.py @@ -0,0 +1,87 @@ +"""Server settings, certificate, upgrade, and uninstall.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from ...ops import install as install_op +from ...ops import settings as settings_op +from ...ops import uninstall as uninstall_op +from ...ops import upgrade as upgrade_op +from ..frame import Row + +if TYPE_CHECKING: + from ..app import App + + +class SettingsView: + name = "settings" + title = "Settings" + + def rows(self, app: "App") -> list[Row]: + status = app.status() + if not status.installed: + return [Row("sbp is not installed", kind="warn", selectable=False)] + return [ + Row("Server", kind="header", selectable=False), + Row(f" host {status.host}", selectable=False), + Row(f" sni {status.sni}", selectable=False), + Row(f" tls {status.tls_mode}", selectable=False), + Row(f" cert until {status.cert_until}", selectable=False), + Row(f" config {status.config_path}", selectable=False), + Row("", selectable=False), + Row("Actions", kind="header", selectable=False), + Row("Set host", action=self._host), + Row("Set SNI", action=self._sni), + Row("Use custom certificate", action=self._cert), + Row("Renew certificate", action=self._renew), + Row("Upgrade sing-box", action=self._upgrade), + Row("Reinstall", action=self._reinstall), + Row("Uninstall", action=self._uninstall), + ] + + def _host(self, app: "App") -> None: + value = app.term.prompt("Set host", "Client-facing host", app.status().host) + if value: + app.external("Set host", lambda progress: settings_op.set_host(app.paths, value)) + + def _sni(self, app: "App") -> None: + value = app.term.prompt("Set SNI", "TLS server name", app.status().sni) + if value: + app.external("Set SNI", lambda progress: settings_op.set_sni(app.paths, value)) + + def _cert(self, app: "App") -> None: + cert = app.term.prompt("Custom certificate", "Certificate path (PEM)", "") + if not cert: + return + key = app.term.prompt("Custom certificate", "Private key path (PEM)", "") + if not key: + return + sni = app.term.prompt("Custom certificate", "SNI", app.status().sni) + app.external( + "Use custom certificate", + lambda progress: settings_op.set_cert(app.paths, cert, key, sni or None), + ) + + def _renew(self, app: "App") -> None: + app.external("Renew certificate", lambda progress: settings_op.regenerate_cert(app.paths)) + + def _upgrade(self, app: "App") -> None: + app.external( + "Upgrade sing-box", + lambda progress: upgrade_op.upgrade(app.paths, progress=progress), + ) + + def _reinstall(self, app: "App") -> None: + if not app.term.confirm("Reinstall", "Reinstall sing-box and regenerate the service?"): + return + options = install_op.InstallOptions(force=True) + app.external( + "Reinstall", lambda progress: install_op.install(app.paths, options, progress=progress) + ) + + def _uninstall(self, app: "App") -> None: + if not app.term.confirm("Uninstall", "Stop and remove the service and binary?"): + return + purge = app.term.confirm("Uninstall", "Also delete configuration, users, and certificates?") + app.external("Uninstall", lambda progress: uninstall_op.uninstall(app.paths, purge=purge)) diff --git a/src/sbp/tui/views/users.py b/src/sbp/tui/views/users.py new file mode 100644 index 0000000..1862536 --- /dev/null +++ b/src/sbp/tui/views/users.py @@ -0,0 +1,89 @@ +"""User list, share links, Clash YAML, and client configs.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Optional + +from ... import protocols +from ...ops import show as show_op +from ...ops import users as users_op +from ..frame import Row + +if TYPE_CHECKING: + from ..app import App + + +class UsersView: + name = "users" + title = "Users" + + def __init__(self) -> None: + self.selected: Optional[str] = None + + def rows(self, app: "App") -> list[Row]: + if not app.status().installed: + return [Row("sbp is not installed", kind="warn", selectable=False)] + if self.selected: + return self._detail(app, self.selected) + users = users_op.list_users(app.paths) + rows = [Row("select a user to see links and clients", kind="dim", selectable=False), Row("", selectable=False)] + rows.extend( + Row(f"{user.name:<20} {user.created[:10]}", action=lambda a, n=user.name: self._open(a, n)) + for user in users + ) + rows.append(Row("", selectable=False)) + rows.append(Row("+ Add user", action=self._add)) + return rows + + def _open(self, app: "App", name: str) -> None: + self.selected = name + app.reset() + + def _close(self, app: "App") -> None: + self.selected = None + app.reset() + + def _add(self, app: "App") -> None: + name = app.term.prompt("Add user", "User name", "") + if not name: + return + app.external("Add user", lambda progress: users_op.add(app.paths, name)) + + def _detail(self, app: "App", name: str) -> list[Row]: + return [ + Row(f"User {name}", kind="header", selectable=False), + Row("", selectable=False), + Row("Show share URIs", action=lambda a: self._uris(a, name)), + Row("Show Clash YAML", action=lambda a: self._clash(a, name)), + Row("Show client config", action=lambda a: self._client(a, name)), + Row("Delete user", action=lambda a: self._delete(a, name)), + Row("Back", action=lambda a: self._close(a)), + ] + + def _uris(self, app: "App", name: str) -> None: + uris = show_op.user_uris(app.paths, name) + text = "\n\n".join(f"{key}\n {uri}" for key, uri in uris) or "no enabled protocols" + app.term.scroll_text(f"Share URIs: {name}", text) + + def _clash(self, app: "App", name: str) -> None: + app.term.scroll_text(f"Clash: {name}", show_op.user_clash(app.paths, name)) + + def _client(self, app: "App", name: str) -> None: + enabled = [row.key for row in app.status().protocols if row.enabled] + if not enabled: + app.term.message("Client config", "no enabled protocols") + return + labels = [protocols.get(key).info.title for key in enabled] + choice = app.term.menu("Client config", labels) + if choice is None: + return + app.term.scroll_text( + f"Client config: {name} / {enabled[choice]}", + show_op.client_config_json(app.paths, name, enabled[choice]), + ) + + def _delete(self, app: "App", name: str) -> None: + if not app.term.confirm("Delete user", f"Delete {name}? Clients using this user stop working."): + return + app.external("Delete user", lambda progress: users_op.remove(app.paths, name)) + self._close(app) diff --git a/tests/test_frame.py b/tests/test_frame.py new file mode 100644 index 0000000..a1ad606 --- /dev/null +++ b/tests/test_frame.py @@ -0,0 +1,115 @@ +import unittest +from unittest import mock + +from sbp.tui import frame +from sbp.tui.term import Term + + +class FakeScreen: + def __init__(self, height: int = 24, width: int = 80) -> None: + self.height = height + self.width = width + self.writes: list[tuple[int, int, str]] = [] + + def getmaxyx(self): + return self.height, self.width + + def addstr(self, y, x, text, attr=0): + if y >= self.height or x + len(text) > self.width: + raise AssertionError(f"write out of bounds: y={y} x={x} len={len(text)}") + self.writes.append((y, x, text)) + + def erase(self): + self.writes.clear() + + def refresh(self): + pass + + def keypad(self, flag): + pass + + def getch(self): + return -1 + + def text(self) -> str: + return "\n".join(text for _, _, text in self.writes) + + +def make_term(screen: FakeScreen) -> Term: + with mock.patch.object(frame_term_curses(), "has_colors", return_value=False), \ + mock.patch.object(frame_term_curses(), "curs_set", return_value=None): + return Term(screen) + + +def frame_term_curses(): + import sbp.tui.term as module + + return module.curses + + +class WindowOffsetTest(unittest.TestCase): + def test_no_scroll_when_everything_fits(self) -> None: + self.assertEqual(frame.window_offset(3, 5, 10), 0) + + def test_keeps_selection_visible(self) -> None: + self.assertEqual(frame.window_offset(7, 20, 5), 3) + + def test_clamps_to_last_page(self) -> None: + self.assertEqual(frame.window_offset(19, 20, 5), 15) + + +class SidebarWidthTest(unittest.TestCase): + def test_minimum(self) -> None: + self.assertEqual(frame.sidebar_width(["A"]), frame.SIDEBAR_MIN) + + def test_grows_with_titles(self) -> None: + self.assertGreater( + frame.sidebar_width(["VeryLongSectionName"]), frame.sidebar_width(["A"]) + ) + + def test_capped(self) -> None: + self.assertEqual(frame.sidebar_width(["x" * 100]), frame.SIDEBAR_MAX) + + +class DrawTest(unittest.TestCase): + def setUp(self) -> None: + self.screen = FakeScreen() + self.term = make_term(self.screen) + + def draw(self, rows=None, index=0, width=80): + rows = rows or [frame.Row("first", action=lambda app: None), frame.Row("second")] + frame.draw( + self.term, + title="sbp 2.0.0", + subtitle="host active", + sidebar=["Dashboard", "Protocols", "Users", "Settings", "Logs"], + sidebar_index=1, + sidebar_focused=False, + content_title="Protocols", + rows=rows, + index=index, + content_focused=True, + footer="keys", + ) + + def test_draws_both_panes(self) -> None: + self.draw() + text = self.screen.text() + for word in ("Dashboard", "Protocols", "Users", "Settings", "Logs", "first", "second", "keys"): + self.assertIn(word, text) + + def test_selected_row_is_padded_for_highlight(self) -> None: + self.draw(index=1) + row = [text for y, _, text in self.screen.writes if text.startswith("second")] + self.assertTrue(row) + self.assertGreater(len(row[0]), len("second")) + + def test_small_terminal_shows_hint(self) -> None: + self.screen = FakeScreen(height=8, width=40) + self.term.screen = self.screen + self.draw() + self.assertIn("terminal too small", self.screen.text()) + + +if __name__ == "__main__": + unittest.main()