aboutsummaryrefslogtreecommitdiffstats
path: root/packages/meshbay-cli/src/meshbay_cli/tui/panes.py
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-10-14 22:20:15 +0200
committerChristophe Besson <cbesson@gmail.com>2026-10-14 22:20:15 +0200
commit422647e7898b8df67cd11f8fb060aec23dec84aa (patch)
tree7ecca0e8df1bc8a7d26cf4b471bd44906dd9cfd1 /packages/meshbay-cli/src/meshbay_cli/tui/panes.py
parent28c88908b88c58cba4de08710f6bb268ede02d57 (diff)
downloadmeshbay-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.py372
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