""" 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