From 422647e7898b8df67cd11f8fb060aec23dec84aa Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Wed, 14 Oct 2026 22:20:15 +0200 Subject: 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 --- .../meshbay-cli/src/meshbay_cli/tui/__init__.py | 1 + packages/meshbay-cli/src/meshbay_cli/tui/app.py | 76 +++++ packages/meshbay-cli/src/meshbay_cli/tui/panes.py | 372 +++++++++++++++++++++ .../meshbay-cli/src/meshbay_cli/tui/screens.py | 354 ++++++++++++++++++++ .../meshbay-cli/src/meshbay_cli/tui/widgets.py | 109 ++++++ 5 files changed, 912 insertions(+) create mode 100644 packages/meshbay-cli/src/meshbay_cli/tui/__init__.py create mode 100644 packages/meshbay-cli/src/meshbay_cli/tui/app.py create mode 100644 packages/meshbay-cli/src/meshbay_cli/tui/panes.py create mode 100644 packages/meshbay-cli/src/meshbay_cli/tui/screens.py create mode 100644 packages/meshbay-cli/src/meshbay_cli/tui/widgets.py (limited to 'packages/meshbay-cli/src/meshbay_cli/tui') diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/__init__.py b/packages/meshbay-cli/src/meshbay_cli/tui/__init__.py new file mode 100644 index 0000000..becd195 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/tui/__init__.py @@ -0,0 +1 @@ +"""The interactive mode: `meshbay-cli` with no argument.""" diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/app.py b/packages/meshbay-cli/src/meshbay_cli/tui/app.py new file mode 100644 index 0000000..dbb773c --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/tui/app.py @@ -0,0 +1,76 @@ +""" +`meshbay-cli` with no argument: a full-screen terminal interface. + +Textual, because it runs on the same asyncio loop as aiortc and httpx rather +than beside them, and on Linux terminals and Windows' consoles alike. Signed in +already, it opens on the two panes; otherwise on the sign-in screen first. + +Nothing is printed while it runs — a line on stderr would tear the screen — so +the log goes to a file in the CLI's data directory instead. +""" + +import logging + +from textual.app import App + +from ..client.account import Account, no_passphrase +from ..client.hub import Hub +from ..client.store import data_dir +from ..commands.context import Context +from ..errors import CliError +from .screens import DeviceCode, MainScreen, SignInScreen + + +class MeshBayApp(App): + TITLE = "MeshBay" + + def __init__(self, ctx: Context): + super().__init__() + self.ctx = ctx + self.hub = ctx.hub + + async def use_hub(self, url: str) -> None: + if url != self.hub.url: + await self.hub.aclose() + self.hub = Hub(url) + self.ctx.hub = self.hub + self.ctx.hub_url = url + + def on_mount(self) -> None: + account = None + if self.ctx.store.peek().get("hub") == self.hub.url: + try: + account = Account.open(self.hub, None, no_passphrase, self.ctx.store) + except CliError as e: + self.notify(e.message, severity="warning", markup=False) + if account is not None and account.remembered: + self.start(account) + else: + self.push_screen(SignInScreen()) + + def start(self, account: Account) -> None: + account.on_device_code = DeviceCode(self) + if isinstance(self.screen, SignInScreen): + self.switch_screen(MainScreen(account)) + else: + self.push_screen(MainScreen(account)) + + +def _log_to_file() -> None: + root = logging.getLogger() + for h in list(root.handlers): + root.removeHandler(h) + try: + directory = data_dir() + directory.mkdir(parents=True, exist_ok=True) + handler = logging.FileHandler(directory / "interactive.log", encoding="utf-8") + except OSError: + handler = logging.NullHandler() + handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")) + root.addHandler(handler) + + +async def run_interactive(ctx: Context) -> int: + _log_to_file() + await MeshBayApp(ctx).run_async() + return 0 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 diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/screens.py b/packages/meshbay-cli/src/meshbay_cli/tui/screens.py new file mode 100644 index 0000000..9d19431 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/tui/screens.py @@ -0,0 +1,354 @@ +""" +The sign-in screen, and the main screen: two panes and the transfers. + +F5 copies the selection to the other pane's folder — a download or an upload, +with the rules `get` and `put` follow: an upload is decided whole before it +starts (names the node refuses stop it, names already there are asked about +once for the batch), nothing on either side is replaced without being asked, +and a name from a group is made writable here before it is written. +""" + +import asyncio +import contextlib +import threading +from pathlib import Path + +from rich.text import Text +from textual.app import ComposeResult +from textual.binding import Binding +from textual.containers import Horizontal, Vertical +from textual.screen import Screen +from textual.widgets import Button, Footer, Input, Static + +from ..client.account import Account +from ..client.names import UnsafeName, local_path +from ..client.paths import handle +from ..client.pool import Pool +from ..client.store import FILE_KEY_NOTICE, KEY_FILE +from ..client.transfers import PART_SUFFIX, TransferError, create_dir, download, upload +from ..client.tree import Tree +from ..client.upload_plan import plan_upload, refused_names, taken_names +from ..errors import CliError +from ..output import safe +from .panes import LocalPane, RemotePane +from .widgets import Choice, TransferStrip + +PER_GROUP = 2 + + +class SignInScreen(Screen): + DEFAULT_CSS = """ + SignInScreen { align: center middle; } + SignInScreen > Vertical { width: 64; height: auto; border: thick $accent; + padding: 1 2; background: $surface; } + SignInScreen Input { margin-bottom: 1; } + SignInScreen #error { color: $error; height: auto; } + """ + + def compose(self) -> ComposeResult: + ctx = self.app.ctx + with Vertical(): + yield Static(Text("Sign in to MeshBay")) + yield Input(value=ctx.hub_url, placeholder="hub", id="hub") + yield Input(value=ctx.username or ctx.stored_username or "", + placeholder="username", id="user") + yield Input(placeholder="passphrase", password=True, id="pass") + yield Button("Sign in", id="go", variant="primary") + yield Static("", id="error", markup=False) + + def on_mount(self) -> None: + empty = next((i for i in self.query(Input) if not i.value), None) + (empty or self.query_one("#pass", Input)).focus() + + def on_input_submitted(self, event: Input.Submitted) -> None: + self._submit() + + def on_button_pressed(self, event: Button.Pressed) -> None: + self._submit() + + def _submit(self) -> None: + hub = self.query_one("#hub", Input).value.strip().rstrip("/") + user = self.query_one("#user", Input).value.strip() + passphrase = self.query_one("#pass", Input).value + if not (hub and user and passphrase): + self.query_one("#error", Static).update("Hub, username and passphrase.") + return + self.query_one("#error", Static).update("Signing in…") + self.run_worker(self._sign_in(hub, user, passphrase), exclusive=True) + + async def _sign_in(self, hub_url: str, user: str, passphrase: str) -> None: + app = self.app + try: + await app.use_hub(hub_url) + account = Account(app.hub, user, lambda: passphrase) + await account.hub.check_version() + await account.sign_in(remember=True) + kind = await account.remember(app.ctx.store) + except CliError as e: + self.query_one("#error", Static).update(e.message) + return + if kind == KEY_FILE: + app.notify(FILE_KEY_NOTICE, timeout=10, markup=False) + app.start(account) + + +class MainScreen(Screen): + BINDINGS = [ + Binding("f5", "copy", "Copy"), + Binding("q", "app.quit", "Quit"), + Binding("tab", "app.focus_next", "Switch", show=False), + ] + + DEFAULT_CSS = """ + MainScreen > Horizontal { height: 1fr; } + """ + + def __init__(self, account: Account): + super().__init__() + self.account = account + self.pool = Pool(account) + self._trees: dict[str, Tree] = {} + self._gates: dict[str, asyncio.Semaphore] = {} + self._watching: set[str] = set() + + def compose(self) -> ComposeResult: + with Horizontal(): + yield LocalPane(Path.cwd(), id="local") + yield RemotePane(self, id="remote") + yield TransferStrip(id="transfers") + yield Footer() + + def on_mount(self) -> None: + self.query_one(LocalPane).table.focus() + + @property + def local(self) -> LocalPane: + return self.query_one(LocalPane) + + @property + def remote(self) -> RemotePane: + return self.query_one(RemotePane) + + @property + def strip(self) -> TransferStrip: + return self.query_one(TransferStrip) + + # ── Groups ─────────────────────────────────────────────────────────────── + + async def tree(self, group: dict) -> Tree: + """The group's index, asked once and then kept current by the node's pushes.""" + if group["id"] not in self._trees: + session = await self.pool.session(group) + self._trees[group["id"]] = Tree.from_index(await session.index()) + if group["id"] not in self._watching: + self._watching.add(group["id"]) + session.ch.on_push("index_delta", lambda _msg, g=group: self._changed(g)) + return self._trees[group["id"]] + + def forget_tree(self, group: dict) -> None: + self._trees.pop(group["id"], None) + + def _changed(self, group: dict) -> None: + """Something changed in the group: the index is asked for again, and the + pane redrawn if it is showing that group.""" + self.forget_tree(group) + remote = self.remote + if remote.group is not None and remote.group["id"] == group["id"]: + remote.load() + + # ── F5 ─────────────────────────────────────────────────────────────────── + + def action_copy(self) -> None: + focused = self.app.focused + if focused is not None and self.remote in focused.ancestors_with_self: + self.run_worker(self._download(), group="copy") + else: + self.run_worker(self._upload(), group="copy") + + def _gate(self, group: dict) -> asyncio.Semaphore: + return self._gates.setdefault(group["id"], asyncio.Semaphore(PER_GROUP)) + + async def _ask(self, message: str, options: list[tuple[str, str]]) -> str: + return await self.app.push_screen(Choice(message, options), wait_for_dismiss=True) + + async def _download(self) -> None: + remote, local = self.remote, self.local + rows = remote.selection() + if remote.group is None or not rows: + self.app.notify("Open a group and choose what to download.", timeout=4) + return + group, dest = remote.group, local.path + try: + tree = await self.tree(group) + except CliError as e: + self.app.notify(e.message, severity="error", markup=False) + return + plan: list[tuple[dict, Path]] = [] + dirs: list[Path] = [] + problems: list[str] = [] + for row in rows: + path = row.data["path"] + top = path.rsplit("/", 1)[-1] + if not tree.root_available(path): + problems.append(f"{top}: its folder on the node is unavailable") + continue + if not row.is_dir: + try: + plan.append((row.data["entry"], local_path(dest, [top])[0])) + except UnsafeName as e: + problems.append(str(e)) + continue + for folder, items in tree.walk(path): + rel = [top] + [p for p in folder[len(path):].strip("/").split("/") if p] + try: + dirs.append(local_path(dest, rel)[0]) + plan += [(i.entry, local_path(dest, rel + [i.name])[0]) + for i in items if not i.is_dir] + except UnsafeName as e: + problems.append(str(e)) + for p in problems: + self.app.notify(p, severity="warning", markup=False) + existing = [t for _, t in plan if t.exists() + and not t.with_name(t.name + PART_SUFFIX).exists()] + overwrite = False + if existing: + names = ", ".join(safe(t.name) for t in existing[:5]) + ( + f" and {len(existing) - 5} more" if len(existing) > 5 else "") + answer = await self._ask( + f"{len(existing)} file(s) already in {dest}: {names}.", + [("skip", "Skip them"), ("overwrite", "Overwrite"), ("cancel", "Cancel")]) + if answer == "cancel": + return + overwrite = answer == "overwrite" + if not overwrite: + plan = [(e, t) for e, t in plan if t not in existing] + for d in dirs: + await asyncio.to_thread(d.mkdir, parents=True, exist_ok=True) + remote.marks.clear() + remote.redraw() + session = await self.pool.session(group) + await asyncio.gather(*(self._one_download(group, session, e, t, overwrite) + for e, t in plan)) + local.action_refresh() + + async def _one_download(self, group, session, entry: dict, target: Path, + overwrite: bool) -> None: + job = self.strip.add("↓", entry["name"], int(entry["size"])) + async with self._gate(group): + self.strip.update_job(job, state="running") + try: + await download( + session, entry, target, overwrite=overwrite, + progress=lambda n: self.strip.update_job(job, done=job.done + n), + on_queued=lambda n: self.strip.update_job(job, state="queued", + note=f"({n} ahead)")) + self.strip.update_job(job, state="done", + note="" if target.name == entry["name"] + else f"saved as {target.name}") + except TransferError as e: + self.strip.update_job(job, state="failed", note=e.message) + + async def _upload(self) -> None: + remote, local = self.remote, self.local + rows = local.selection() + if not rows: + return + if remote.group is None or not remote.path: + self.app.notify("Open a folder of a group on the right to upload into.", + timeout=4) + return + group, folder = remote.group, remote.path + try: + tree = await self.tree(group) + except CliError as e: + self.app.notify(e.message, severity="error", markup=False) + return + root = tree.roots.get(folder.split("/", 1)[0], {}) + if not root.get("writable") or not root.get("available", True): + self.app.notify(f"{safe(handle(group))}/{safe(folder)} does not accept " + "uploads (read-only or unavailable).", severity="error", + markup=False) + return + try: + dirs, files, problems = plan_upload([str(local.path / r.name) for r in rows], + recursive=True) + except CliError as e: + self.app.notify(e.message, severity="error", markup=False) + return + problems += refused_names([f.rel[-1] for f in files], dirs) + if problems: + await self._ask("Nothing was sent:\n" + "\n".join(safe(p) for p in problems), + [("ok", "OK")]) + return + tops = sorted({d[0] for d in dirs} | {f.rel[0] for f in files}) + taken = taken_names(tree, folder, tops) + if taken: + answer = await self._ask( + f"Already in {safe(folder)}: {', '.join(safe(t) for t in sorted(taken))}. " + "The node never replaces a file.", + [("skip", "Skip them"), ("cancel", "Cancel")]) + if answer != "skip": + return + dirs = [d for d in dirs if d[0] not in taken] + files = [f for f in files if f.rel[0] not in taken] + local.marks.clear() + local.redraw() + session = await self.pool.session(group) + remote_dir = {(): folder} + try: + for d in dirs: + remote_dir[tuple(d)] = await create_dir(session, remote_dir[tuple(d[:-1])], + d[-1]) + except TransferError as e: + self.app.notify(e.message, severity="error", markup=False) + return + await asyncio.gather(*(self._one_upload(group, session, f, + remote_dir[tuple(f.rel[:-1])]) + for f in files)) + self._changed(group) + + async def _one_upload(self, group, session, item, folder: str) -> None: + name = item.rel[-1] + job = self.strip.add("↑", name, item.size) + async with self._gate(group): + self.strip.update_job(job, state="running") + lock = threading.Lock() + try: + with open(item.local, "rb") as fh: + def read(offset: int, length: int) -> bytes: + with lock: + fh.seek(offset) + return fh.read(length) + stored = await upload( + session, read, item.size, name, folder, + progress=lambda n: self.strip.update_job(job, done=job.done + n), + on_queued=lambda n: self.strip.update_job(job, state="queued", + note=f"({n} ahead)")) + self.strip.update_job(job, state="done", note="" if stored == name + else f"stored as {stored}") + except OSError as e: + self.strip.update_job(job, state="failed", note=e.strerror or str(e)) + except TransferError as e: + self.strip.update_job(job, state="failed", note=e.message) + + async def on_unmount(self) -> None: + with contextlib.suppress(Exception): + await self.pool.close() + + +class DeviceCode: + """`account.on_device_code` for the interactive mode: the code in a + notification that stays until the approval lands; the node is retried meanwhile.""" + + def __init__(self, app): + self.app = app + + def __call__(self, group: dict, pending, *, first: bool) -> bool: + if first: + self.app.notify( + f"This machine is not yet a device of yours on the node hosting " + f"{safe(handle(group))}. Approve it from the MeshBay application or a " + f"browser: the group's Settings → \"Your devices on this node\" → " + f"enter {pending.link_code}. Waiting…", + title="Approve this machine", timeout=600, markup=False) + return True diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/widgets.py b/packages/meshbay-cli/src/meshbay_cli/tui/widgets.py new file mode 100644 index 0000000..66fca4e --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/tui/widgets.py @@ -0,0 +1,109 @@ +"""A question with a few answers, and the strip that shows transfers.""" + +import time +from dataclasses import dataclass + +from rich.text import Text +from textual.app import ComposeResult +from textual.containers import Horizontal, Vertical +from textual.screen import ModalScreen +from textual.widgets import Button, Static + +from ..output import human_size, safe + + +class Choice(ModalScreen[str]): + """One question, asked once for a whole batch — never once per file.""" + + DEFAULT_CSS = """ + Choice { align: center middle; } + Choice > Vertical { width: 72; height: auto; max-height: 80%; + border: thick $accent; background: $surface; padding: 1 2; } + Choice Static { height: auto; margin-bottom: 1; } + Choice Horizontal { height: auto; align: center middle; } + Choice Button { margin: 0 1; } + """ + BINDINGS = [("escape", "dismiss('cancel')", "Cancel")] + + def __init__(self, message: str, options: list[tuple[str, str]]): + super().__init__() + self.message = message + self.options = options + + def compose(self) -> ComposeResult: + with Vertical(): + yield Static(Text(self.message)) + with Horizontal(): + for value, label in self.options: + yield Button(label, id=f"choice-{value}") + + def on_mount(self) -> None: + self.query(Button).first().focus() + + def on_button_pressed(self, event: Button.Pressed) -> None: + self.dismiss(event.button.id.removeprefix("choice-")) + + +@dataclass +class Job: + arrow: str + name: str + size: int + done: int = 0 + state: str = "waiting" # waiting, queued, running, done, failed + note: str = "" + started: float = 0.0 + + +class TransferStrip(Static): + """The running transfers, then the latest finished ones.""" + + DEFAULT_CSS = """ + TransferStrip { height: 6; border-top: solid $panel; padding: 0 1; } + """ + SHOWN = 5 + + def __init__(self, **kwargs): + super().__init__("", markup=False, **kwargs) + self.jobs: list[Job] = [] + + def on_mount(self) -> None: + self.refresh_text() + + def add(self, arrow: str, name: str, size: int) -> Job: + job = Job(arrow, name, size) + self.jobs.append(job) + self.refresh_text() + return job + + def update_job(self, job: Job, **fields) -> None: + for k, v in fields.items(): + setattr(job, k, v) + if job.state == "running" and not job.started: + job.started = time.monotonic() + self.refresh_text() + + def refresh_text(self) -> None: + active = [j for j in self.jobs if j.state in ("running", "queued", "waiting")] + ended = [j for j in self.jobs if j.state in ("done", "failed")] + lines = [self._line(j) for j in active[:self.SHOWN]] + if len(active) > self.SHOWN: + lines.append(f" … and {len(active) - self.SHOWN} more waiting") + for j in reversed(ended[-(self.SHOWN - len(lines)):] if len(lines) < self.SHOWN + else []): + lines.append(self._line(j)) + self.update(Text("\n".join(lines) or "No transfers.")) + + def _line(self, j: Job) -> str: + name = safe(j.name) + if j.state == "running": + pct = 100 * j.done // j.size if j.size else 100 + rate = j.done / max(time.monotonic() - j.started, 0.001) + return f"{j.arrow} {name} {pct}% {human_size(int(rate))}/s" + if j.state == "queued": + return f"{j.arrow} {name} waiting for a free slot on the node {j.note}" + if j.state == "waiting": + return f"{j.arrow} {name} waiting" + if j.state == "done": + return f"{j.arrow} {name} done{(' — ' + j.note) if j.note else ''}" + return f"{j.arrow} {name} FAILED: {j.note}" -- cgit v1.2.3