prole/status.py

#!/usr/bin/env python3
"""
Knoe Status Viewer — tk / ncurses status window.

A standalone top-level script that displays cluster service status through
a navigation sidebar of milestones and a live console output panel.

Usage:
    python -m knoe.status            # GUI mode (tk)
    python -m knoe.status --no-gui   # ncurses fallback
    python knoe/status.py            # direct invocation
"""

from __future__ import annotations

import argparse
import configparser
import curses
import os
import platform
import subprocess
import sys
import threading
import time
import webbrowser
from pathlib import Path

# ---------------------------------------------------------------------------
# Project root detection
# ---------------------------------------------------------------------------
_SELF = Path(__file__).resolve()
PROJECT_ROOT = _SELF.parent  # status.py lives at the project root

# ---------------------------------------------------------------------------
# Configuration helpers
# ---------------------------------------------------------------------------


def _load_knoe_cfg(cfg_path: Path | None = None) -> configparser.ConfigParser:
    """Load knoe.cfg (with per-environment override layering).

    Resolution order for the config file:
      1. Explicit *cfg_path* argument (from ``-c`` CLI flag).
      2. ``$KNOE_CONF/{k3d|k3s|gke}.cfg`` environment variable directory.
      3. ``<PROJECT_ROOT>/conf/{k3d|k3s|gke}.cfg`` fallback.
    """
    if cfg_path is None:
        try:
            from knoe import knoe_conf as knoe_conf_mgr

            conf_dir = knoe_conf_mgr.resolve_knoe_conf_dir(PROJECT_ROOT)
            cfg_path = knoe_conf_mgr.entrypoint_path(conf_dir)
        except Exception:
            cfg_path = PROJECT_ROOT / "conf" / "knoe.cfg"
    cp = configparser.ConfigParser(interpolation=configparser.BasicInterpolation())
    try:
        from knoe import knoe_conf as knoe_conf_mgr

        files = [
            str(p)
            for p in knoe_conf_mgr.layered_cfg_files(cfg_path)
            if p.exists()
        ]
        cp.read(files)
    except Exception:
        cp.read(str(cfg_path))
    return cp


def _cfg_get(
    cp: configparser.ConfigParser, section: str, key: str, fallback: str = ""
) -> str:
    try:
        raw = cp.get(section, key, fallback=fallback)
        # Resolve simple ${VARIABLE} references within [User] section
        while "${" in raw:
            start = raw.index("${")
            end = raw.index("}", start)
            var = raw[start + 2 : end]
            val = cp.get("User", var, fallback=var)
            raw = raw[:start] + val + raw[end + 1 :]
        return raw
    except Exception:
        return fallback


def _get_namespace(cp: configparser.ConfigParser) -> str:
    ns = _cfg_get(cp, "User", "NAMESPACE", "default")
    return ns or "default"


def _get_service_namespace(cp: configparser.ConfigParser) -> str:
    ns = _cfg_get(cp, "User", "SERVICE_NAMESPACE", "default")
    return ns or "default"


def _get_etc_dir(cp: configparser.ConfigParser) -> Path:
    svc = _cfg_get(cp, "User", "KNOE_SERVICE", "")
    if svc:
        p = Path(svc)
        if p.is_dir():
            return p
    return PROJECT_ROOT / "etc"


# ---------------------------------------------------------------------------
# Status milestones — each is a (label, command_factory) pair.
# command_factory receives (namespace, service_namespace, etc_dir) and returns
# a list[str] command suitable for subprocess.
# ---------------------------------------------------------------------------

# The canonical list of service init scripts checked by etc/status.sh.
# Order matches the deployment sequence used by knoe.sh / install.py.
_STATUS_SCRIPTS: list[str] = [
    "init_common_services.sh",
    "init_openbao.sh",
    "init_kong.sh",
    "init_postgrest.sh",
    "init_monitoring.sh",
    "init_cnpg_backup.sh",
    "init_port_forwards.sh",
]


def _service_label(script_name: str) -> str:
    """Derive a human-readable sidebar label from a script filename."""
    stem = script_name.removesuffix(".sh")
    stem = stem.removeprefix("init_")
    return stem.replace("_", " ").replace("-", " ").title()


def _build_milestones(
    namespace: str,
    svc_namespace: str,
    etc_dir: Path,
    cp: configparser.ConfigParser | None = None,
) -> list[tuple[str, list[str]]]:
    """Return an ordered list of (display_name, command) milestones.

    Navigation order:
      1. ``./knoe.sh status``
      2. Kubernetes pod overview (``kubecolor get pods -o wide -A``)
      3. ``cnpg status knoe-db``
      4. One entry per active service init script (same list as status.sh)
    """
    milestones: list[tuple[str, list[str]]] = []
    knoe_sh = PROJECT_ROOT / "knoe.sh"

    # 1. Overall knoe.sh status
    milestones.append(
        (
            "Knoe Status",
            [str(knoe_sh), "status"],
        )
    )

    # 2. Kubernetes pod overview
    milestones.append(
        (
            "Kubernetes Pods",
            ["kubecolor", "get", "pods", "-o", "wide", "-A"],
        )
    )

    # 3. CloudNative-PG cluster status
    milestones.append(
        (
            "CloudNative-PG",
            ["kubecolor", "cnpg", "status", "knoe-db", "-n", namespace],
        )
    )

    # 4. Per-service status checks — mirrors etc/status.sh STATUS_SCRIPTS
    scripts_to_check = list(_STATUS_SCRIPTS)

    # Conditional: kerberos
    if cp is not None:
        kerberos_enabled = _cfg_get(cp, "Inputs", "kerberos_config.enabled", "false")
        if kerberos_enabled.lower() in ("true", "1", "yes"):
            scripts_to_check.append("init_kerberos.sh")

    for script_name in scripts_to_check:
        script_path = etc_dir / script_name
        if script_path.is_file():
            milestones.append(
                (
                    _service_label(script_name),
                    [str(script_path), "status"],
                )
            )

    return milestones


def _script_supports_status(path: Path) -> bool:
    """Quick heuristic: check if a shell script mentions 'status' in its usage."""
    try:
        head = path.read_text(errors="replace")[:2048]
        return "status" in head.lower()
    except Exception:
        return False


# ---------------------------------------------------------------------------
# Monitoring / footer endpoints
# ---------------------------------------------------------------------------


def _build_endpoints(cp: configparser.ConfigParser) -> list[tuple[str, str]]:
    """Parse PORT_FORWARD mappings and return launchable (label, url) pairs."""
    endpoints: list[tuple[str, str]] = []
    # Collect port-forward mappings from all sections
    for section in cp.sections():
        for key, raw in cp.items(section):
            if not key.startswith("port_forward_") or "mapping" not in key.lower():
                continue
            fields = _parse_pf_mapping(raw)
            if not fields:
                continue
            desc = fields.get("description", fields.get("id", ""))
            address = fields.get("address", "127.0.0.1")
            port = fields.get("hostport", "")
            protocol = fields.get("protocol", "TCP").upper()
            if not port:
                continue
            scheme = "https" if port in ("443", "8443") else "http"
            if address in ("0.0.0.0", ""):
                address = "127.0.0.1"
            url = f"{scheme}://{address}:{port}"
            # Only include web-accessible services
            if fields.get("id", "") in (
                "grafana",
                "prometheus",
                "dashboard",
                "argocd",
                "openbao",
                "opentofu",
                "knoe-db-manager",
            ):
                endpoints.append((desc or fields.get("id", "Service"), url))
    return endpoints


def _parse_pf_mapping(raw: str) -> dict[str, str]:
    """Parse a port-forward mapping string of the form key=val;key=val;…"""
    fields: dict[str, str] = {}
    for part in raw.split(";"):
        part = part.strip()
        if "=" in part:
            k, v = part.split("=", 1)
            fields[k.strip().lower()] = v.strip()
    return fields


# ===================================================================
# TK GUI
# ===================================================================


def _run_tk(
    milestones: list[tuple[str, list[str]]],
    endpoints: list[tuple[str, str]],
    namespace: str,
):
    """Launch the tk-based status viewer."""
    import tkinter as tk

    # macOS process name
    if platform.system() == "Darwin":
        try:
            from Foundation import NSProcessInfo

            NSProcessInfo.processInfo().setProcessName_("Knoe Status")
        except Exception:
            pass

    root = tk.Tk()
    root.title("Knoe Status Viewer")

    # Center on screen
    win_w, win_h = 1300, 910
    sw = root.winfo_screenwidth()
    sh = root.winfo_screenheight()
    root.geometry(f"{win_w}x{win_h}+{(sw - win_w) // 2}+{(sh - win_h) // 2}")
    root.resizable(False, False)
    root.configure(bg="white")

    # ---- state ----
    active_milestone_idx = tk.IntVar(value=0)
    refresh_seconds = tk.IntVar(value=0)  # 0 = off
    _refresh_job: list[str | None] = [None]  # mutable container for after-id
    _proc_lock = threading.Lock()
    _current_proc: list[subprocess.Popen | None] = [None]

    # ---- layout: sidebar | divider | content ----
    main_frame = tk.Frame(root, bg="white")
    main_frame.pack(fill="both", expand=True)

    sidebar = tk.Frame(main_frame, bg="#F5F5DC", width=325)
    sidebar.pack(side="left", fill="y")
    sidebar.pack_propagate(False)

    tk.Frame(main_frame, bg="#CCCCCC", width=1).pack(side="left", fill="y")

    content = tk.Frame(main_frame, bg="white")
    content.pack(side="left", fill="both", expand=True)

    # ---- header bar (refresh selector) ----
    header = tk.Frame(content, bg="white", height=48)
    header.pack(fill="x", side="top")
    header.pack_propagate(False)

    tk.Label(
        header, text="⟳ Refresh:", bg="white", fg="#555", font=("SF Pro Text", 11)
    ).pack(side="right", padx=(0, 4), pady=8)

    refresh_options = [("Off", 0), ("5s", 5), ("10s", 10), ("30s", 30), ("60s", 60)]
    for label, secs in reversed(refresh_options):
        tk.Radiobutton(
            header,
            text=label,
            variable=refresh_seconds,
            value=secs,
            bg="white",
            fg="black",
            selectcolor="#E5E5D5",
            activebackground="white",
            highlightthickness=0,
            font=("SF Pro Text", 10),
            command=lambda: _schedule_refresh(),
        ).pack(side="right", padx=2, pady=8)

    # Custom seconds entry
    tk.Label(header, text="sec", bg="white", fg="#888", font=("SF Pro Text", 10)).pack(
        side="right", padx=(0, 2), pady=8
    )
    custom_entry = tk.Entry(
        header,
        width=4,
        font=("SF Pro Text", 10),
        relief="solid",
        bd=1,
        bg="white",
        fg="black",
        insertbackground="black",
    )
    custom_entry.pack(side="right", padx=(0, 2), pady=8)

    def _apply_custom(*_args):
        try:
            val = int(custom_entry.get().strip())
            if val > 0:
                refresh_seconds.set(val)
                _schedule_refresh()
        except ValueError:
            pass

    custom_entry.bind("<Return>", _apply_custom)

    # ---- console output ----
    console_frame = tk.Frame(content, bg="#F5F5DC", bd=1, relief="sunken")
    console_frame.pack(fill="both", expand=True, padx=12, pady=(4, 0))

    console_text = tk.Text(
        console_frame,
        bg="#F5F5DC",
        fg="#1d1d1f",
        font=("Menlo", 10) if platform.system() == "Darwin" else ("Consolas", 10),
        padx=10,
        pady=10,
        insertbackground="#1d1d1f",
        highlightthickness=0,
        bd=0,
        relief="flat",
        state="disabled",
        wrap="none",
    )
    yscroll = tk.Scrollbar(console_frame, orient="vertical", command=console_text.yview)
    xscroll = tk.Scrollbar(
        console_frame, orient="horizontal", command=console_text.xview
    )
    console_text.configure(yscrollcommand=yscroll.set, xscrollcommand=xscroll.set)
    yscroll.pack(side="right", fill="y")
    xscroll.pack(side="bottom", fill="x")
    console_text.pack(side="left", fill="both", expand=True)

    def console_write(text: str):
        def _do():
            if not console_text.winfo_exists():
                return
            console_text.configure(state="normal")
            console_text.insert("end", text)
            console_text.see("end")
            console_text.configure(state="disabled")

        root.after(0, _do)

    def console_clear():
        def _do():
            if not console_text.winfo_exists():
                return
            console_text.configure(state="normal")
            console_text.delete("1.0", "end")
            console_text.configure(state="disabled")

        root.after(0, _do)

    # ---- command execution ----
    def _run_milestone(idx: int):
        if idx < 0 or idx >= len(milestones):
            return
        label, cmd = milestones[idx]
        console_clear()
        console_write(f"▶ {label}\n")
        console_write(f"  $ {' '.join(cmd)}\n{'─' * 72}\n")

        def _exec():
            env = os.environ.copy()
            env["PROLE_NAMESPACE"] = namespace
            try:
                with _proc_lock:
                    _current_proc[0] = subprocess.Popen(
                        cmd,
                        stdout=subprocess.PIPE,
                        stderr=subprocess.STDOUT,
                        env=env,
                        text=True,
                    )
                proc = _current_proc[0]
                for line in proc.stdout:  # type: ignore[union-attr]
                    console_write(line)
                proc.wait()
                rc = proc.returncode
                console_write(f"\n{'─' * 72}\n")
                console_write(f"Exit code: {rc}\n")
            except FileNotFoundError:
                console_write(f"\n⚠ Command not found: {cmd[0]}\n")
            except Exception as exc:
                console_write(f"\n⚠ Error: {exc}\n")
            finally:
                with _proc_lock:
                    _current_proc[0] = None

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

    # ---- sidebar navigation ----
    nav_labels: list[tk.Label] = []

    tk.Label(
        sidebar,
        text="STATUS",
        bg="#F5F5DC",
        fg="#8B8B7A",
        font=("SF Pro Text", 10, "bold"),
        anchor="w",
    ).pack(fill="x", padx=20, pady=(20, 10))

    def _on_nav_click(idx: int):
        active_milestone_idx.set(idx)
        _highlight_nav(idx)
        _run_milestone(idx)

    def _highlight_nav(idx: int):
        for i, lbl in enumerate(nav_labels):
            if i == idx:
                lbl.configure(bg="#E5E5D5", font=("SF Pro Text", 11, "bold"))
            else:
                lbl.configure(bg="#F5F5DC", font=("SF Pro Text", 11))

    for i, (label, _cmd) in enumerate(milestones):
        lbl = tk.Label(
            sidebar,
            text=label,
            bg="#F5F5DC",
            fg="black",
            font=("SF Pro Text", 11),
            anchor="w",
            padx=20,
            pady=5,
            cursor="hand2",
        )
        lbl.pack(fill="x")
        lbl.bind("<Button-1>", lambda _e, idx=i: _on_nav_click(idx))
        nav_labels.append(lbl)

    # ---- refresh scheduling ----
    def _schedule_refresh():
        # Cancel any pending refresh
        if _refresh_job[0] is not None:
            root.after_cancel(_refresh_job[0])
            _refresh_job[0] = None
        secs = refresh_seconds.get()
        if secs > 0:

            def _tick():
                _run_milestone(active_milestone_idx.get())
                _refresh_job[0] = root.after(secs * 1000, _tick)

            _refresh_job[0] = root.after(secs * 1000, _tick)

    # ---- footer ----
    footer = tk.Frame(content, bg="#F5F5DC", height=64)
    footer.pack(fill="x", side="bottom")
    footer.pack_propagate(False)
    tk.Frame(footer, bg="#CCCCCC", height=1).pack(side="top", fill="x")

    for ep_label, ep_url in endpoints:
        btn = tk.Button(
            footer,
            text=f"🔗 {ep_label}",
            command=lambda u=ep_url: webbrowser.open(u),
            bg="#F5F5DC",
            fg="black",
            activebackground="#E5E5D5",
            activeforeground="black",
            highlightbackground="#F5F5DC",
            highlightthickness=0,
            relief="flat",
            font=("SF Pro Text", 10),
            padx=10,
            pady=6,
        )
        btn.pack(side="left", padx=(10, 2), pady=10)

    # Exit button on the right
    tk.Button(
        footer,
        text="Exit",
        command=root.destroy,
        bg="#F5F5DC",
        fg="black",
        activebackground="#E5E5D5",
        activeforeground="black",
        highlightbackground="#F5F5DC",
        highlightthickness=0,
        relief="flat",
        font=("SF Pro Text", 11),
        padx=16,
        pady=8,
    ).pack(side="right", padx=20, pady=10)

    # ---- initial selection ----
    if milestones:
        _highlight_nav(0)
        _run_milestone(0)

    def _on_close():
        with _proc_lock:
            p = _current_proc[0]
            if p is not None:
                try:
                    p.terminate()
                except Exception:
                    pass
        root.destroy()

    root.protocol("WM_DELETE_WINDOW", _on_close)
    root.mainloop()


# ===================================================================
# NCURSES FALLBACK
# ===================================================================


def _run_ncurses(
    milestones: list[tuple[str, list[str]]],
    endpoints: list[tuple[str, str]],
    namespace: str,
):
    """Minimal curses-based status viewer."""

    def _main(stdscr):
        curses.curs_set(0)
        curses.use_default_colors()
        curses.init_pair(1, curses.COLOR_BLACK, curses.COLOR_YELLOW)  # nav highlight
        curses.init_pair(2, curses.COLOR_WHITE, curses.COLOR_BLUE)  # header
        curses.init_pair(3, curses.COLOR_BLACK, curses.COLOR_WHITE)  # footer

        selected = 0
        output_lines: list[str] = []
        scroll_offset = 0
        running = False
        refresh_secs = 0
        last_run = 0.0

        def _execute(idx: int):
            nonlocal output_lines, running, last_run
            if idx < 0 or idx >= len(milestones):
                return
            running = True
            label, cmd = milestones[idx]
            output_lines = [f"▶ {label}", f"  $ {' '.join(cmd)}", "─" * 60]
            try:
                env = os.environ.copy()
                env["PROLE_NAMESPACE"] = namespace
                result = subprocess.run(
                    cmd,
                    capture_output=True,
                    text=True,
                    env=env,
                    timeout=120,
                )
                output_lines.extend(result.stdout.splitlines())
                if result.stderr:
                    output_lines.append("── stderr ──")
                    output_lines.extend(result.stderr.splitlines())
                output_lines.append("─" * 60)
                output_lines.append(f"Exit code: {result.returncode}")
            except FileNotFoundError:
                output_lines.append(f"⚠ Command not found: {cmd[0]}")
            except subprocess.TimeoutExpired:
                output_lines.append("⚠ Command timed out (120s)")
            except Exception as exc:
                output_lines.append(f"⚠ Error: {exc}")
            running = False
            last_run = time.time()

        # Initial run
        _execute(selected)

        while True:
            stdscr.clear()
            max_y, max_x = stdscr.getmaxyx()
            nav_width = min(30, max_x // 3)
            console_x = nav_width + 1

            # Header
            hdr = f" Knoe Status Viewer   [Refresh: {'off' if refresh_secs == 0 else f'{refresh_secs}s'}]"
            stdscr.addnstr(
                0, 0, hdr.ljust(max_x), max_x, curses.color_pair(2) | curses.A_BOLD
            )

            # Sidebar
            for i, (label, _cmd) in enumerate(milestones):
                y = i + 2
                if y >= max_y - 2:
                    break
                attr = curses.color_pair(1) | curses.A_BOLD if i == selected else 0
                text = label[: nav_width - 2]
                stdscr.addnstr(y, 1, text.ljust(nav_width - 2), nav_width - 2, attr)

            # Vertical divider
            for y in range(1, max_y - 1):
                try:
                    stdscr.addch(y, nav_width, "│")
                except curses.error:
                    pass

            # Console output
            visible_h = max_y - 3
            visible_w = max_x - console_x - 1
            for i, line in enumerate(
                output_lines[scroll_offset : scroll_offset + visible_h]
            ):
                row = i + 2
                if row >= max_y - 1:
                    break
                stdscr.addnstr(row, console_x, line, visible_w)

            # Footer
            ft = " [↑↓] Navigate  [Enter] Run  [r] Refresh interval  [q] Quit"
            ep_text = "  ".join(f"[{l}]" for l, _u in endpoints[:4])
            if ep_text:
                ft += "  │  " + ep_text
            stdscr.addnstr(max_y - 1, 0, ft.ljust(max_x), max_x, curses.color_pair(3))

            stdscr.refresh()

            # Auto-refresh
            if refresh_secs > 0 and not running:
                if time.time() - last_run >= refresh_secs:
                    _execute(selected)
                    scroll_offset = 0
                    continue

            stdscr.timeout(500)  # poll every 500ms
            ch = stdscr.getch()

            if ch == ord("q"):
                break
            elif ch == curses.KEY_UP and selected > 0:
                selected -= 1
                scroll_offset = 0
                _execute(selected)
            elif ch == curses.KEY_DOWN and selected < len(milestones) - 1:
                selected += 1
                scroll_offset = 0
                _execute(selected)
            elif ch in (curses.KEY_ENTER, 10, 13):
                scroll_offset = 0
                _execute(selected)
            elif ch == ord("r"):
                # Cycle through refresh intervals
                cycle = [0, 5, 10, 30, 60]
                try:
                    idx = cycle.index(refresh_secs)
                except ValueError:
                    idx = 0
                refresh_secs = cycle[(idx + 1) % len(cycle)]
            elif ch == curses.KEY_PPAGE:
                scroll_offset = max(0, scroll_offset - visible_h)
            elif ch == curses.KEY_NPAGE:
                scroll_offset = min(
                    max(0, len(output_lines) - visible_h), scroll_offset + visible_h
                )

    curses.wrapper(_main)


# ===================================================================
# CLI entry point
# ===================================================================


def _has_display() -> bool:
    if os.environ.get("DISPLAY"):
        return True
    if platform.system() == "Darwin":
        try:
            import tkinter as tk

            r = tk.Tk()
            r.withdraw()
            r.destroy()
            return True
        except Exception:
            return False
    return False


def main():
    parser = argparse.ArgumentParser(description="Knoe Status Viewer")
    parser.add_argument("-c", "--config", default=None, help="Path to knoe.cfg")
    parser.add_argument(
        "--no-gui", action="store_true", help="Use ncurses terminal interface"
    )
    parser.add_argument("--gui", action="store_true", help="Force GUI mode")
    args = parser.parse_args()

    cfg_path = Path(args.config) if args.config else None
    cp = _load_knoe_cfg(cfg_path)
    namespace = _get_namespace(cp)
    svc_namespace = _get_service_namespace(cp)
    etc_dir = _get_etc_dir(cp)

    milestones = _build_milestones(namespace, svc_namespace, etc_dir, cp)
    endpoints = _build_endpoints(cp)

    use_gui = args.gui or (not args.no_gui and _has_display())

    if use_gui:
        _run_tk(milestones, endpoints, namespace)
    else:
        _run_ncurses(milestones, endpoints, namespace)


if __name__ == "__main__":
    main()