Replace content CLI with Textual TUI

tools/content/tui/content_pane.py

@@ -0,0 +1,65 @@
+from typing import Any, Callable
+
+from textual import work
+from textual.containers import Vertical
+from textual.widgets import DataTable
+
+
+class ContentPane(Vertical):
+    content_file = ""
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.set_content_file(self.content_file)
+
+    def set_content_file(self, name: str):
+        self.content_file = name
+        self.border_title = f"[b]kitty[/b] ~/content/{name}"
+
+    def table(self) -> DataTable:
+        return self.query_one(DataTable)
+
+    def selected_key(self) -> str | None:
+        table = self.table()
+        if table.row_count == 0:
+            return None
+        row_key, _ = table.coordinate_to_cell_key(table.cursor_coordinate)
+        return row_key.value
+
+    def restore_cursor(self, key: str | None):
+        table = self.table()
+        if key is None or table.row_count == 0:
+            return
+        try:
+            table.move_cursor(row=table.get_row_index(key))
+        except Exception:
+            table.move_cursor(row=0)
+
+    def reload(self, focus_key: str | None = None):
+        raise NotImplementedError
+
+    def action_reload(self):
+        self.reload(self.selected_key())
+        self.notify("Reloaded from disk")
+
+    def run_now(self, task: Callable[[], Any], message: str | Callable[[Any], str], focus_key: str | None = None):
+        try:
+            result = task()
+        except (OSError, ValueError) as error:
+            self.notify(str(error), title="Error", severity="error")
+            return
+        self.reload(focus_key)
+        self.notify(message(result) if callable(message) else message)
+
+    def run_in_background(self, waiting_message: str, task: Callable[[], Any], on_success: Callable[[Any], None]):
+        self.notify(waiting_message)
+        self.background_task(task, on_success)
+
+    @work(thread=True)
+    def background_task(self, task: Callable[[], Any], on_success: Callable[[Any], None]):
+        try:
+            result = task()
+        except (OSError, ValueError) as error:
+            self.app.call_from_thread(self.notify, str(error), title="Error", severity="error")
+            return
+        self.app.call_from_thread(on_success, result)