from __future__ import annotations from dataclasses import dataclass from pathlib import Path from typing import Any from rich.text import Text from textual.app import App, ComposeResult from textual.binding import Binding from textual.widgets import DataTable, Footer, Header, Static from .installer import disable_plugin from .models import load_lockfile, load_registry from .operations import enable_disabled_plugin from .reports import installed_plugins_report from .state import load_installed_state @dataclass(frozen=True) class InstallationChoice: install_id: str install_label: str instance_name: str instance_path: Path state_root: Path class PluginHelperTui(App[int]): CSS = """ #title { padding: 0 1; text-style: bold; } #status { padding: 0 1; color: $text-muted; } """ BINDINGS = [ Binding("enter", "select", "Select", priority=True), Binding("space", "toggle_plugin", "Toggle", priority=True), Binding("d", "disable_all_plugins", "Disable all", priority=True), Binding("e", "enable_all_plugins", "Enable all", priority=True), Binding("r", "refresh", "Refresh"), Binding("b", "back", "Back"), Binding("q", "quit", "Quit"), Binding("ctrl+q", "quit", "Quit", show=False, priority=True), ] def __init__( self, *, choices: list[InstallationChoice], repo_root: Path, setup_hint: str | None = None, ) -> None: super().__init__() self.choices = choices self.repo_root = repo_root self.setup_hint = setup_hint self.mode = "installations" self.selected_installation: InstallationChoice | None = None self.plugin_rows: list[dict[str, Any]] = [] self.status_message = "" def compose(self) -> ComposeResult: yield Header(show_clock=False) yield Static("", id="title") yield DataTable(id="table") yield Static("", id="status") yield Footer() def on_mount(self) -> None: table = self.query_one(DataTable) table.cursor_type = "row" self._show_installations() def action_select(self) -> None: if self.mode != "installations": return index = self._cursor_index(len(self.choices)) if index is None: return self.selected_installation = self.choices[index] self._show_plugins() def action_back(self) -> None: if self.mode == "plugins": self._show_installations() def action_refresh(self) -> None: if self.mode == "plugins": self._show_plugins() else: self._show_installations() def action_toggle_plugin(self) -> None: if self.mode != "plugins" or self.selected_installation is None: return index = self._cursor_index(len(self.plugin_rows)) if index is None: return plugin = self.plugin_rows[index] plugin_id = plugin["id"] target = self.selected_installation try: if plugin["status"] == "enabled": result = disable_plugin( target.instance_name, target.instance_path, target.state_root, plugin_id, force=False, ) if not result["stateUpdated"]: self._set_status(f"Could not disable {plugin_id}: {self._format_skipped(result['skipped'])}") return self._set_status(f"Disabled {plugin_id}; removed {len(result['removed'])} files.") elif plugin["status"] == "disabled": result = enable_disabled_plugin( instance=target.instance_name, instance_path=target.instance_path, state_root=target.state_root, plugin_id=plugin_id, repo=self.repo_root, ) self._set_status(f"Enabled {plugin_id}; applied {len(result['applied'])} files.") else: self._set_status(f"Cannot toggle {plugin_id}: unknown status {plugin['status']}.") return except Exception as exc: self._set_status(f"Could not toggle {plugin_id}: {exc}") return self._show_plugins(preserve_status=True) def action_disable_all_plugins(self) -> None: if self.mode != "plugins" or self.selected_installation is None: return target = self.selected_installation enabled = [plugin for plugin in self.plugin_rows if plugin["status"] == "enabled"] changed = 0 errors: list[str] = [] for plugin in enabled: plugin_id = plugin["id"] try: result = disable_plugin( target.instance_name, target.instance_path, target.state_root, plugin_id, force=False, ) if result["stateUpdated"]: changed += 1 else: errors.append(f"{plugin_id}: {self._format_skipped(result['skipped'])}") except Exception as exc: errors.append(f"{plugin_id}: {exc}") self._set_bulk_status("Disabled", changed, errors) self._show_plugins(preserve_status=True) def action_enable_all_plugins(self) -> None: if self.mode != "plugins" or self.selected_installation is None: return target = self.selected_installation disabled = [plugin for plugin in self.plugin_rows if plugin["status"] == "disabled"] changed = 0 errors: list[str] = [] for plugin in disabled: plugin_id = plugin["id"] try: enable_disabled_plugin( instance=target.instance_name, instance_path=target.instance_path, state_root=target.state_root, plugin_id=plugin_id, repo=self.repo_root, ) changed += 1 except Exception as exc: errors.append(f"{plugin_id}: {exc}") self._set_bulk_status("Enabled", changed, errors) self._show_plugins(preserve_status=True) def _show_installations(self) -> None: self.mode = "installations" self.plugin_rows = [] self._set_title("Choose Beat Saber Installation") table = self.query_one(DataTable) table.clear(columns=True) table.add_columns("Install", "Version", "Instance Path", "State Dir") for choice in self.choices: table.add_row( choice.install_label, choice.instance_name, str(choice.instance_path), str(choice.state_root), ) if self.setup_hint: self._set_status(self.setup_hint) else: self._set_status("Enter selects an installation. q quits.") def _show_plugins(self, *, preserve_status: bool = False) -> None: if self.selected_installation is None: self._show_installations() return target = self.selected_installation self.mode = "plugins" self._set_title(f"{target.install_label} / {target.instance_name}") table = self.query_one(DataTable) table.clear(columns=True) table.add_columns("Status", "Name", "ID", "Version", "Files", "Asset") try: lockfile = load_lockfile(self.repo_root / "locks" / f"{target.instance_name}.lock.toml") report = installed_plugins_report( installed_state=load_installed_state(target.state_root, target.instance_name), registry=load_registry(self.repo_root / "registry" / "plugins.toml"), lockfile=lockfile, ) self.plugin_rows = report["plugins"] except Exception as exc: self.plugin_rows = [] if not preserve_status: self._set_status(f"Could not load plugins: {exc}") return for plugin in self.plugin_rows: table.add_row( self._status_marker(plugin["status"]), plugin["name"], plugin["id"], plugin["version"], str(plugin["fileCount"]), plugin["asset"], ) if not preserve_status: if self.plugin_rows: self._set_status("Space toggles selected. d disables all. e enables all. b returns to installations.") else: self._set_status("No managed plugins recorded for this installation.") def _cursor_index(self, row_count: int) -> int | None: table = self.query_one(DataTable) row = table.cursor_coordinate.row if 0 <= row < row_count: return row return None def _set_title(self, message: str) -> None: self.query_one("#title", Static).update(message) def _set_status(self, message: str) -> None: self.status_message = message self.query_one("#status", Static).update(message) def _set_bulk_status(self, verb: str, changed: int, errors: list[str]) -> None: if errors: preview = "; ".join(errors[:3]) suffix = f"; {len(errors) - 3} more" if len(errors) > 3 else "" self._set_status(f"{verb} {changed} plugins; {len(errors)} failed: {preview}{suffix}") else: self._set_status(f"{verb} {changed} plugins.") @staticmethod def _status_marker(status: str) -> Text: return Text("[x]" if status == "enabled" else "[ ]", no_wrap=True) @staticmethod def _format_skipped(skipped: list[dict[str, str]]) -> str: if not skipped: return "no files changed" return "; ".join(f"{item['path']} {item['reason']}" for item in skipped)