1603 lines
56 KiB
Python
1603 lines
56 KiB
Python
from __future__ import annotations
|
||
|
||
from concurrent.futures import Future, ThreadPoolExecutor
|
||
from dataclasses import dataclass, replace
|
||
import curses
|
||
import unicodedata
|
||
from typing import Callable, Iterable
|
||
|
||
from .aliases import MAX_ALIAS_LENGTH, normalize_alias
|
||
from .domain import is_valid_session_name
|
||
from .remotes import (
|
||
RegistryError,
|
||
RemoteEntry,
|
||
RemoteQueryPool,
|
||
RemoteQueryResult,
|
||
RemoteRegistry,
|
||
)
|
||
|
||
|
||
RenameHandler = Callable[[str, str], None]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class DeleteSessionOutcome:
|
||
warning: str | None = None
|
||
windows: tuple["CatalogWindow", ...] | None = None
|
||
tab_closed: bool = False
|
||
|
||
|
||
def terminal_display_width(text: str) -> int:
|
||
"""Return the number of terminal cells occupied by Unicode text."""
|
||
width = 0
|
||
for character in text:
|
||
if unicodedata.combining(character):
|
||
continue
|
||
if unicodedata.category(character).startswith("C"):
|
||
continue
|
||
width += 2 if unicodedata.east_asian_width(character) in {"W", "F"} else 1
|
||
return width
|
||
|
||
|
||
def truncate_terminal_text(text: str, max_width: int) -> str:
|
||
if max_width <= 0:
|
||
return ""
|
||
result: list[str] = []
|
||
width = 0
|
||
for character in text:
|
||
character_width = terminal_display_width(character)
|
||
if width + character_width > max_width:
|
||
break
|
||
result.append(character)
|
||
width += character_width
|
||
return "".join(result)
|
||
|
||
|
||
def terminal_text_tail(text: str, max_width: int) -> str:
|
||
if max_width <= 0:
|
||
return ""
|
||
start = len(text)
|
||
width = 0
|
||
for index in range(len(text) - 1, -1, -1):
|
||
character_width = terminal_display_width(text[index])
|
||
if width + character_width > max_width:
|
||
break
|
||
start = index
|
||
width += character_width
|
||
return text[start:]
|
||
|
||
|
||
def pad_terminal_text(text: str, width: int) -> str:
|
||
truncated = truncate_terminal_text(text, width)
|
||
return truncated + " " * max(0, width - terminal_display_width(truncated))
|
||
|
||
|
||
def input_display_line(prompt: str, value: str, width: int) -> str:
|
||
prefix = f"{prompt}: "
|
||
if terminal_display_width(prefix) >= width:
|
||
return truncate_terminal_text(prefix, width)
|
||
value_width = width - terminal_display_width(prefix)
|
||
if terminal_display_width(value) <= value_width:
|
||
visible_value = value
|
||
elif value_width <= 1:
|
||
visible_value = "…"[:value_width]
|
||
else:
|
||
visible_value = "…" + terminal_text_tail(value, value_width - 1)
|
||
return prefix + visible_value
|
||
|
||
|
||
class SelectorError(RuntimeError):
|
||
"""The interactive selector could not be displayed."""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CatalogWindow:
|
||
window_id: str
|
||
key: str
|
||
label: str
|
||
dead: bool
|
||
active: bool
|
||
kind: str | None = None
|
||
target: str | None = None
|
||
session: str | None = None
|
||
default_label: str | None = None
|
||
remote_transport: str = "native"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class StopSessionOutcome:
|
||
windows: tuple[CatalogWindow, ...]
|
||
tab_closed: bool = True
|
||
warning: str | None = None
|
||
|
||
|
||
DeleteSessionHandler = Callable[[str, str, str | None], DeleteSessionOutcome | None]
|
||
StopSessionHandler = Callable[[str, str, str], StopSessionOutcome | list[CatalogWindow]]
|
||
CloseWindowHandler = Callable[[str], Iterable[CatalogWindow]]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class _PendingAction:
|
||
kind: str
|
||
target: str
|
||
session: str
|
||
window_id: str | None = None
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SelectorResult:
|
||
kind: str
|
||
window_id: str | None = None
|
||
dead: bool = False
|
||
target: str | None = None
|
||
session: str | None = None
|
||
remote_keybindings: str = "local"
|
||
remote_transport: str = "native"
|
||
endpoint_key: str | None = None
|
||
alias: str | None = None
|
||
|
||
@classmethod
|
||
def window(cls, window_id: str, *, dead: bool) -> SelectorResult:
|
||
return cls(kind="window", window_id=window_id, dead=dead)
|
||
|
||
@classmethod
|
||
def close(cls, window_id: str) -> SelectorResult:
|
||
return cls(kind="close", window_id=window_id)
|
||
|
||
@classmethod
|
||
def local(cls, session: str) -> SelectorResult:
|
||
return cls(kind="local", session=session)
|
||
|
||
@classmethod
|
||
def rename(cls, endpoint_key: str, alias: str) -> SelectorResult:
|
||
return cls(kind="rename", endpoint_key=endpoint_key, alias=alias)
|
||
|
||
@classmethod
|
||
def remote(
|
||
cls,
|
||
target: str,
|
||
session: str,
|
||
remote_keybindings: str,
|
||
remote_transport: str = "native",
|
||
) -> SelectorResult:
|
||
return cls(
|
||
kind="remote",
|
||
target=target,
|
||
session=session,
|
||
remote_keybindings=remote_keybindings,
|
||
remote_transport=remote_transport,
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PickerRow:
|
||
key: str
|
||
kind: str
|
||
main: str
|
||
detail: str = ""
|
||
selectable: bool = True
|
||
window_id: str | None = None
|
||
dead: bool = False
|
||
target: str | None = None
|
||
session: str | None = None
|
||
remote_keybindings: str = "local"
|
||
remote_transport: str = "native"
|
||
endpoint_key: str | None = None
|
||
deletable: bool = False
|
||
running: bool | None = None
|
||
error: str | None = None
|
||
|
||
|
||
def build_main_rows(
|
||
windows: Iterable[CatalogWindow],
|
||
entries: Iterable[RemoteEntry],
|
||
results: dict[str, RemoteQueryResult],
|
||
loading_targets: set[str],
|
||
*,
|
||
local_sessions: Iterable[str] = (),
|
||
registry_error: str | None = None,
|
||
) -> list[PickerRow]:
|
||
window_list = list(windows)
|
||
entry_list = sorted(entries, key=lambda entry: entry.target)
|
||
registered = {entry.target for entry in entry_list}
|
||
rows: list[PickerRow] = []
|
||
|
||
standalone = sorted(
|
||
[
|
||
window
|
||
for window in window_list
|
||
if window.kind != "remote"
|
||
or window.target is None
|
||
or window.target not in registered
|
||
],
|
||
key=lambda window: (not window.active, window.label.casefold()),
|
||
)
|
||
rows.extend(_window_row(window) for window in standalone)
|
||
|
||
open_local_sessions = {
|
||
window.session
|
||
for window in window_list
|
||
if window.kind == "local" and window.session is not None
|
||
}
|
||
available_local_sessions = sorted(
|
||
{
|
||
session
|
||
for session in local_sessions
|
||
if is_valid_session_name(session) and session not in open_local_sessions
|
||
},
|
||
key=lambda name: (name != "default", name),
|
||
)
|
||
rows.extend(
|
||
PickerRow(
|
||
f"local:{session}",
|
||
"local",
|
||
f"○ {_local_session_label(session)}",
|
||
"未打开",
|
||
session=session,
|
||
running=True,
|
||
)
|
||
for session in available_local_sessions
|
||
)
|
||
|
||
remote_windows = {
|
||
(window.target, window.session): window
|
||
for window in window_list
|
||
if window.kind == "remote"
|
||
and window.target is not None
|
||
and window.session is not None
|
||
}
|
||
for entry in entry_list:
|
||
result = results.get(entry.target)
|
||
preferred_transport = _preferred_remote_transport(entry, result, window_list)
|
||
state_known = result is not None and result.error is None
|
||
session_by_name = (
|
||
{
|
||
session.name: session
|
||
for session in result.sessions
|
||
if is_valid_session_name(session.name)
|
||
}
|
||
if state_known and result is not None
|
||
else {}
|
||
)
|
||
for target, session_name in remote_windows:
|
||
if target == entry.target and session_name is not None:
|
||
session_by_name.setdefault(session_name, None)
|
||
|
||
for session_name in sorted(
|
||
session_by_name,
|
||
key=lambda name: (name != "default", name),
|
||
):
|
||
window = remote_windows.get((entry.target, session_name))
|
||
session = session_by_name[session_name]
|
||
running: bool | None
|
||
if session is not None:
|
||
running = session.running
|
||
elif state_known and window is not None:
|
||
assert window is not None
|
||
running = not window.dead
|
||
else:
|
||
running = None
|
||
deletable = running is False and session_name != "default"
|
||
transport = preferred_transport
|
||
# Automatic transport reconciliation is intentionally one-way.
|
||
# A confirmed/manual SSH window must never be downgraded merely
|
||
# because platform detection is loading, failed or inconclusive.
|
||
if window is not None and not (
|
||
transport == "ssh" and window.remote_transport != "ssh"
|
||
):
|
||
rows.append(
|
||
_window_row(
|
||
window,
|
||
running=running,
|
||
error=result.error if result is not None else None,
|
||
deletable=deletable,
|
||
)
|
||
)
|
||
continue
|
||
if window is not None and transport == "ssh":
|
||
symbol = "↻"
|
||
detail = "Windows · 切换为 SSH 模式"
|
||
else:
|
||
symbol = "○" if running else "■"
|
||
if transport == "ssh":
|
||
detail = (
|
||
"Windows / SSH · 远端运行"
|
||
if running
|
||
else "Windows / SSH · 已停止 · default 保留项"
|
||
if session_name == "default"
|
||
else "Windows / SSH · 已停止"
|
||
)
|
||
else:
|
||
detail = (
|
||
"远端运行"
|
||
if running
|
||
else "已停止 · default 保留项"
|
||
if session_name == "default"
|
||
else "已停止"
|
||
)
|
||
rows.append(
|
||
PickerRow(
|
||
f"remote:{entry.target}:{session_name}",
|
||
"remote",
|
||
f"{symbol} {_remote_session_label(entry.target, session_name)}",
|
||
detail,
|
||
window_id=window.window_id if window is not None else None,
|
||
target=entry.target,
|
||
session=session_name,
|
||
remote_keybindings=entry.remote_keybindings,
|
||
remote_transport=transport,
|
||
endpoint_key=window.key if window is not None else None,
|
||
deletable=deletable,
|
||
running=running,
|
||
)
|
||
)
|
||
|
||
if session_by_name:
|
||
continue
|
||
if entry.target in loading_targets:
|
||
rows.append(
|
||
PickerRow(
|
||
f"loading:{entry.target}",
|
||
"loading",
|
||
f"… {entry.target}",
|
||
"正在查询",
|
||
selectable=False,
|
||
target=entry.target,
|
||
remote_keybindings=entry.remote_keybindings,
|
||
remote_transport=preferred_transport,
|
||
)
|
||
)
|
||
elif result is not None and result.error is not None:
|
||
rows.append(
|
||
PickerRow(
|
||
f"retry:{entry.target}",
|
||
"retry",
|
||
f"! {entry.target}",
|
||
"无法连接",
|
||
target=entry.target,
|
||
remote_keybindings=entry.remote_keybindings,
|
||
remote_transport=preferred_transport,
|
||
error=result.error,
|
||
)
|
||
)
|
||
elif state_known:
|
||
rows.append(
|
||
PickerRow(
|
||
f"empty:{entry.target}",
|
||
"remote-empty",
|
||
f"○ {entry.target}",
|
||
"没有 Session",
|
||
target=entry.target,
|
||
remote_keybindings=entry.remote_keybindings,
|
||
remote_transport=preferred_transport,
|
||
)
|
||
)
|
||
|
||
if entry_list:
|
||
rows.append(
|
||
PickerRow(
|
||
"new-session",
|
||
"new-session",
|
||
"+ 新建 Session…",
|
||
"选择远端",
|
||
)
|
||
)
|
||
if registry_error is not None:
|
||
rows.append(
|
||
PickerRow(
|
||
"registry-error",
|
||
"error",
|
||
"! 远端配置不可用",
|
||
registry_error,
|
||
selectable=False,
|
||
)
|
||
)
|
||
else:
|
||
rows.append(PickerRow("add-remote", "add-remote", "+ 添加远端…"))
|
||
return rows
|
||
|
||
|
||
def _preferred_remote_transport(
|
||
entry: RemoteEntry,
|
||
result: RemoteQueryResult | None,
|
||
windows: Iterable[CatalogWindow],
|
||
) -> str:
|
||
if (
|
||
entry.remote_transport == "ssh"
|
||
or (result is not None and result.platform == "windows")
|
||
or any(
|
||
window.target == entry.target and window.remote_transport == "ssh"
|
||
for window in windows
|
||
)
|
||
):
|
||
return "ssh"
|
||
return "native"
|
||
|
||
|
||
def _local_session_label(session: str) -> str:
|
||
return "本机" if session == "default" else f"本机 / {session}"
|
||
|
||
|
||
def _remote_session_label(target: str, session: str) -> str:
|
||
return target if session == "default" else f"{target} / {session}"
|
||
|
||
|
||
def _window_display_label(window: CatalogWindow) -> str:
|
||
if window.default_label is not None and window.label != window.default_label:
|
||
return window.label
|
||
if window.kind == "local" and window.session is not None:
|
||
return _local_session_label(window.session)
|
||
if (
|
||
window.kind == "remote"
|
||
and window.target is not None
|
||
and window.session is not None
|
||
):
|
||
return _remote_session_label(window.target, window.session)
|
||
return window.label
|
||
|
||
|
||
def _window_row(
|
||
window: CatalogWindow,
|
||
*,
|
||
running: bool | None = None,
|
||
error: str | None = None,
|
||
deletable: bool = False,
|
||
) -> PickerRow:
|
||
if running is False:
|
||
symbol = "■"
|
||
detail = (
|
||
"已停止 · default 保留项"
|
||
if window.session == "default"
|
||
else "已停止"
|
||
)
|
||
else:
|
||
symbol = "▶" if window.active else "×" if window.dead else "●"
|
||
detail = "当前" if window.active else "已断线" if window.dead else "已打开"
|
||
if error is not None and window.dead:
|
||
detail = "无法连接"
|
||
return PickerRow(
|
||
f"window:{window.window_id}",
|
||
"window",
|
||
f"{symbol} {_window_display_label(window)}",
|
||
detail,
|
||
window_id=window.window_id,
|
||
dead=window.dead,
|
||
target=window.target,
|
||
session=window.session,
|
||
remote_transport=window.remote_transport,
|
||
endpoint_key=window.key,
|
||
deletable=deletable,
|
||
running=running,
|
||
error=error,
|
||
)
|
||
|
||
|
||
def filter_rows(rows: Iterable[PickerRow], query: str) -> list[PickerRow]:
|
||
if not query:
|
||
return list(rows)
|
||
needle = query.casefold()
|
||
return [
|
||
row
|
||
for row in rows
|
||
if needle in f"{row.main} {row.detail} {row.error or ''}".casefold()
|
||
]
|
||
|
||
|
||
def _row_subject(row: PickerRow) -> str:
|
||
return row.main[2:] if len(row.main) > 2 and row.main[1] == " " else row.main
|
||
|
||
|
||
def _window_is_active(
|
||
windows: Iterable[CatalogWindow],
|
||
window_id: str | None,
|
||
) -> bool:
|
||
return any(window.window_id == window_id and window.active for window in windows)
|
||
|
||
|
||
def _action_row(
|
||
source: PickerRow,
|
||
suffix: str,
|
||
kind: str,
|
||
main: str,
|
||
detail: str = "",
|
||
) -> PickerRow:
|
||
return replace(
|
||
source,
|
||
key=f"action:{suffix}:{source.key}",
|
||
kind=kind,
|
||
main=main,
|
||
detail=detail,
|
||
selectable=True,
|
||
)
|
||
|
||
|
||
class SessionSelector:
|
||
def __init__(
|
||
self,
|
||
windows: Iterable[CatalogWindow],
|
||
registry: RemoteRegistry,
|
||
ssh_bin: str,
|
||
local_sessions: Iterable[str] = (),
|
||
aliases: dict[str, str] | None = None,
|
||
rename_handler: RenameHandler | None = None,
|
||
close_window_handler: CloseWindowHandler | None = None,
|
||
delete_session_handler: DeleteSessionHandler | None = None,
|
||
stop_session_handler: StopSessionHandler | None = None,
|
||
) -> None:
|
||
self.windows = list(windows)
|
||
self.local_sessions = list(local_sessions)
|
||
self.aliases = dict(aliases or {})
|
||
self.rename_handler = rename_handler
|
||
self.close_window_handler = close_window_handler
|
||
self.delete_session_handler = delete_session_handler
|
||
self.stop_session_handler = stop_session_handler
|
||
self.registry = registry
|
||
self.pool = RemoteQueryPool(ssh_bin)
|
||
self.action_executor = ThreadPoolExecutor(
|
||
max_workers=1,
|
||
thread_name_prefix="herdr-hub-action",
|
||
)
|
||
self.action_future: Future[object] | None = None
|
||
self.pending_action: _PendingAction | None = None
|
||
self.entries: dict[str, RemoteEntry] = {}
|
||
self.results: dict[str, RemoteQueryResult] = {}
|
||
self.registry_error: str | None = None
|
||
self.message = ""
|
||
self.query = ""
|
||
self.selected_key: str | None = None
|
||
self.scroll = 0
|
||
self.view = "main"
|
||
self.input_kind: str | None = None
|
||
self.confirm_kind: str | None = None
|
||
self.input_value = ""
|
||
self.input_target: str | None = None
|
||
self.input_session: str | None = None
|
||
self.input_window_id: str | None = None
|
||
self.input_key: str | None = None
|
||
self.input_remote_keybindings = "local"
|
||
self.input_remote_transport = "native"
|
||
self.context_row: PickerRow | None = None
|
||
self._drawn_rows: dict[int, PickerRow] = {}
|
||
self._action_button: tuple[int, int, int] | None = None
|
||
self._row_action_buttons: dict[int, tuple[int, int, PickerRow]] = {}
|
||
self._load_registry()
|
||
|
||
def _load_registry(self) -> None:
|
||
try:
|
||
entries = self.registry.load()
|
||
except RegistryError as error:
|
||
self.registry_error = str(error)
|
||
return
|
||
self.registry_error = None
|
||
self.entries = {entry.target: entry for entry in entries}
|
||
self.pool.enqueue_many(entries)
|
||
|
||
def run(self) -> SelectorResult | None:
|
||
try:
|
||
return curses.wrapper(self._curses_main)
|
||
except curses.error as error:
|
||
raise SelectorError(f"could not initialize the terminal selector: {error}") from error
|
||
finally:
|
||
self.pool.close()
|
||
self.action_executor.shutdown(wait=False, cancel_futures=True)
|
||
|
||
def _curses_main(self, screen) -> SelectorResult | None: # type: ignore[no-untyped-def]
|
||
screen.keypad(True)
|
||
screen.timeout(100)
|
||
try:
|
||
curses.curs_set(0)
|
||
except curses.error:
|
||
pass
|
||
try:
|
||
curses.use_default_colors()
|
||
except curses.error:
|
||
pass
|
||
curses.mouseinterval(150)
|
||
curses.mousemask(curses.ALL_MOUSE_EVENTS)
|
||
|
||
while True:
|
||
for result in self.pool.poll():
|
||
self.results[result.entry.target] = result
|
||
self._poll_action()
|
||
rows = self._rows()
|
||
filtered = (
|
||
filter_rows(rows, self.query)
|
||
if self.view in {"main", "search"}
|
||
else rows
|
||
)
|
||
self._normalize_selection(filtered)
|
||
self._draw(screen, filtered)
|
||
try:
|
||
key = screen.get_wch()
|
||
except curses.error:
|
||
continue
|
||
|
||
outcome = self._handle_key(key, filtered)
|
||
if isinstance(outcome, SelectorResult) or outcome is False:
|
||
return outcome if isinstance(outcome, SelectorResult) else None
|
||
|
||
def _rows(self) -> list[PickerRow]:
|
||
if self.view == "actions":
|
||
return self._action_rows()
|
||
if self.view == "confirm":
|
||
return self._confirmation_rows()
|
||
if self.view == "working":
|
||
action = self.pending_action
|
||
target = action.target if action is not None else ""
|
||
session = action.session if action is not None else ""
|
||
verb = "删除" if action is not None and action.kind == "delete" else "停止"
|
||
return [
|
||
PickerRow(
|
||
"working-heading",
|
||
"heading",
|
||
f"正在{verb} {target}/{session}",
|
||
"SSH 操作进行中，请稍候…",
|
||
selectable=False,
|
||
)
|
||
]
|
||
if self.view == "input":
|
||
return []
|
||
return build_main_rows(
|
||
self.windows,
|
||
self.entries.values(),
|
||
self.results,
|
||
{target for target in self.entries if self.pool.is_loading(target)},
|
||
local_sessions=self.local_sessions,
|
||
registry_error=self.registry_error,
|
||
)
|
||
|
||
def _confirmation_rows(self) -> list[PickerRow]:
|
||
target = self.input_target or ""
|
||
session = self.input_session or ""
|
||
if self.confirm_kind == "remove":
|
||
return [
|
||
PickerRow(
|
||
"confirm-heading",
|
||
"heading",
|
||
f"移除 {target}？",
|
||
"只删除 Hub 记录，不修改远端数据",
|
||
selectable=False,
|
||
),
|
||
PickerRow("confirm-yes", "confirm-yes", "确认移除", target=target),
|
||
PickerRow("confirm-no", "back", "取消"),
|
||
]
|
||
if self.confirm_kind == "delete":
|
||
return [
|
||
PickerRow(
|
||
"confirm-heading",
|
||
"heading",
|
||
f"永久删除 {target}/{session}？",
|
||
(
|
||
"关联 Hub Tab 也会关闭"
|
||
if self.input_window_id is not None
|
||
else "工作现场和历史数据无法恢复"
|
||
),
|
||
selectable=False,
|
||
),
|
||
PickerRow(
|
||
"confirm-yes",
|
||
"delete-session-yes",
|
||
"确认永久删除",
|
||
window_id=self.input_window_id,
|
||
target=target,
|
||
session=session,
|
||
),
|
||
PickerRow("confirm-no", "back", "取消"),
|
||
]
|
||
closes_tab = len(self.windows) > 1
|
||
return [
|
||
PickerRow(
|
||
"confirm-heading",
|
||
"heading",
|
||
f"关闭并停止 {target}/{session}？",
|
||
(
|
||
"Hub Tab 将关闭，远端任务会中断"
|
||
if closes_tab
|
||
else "最后一个 Tab 将保留，远端任务会中断"
|
||
),
|
||
selectable=False,
|
||
),
|
||
PickerRow(
|
||
"confirm-yes",
|
||
"stop-session-yes",
|
||
"确认停止",
|
||
window_id=self.input_window_id,
|
||
target=target,
|
||
session=session,
|
||
),
|
||
PickerRow("confirm-no", "back", "取消"),
|
||
]
|
||
|
||
def _action_rows(self) -> list[PickerRow]:
|
||
source = self.context_row
|
||
if source is None:
|
||
return [PickerRow("actions-back", "back", "← 返回 Sessions")]
|
||
|
||
if source.kind == "new-session":
|
||
rows = [
|
||
PickerRow(
|
||
"actions-heading",
|
||
"heading",
|
||
"在哪台远端新建 Session？",
|
||
selectable=False,
|
||
)
|
||
]
|
||
for entry in sorted(self.entries.values(), key=lambda item: item.target):
|
||
result = self.results.get(entry.target)
|
||
transport = _preferred_remote_transport(entry, result, self.windows)
|
||
rows.append(
|
||
PickerRow(
|
||
f"action-new:{entry.target}",
|
||
"action-new",
|
||
entry.target,
|
||
"Windows / SSH" if transport == "ssh" else "自动检测",
|
||
target=entry.target,
|
||
remote_keybindings=entry.remote_keybindings,
|
||
remote_transport=transport,
|
||
)
|
||
)
|
||
rows.append(PickerRow("actions-back", "back", "← 返回 Sessions"))
|
||
return rows
|
||
|
||
if source.kind == "add-remote":
|
||
return [
|
||
PickerRow(
|
||
"actions-heading",
|
||
"heading",
|
||
"添加远端",
|
||
"选择连接模式",
|
||
selectable=False,
|
||
),
|
||
PickerRow(
|
||
"action-add-auto",
|
||
"action-add",
|
||
"自动检测（推荐）",
|
||
"Linux / macOS",
|
||
remote_transport="native",
|
||
),
|
||
PickerRow(
|
||
"action-add-ssh",
|
||
"action-add",
|
||
"Windows / SSH",
|
||
"强制 SSH TTY 模式",
|
||
remote_transport="ssh",
|
||
),
|
||
PickerRow("actions-back", "back", "← 返回 Sessions"),
|
||
]
|
||
|
||
rows = [
|
||
PickerRow(
|
||
"actions-heading",
|
||
"heading",
|
||
_row_subject(source),
|
||
source.error or source.detail,
|
||
selectable=False,
|
||
)
|
||
]
|
||
if source.kind == "window":
|
||
label = (
|
||
"启动并打开"
|
||
if source.running is False
|
||
else "重新连接"
|
||
if source.dead
|
||
else "返回当前 Session"
|
||
if _window_is_active(self.windows, source.window_id)
|
||
else "切换到此 Session"
|
||
)
|
||
rows.append(_action_row(source, "open", "action-window", label))
|
||
elif source.kind == "local":
|
||
rows.append(_action_row(source, "open", "action-local", "打开 Session"))
|
||
elif source.kind == "remote":
|
||
label = "启动并打开" if source.running is False else "打开 Session"
|
||
rows.append(_action_row(source, "open", "action-remote", label))
|
||
elif source.kind == "retry":
|
||
rows.append(_action_row(source, "retry", "action-retry", "重试连接"))
|
||
|
||
if source.kind != "retry" and source.error is not None and source.target is not None:
|
||
rows.append(_action_row(source, "retry", "action-retry", "重试连接"))
|
||
|
||
if source.kind == "window" and source.endpoint_key is not None:
|
||
rows.append(_action_row(source, "rename", "action-rename", "修改显示名称"))
|
||
if len(self.windows) > 1:
|
||
rows.append(
|
||
_action_row(
|
||
source,
|
||
"close",
|
||
"action-close",
|
||
"关闭 Hub Tab",
|
||
"远端 Session 保持运行" if source.target is not None else "",
|
||
)
|
||
)
|
||
if (
|
||
source.window_id is not None
|
||
and source.target is not None
|
||
and source.session is not None
|
||
and (source.running is True or (source.running is None and not source.dead))
|
||
):
|
||
rows.append(
|
||
_action_row(
|
||
source,
|
||
"stop",
|
||
"action-stop",
|
||
"关闭并停止 Session",
|
||
"会中断远端任务",
|
||
)
|
||
)
|
||
if source.deletable:
|
||
rows.append(
|
||
_action_row(
|
||
source,
|
||
"delete",
|
||
"action-delete",
|
||
"永久删除 Session",
|
||
"工作现场无法恢复",
|
||
)
|
||
)
|
||
if source.target is not None and source.target in self.entries:
|
||
entry = self.entries[source.target]
|
||
rows.append(
|
||
replace(
|
||
_action_row(
|
||
source,
|
||
"new",
|
||
"action-new",
|
||
"在此远端新建 Session",
|
||
),
|
||
remote_keybindings=entry.remote_keybindings,
|
||
)
|
||
)
|
||
rows.append(
|
||
_action_row(
|
||
source,
|
||
"remove",
|
||
"action-remove",
|
||
"移除远端记录",
|
||
"不修改远端数据",
|
||
)
|
||
)
|
||
rows.append(PickerRow("actions-back", "back", "← 返回 Sessions"))
|
||
return rows
|
||
|
||
def _handle_key(self, key, rows: list[PickerRow]): # type: ignore[no-untyped-def]
|
||
if self.action_future is not None:
|
||
if key == "\x1b":
|
||
self.message = "远端操作正在进行，完成后可继续管理"
|
||
return None
|
||
if self.view == "input":
|
||
return self._handle_input_key(key)
|
||
if key == curses.KEY_MOUSE:
|
||
return self._handle_mouse(rows)
|
||
if key in {curses.KEY_UP}:
|
||
self._move(rows, -1)
|
||
return None
|
||
if key in {curses.KEY_DOWN}:
|
||
self._move(rows, 1)
|
||
return None
|
||
if key == curses.KEY_PPAGE:
|
||
self._move(rows, -8)
|
||
return None
|
||
if key == curses.KEY_NPAGE:
|
||
self._move(rows, 8)
|
||
return None
|
||
if self.view == "search":
|
||
if key in {curses.KEY_ENTER, "\n", "\r"}:
|
||
return self._activate(self._selected_row(rows))
|
||
if key == "\x1b":
|
||
selected_key = self.selected_key
|
||
self.query = ""
|
||
self._show_main(selected_key=selected_key)
|
||
return None
|
||
if key in {curses.KEY_BACKSPACE, "\x7f", "\b"}:
|
||
self.query = self.query[:-1]
|
||
self.selected_key = None
|
||
return None
|
||
if isinstance(key, str) and key.isprintable() and len(self.query) < 128:
|
||
self.query += key
|
||
self.selected_key = None
|
||
return None
|
||
if key in {curses.KEY_ENTER, "\n", "\r"}:
|
||
return self._activate(self._selected_row(rows))
|
||
if key == " " and self.view == "main":
|
||
self._show_actions(self._selected_row(rows))
|
||
return None
|
||
if key == "/" and self.view == "main":
|
||
self.view = "search"
|
||
self.query = ""
|
||
self.selected_key = None
|
||
self.message = ""
|
||
return None
|
||
if key == "\x12" and self.view == "main":
|
||
self.results.clear()
|
||
self.pool.enqueue_many(self.entries.values())
|
||
self.message = "正在刷新所有远端"
|
||
return None
|
||
if key == "\x1b":
|
||
return self._go_back()
|
||
return None
|
||
|
||
def _handle_input_key(self, key): # type: ignore[no-untyped-def]
|
||
if key == "\x1b":
|
||
return self._go_back()
|
||
if key in {curses.KEY_ENTER, "\n", "\r"}:
|
||
value = self.input_value.strip()
|
||
if self.input_kind == "add":
|
||
return self._finish_add(value)
|
||
if self.input_kind == "rename":
|
||
return self._finish_rename(value)
|
||
return self._finish_new(value)
|
||
if key == "\x15" and self.input_kind == "rename":
|
||
self.input_value = ""
|
||
return None
|
||
if key in {curses.KEY_BACKSPACE, "\x7f", "\b"}:
|
||
self.input_value = self.input_value[:-1]
|
||
return None
|
||
limit = MAX_ALIAS_LENGTH if self.input_kind == "rename" else 512
|
||
if isinstance(key, str) and key.isprintable() and len(self.input_value) < limit:
|
||
self.input_value += key
|
||
return None
|
||
|
||
def _handle_mouse(self, rows: list[PickerRow]):
|
||
try:
|
||
_, x, y, _, state = curses.getmouse()
|
||
except curses.error:
|
||
return None
|
||
if state & getattr(curses, "BUTTON4_PRESSED", 0):
|
||
self._move(rows, -3)
|
||
return None
|
||
if state & getattr(curses, "BUTTON5_PRESSED", 0):
|
||
self._move(rows, 3)
|
||
return None
|
||
activated = (
|
||
getattr(curses, "BUTTON1_RELEASED", 0)
|
||
| getattr(curses, "BUTTON1_CLICKED", 0)
|
||
| getattr(curses, "BUTTON1_DOUBLE_CLICKED", 0)
|
||
)
|
||
if state & activated:
|
||
row_action = self._row_action_buttons.get(y)
|
||
if row_action is not None:
|
||
button_start, button_end, action_row = row_action
|
||
if button_start <= x < button_end:
|
||
self.selected_key = action_row.key
|
||
self._show_actions(action_row)
|
||
return None
|
||
if self._action_button is not None and self.view in {"main", "search"}:
|
||
button_y, button_start, button_end = self._action_button
|
||
if y == button_y and button_start <= x < button_end:
|
||
self._show_actions(self._selected_row(rows))
|
||
return None
|
||
row = self._drawn_rows.get(y)
|
||
if row is not None and row.selectable:
|
||
self.selected_key = row.key
|
||
return self._activate(row)
|
||
return None
|
||
|
||
def _activate(self, row: PickerRow | None):
|
||
if row is None:
|
||
return None
|
||
if row.kind in {"window", "action-window"}:
|
||
assert row.window_id is not None
|
||
return SelectorResult.window(row.window_id, dead=row.dead)
|
||
if row.kind == "action-close":
|
||
assert row.window_id is not None
|
||
if self.close_window_handler is None:
|
||
return SelectorResult.close(row.window_id)
|
||
self._finish_close_window(row.window_id)
|
||
return None
|
||
if row.kind in {"local", "action-local"}:
|
||
assert row.session is not None
|
||
return SelectorResult.local(row.session)
|
||
if row.kind in {"remote", "action-remote"}:
|
||
assert row.target is not None and row.session is not None
|
||
return SelectorResult.remote(
|
||
row.target,
|
||
row.session,
|
||
row.remote_keybindings,
|
||
row.remote_transport,
|
||
)
|
||
if row.kind in {"new-session", "add-remote", "remote-empty"}:
|
||
self._show_actions(row)
|
||
elif row.kind in {"retry", "action-retry"}:
|
||
assert row.target is not None
|
||
self.results.pop(row.target, None)
|
||
self.pool.enqueue(self.entries[row.target])
|
||
self._show_main()
|
||
self.message = f"正在重试 {row.target}"
|
||
elif row.kind == "action-new":
|
||
self.view = "input"
|
||
self.input_kind = "new"
|
||
self.input_target = row.target
|
||
self.input_remote_keybindings = row.remote_keybindings
|
||
self.input_remote_transport = row.remote_transport
|
||
self.input_value = ""
|
||
self.message = ""
|
||
elif row.kind == "action-add":
|
||
self.view = "input"
|
||
self.input_kind = "add"
|
||
self.input_target = None
|
||
self.input_remote_keybindings = "local"
|
||
self.input_remote_transport = row.remote_transport
|
||
self.input_value = ""
|
||
self.message = ""
|
||
elif row.kind == "action-rename":
|
||
assert row.endpoint_key is not None
|
||
self.view = "input"
|
||
self.input_kind = "rename"
|
||
self.input_key = row.endpoint_key
|
||
self.input_target = next(
|
||
(
|
||
window.default_label
|
||
for window in self.windows
|
||
if window.key == row.endpoint_key
|
||
),
|
||
row.endpoint_key,
|
||
)
|
||
self.input_value = self.aliases.get(row.endpoint_key, "")
|
||
self.message = ""
|
||
elif row.kind == "action-remove":
|
||
assert row.target is not None
|
||
self._show_remove_confirmation(row.target)
|
||
elif row.kind == "action-stop":
|
||
assert (
|
||
row.window_id is not None
|
||
and row.target is not None
|
||
and row.session is not None
|
||
)
|
||
self.view = "confirm"
|
||
self.confirm_kind = "stop"
|
||
self.input_window_id = row.window_id
|
||
self.input_target = row.target
|
||
self.input_session = row.session
|
||
self.selected_key = "confirm-no"
|
||
self.message = ""
|
||
elif row.kind == "action-delete":
|
||
self._show_delete_session_confirmation(row)
|
||
elif row.kind == "confirm-yes":
|
||
self._finish_remove(row.target or "")
|
||
elif row.kind == "delete-session-yes":
|
||
self._finish_delete_session(
|
||
row.target or "",
|
||
row.session or "",
|
||
row.window_id,
|
||
)
|
||
elif row.kind == "stop-session-yes":
|
||
self._finish_stop_session(
|
||
row.window_id or "",
|
||
row.target or "",
|
||
row.session or "",
|
||
)
|
||
elif row.kind == "back":
|
||
return self._go_back()
|
||
return None
|
||
|
||
def _finish_add(self, target: str):
|
||
try:
|
||
entry = RemoteEntry(
|
||
target,
|
||
self.input_remote_keybindings,
|
||
self.input_remote_transport,
|
||
)
|
||
self.registry.upsert(entry)
|
||
except (ValueError, RegistryError) as error:
|
||
self.message = str(error)
|
||
return None
|
||
self.entries[target] = entry
|
||
self.results.pop(target, None)
|
||
self.pool.enqueue(entry)
|
||
self._show_main()
|
||
self.message = f"已添加 {target}，正在查询"
|
||
return None
|
||
|
||
def _finish_new(self, session: str):
|
||
if not is_valid_session_name(session):
|
||
self.message = "Session 名称需为 1-64 位字母、数字、. _ -"
|
||
return None
|
||
target = self.input_target
|
||
assert target is not None
|
||
entry = self.entries[target]
|
||
return SelectorResult.remote(
|
||
target,
|
||
session,
|
||
entry.remote_keybindings,
|
||
self.input_remote_transport,
|
||
)
|
||
|
||
def _finish_rename(self, alias: str):
|
||
try:
|
||
normalized = normalize_alias(alias)
|
||
except ValueError as error:
|
||
self.message = str(error)
|
||
return None
|
||
assert self.input_key is not None
|
||
if self.rename_handler is not None:
|
||
try:
|
||
self.rename_handler(self.input_key, normalized)
|
||
except (OSError, RuntimeError, ValueError) as error:
|
||
self.message = str(error)
|
||
return None
|
||
|
||
endpoint_key = self.input_key
|
||
if normalized:
|
||
self.aliases[endpoint_key] = normalized
|
||
else:
|
||
self.aliases.pop(endpoint_key, None)
|
||
display_label = normalized
|
||
updated_windows: list[CatalogWindow] = []
|
||
for window in self.windows:
|
||
if window.key == endpoint_key:
|
||
display_label = normalized or window.default_label or window.label
|
||
window = replace(window, label=display_label)
|
||
updated_windows.append(window)
|
||
self.windows = updated_windows
|
||
selected_key = (
|
||
self.context_row.key
|
||
if self.context_row is not None
|
||
else next(
|
||
(
|
||
f"window:{window.window_id}"
|
||
for window in self.windows
|
||
if window.key == endpoint_key
|
||
),
|
||
None,
|
||
)
|
||
)
|
||
self._show_main(selected_key=selected_key)
|
||
self.message = (
|
||
f"已保存：{display_label}"
|
||
if normalized
|
||
else f"已恢复默认名称：{display_label}"
|
||
)
|
||
return None
|
||
return SelectorResult.rename(self.input_key, normalized)
|
||
|
||
def _show_delete_session_confirmation(self, row: PickerRow) -> None:
|
||
assert row.deletable and row.target is not None and row.session is not None
|
||
self.view = "confirm"
|
||
self.confirm_kind = "delete"
|
||
self.input_target = row.target
|
||
self.input_session = row.session
|
||
self.input_window_id = row.window_id
|
||
self.selected_key = "confirm-no"
|
||
self.message = ""
|
||
|
||
def _show_remove_confirmation(self, target: str) -> None:
|
||
self.view = "confirm"
|
||
self.confirm_kind = "remove"
|
||
self.input_target = target
|
||
self.selected_key = "confirm-no"
|
||
self.message = ""
|
||
|
||
def _show_actions(self, row: PickerRow | None) -> None:
|
||
if row is None or not row.selectable:
|
||
return
|
||
self.context_row = row
|
||
self.view = "actions"
|
||
self.selected_key = None
|
||
self.scroll = 0
|
||
self.message = ""
|
||
|
||
def _go_back(self):
|
||
if self.view == "main":
|
||
return False
|
||
if self.view == "actions":
|
||
selected_key = self.context_row.key if self.context_row is not None else None
|
||
self._show_main(selected_key=selected_key)
|
||
return None
|
||
if self.context_row is not None and self.view in {"confirm", "input"}:
|
||
self.view = "actions"
|
||
self.input_kind = None
|
||
self.confirm_kind = None
|
||
self.input_target = None
|
||
self.input_session = None
|
||
self.input_window_id = None
|
||
self.input_value = ""
|
||
self.input_key = None
|
||
self.selected_key = None
|
||
self.scroll = 0
|
||
self.message = ""
|
||
return None
|
||
self._show_main()
|
||
return None
|
||
|
||
def _finish_delete_session(
|
||
self,
|
||
target: str,
|
||
session: str,
|
||
window_id: str | None = None,
|
||
) -> None:
|
||
if self.delete_session_handler is None:
|
||
self.message = "Session 删除功能不可用"
|
||
return
|
||
self.pending_action = _PendingAction("delete", target, session, window_id)
|
||
self.view = "working"
|
||
self.selected_key = None
|
||
self.message = "正在通过 SSH 永久删除 Session…"
|
||
self.action_future = self.action_executor.submit(
|
||
self.delete_session_handler,
|
||
target,
|
||
session,
|
||
window_id,
|
||
)
|
||
|
||
def _finish_stop_session(
|
||
self,
|
||
window_id: str,
|
||
target: str,
|
||
session: str,
|
||
) -> None:
|
||
if self.stop_session_handler is None:
|
||
self.message = "Session 停止功能不可用"
|
||
return
|
||
self.pending_action = _PendingAction("stop", target, session, window_id)
|
||
self.view = "working"
|
||
self.selected_key = None
|
||
self.message = "正在停止远端 Session…"
|
||
self.action_future = self.action_executor.submit(
|
||
self.stop_session_handler,
|
||
window_id,
|
||
target,
|
||
session,
|
||
)
|
||
|
||
def _poll_action(self) -> None:
|
||
future = self.action_future
|
||
action = self.pending_action
|
||
if future is None or action is None or not future.done():
|
||
return
|
||
self.action_future = None
|
||
self.pending_action = None
|
||
try:
|
||
outcome = future.result()
|
||
except (OSError, RuntimeError, ValueError) as error:
|
||
self._show_main()
|
||
entry = self.entries.get(action.target)
|
||
if entry is not None:
|
||
self.pool.enqueue(entry)
|
||
self.message = f"{action.target}/{action.session} 操作失败：{error}"
|
||
return
|
||
|
||
if action.kind == "delete":
|
||
delete_outcome = (
|
||
outcome
|
||
if isinstance(outcome, DeleteSessionOutcome)
|
||
else DeleteSessionOutcome()
|
||
)
|
||
if delete_outcome.windows is not None:
|
||
self.windows = list(delete_outcome.windows)
|
||
result = self.results.get(action.target)
|
||
if result is not None:
|
||
self.results[action.target] = replace(
|
||
result,
|
||
sessions=tuple(
|
||
item
|
||
for item in result.sessions
|
||
if item.name != action.session
|
||
),
|
||
)
|
||
warning = delete_outcome.warning
|
||
message = (
|
||
f"已关闭 Tab 并删除历史 Session：{action.target}/{action.session}"
|
||
if delete_outcome.tab_closed
|
||
else f"已删除历史 Session：{action.target}/{action.session}"
|
||
)
|
||
else:
|
||
stop_outcome = (
|
||
outcome
|
||
if isinstance(outcome, StopSessionOutcome)
|
||
else StopSessionOutcome(tuple(outcome))
|
||
)
|
||
self.windows = list(stop_outcome.windows)
|
||
result = self.results.get(action.target)
|
||
if result is not None:
|
||
self.results[action.target] = replace(
|
||
result,
|
||
sessions=tuple(
|
||
replace(item, running=False)
|
||
if item.name == action.session
|
||
else item
|
||
for item in result.sessions
|
||
),
|
||
)
|
||
warning = stop_outcome.warning
|
||
message = (
|
||
f"已关闭 Tab 并停止远端 Session：{action.target}/{action.session}"
|
||
if stop_outcome.tab_closed
|
||
else f"已停止远端 Session：{action.target}/{action.session}"
|
||
)
|
||
|
||
entry = self.entries.get(action.target)
|
||
if entry is not None:
|
||
self.pool.enqueue(entry)
|
||
self._show_main()
|
||
self.message = f"{message}；{warning}" if warning else message
|
||
|
||
def _finish_remove(self, target: str) -> None:
|
||
try:
|
||
self.registry.remove(target)
|
||
except RegistryError as error:
|
||
self.message = str(error)
|
||
self._show_main()
|
||
return
|
||
self.pool.cancel(target)
|
||
self.entries.pop(target, None)
|
||
self.results.pop(target, None)
|
||
self._show_main()
|
||
self.message = f"已移除 {target}；已有 window 保持不变"
|
||
|
||
def _finish_close_window(self, window_id: str) -> None:
|
||
assert self.close_window_handler is not None
|
||
label = next(
|
||
(
|
||
_window_display_label(window)
|
||
for window in self.windows
|
||
if window.window_id == window_id
|
||
),
|
||
window_id,
|
||
)
|
||
try:
|
||
self.windows = list(self.close_window_handler(window_id))
|
||
except (OSError, RuntimeError, ValueError) as error:
|
||
self.message = f"关闭 Tab 失败：{error}"
|
||
return
|
||
self._show_main()
|
||
self.message = f"已关闭 Hub Tab：{label}；Session 保持运行"
|
||
|
||
def _show_main(self, selected_key: str | None = None) -> None:
|
||
self.view = "main"
|
||
self.query = ""
|
||
self.input_kind = None
|
||
self.confirm_kind = None
|
||
self.input_target = None
|
||
self.input_session = None
|
||
self.input_window_id = None
|
||
self.input_value = ""
|
||
self.input_key = None
|
||
self.input_remote_keybindings = "local"
|
||
self.input_remote_transport = "native"
|
||
self.context_row = None
|
||
self.selected_key = selected_key
|
||
self.scroll = 0
|
||
|
||
def _normalize_selection(self, rows: list[PickerRow]) -> None:
|
||
selectable = [row for row in rows if row.selectable]
|
||
if not selectable:
|
||
self.selected_key = None
|
||
self.scroll = 0
|
||
return
|
||
if self.selected_key not in {row.key for row in selectable}:
|
||
self.selected_key = selectable[0].key
|
||
|
||
def _selected_row(self, rows: list[PickerRow]) -> PickerRow | None:
|
||
return next((row for row in rows if row.key == self.selected_key), None)
|
||
|
||
def _move(self, rows: list[PickerRow], delta: int) -> None:
|
||
selectable = [row for row in rows if row.selectable]
|
||
if not selectable:
|
||
return
|
||
current = next(
|
||
(index for index, row in enumerate(selectable) if row.key == self.selected_key),
|
||
0,
|
||
)
|
||
next_index = max(0, min(len(selectable) - 1, current + delta))
|
||
self.selected_key = selectable[next_index].key
|
||
|
||
def _draw(self, screen, rows: list[PickerRow]) -> None: # type: ignore[no-untyped-def]
|
||
screen.erase()
|
||
height, width = screen.getmaxyx()
|
||
self._drawn_rows = {}
|
||
self._action_button = None
|
||
self._row_action_buttons = {}
|
||
if height < 8 or width < 28:
|
||
self._safe_add(screen, 0, 0, "终端太小，请旋转屏幕", curses.A_BOLD)
|
||
screen.refresh()
|
||
return
|
||
|
||
if self.view == "input":
|
||
self._draw_input(screen, height, width)
|
||
return
|
||
try:
|
||
curses.curs_set(0)
|
||
except curses.error:
|
||
pass
|
||
|
||
title = {
|
||
"main": "Sessions",
|
||
"search": "Sessions",
|
||
"actions": "操作",
|
||
"confirm": {
|
||
"remove": "移除远端记录",
|
||
"delete": "永久删除 Session",
|
||
"stop": "关闭并停止 Session",
|
||
}.get(self.confirm_kind, "确认操作"),
|
||
"working": (
|
||
"删除 Session"
|
||
if self.pending_action is not None
|
||
and self.pending_action.kind == "delete"
|
||
else "停止 Session"
|
||
),
|
||
}.get(self.view, "Sessions")
|
||
self._safe_add(screen, 1, 2, title, curses.A_BOLD)
|
||
if self.view in {"main", "search"}:
|
||
action_label = "[操作]"
|
||
action_width = terminal_display_width(action_label)
|
||
action_x = max(2, width - action_width - 2)
|
||
self._safe_add(screen, 1, action_x, action_label, curses.A_BOLD)
|
||
self._action_button = (1, action_x, action_x + action_width)
|
||
if self.view == "search":
|
||
search_line = input_display_line("搜索", self.query, width - 4)
|
||
self._safe_add(screen, 2, 2, search_line)
|
||
try:
|
||
curses.curs_set(1)
|
||
screen.move(
|
||
2,
|
||
min(width - 2, 2 + terminal_display_width(search_line)),
|
||
)
|
||
except curses.error:
|
||
pass
|
||
else:
|
||
self._safe_add(screen, 2, 2, "按 / 搜索 Session", curses.A_DIM)
|
||
if self.message:
|
||
self._safe_add(screen, 3, 2, self.message, curses.A_DIM)
|
||
|
||
list_top = 4
|
||
list_height = max(1, height - list_top - 2)
|
||
selected_index = next(
|
||
(index for index, row in enumerate(rows) if row.key == self.selected_key),
|
||
0,
|
||
)
|
||
if selected_index < self.scroll:
|
||
self.scroll = selected_index
|
||
elif selected_index >= self.scroll + list_height:
|
||
self.scroll = selected_index - list_height + 1
|
||
self.scroll = max(0, min(self.scroll, max(0, len(rows) - list_height)))
|
||
|
||
for offset, row in enumerate(rows[self.scroll : self.scroll + list_height]):
|
||
y = list_top + offset
|
||
self._drawn_rows[y] = row
|
||
selected = row.key == self.selected_key
|
||
attribute = curses.A_REVERSE if selected else curses.A_NORMAL
|
||
if row.kind == "heading":
|
||
attribute |= curses.A_BOLD
|
||
elif not row.selectable:
|
||
attribute |= curses.A_DIM
|
||
has_row_action = (
|
||
self.view in {"main", "search"}
|
||
and row.selectable
|
||
and row.kind not in {"new-session", "add-remote"}
|
||
)
|
||
action_room = 4 if has_row_action else 0
|
||
if width < 48:
|
||
# On phones, identification matters more than repeating state
|
||
# text that is also available in the action menu.
|
||
detail_room = 0
|
||
main_room = max(8, width - 3 - action_room)
|
||
else:
|
||
detail_room = min(24, max(8, width // 3))
|
||
main_room = max(8, width - detail_room - 6 - action_room)
|
||
main = pad_terminal_text(row.main, main_room)
|
||
detail = truncate_terminal_text(row.detail, detail_room)
|
||
line = f" {main}" if detail_room == 0 else f" {main} {detail}"
|
||
self._safe_add(screen, y, 0, line, attribute)
|
||
if has_row_action:
|
||
action_label = "[⋯]"
|
||
action_width = terminal_display_width(action_label)
|
||
action_x = max(0, width - action_width - 1)
|
||
self._safe_add(screen, y, action_x, action_label, attribute)
|
||
self._row_action_buttons[y] = (
|
||
action_x,
|
||
action_x + action_width,
|
||
row,
|
||
)
|
||
|
||
footer = (
|
||
"Enter/单击 打开 · Space/[操作] 管理 · / 搜索 · Esc 关闭"
|
||
if width >= 64
|
||
else "Enter 打开 · Space 操作"
|
||
)
|
||
if self.view == "working":
|
||
footer = "远端操作进行中，请稍候"
|
||
elif self.view == "search":
|
||
footer = (
|
||
"输入筛选 · Enter 打开 · Esc 清除"
|
||
if width >= 48
|
||
else "输入筛选 · Enter 打开"
|
||
)
|
||
elif self.view != "main":
|
||
footer = (
|
||
"单击/Enter 选择 · Esc 返回"
|
||
if width >= 48
|
||
else "Enter 选择 · Esc 返回"
|
||
)
|
||
self._safe_add(screen, height - 1, 1, footer, curses.A_DIM)
|
||
if self.view == "search":
|
||
search_line = input_display_line("搜索", self.query, width - 4)
|
||
try:
|
||
screen.move(
|
||
2,
|
||
min(width - 2, 2 + terminal_display_width(search_line)),
|
||
)
|
||
except curses.error:
|
||
pass
|
||
screen.refresh()
|
||
|
||
def _draw_input(self, screen, height: int, width: int) -> None: # type: ignore[no-untyped-def]
|
||
if self.input_kind == "rename":
|
||
remaining = MAX_ALIAS_LENGTH - len(self.input_value)
|
||
self._safe_add(screen, 1, 2, "修改 Hub 显示名称", curses.A_BOLD)
|
||
self._safe_add(
|
||
screen,
|
||
2,
|
||
2,
|
||
f"底层 Session：{self.input_target}",
|
||
curses.A_DIM,
|
||
)
|
||
prompt = f"显示名称（剩余 {remaining} 字符）"
|
||
input_line = input_display_line(prompt, self.input_value, width - 3)
|
||
self._safe_add(screen, 4, 2, input_line)
|
||
self._safe_add(
|
||
screen,
|
||
5,
|
||
2,
|
||
"留空并确认可恢复默认名称；不会修改底层 Session ID",
|
||
curses.A_DIM,
|
||
)
|
||
if self.message:
|
||
self._safe_add(screen, 6, 2, self.message, curses.A_DIM)
|
||
self._safe_add(
|
||
screen,
|
||
height - 1,
|
||
1,
|
||
"Enter 保存 · Ctrl+U 清空 · Esc 返回",
|
||
curses.A_DIM,
|
||
)
|
||
try:
|
||
curses.curs_set(1)
|
||
screen.move(
|
||
4,
|
||
min(width - 2, 2 + terminal_display_width(input_line)),
|
||
)
|
||
except curses.error:
|
||
pass
|
||
screen.refresh()
|
||
return
|
||
is_add = self.input_kind == "add"
|
||
title = (
|
||
"添加 Windows/SSH 远端"
|
||
if is_add and self.input_remote_transport == "ssh"
|
||
else "添加远端"
|
||
if is_add
|
||
else f"在 {self.input_target} 新建 Session"
|
||
)
|
||
prompt = "SSH target" if is_add else "Session 名称"
|
||
self._safe_add(screen, 1, 2, title, curses.A_BOLD)
|
||
input_line = input_display_line(prompt, self.input_value, width - 3)
|
||
self._safe_add(screen, 3, 2, input_line)
|
||
if self.message:
|
||
self._safe_add(screen, 5, 2, self.message, curses.A_DIM)
|
||
self._safe_add(screen, height - 1, 1, "Enter 确认 · Esc 返回", curses.A_DIM)
|
||
try:
|
||
curses.curs_set(1)
|
||
screen.move(
|
||
3,
|
||
min(width - 2, 2 + terminal_display_width(input_line)),
|
||
)
|
||
except curses.error:
|
||
pass
|
||
screen.refresh()
|
||
|
||
@staticmethod
|
||
def _safe_add( # type: ignore[no-untyped-def]
|
||
screen,
|
||
y: int,
|
||
x: int,
|
||
text: str,
|
||
attribute: int = curses.A_NORMAL,
|
||
) -> None:
|
||
height, width = screen.getmaxyx()
|
||
if y < 0 or y >= height or x >= width:
|
||
return
|
||
try:
|
||
visible = truncate_terminal_text(text, max(0, width - x - 1))
|
||
screen.addstr(y, x, visible, attribute)
|
||
except curses.error:
|
||
pass
|