server-manager/gui/tabs/redis_tab.py

"""
Redis tab — interactive Redis CLI with DB selector, command history, and output console.
"""

import threading
import customtkinter as ctk
from core.redis_client import RedisClient
from core.i18n import t


class RedisTab(ctk.CTkFrame):
    def __init__(self, master, store):
        super().__init__(master, fg_color="transparent")
        self.store = store
        self._current_alias: str | None = None
        self._client: RedisClient | None = None
        self._command_history: list[str] = []
        self._history_index: int = -1

        self._build_ui()

    def _build_ui(self):
        # ── Top bar: DB selector + stats ──
        top_frame = ctk.CTkFrame(self, fg_color="transparent")
        top_frame.pack(fill="x", padx=15, pady=(15, 5))

        # DB selector
        db_label = ctk.CTkLabel(top_frame, text=t("redis_db"), anchor="w",
                                font=ctk.CTkFont(size=12, weight="bold"))
        db_label.pack(side="left", padx=(0, 5))

        self._db_var = ctk.StringVar(value="0")
        self._db_selector = ctk.CTkOptionMenu(
            top_frame, values=[str(i) for i in range(16)],
            variable=self._db_var, width=70,
            command=self._on_db_changed,
        )
        self._db_selector.pack(side="left", padx=(0, 15))

        # Keys count
        self._keys_label = ctk.CTkLabel(top_frame, text=t("redis_keys") + ": —",
                                        font=ctk.CTkFont(size=12), text_color="#9ca3af")
        self._keys_label.pack(side="left", padx=(0, 15))

        # Memory usage
        self._memory_label = ctk.CTkLabel(top_frame, text=t("redis_memory") + ": —",
                                          font=ctk.CTkFont(size=12), text_color="#9ca3af")
        self._memory_label.pack(side="left")

        # ── Command input row ──
        cmd_frame = ctk.CTkFrame(self, fg_color="transparent")
        cmd_frame.pack(fill="x", padx=15, pady=5)

        prompt_label = ctk.CTkLabel(cmd_frame, text="redis>", font=ctk.CTkFont(family="Consolas", size=13),
                                    text_color="#ef4444")
        prompt_label.pack(side="left", padx=(0, 5))

        self._cmd_entry = ctk.CTkEntry(cmd_frame, placeholder_text=t("redis_command_placeholder"),
                                       font=ctk.CTkFont(family="Consolas", size=13))
        self._cmd_entry.pack(side="left", fill="x", expand=True, padx=(0, 10))
        self._cmd_entry.bind("<Return>", lambda e: self._execute_command())
        self._cmd_entry.bind("<Up>", self._history_up)
        self._cmd_entry.bind("<Down>", self._history_down)

        # ── Buttons row ──
        btn_frame = ctk.CTkFrame(self, fg_color="transparent")
        btn_frame.pack(fill="x", padx=15, pady=5)

        self._exec_btn = ctk.CTkButton(btn_frame, text=t("redis_execute"), width=90,
                                       command=self._execute_command)
        self._exec_btn.pack(side="left", padx=(0, 5))

        self._info_btn = ctk.CTkButton(btn_frame, text="INFO", width=70,
                                       fg_color="#6b7280", hover_color="#4b5563",
                                       command=lambda: self._run_quick("INFO"))
        self._info_btn.pack(side="left", padx=(0, 5))

        self._dbsize_btn = ctk.CTkButton(btn_frame, text="DBSIZE", width=80,
                                         fg_color="#6b7280", hover_color="#4b5563",
                                         command=lambda: self._run_quick("DBSIZE"))
        self._dbsize_btn.pack(side="left", padx=(0, 5))

        self._scan_btn = ctk.CTkButton(btn_frame, text="SCAN", width=70,
                                       fg_color="#6b7280", hover_color="#4b5563",
                                       command=lambda: self._run_quick("SCAN 0 COUNT 100"))
        self._scan_btn.pack(side="left", padx=(0, 5))

        self._clear_btn = ctk.CTkButton(btn_frame, text=t("redis_clear"), width=70,
                                        fg_color="#374151", hover_color="#1f2937",
                                        command=self._clear_output)
        self._clear_btn.pack(side="right")

        # ── Output console ──
        self._output = ctk.CTkTextbox(self, font=ctk.CTkFont(family="Consolas", size=12),
                                      state="disabled")
        self._output.pack(fill="both", expand=True, padx=15, pady=(5, 5))

        # ── Status bar ──
        self._status_bar = ctk.CTkLabel(self, text=t("redis_disconnected"), anchor="w",
                                        font=ctk.CTkFont(size=11), text_color="#9ca3af")
        self._status_bar.pack(fill="x", padx=15, pady=(0, 10))

    # ── Public API ──

    def set_server(self, alias: str | None):
        """Called when user selects a server in sidebar."""
        self._current_alias = alias
        self._client = None
        self._command_history.clear()
        self._history_index = -1
        self._clear_output()

        if alias:
            self._set_status(t("redis_ready").format(alias=alias), "#22c55e")
            self._refresh_stats()
        else:
            self._set_status(t("redis_disconnected"), "#9ca3af")
            self._keys_label.configure(text=t("redis_keys") + ": —")
            self._memory_label.configure(text=t("redis_memory") + ": —")

    # ── Command execution ──

    def _execute_command(self):
        cmd = self._cmd_entry.get().strip()
        if not cmd:
            return
        if not self._current_alias:
            self._append_output(t("no_server_selected"))
            return

        # Add to history
        if not self._command_history or self._command_history[-1] != cmd:
            self._command_history.append(cmd)
        self._history_index = -1

        self._cmd_entry.delete(0, "end")
        self._append_output(f"redis> {cmd}")
        self._set_buttons_state("disabled")

        db = int(self._db_var.get())

        def _do():
            try:
                client = self._get_client()
                result = client.execute(cmd, db=db)
                formatted = self._format_result(result)
                self.after(0, lambda: self._append_output(formatted))
            except Exception as e:
                self.after(0, lambda: self._append_output(f"(error) {e}"))
            finally:
                self.after(0, lambda: self._set_buttons_state("normal"))
                self.after(0, self._refresh_stats)

        threading.Thread(target=_do, daemon=True).start()

    def _run_quick(self, cmd: str):
        """Execute a preset command."""
        self._cmd_entry.delete(0, "end")
        self._cmd_entry.insert(0, cmd)
        self._execute_command()

    def _get_client(self) -> RedisClient:
        if self._client is None:
            self._client = RedisClient(self._current_alias, self.store)
        return self._client

    # ── Stats refresh ──

    def _refresh_stats(self):
        if not self._current_alias:
            return

        def _do():
            try:
                client = self._get_client()
                db = int(self._db_var.get())
                keys_count = client.execute("DBSIZE", db=db)
                info = client.execute("INFO memory", db=db)

                # Parse memory from INFO output
                memory = "—"
                if isinstance(info, str):
                    for line in info.split("\r\n"):
                        if line.startswith("used_memory_human:"):
                            memory = line.split(":")[1].strip()
                            break

                keys_text = str(keys_count) if keys_count is not None else "—"
                self.after(0, lambda: self._keys_label.configure(
                    text=t("redis_keys") + f": {keys_text}"))
                self.after(0, lambda: self._memory_label.configure(
                    text=t("redis_memory") + f": {memory}"))
            except Exception:
                pass

        threading.Thread(target=_do, daemon=True).start()

    def _on_db_changed(self, _value: str):
        self._refresh_stats()

    # ── History navigation ──

    def _history_up(self, _event):
        if not self._command_history:
            return "break"
        if self._history_index == -1:
            self._history_index = len(self._command_history) - 1
        elif self._history_index > 0:
            self._history_index -= 1
        self._cmd_entry.delete(0, "end")
        self._cmd_entry.insert(0, self._command_history[self._history_index])
        return "break"

    def _history_down(self, _event):
        if not self._command_history:
            return "break"
        if self._history_index == -1:
            return "break"
        if self._history_index < len(self._command_history) - 1:
            self._history_index += 1
            self._cmd_entry.delete(0, "end")
            self._cmd_entry.insert(0, self._command_history[self._history_index])
        else:
            self._history_index = -1
            self._cmd_entry.delete(0, "end")
        return "break"

    # ── Output helpers ──

    def _format_result(self, result) -> str:
        """Format Redis response for display."""
        if result is None:
            return "(nil)"
        if isinstance(result, bytes):
            return result.decode("utf-8", errors="replace")
        if isinstance(result, int):
            return f"(integer) {result}"
        if isinstance(result, list):
            if not result:
                return "(empty list or set)"
            lines = []
            for i, item in enumerate(result, 1):
                val = item.decode("utf-8", errors="replace") if isinstance(item, bytes) else str(item)
                lines.append(f"{i}) \"{val}\"")
            return "\n".join(lines)
        if isinstance(result, str):
            return result
        return str(result)

    def _append_output(self, text: str):
        self._output.configure(state="normal")
        self._output.insert("end", text + "\n")
        self._output.configure(state="disabled")
        self._output.see("end")

    def _clear_output(self):
        self._output.configure(state="normal")
        self._output.delete("1.0", "end")
        self._output.configure(state="disabled")

    def _set_status(self, text: str, color: str = "#9ca3af"):
        self._status_bar.configure(text=text, text_color=color)

    def _set_buttons_state(self, state: str):
        for btn in (self._exec_btn, self._info_btn, self._dbsize_btn, self._scan_btn):
            btn.configure(state=state)