shell/plex/tui/app.py

#!/usr/bin/env python3
"""Plex Management TUI — a Textual-based terminal interface for Plex operations."""

from __future__ import annotations

import asyncio
import sys
from pathlib import Path
from typing import Callable, Coroutine

from textual import on, work
from textual.app import App, ComposeResult
from textual.containers import Horizontal, Vertical, VerticalScroll
from textual.screen import ModalScreen
from textual.widgets import (
    Button,
    Footer,
    Header,
    Input,
    Label,
    ListItem,
    ListView,
    Log,
    Static,
)

from backend import CommandResult, run_command  # noqa: local import
from backend import cache_sudo, is_sudo_failure  # noqa: local import

# ── Path to stylesheets ─────────────────────────────────────────────────
CSS_PATH = Path(__file__).parent / "plex_tui.tcss"


class NavListView(ListView):
    """ListView that only changes selection on click/keyboard, not mouse hover."""

    def on_mouse_move(self, event) -> None:
        """Swallow mouse-move so the cursor doesn't follow the pointer."""
        event.stop()

# ── Navigation items ────────────────────────────────────────────────────

NAV_SECTIONS: list[tuple[str, str]] = [
    ("service", "⏻  Service Control"),
    ("libraries", "📚  Library Scanner"),
    ("backup", "💾  Backup & Validate"),
    ("monitor", "📊  Monitoring"),
    ("database", "🗄️  Database Management"),
    ("recovery", "🔧  Recovery"),
    ("queries", "🔍  Queries & Stats"),
    ("testing", "🧪  Testing"),
]


# ── Confirmation dialog ─────────────────────────────────────────────────


class ConfirmDialog(ModalScreen[bool]):
    """A modal yes/no confirmation dialog."""

    def __init__(self, title: str, body: str) -> None:
        super().__init__()
        self._title = title
        self._body = body

    def compose(self) -> ComposeResult:
        with Vertical(id="dialog-container"):
            yield Label(self._title, id="dialog-title")
            yield Static(self._body, id="dialog-body")
            with Horizontal(classes="dialog-buttons"):
                yield Button("Cancel", variant="default", id="dialog-cancel")
                yield Button("Confirm", variant="error", id="dialog-confirm")

    @on(Button.Pressed, "#dialog-confirm")
    def _confirm(self) -> None:
        self.dismiss(True)

    @on(Button.Pressed, "#dialog-cancel")
    def _cancel(self) -> None:
        self.dismiss(False)


# ── Input dialog ─────────────────────────────────────────────────────────


class InputDialog(ModalScreen[str | None]):
    """Modal dialog that asks for a text value."""

    def __init__(self, title: str, body: str, placeholder: str = "") -> None:
        super().__init__()
        self._title = title
        self._body = body
        self._placeholder = placeholder

    def compose(self) -> ComposeResult:
        with Vertical(id="input-dialog-container"):
            yield Label(self._title, id="input-dialog-title")
            yield Static(self._body, id="input-dialog-body")
            yield Input(placeholder=self._placeholder, id="input-value")
            with Horizontal(classes="dialog-buttons"):
                yield Button("Cancel", variant="default", id="input-cancel")
                yield Button("OK", variant="primary", id="input-ok")

    @on(Button.Pressed, "#input-ok")
    def _ok(self) -> None:
        value = self.query_one("#input-value", Input).value.strip()
        self.dismiss(value if value else None)

    @on(Button.Pressed, "#input-cancel")
    def _cancel(self) -> None:
        self.dismiss(None)

    @on(Input.Submitted)
    def _submit(self) -> None:
        self._ok()


# ── Password dialog ──────────────────────────────────────────────────────


class PasswordDialog(ModalScreen[str | None]):
    """Modal dialog that asks for a password (masked input)."""

    def __init__(self, title: str, body: str) -> None:
        super().__init__()
        self._title = title
        self._body = body

    def compose(self) -> ComposeResult:
        with Vertical(id="password-dialog-container"):
            yield Label(self._title, id="password-dialog-title")
            yield Static(self._body, id="password-dialog-body")
            yield Input(placeholder="Password", password=True, id="password-value")
            with Horizontal(classes="dialog-buttons"):
                yield Button("Cancel", variant="default", id="password-cancel")
                yield Button("Authenticate", variant="primary", id="password-ok")

    @on(Button.Pressed, "#password-ok")
    def _ok(self) -> None:
        value = self.query_one("#password-value", Input).value
        self.dismiss(value if value else None)

    @on(Button.Pressed, "#password-cancel")
    def _cancel(self) -> None:
        self.dismiss(None)

    @on(Input.Submitted)
    def _submit(self) -> None:
        self._ok()


# ── Section panels ───────────────────────────────────────────────────────


def _section_header(text: str) -> Static:
    return Static(text, classes="section-header")


def _btn(label: str, btn_id: str, classes: str = "action-btn") -> Button:
    return Button(label, id=btn_id, classes=classes)


class ServicePanel(Vertical):
    def compose(self) -> ComposeResult:
        yield _section_header("Plex Media Server — Service Control")
        with Horizontal(classes="button-row"):
            yield _btn("▶  Start", "svc-start", "action-btn-success")
            yield _btn("⏹  Stop", "svc-stop", "action-btn-danger")
            yield _btn("🔄  Restart", "svc-restart", "action-btn-warning")
        with Horizontal(classes="button-row"):
            yield _btn("ℹ️  Status", "svc-status")


class LibraryPanel(Vertical):
    def compose(self) -> ComposeResult:
        yield _section_header("Library Scanner")
        with Horizontal(classes="button-row"):
            yield _btn("📋  List Libraries", "lib-list")
            yield _btn("🔍  Scan All", "lib-scan-all")
            yield _btn("🔍  Scan Section…", "lib-scan-id")
        with Horizontal(classes="button-row"):
            yield _btn("🔄  Refresh All", "lib-refresh-all")
            yield _btn("🔄  Refresh Section…", "lib-refresh-id")
            yield _btn("⚡  Force Refresh All", "lib-force-refresh")
        with Horizontal(classes="button-row"):
            yield _btn("📊  Analyze All", "lib-analyze-all")
            yield _btn("📊  Analyze Section…", "lib-analyze-id")


class BackupPanel(Vertical):
    def compose(self) -> ComposeResult:
        yield _section_header("Backup & Validation")
        with Horizontal(classes="button-row"):
            yield _btn("💾  Run Backup", "bak-run")
            yield _btn("💾  Backup + Auto-Repair", "bak-run-repair")
            yield _btn("🔍  Integrity Check", "bak-integrity")
        with Horizontal(classes="button-row"):
            yield _btn("📋  List Backups", "bak-list")
            yield _btn("✅  Validate Latest", "bak-validate-latest")
            yield _btn("📝  Validation Report", "bak-validate-report")
        with Horizontal(classes="button-row"):
            yield _btn("📦  Built-in Status", "bak-builtin")
            yield _btn("📦  Built-in Detailed", "bak-builtin-detail")


class MonitorPanel(Vertical):
    def compose(self) -> ComposeResult:
        yield _section_header("Backup Monitoring Dashboard")
        with Horizontal(classes="button-row"):
            yield _btn("📊  Show Dashboard", "mon-dashboard")


class DatabasePanel(Vertical):
    def compose(self) -> ComposeResult:
        yield _section_header("Database Management")
        with Horizontal(classes="button-row"):
            yield _btn("🔍  Integrity Check", "db-check")
            yield _btn("🔧  Gentle Repair", "db-repair-gentle", "action-btn-warning")
            yield _btn("⚠️  Force Repair", "db-repair-force", "action-btn-danger")
        with Horizontal(classes="button-row"):
            yield _btn("🧹  Cleanup (dry-run)", "db-cleanup-dry")
            yield _btn("🧹  Cleanup (apply)", "db-cleanup-apply", "action-btn-warning")
            yield _btn("📥  Install/Update DBRepair", "db-install-dbrepair")


class RecoveryPanel(Vertical):
    def compose(self) -> ComposeResult:
        yield _section_header("Recovery Operations")
        with Horizontal(classes="button-row"):
            yield _btn("🔎  Verify Backup Only", "rec-verify")
            yield _btn("🧪  Nuclear Dry-Run", "rec-nuclear-dry")
        with Horizontal(classes="button-row"):
            yield _btn("☢️  Nuclear Recovery", "rec-nuclear-auto", "action-btn-danger")
        yield Static(
            "  ⚠ Nuclear recovery is a last resort — it replaces your entire database from backup.",
            classes="status-warning",
        )
        yield _section_header("Post-Recovery Validation")
        with Horizontal(classes="button-row"):
            yield _btn("⚡  Quick Validate", "rec-validate-quick")
            yield _btn("🔍  Detailed Validate", "rec-validate-detailed")
            yield _btn("📈  Performance Validate", "rec-validate-perf")


class QueryPanel(Vertical):
    def compose(self) -> ComposeResult:
        yield _section_header("Queries & Statistics")
        with Horizontal(classes="button-row"):
            yield _btn("🆕  Recent Additions (7d)", "qry-recent-7")
            yield _btn("🆕  Recent Additions (30d)", "qry-recent-30")
            yield _btn("🆕  Recent Additions…", "qry-recent-custom")
        with Horizontal(classes="button-row"):
            yield _btn("📊  Library Stats", "qry-stats")
            yield _btn("🔢  Media Counts", "qry-counts")
            yield _btn("📋  List Libraries", "qry-libraries")
        with Horizontal(classes="button-row"):
            yield _btn("💬  Custom SQL Query…", "qry-custom")


class TestingPanel(Vertical):
    def compose(self) -> ComposeResult:
        yield _section_header("Testing & Diagnostics")
        with Horizontal(classes="button-row"):
            yield _btn("⚡  Quick Smoke Tests", "test-quick")
            yield _btn("🧪  Unit Tests", "test-unit")
            yield _btn("🔗  Integration Tests", "test-integration")
        with Horizontal(classes="button-row"):
            yield _btn("🧹  Cleanup Test Artifacts", "test-cleanup")


# Map section key -> panel class
PANELS: dict[str, type] = {
    "service": ServicePanel,
    "libraries": LibraryPanel,
    "backup": BackupPanel,
    "monitor": MonitorPanel,
    "database": DatabasePanel,
    "recovery": RecoveryPanel,
    "queries": QueryPanel,
    "testing": TestingPanel,
}


# ── Main application ────────────────────────────────────────────────────


class PlexTUI(App):
    """Plex Management TUI."""

    TITLE = "Plex Management Console"
    CSS_PATH = CSS_PATH
    BINDINGS = [
        ("q", "quit", "Quit"),
        ("d", "toggle_dark", "Toggle Dark"),
        ("c", "clear_log", "Clear Log"),
        ("s", "authenticate_sudo", "Sudo Auth"),
        ("1", "nav('service')", "Service"),
        ("2", "nav('libraries')", "Libraries"),
        ("3", "nav('backup')", "Backup"),
        ("4", "nav('monitor')", "Monitor"),
        ("5", "nav('database')", "Database"),
        ("6", "nav('recovery')", "Recovery"),
        ("7", "nav('queries')", "Queries"),
        ("8", "nav('testing')", "Testing"),
    ]

    def __init__(self) -> None:
        super().__init__()
        self._current_section = "service"

    # ── Composition ───────────────────────────────────────────────────

    def compose(self) -> ComposeResult:
        yield Header()
        with Horizontal():
            with Vertical(id="sidebar"):
                yield Static("Plex Manager", id="sidebar-title")
                yield NavListView(
                    *[
                        ListItem(Label(label, classes="nav-label"), id=f"nav-{key}")
                        for key, label in NAV_SECTIONS
                    ],
                    id="nav-list",
                )
            with Vertical(id="main-content"):
                # A scrollable area where the active section panel lives
                yield VerticalScroll(ServicePanel(), id="panel-area")
                yield Log(id="output-log", highlight=True, auto_scroll=True)
        yield Footer()

    def on_mount(self) -> None:
        self._select_nav(0)
        # Disable any-event mouse tracking (mode 1003) to prevent hover effects.
        # Basic click tracking (mode 1000) remains active.
        self.set_timer(0.1, self._disable_mouse_move_tracking)

    def _disable_mouse_move_tracking(self) -> None:
        sys.stdout.write("\x1b[?1003l")
        sys.stdout.flush()

    # ── Navigation ────────────────────────────────────────────────────

    def _select_nav(self, index: int) -> None:
        nav_list = self.query_one("#nav-list", ListView)
        nav_list.index = index

    @on(ListView.Selected, "#nav-list")
    def _on_nav_selected(self, event: ListView.Selected) -> None:
        item_id = event.item.id or ""
        section_key = item_id.removeprefix("nav-")
        if section_key in PANELS:
            self._switch_section(section_key)

    def _switch_section(self, key: str) -> None:
        if key == self._current_section:
            return
        self._current_section = key
        panel_area = self.query_one("#panel-area", VerticalScroll)
        panel_area.remove_children()
        panel_area.mount(PANELS[key]())

    def action_nav(self, section: str) -> None:
        keys = [k for k, _ in NAV_SECTIONS]
        if section in keys:
            self._select_nav(keys.index(section))
            self._switch_section(section)

    def action_clear_log(self) -> None:
        self.query_one("#output-log", Log).clear()

    # ── Sudo authentication ─────────────────────────────────────

    def action_authenticate_sudo(self) -> None:
        self._prompt_sudo()

    def _prompt_sudo(self) -> None:
        async def _cb(password: str | None) -> None:
            if password is not None:
                success = await cache_sudo(password)
                if success:
                    self._log("🔓 Sudo credentials cached successfully.")
                else:
                    self._log("✗ Sudo authentication failed. Wrong password?")

        self.push_screen(
            PasswordDialog(
                "🔒 Sudo Authentication",
                "Enter your password to cache sudo credentials:",
            ),
            _cb,  # type: ignore[arg-type]
        )

    # ── Logging helper ────────────────────────────────────────────────

    def _log(self, text: str) -> None:
        log_widget = self.query_one("#output-log", Log)
        log_widget.write_line(text)

    def _log_result(self, label: str, result: CommandResult) -> None:
        status = "✓" if result.ok else "✗"
        self._log(f"[{status}] {label}")
        if result.output:
            for line in result.output.splitlines():
                self._log(f"  {line}")
        self._log("")

    # ── Async operation runner ────────────────────────────────────────

    @work(thread=False)
    async def _run_op(
        self,
        label: str,
        coro: Coroutine[None, None, CommandResult],
    ) -> None:
        self._log(f"⏳ {label}...")
        result = await coro
        if is_sudo_failure(result):
            self._log(f"[✗] {label}")
            self._log("  🔒 This command requires sudo. Press 's' to authenticate, then try again.")
            self._log("")
        else:
            self._log_result(label, result)

    def _ask_section_then_run(
        self,
        title: str,
        op: Callable[[str], Coroutine[None, None, CommandResult]],
    ) -> None:
        """Prompt for a section ID, then run an async operation with it."""

        async def _do(dialog_result: str | None) -> None:
            if dialog_result is not None:
                self._run_op(f"{title} (section {dialog_result})", op(dialog_result))

        self.app.push_screen(
            InputDialog(title, "Enter library section ID:", placeholder="e.g. 1"),
            _do,  # type: ignore[arg-type]
        )

    # ── Button dispatch ───────────────────────────────────────────────

    @on(Button.Pressed)
    def _on_button(self, event: Button.Pressed) -> None:
        from backend import (  # local import to keep top-level light
            backup_builtin_detailed,
            backup_builtin_status,
            backup_integrity_check,
            backup_list,
            backup_run,
            backup_validate,
            backup_validate_report,
            custom_query,
            db_check,
            db_cleanup,
            db_install_dbrepair,
            db_repair_force,
            db_repair_gentle,
            library_analyze,
            library_list,
            library_refresh,
            library_scan,
            library_stats,
            list_libraries_query,
            media_counts,
            monitor_dashboard,
            nuclear_recovery_auto,
            nuclear_recovery_dry_run,
            nuclear_recovery_verify,
            plex_restart,
            plex_start,
            plex_status,
            plex_stop,
            recent_additions,
            run_tests_cleanup,
            run_tests_integration,
            run_tests_quick,
            run_tests_unit,
            validate_recovery,
        )

        bid = event.button.id or ""

        # ── Service ────────────────────────────────
        if bid == "svc-start":
            self._run_op("Start Plex", plex_start())
        elif bid == "svc-stop":
            self._confirm_then_run(
                "Stop Plex?",
                "This will stop the Plex Media Server service.",
                "Stop Plex",
                plex_stop(),
            )
        elif bid == "svc-restart":
            self._confirm_then_run(
                "Restart Plex?",
                "This will restart the Plex Media Server service.",
                "Restart Plex",
                plex_restart(),
            )
        elif bid == "svc-status":
            self._run_op("Plex Status", plex_status())

        # ── Libraries ──────────────────────────────
        elif bid == "lib-list":
            self._run_op("List Libraries", library_list())
        elif bid == "lib-scan-all":
            self._run_op("Scan All Libraries", library_scan())
        elif bid == "lib-scan-id":
            self._ask_section_then_run("Scan Library", library_scan)
        elif bid == "lib-refresh-all":
            self._run_op("Refresh All Libraries", library_refresh())
        elif bid == "lib-refresh-id":
            self._ask_section_then_run("Refresh Library", library_refresh)
        elif bid == "lib-force-refresh":
            self._run_op("Force Refresh All", library_refresh(force=True))
        elif bid == "lib-analyze-all":
            self._run_op("Analyze All Libraries", library_analyze())
        elif bid == "lib-analyze-id":
            self._ask_section_then_run("Analyze Library", library_analyze)

        # ── Backup ─────────────────────────────────
        elif bid == "bak-run":
            self._run_op("Run Backup", backup_run())
        elif bid == "bak-run-repair":
            self._run_op("Backup + Auto-Repair", backup_run(auto_repair=True))
        elif bid == "bak-integrity":
            self._run_op("Integrity Check", backup_integrity_check())
        elif bid == "bak-list":
            self._run_op("List Backups", backup_list())
        elif bid == "bak-validate-latest":
            self._run_op("Validate Latest Backup", backup_validate(latest_only=True))
        elif bid == "bak-validate-report":
            self._run_op("Full Validation Report", backup_validate_report())
        elif bid == "bak-builtin":
            self._run_op("Built-in Backup Status", backup_builtin_status())
        elif bid == "bak-builtin-detail":
            self._run_op("Built-in Backup (Detailed)", backup_builtin_detailed())

        # ── Monitor ────────────────────────────────
        elif bid == "mon-dashboard":
            self._run_op("Monitoring Dashboard", monitor_dashboard())

        # ── Database ───────────────────────────────
        elif bid == "db-check":
            self._run_op("Database Integrity Check", db_check())
        elif bid == "db-repair-gentle":
            self._confirm_then_run(
                "Gentle DB Repair?",
                "This will attempt a gentle repair of the Plex database.",
                "Gentle DB Repair",
                db_repair_gentle(),
            )
        elif bid == "db-repair-force":
            self._confirm_then_run(
                "Force DB Repair?",
                "This will aggressively repair the Plex database. "
                "The Plex service will be stopped during repair.",
                "Force DB Repair",
                db_repair_force(),
            )
        elif bid == "db-cleanup-dry":
            self._run_op("DB Cleanup (dry-run)", db_cleanup(dry_run=True))
        elif bid == "db-cleanup-apply":
            self._confirm_then_run(
                "Apply DB Cleanup?",
                "This will permanently remove temporary and recovery files "
                "from the Plex database directory.",
                "DB Cleanup",
                db_cleanup(dry_run=False),
            )
        elif bid == "db-install-dbrepair":
            self._run_op("Install/Update DBRepair", db_install_dbrepair())

        # ── Recovery ───────────────────────────────
        elif bid == "rec-verify":
            self._run_op("Verify Backup Integrity", nuclear_recovery_verify())
        elif bid == "rec-nuclear-dry":
            self._run_op("Nuclear Recovery (dry-run)", nuclear_recovery_dry_run())
        elif bid == "rec-nuclear-auto":
            self._confirm_then_run(
                "☢️  NUCLEAR RECOVERY",
                "This will REPLACE YOUR ENTIRE DATABASE from the best available backup.\n\n"
                "This is a LAST RESORT operation. Plex will be stopped during recovery.\n"
                "Are you absolutely sure?",
                "Nuclear Recovery",
                nuclear_recovery_auto(),
            )
        elif bid == "rec-validate-quick":
            self._run_op("Quick Recovery Validation", validate_recovery("--quick"))
        elif bid == "rec-validate-detailed":
            self._run_op(
                "Detailed Recovery Validation", validate_recovery("--detailed")
            )
        elif bid == "rec-validate-perf":
            self._run_op(
                "Performance Recovery Validation", validate_recovery("--performance")
            )

        # ── Queries ────────────────────────────────
        elif bid == "qry-recent-7":
            self._run_op("Recent Additions (7 days)", recent_additions(7))
        elif bid == "qry-recent-30":
            self._run_op("Recent Additions (30 days)", recent_additions(30))
        elif bid == "qry-recent-custom":
            self._ask_days_then_run()
        elif bid == "qry-stats":
            self._run_op("Library Stats", library_stats())
        elif bid == "qry-counts":
            self._run_op("Media Counts", media_counts())
        elif bid == "qry-libraries":
            self._run_op("List Libraries", list_libraries_query())
        elif bid == "qry-custom":
            self._ask_sql_then_run()

        # ── Testing ────────────────────────────────
        elif bid == "test-quick":
            self._run_op("Quick Smoke Tests", run_tests_quick())
        elif bid == "test-unit":
            self._run_op("Unit Tests", run_tests_unit())
        elif bid == "test-integration":
            self._run_op("Integration Tests", run_tests_integration())
        elif bid == "test-cleanup":
            self._run_op("Cleanup Test Artifacts", run_tests_cleanup())

    # ── Confirmation helper ───────────────────────────────────────────

    def _confirm_then_run(
        self,
        title: str,
        body: str,
        label: str,
        coro: Coroutine[None, None, CommandResult],
    ) -> None:
        async def _callback(confirmed: bool) -> None:
            if confirmed:
                self._run_op(label, coro)

        self.push_screen(ConfirmDialog(title, body), _callback)  # type: ignore[arg-type]

    # ── Input prompt helpers ──────────────────────────────────────────

    def _ask_days_then_run(self) -> None:
        from backend import recent_additions

        async def _cb(val: str | None) -> None:
            if val is not None and val.isdigit():
                self._run_op(
                    f"Recent Additions ({val} days)", recent_additions(int(val))
                )

        self.push_screen(
            InputDialog(
                "Recent Additions",
                "Enter number of days:",
                placeholder="e.g. 14",
            ),
            _cb,  # type: ignore[arg-type]
        )

    def _ask_sql_then_run(self) -> None:
        from backend import custom_query

        async def _cb(val: str | None) -> None:
            if val is not None:
                self._run_op(f"Custom Query", custom_query(val))

        self.push_screen(
            InputDialog(
                "Custom SQL Query",
                "Enter a SQL query to run against the Plex database:",
                placeholder="SELECT count(*) FROM metadata_items",
            ),
            _cb,  # type: ignore[arg-type]
        )


# ── Entry point ──────────────────────────────────────────────────────────

def main() -> None:
    app = PlexTUI()
    app.run()


if __name__ == "__main__":
    main()