path: root/packages/meshbay-cli/src/meshbay_cli/tui/panes.py

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