rmi-backend/app/_archive/legacy_2026_07/rmi_dashboard.py

"""
RMI TUI Dashboard - Terminal User Interface
===========================================

Interactive CLI dashboard for RMI platform.
Features:
- System status display
- Menu-driven navigation
- Quick commands for common actions
- Live backend status monitoring
"""

import os
import subprocess
import time
from typing import Any

import httpx

from app.core.logging import get_logger

logger = get_logger(__name__)


# Terminal colors
class Colors:
    RED = "\033[0;31m"
    GREEN = "\033[0;32m"
    YELLOW = "\033[1;33m"
    BLUE = "\033[0;34m"
    PURPLE = "\033[0;35m"
    CYAN = "\033[0;36m"
    WHITE = "\033[0;37m"
    BOLD = "\033[1m"
    RESET = "\033[0m"


# ─── UI UTILITIES ──────────────────────────────────────────────────


def clear_screen():
    """Clear terminal screen."""
    os.system("clear" if os.name == "posix" else "cls")


def print_header(title: str):
    """Print centered header."""
    width = 60
    logger.info(f"\n{Colors.BLUE}{'=' * width}{Colors.RESET}")
    print(
        f"{Colors.BLUE}║{Colors.RESET}{Colors.BOLD}{title.center(width - 2)}{Colors.RESET}{Colors.BLUE}║{Colors.RESET}"
    )
    logger.info(f"{Colors.BLUE}{'=' * width}{Colors.RESET}\n")
def print_box(title: str, content: str, color=Colors.WHITE):
    """Print content in a box."""
    lines = content.split("\n")
    max_len = max(len(l) for l in lines) if lines else 0  # noqa: E741
    width = max_len + 4

    logger.info(f"\n{color}┌{'─' * width}┐{Colors.RESET}")
    logger.info(f"{color}│{Colors.RESET} {Colors.BOLD}{title.center(width - 2)}{Colors.RESET} {color}│{Colors.RESET}")
    logger.info(f"{color}├{'─' * width}┤{Colors.RESET}")
    for line in lines:
        logger.info(f"{color}│{Colors.RESET} {line.ljust(width - 2)} {color}│{Colors.RESET}")
    logger.info(f"{color}└{'─' * width}┘{Colors.RESET}\n")
# ─── STATUS MONITOR ────────────────────────────────────────────────


def get_backend_status(url: str = "http://127.0.0.1:8010") -> dict[str, Any]:
    """Get backend health status."""
    try:
        response = httpx.get(f"{url}/health", timeout=2)
        if response.status_code == 200:
            return response.json()
    except Exception:
        pass
    return {"error": "Backend not reachable"}


def get_redis_status() -> dict[str, Any]:
    """Get Redis connection status."""
    try:
        import redis

        r = redis.Redis(host="localhost", port=6379, db=0, socket_timeout=2)
        if r.ping():
            info = r.info()
            return {
                "connected": True,
                "version": info.get("redis_version", "unknown"),
                "memory_used": info.get("used_memory_human", "unknown"),
                "connections": info.get("connected_clients", 0),
            }
    except Exception:
        pass
    return {"error": "Redis not reachable"}


# ─── DASHBOARD UI ──────────────────────────────────────────────────


class Dashboard:
    """Interactive TUI Dashboard."""

    def __init__(self):
        self.running = True
        self.current_view = "main"

    def display_main(self):
        """Display main menu."""
        clear_screen()

        # Header
        logger.info(f"{Colors.CYAN}{Colors.BOLD}")
        logger.info("╔════════════════════════════════════════════════════════════╗")
        logger.info("║  RMI INTELLIGENCE PLATFORM           v2.0.0                ║")
        logger.info("╚════════════════════════════════════════════════════════════╝")
        logger.info(f"{Colors.RESET}")
        # Status
        backend = get_backend_status()
        redis = get_redis_status()

        backend_status = (
            f"{Colors.GREEN}● Online{Colors.RESET}"
            if backend.get("status") == "ok"
            else f"{Colors.RED}● Offline{Colors.RESET}"
        )
        redis_status = (
            f"{Colors.GREEN}● Connected{Colors.RESET}"
            if redis.get("connected")
            else f"{Colors.RED}● Disconnected{Colors.RESET}"
        )

        logger.info(f"{Colors.YELLOW}System Status:{Colors.RESET}")
        logger.info(f"  Backend: {backend_status} | Redis: {redis_status}")
        if backend.get("status") == "ok":
            logger.info(f"  Service: {backend.get('service', 'unknown')}")
            logger.info(f"  Version: {backend.get('version', 'unknown')}")
            logger.info(f"  Timestamp: {backend.get('timestamp', 'N/A')}")
        logger.info(f"\n{Colors.YELLOW}Available Ports:{Colors.RESET}")
        logger.info("  8010 - Main Backend API")
        logger.info("  6379 - Redis Server")
        logger.info("  22 - SSH (if enabled)")
        # Features
        logger.info(f"\n{Colors.YELLOW}Built-in Features:{Colors.RESET}")
        features = [
            ("[1] Server Control", "Start/stop/restart backend server"),
            ("[2] Auth System", "User authentication & OAuth management"),
            ("[3] Security Intel", "Threat intel & contract scanning"),
            ("[4] x402 Micropayments", "Micropayment enforcement & tools"),
            ("[5] Chat Interface", "Direct chat with RMI assistant"),
            ("[6] Dashboard TUI", "Interactive terminal dashboard"),
            ("[7] Network Status", "Check all service connectivity"),
            ("[0] Quit", "Exit to shell"),
        ]

        for key, (name, desc) in enumerate(features):
            logger.info(f"  {Colors.GREEN}{key}{Colors.RESET}. {Colors.CYAN}{name:<20}{Colors.RESET} - {desc}")
        logger.info(f"\n{Colors.PURPLE}Use 'help' for more information{Colors.RESET}\n")
    def display_server_control(self):
        """Display server control menu."""
        print_header("Server Control")

        pid = self._get_backend_pid()
        status = "Running" if pid else "Stopped"

        logger.info(f"  Backend Status: {Colors.GREEN if pid else Colors.RED}{status}{Colors.RESET}")
        if pid:
            logger.info(f"  PID: {pid}")
        logger.info("\n  Actions:")
        logger.info(f"    {Colors.GREEN}1{Colors.RESET}. Start Backend")
        logger.info(f"    {Colors.GREEN}2{Colors.RESET}. Stop Backend")
        logger.info(f"    {Colors.GREEN}3{Colors.RESET}. Restart Backend")
        logger.info(f"    {Colors.GREEN}0{Colors.RESET}. Back")
        choice = input("\n  Select: ").strip()

        if choice == "1":
            self._start_backend()
        elif choice == "2":
            self._stop_backend()
        elif choice == "3":
            self._restart_backend()

    def _get_backend_pid(self) -> int | None:
        """Get backend process ID."""
        try:
            result = subprocess.run(["pgrep", "-f", "uvicorn main:app"], capture_output=True, text=True, timeout=5)
            if result.stdout.strip():
                return int(result.stdout.strip())
        except Exception:
            pass
        return None

    def _start_backend(self):
        """Start backend server."""
        logger.info("\n{Colors.YELLOW}Starting backend...{Colors.RESET}")
        try:
            subprocess.Popen(
                ["python3", "-m", "uvicorn", "main:app", "--host", "127.0.0.1", "--port", "8010"],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                start_new_session=True,
            )
            time.sleep(3)
            logger.info(f"{Colors.GREEN}Backend started!{Colors.RESET}")
        except Exception as e:
            logger.warning(f"{Colors.RED}Error starting backend: {e}{Colors.RESET}")
    def _stop_backend(self):
        """Stop backend server."""
        pid = self._get_backend_pid()
        if pid:
            try:
                os.kill(pid, 15)
                logger.info(f"{Colors.GREEN}Backend stopped (PID: {pid}){Colors.RESET}")
            except Exception as e:
                logger.warning(f"{Colors.RED}Error stopping backend: {e}{Colors.RESET}")
    def _restart_backend(self):
        """Restart backend server."""
        self._stop_backend()
        time.sleep(1)
        self._start_backend()

    def display_auth_system(self):
        """Display auth system info."""
        print_header("Authentication System")

        logger.info(f"  {Colors.CYAN}OAuth Providers:{Colors.RESET}")
        logger.info("    • Email + Password")
        logger.info("    • Google")
        logger.info("    • Telegram")
        logger.info("    • GitHub")
        logger.info("    • Wallet Signature (EIP-4361)")
        logger.info(f"\n  {Colors.CYAN}Features:{Colors.RESET}")
        logger.info("    • Session management (JWT)")
        logger.info("    • Rate limiting per user")
        logger.info("    • x402 micropayment enforcement")
        logger.info(f"\n  {Colors.CYAN}Endpoints:{Colors.RESET}")
        logger.info("    POST /api/v1/auth/register   - Register new user")
        logger.info("    POST /api/v1/auth/login      - Login user")
        logger.info("    GET  /api/v1/auth/status     - Get current session")
        input(f"\n{Colors.PURPLE}Press Enter to return...{Colors.RESET}")

    def display_security_intel(self):
        """Display security intel info."""
        print_header("Security Intelligence")

        logger.info(f"  {Colors.CYAN}Modules:{Colors.RESET}")
        logger.info("    • Slither - Static analysis")
        logger.info("    • Mythril - Symbolic execution")
        logger.info("    • OFAC Sanctions - Blocklist")
        logger.info("    • Threat Intel - Risk scoring")
        logger.info(f"\n  {Colors.CYAN}Contract Scanning:{Colors.RESET}")
        logger.info("    • Reentrancy detection")
        logger.info("    • Integer overflow/underflow")
        logger.info("    • Unchecked calls")
        logger.info("    • DAO patterns")
        input(f"\n{Colors.PURPLE}Press Enter to return...{Colors.RESET}")

    def display_x402_menu(self):
        """Display x402 micropayments info."""
        print_header("x402 Micropayments")

        logger.info(f"  {Colors.CYAN}Features:{Colors.RESET}")
        logger.info("    • Micropayment enforcement")
        logger.info("    • Rate-based pricing")
        logger.info("    • Forensic analysis")
        logger.info("    • Redis-backed tracking")
        logger.info(f"\n  {Colors.CYAN}Endpoints:{Colors.RESET}")
        logger.info("    GET  /api/v1/x402/tools      - Available tools catalog")
        logger.info("    GET  /api/v1/x402/status     - Payment status")
        logger.info("    POST /api/v1/x402/enforce    - Enforce micropayment")
        input(f"\n{Colors.PURPLE}Press Enter to return...{Colors.RESET}")

    def display_chat(self):
        """Launch chat interface."""
        print_header("Chat Interface")
        logger.info("\n{Colors.YELLOW}Direct chat with RMI assistant{Colors.RESET}")
        logger.info("\n{Colors.RED}Note:{Colors.RESET} This launches the system shell")
        input(f"\n{Colors.PURPLE}Press Enter to return...{Colors.RESET}")

    def display_network_status(self):
        """Display network connectivity status."""
        print_header("Network Status")

        tests = [
            ("Local Backend", "http://127.0.0.1:8010", "/health"),
            ("Redis", "localhost", None),
            ("Ethereum RPC", "https://ethereum-rpc.publicnode.com", None),
            ("Base RPC", "https://base-rpc.publicnode.com", None),
        ]

        for name, url, path in tests:
            try:
                full_url = f"{url}{path}" if path else url
                response = httpx.get(full_url, timeout=3)
                status = f"{Colors.GREEN}✓ OK{Colors.RESET} ({response.status_code})"
            except Exception:
                status = f"{Colors.RED}✗ Failed{Colors.RESET}"

            logger.info(f"  {name:<20} {status}")
        logger.info()
        input(f"{Colors.PURPLE}Press Enter to return...{Colors.RESET}")

    def run(self):
        """Main dashboard loop."""
        while self.running:
            self.display_main()

            choice = input("Select an option: ").strip().lower()

            if choice == "0":
                self.running = False
            elif choice == "1":
                self.display_server_control()
            elif choice == "2":
                self.display_auth_system()
            elif choice == "3":
                self.display_security_intel()
            elif choice == "4":
                self.display_x402_menu()
            elif choice == "5":
                self.display_chat()
            elif choice == "6":
                logger.info("{TUI already running}")
            elif choice == "7":
                self.display_network_status()
            elif choice in ["help", "?"]:
                logger.info(f"{Colors.YELLOW}Commands:{Colors.RESET}")
                logger.info("  help  - Show this help")
                logger.info("  clear - Clear screen")
                logger.info("  quit  - Exit dashboard")
            elif choice == "clear":
                clear_screen()
            elif choice in ["quit", "exit"]:
                self.running = False
            else:
                logger.info(f"{Colors.RED}Invalid option{Colors.RESET}")
# ─── SINGLETON LAUNCHER ────────────────────────────────────────────


def launch_dashboard():
    """Launch the interactive dashboard."""
    try:
        dashboard = Dashboard()
        dashboard.run()
    except KeyboardInterrupt:
        pass
    except Exception as e:
        logger.warning(f"{Colors.RED}Dashboard error: {e}{Colors.RESET}")
# ─── CLI ENTRY POINT ───────────────────────────────────────────────

if __name__ == "__main__":
    launch_dashboard()