""" 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 meshbay_common.paths import long_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(long_path(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