diff options
| author | Christophe Besson <cbesson@gmail.com> | 2026-10-14 22:20:15 +0200 |
|---|---|---|
| committer | Christophe Besson <cbesson@gmail.com> | 2026-10-14 22:20:15 +0200 |
| commit | 422647e7898b8df67cd11f8fb060aec23dec84aa (patch) | |
| tree | 7ecca0e8df1bc8a7d26cf4b471bd44906dd9cfd1 /packages/meshbay-cli/src/meshbay_cli/tui/panes.py | |
| parent | 28c88908b88c58cba4de08710f6bb268ede02d57 (diff) | |
| download | meshbay-422647e7898b8df67cd11f8fb060aec23dec84aa.tar.gz | |
feat(cli): meshbay-cli, a command-line client for members
Signs in to the hub, lists the account's groups and their folders, and
downloads and uploads files directly with each group's node, end-to-end
encrypted. With no argument it opens a two-pane interactive mode (Textual).
Session and device keys are kept encrypted under the system's key storage,
or a 0600 key file where there is none; node keys are pinned at first
contact; a new machine links as a device.
Packaged on its own on Linux (deb and rpm, on meshbay-common's venv, with a
man page generated from the parser) and frozen next to meshbay-node in the
full Windows installer, not the light one, with a self-test the build runs.
client/keys.py ports keyderive.js and is LGPL like it.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Diffstat (limited to 'packages/meshbay-cli/src/meshbay_cli/tui/panes.py')
| -rw-r--r-- | packages/meshbay-cli/src/meshbay_cli/tui/panes.py | 372 |
1 files changed, 372 insertions, 0 deletions
diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/panes.py b/packages/meshbay-cli/src/meshbay_cli/tui/panes.py new file mode 100644 index 0000000..d1e3dd0 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/tui/panes.py @@ -0,0 +1,372 @@ +""" +The two panes: this machine's folders on the left, the groups on the right. + +Both draw rows the same way — folders first, then files, each sorted by the +pane's key — and both mark a selection with Space. Names are someone else's +text on the right and may be on the left: every name is escaped (`safe`) and +handed to the table as plain text, never as markup. +""" + +import asyncio +import os +import stat +from dataclasses import dataclass, field +from pathlib import Path + +from rich.text import Text +from textual.app import ComposeResult +from textual.binding import Binding +from textual.containers import Vertical +from textual.widgets import DataTable, Input, Static + +from ..client.groups import my_groups +from ..client.paths import handle +from ..client.tree import Tree +from ..errors import CliError +from ..output import date, human_size, safe + +PARENT = ".." +SORTS = ("name", "size", "date") +SIZE_WIDTH, DATE_WIDTH = 9, 16 +# Cell padding on both sides of three columns, and the vertical scrollbar. +CHROME = 2 * 3 + 2 + + +@dataclass +class Row: + key: str + name: str + is_dir: bool + size: int | None = None + when: float | None = None + note: str = "" # "down", "read-only", ... + dim: bool = False + data: dict = field(default_factory=dict) + + +class Pane(Vertical, can_focus=False): + """A titled table with marks, a sort key and a filter.""" + + BINDINGS = [ + Binding("space", "mark", "Select"), + Binding("backspace", "up", "Up"), + Binding("slash", "filter", "Filter"), + Binding("s", "sort", "Sort"), + Binding("r", "refresh", "Refresh"), + ] + + DEFAULT_CSS = """ + Pane { width: 1fr; border: round $panel; } + Pane:focus-within { border: round $accent; } + Pane > .title { height: 1; padding: 0 1; background: $panel; } + Pane > DataTable { height: 1fr; } + Pane > Input { display: none; } + Pane > Input.shown { display: block; } + """ + + def __init__(self, **kwargs): + super().__init__(**kwargs) + self.rows: list[Row] = [] + self.marks: set[str] = set() + self.sort = "name" + self.filter_text = "" + self.name_width = 30 + self.show_date = True + + def compose(self) -> ComposeResult: + yield Static("", classes="title", markup=False) + table = DataTable(cursor_type="row", zebra_stripes=False) + self._add_columns(table) + yield table + yield Input(placeholder="filter — Enter to apply, Esc to clear") + + def _add_columns(self, table: DataTable) -> None: + table.add_column("Name", key="name", width=self.name_width) + table.add_column(Text("Size", justify="right"), key="size", width=SIZE_WIDTH) + if self.show_date: + table.add_column("Date", key="date", width=DATE_WIDTH) + + def on_resize(self) -> None: + """ + The name takes what the size and the date leave, never more: a long name + with no spaces in it must not push them out of the pane. In a pane too + narrow for all three (two panes in an 80-column terminal), the date goes. + """ + self._draw_title() + inner = self.table.size.width + show_date = inner - SIZE_WIDTH - DATE_WIDTH - CHROME >= 16 + width = max(8, inner - SIZE_WIDTH - (DATE_WIDTH if show_date else 0) + - (CHROME if show_date else CHROME - 2)) + if (width, show_date) != (self.name_width, self.show_date): + self.name_width, self.show_date = width, show_date + self.table.clear(columns=True) + self._add_columns(self.table) + self.redraw() + + @property + def table(self) -> DataTable: + return self.query_one(DataTable) + + def set_title(self, text: str) -> None: + self._title = text + self._draw_title() + + def _draw_title(self) -> None: + """One line; a long path loses its beginning, which is the part known.""" + text = getattr(self, "_title", "") + room = max(8, self.size.width - 4) + if len(text) > room: + text = "…" + text[-(room - 1):] + self.query_one(".title", Static).update(Text(text, no_wrap=True)) + + # ── Drawing ────────────────────────────────────────────────────────────── + + def show(self, rows: list[Row], *, keep_cursor: str | None = None) -> None: + self.rows = rows + self.marks &= {r.key for r in rows} + self.redraw(keep_cursor=keep_cursor) + + def _ordered(self) -> list[Row]: + def k(r: Row): + if self.sort == "size": + return (-(r.size or 0), r.name.lower()) + if self.sort == "date": + return (-(r.when or 0), r.name.lower()) + return (r.name.lower(), r.name) + visible = [r for r in self.rows if r.key != PARENT + and self.filter_text.lower() in r.name.lower()] + parent = [r for r in self.rows if r.key == PARENT] + return (parent + sorted((r for r in visible if r.is_dir), key=k) + + sorted((r for r in visible if not r.is_dir), key=k)) + + def redraw(self, *, keep_cursor: str | None = None) -> None: + table = self.table + current = keep_cursor or self.cursor_key() + table.clear() + for r in self._ordered(): + mark = "* " if r.key in self.marks else " " + name = Text(mark + safe(r.name) + ("/" if r.is_dir and r.key != PARENT else ""), + style="dim" if r.dim else ("bold" if r.key in self.marks else ""), + no_wrap=True, overflow="ellipsis") + if r.note: + name.append(f" ({r.note})", style="italic dim") + size = Text("" if r.is_dir or r.size is None else human_size(r.size), + justify="right") + cells = [name, size] + if self.show_date: + cells.append(Text(date(int(r.when)) if r.when else "")) + table.add_row(*cells, key=r.key) + keys = [r.key for r in self._ordered()] + if current in keys: + table.move_cursor(row=keys.index(current)) + + def cursor_key(self) -> str | None: + table = self.table + if table.row_count == 0: + return None + try: + return table.coordinate_to_cell_key(table.cursor_coordinate).row_key.value + except Exception: + return None + + def row(self, key: str | None) -> Row | None: + return next((r for r in self.rows if r.key == key), None) + + def selection(self) -> list[Row]: + """The marked rows, or the one under the cursor.""" + if self.marks: + return [r for r in self.rows if r.key in self.marks] + r = self.row(self.cursor_key()) + return [r] if r and r.key != PARENT else [] + + # ── Actions ────────────────────────────────────────────────────────────── + + def action_mark(self) -> None: + key = self.cursor_key() + if key is None or key == PARENT: + return + self.marks ^= {key} + row = self.table.cursor_row + self.redraw(keep_cursor=key) + self.table.move_cursor(row=min(row + 1, self.table.row_count - 1)) + + def action_sort(self) -> None: + self.sort = SORTS[(SORTS.index(self.sort) + 1) % len(SORTS)] + self.app.notify(f"Sorted by {self.sort}", timeout=2) + self.redraw() + + def action_filter(self) -> None: + box = self.query_one(Input) + box.add_class("shown") + box.value = self.filter_text + box.focus() + + def on_input_changed(self, event: Input.Changed) -> None: + self.filter_text = event.value + self.redraw() + + def on_input_submitted(self, event: Input.Submitted) -> None: + self.table.focus() + + def key_escape(self) -> None: + box = self.query_one(Input) + if box.has_class("shown"): + box.value = "" + box.remove_class("shown") + self.filter_text = "" + self.redraw() + self.table.focus() + + def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: + row = self.row(event.row_key.value) + if row is None: + return + if row.key == PARENT: + self.action_up() + elif row.is_dir: + self.enter(row) + + # Subclasses + def enter(self, row: Row) -> None: ... + def action_up(self) -> None: ... + def action_refresh(self) -> None: ... + + +class LocalPane(Pane): + def __init__(self, start: Path, **kwargs): + super().__init__(**kwargs) + self.path = start.resolve() + + def on_mount(self) -> None: + self.load() + + def load(self, keep: str | None = None) -> None: + self.set_title(f"Local: {self.path}") + self.run_worker(self._load(keep), exclusive=True, group="local") + + async def _load(self, keep: str | None) -> None: + try: + rows = await asyncio.to_thread(_scan, self.path) + except OSError as e: + self.app.notify(f"{self.path}: {e.strerror}", severity="error", markup=False) + rows = [] + if self.path.parent != self.path: + rows.insert(0, Row(PARENT, PARENT, True)) + self.show(rows, keep_cursor=keep) + + def enter(self, row: Row) -> None: + self.path = self.path / row.name + self.marks.clear() + self.load() + + def action_up(self) -> None: + if self.path.parent != self.path: + came_from = self.path.name + self.path = self.path.parent + self.marks.clear() + self.load(keep=came_from) + + def action_refresh(self) -> None: + # No position captured now: the cursor may move while the folder is read, + # and the redraw keeps wherever it is when the rows arrive. + self.load() + + +def _scan(path: Path) -> list[Row]: + rows = [] + with os.scandir(path) as it: + for e in it: + try: + st = e.stat() + except OSError: + continue + is_dir = stat.S_ISDIR(st.st_mode) + if not is_dir and not stat.S_ISREG(st.st_mode): + continue + rows.append(Row(e.name, e.name, is_dir, None if is_dir else st.st_size, + st.st_mtime)) + return rows + + +class RemotePane(Pane): + """Groups, then a group's roots and folders — the Files tab, never merged.""" + + def __init__(self, screen, **kwargs): + super().__init__(**kwargs) + self.main = screen + self.groups: list[dict] = [] + self.group: dict | None = None # None: the list of groups + self.path = "" # inside the group: root/folders + + def on_mount(self) -> None: + self.load() + + @property + def location(self) -> str: + if self.group is None: + return "Your groups" + return f"{handle(self.group)}/{self.path}".rstrip("/") + + def load(self, keep: str | None = None) -> None: + self.set_title(f"{safe(self.location)} (loading…)") + self.run_worker(self._load(keep), exclusive=True, group="remote") + + async def _load(self, keep: str | None) -> None: + try: + if self.group is None: + self.groups = await my_groups(self.main.account) + rows = [Row(g["id"], handle(g), True, + note="" if g.get("node_online") else "down", + dim=not g.get("node_online"), data=g) + for g in self.groups] + else: + tree = await self.main.tree(self.group) + rows = [Row(PARENT, PARENT, True)] + _tree_rows(tree, self.path) + except CliError as e: + self.set_title(safe(self.location)) + self.app.notify(e.message, severity="error", markup=False, timeout=8) + if self.group is not None: + self.group, self.path = None, "" + self.load() + return + self.set_title(safe(self.location)) + self.show(rows, keep_cursor=keep) + + def enter(self, row: Row) -> None: + self.marks.clear() + if self.group is None: + self.group, self.path = row.data, "" + else: + self.path = row.data["path"] + self.load() + + def action_up(self) -> None: + if self.group is None: + return + self.marks.clear() + if not self.path: + came_from, self.group = self.group["id"], None + else: + came_from = self.path + self.path = self.path.rpartition("/")[0] + self.load(keep=came_from) + + def action_refresh(self) -> None: + if self.group is not None: + self.main.forget_tree(self.group) + self.load() + + +def _tree_rows(tree: Tree, path: str) -> list[Row]: + rows = [] + for item in tree.children(path): + note = "" + if item.root is not None: + note = "" if item.root.get("writable") else "read-only" + if not item.available: + note = "unavailable" + rows.append(Row(item.path, item.name, item.is_dir, + None if item.is_dir else item.size, + item.added_at or None, note=note, dim=not item.available, + data={"path": item.path, "entry": item.entry})) + return rows |