Tatsumi Crew - File Manager

Current Directory

/opt/sharedrads
Type	Name	Operations
	__pycache__ Open
	account_review Open
	autosuspend Open
	check_software_mods Open
	cms_tools Open
	domainchecker Open
	etc Open
	extras Open
	failsuspend Open
	guds_modules Open
	mailers Open
	mitigatord Open
	mysql Open
	nlp_scripts Open
	oldrads Open
	ops Open
	perl Open
	python Open
	suspended Open
	temporary Open
	README	Edit Get
	account-review	Edit Get
	alp.py	Edit Get
	autossl_runner.sh	Edit Get
	autosusprunner.sh	Edit Get
	backup_scan.sh	Edit Get
	blockip	Edit Get
	check_apache	Edit Get
	check_autossl	Edit Get
	check_bandwidth	Edit Get
	check_boxtrapper	Edit Get
	check_cpu	Edit Get
	check_crons	Edit Get
	check_darkmailer.py	Edit Get
	check_dcpumon	Edit Get
	check_dns	Edit Get
	check_domcount.sh	Edit Get
	check_exim	Edit Get
	check_hacks	Edit Get
	check_imap	Edit Get
	check_io	Edit Get
	check_lve	Edit Get
	check_mailchannels_dns	Edit Get
	check_max_children	Edit Get
	check_mem	Edit Get
	check_misc	Edit Get
	check_mysql	Edit Get
	check_pacct	Edit Get
	check_pop3	Edit Get
	check_raid	Edit Get
	check_server	Edit Get
	check_size	Edit Get
	check_software	Edit Get
	check_spamd	Edit Get
	check_traffic	Edit Get
	check_user	Edit Get
	check_zoneh	Edit Get
	clean_exim.py	Edit Get
	clean_moveuser	Edit Get
	cms_counter.py	Edit Get
	cms_creds	Edit Get
	cms_dumpdb	Edit Get
	cms_pw	Edit Get
	cmspass.py	Edit Get
	cpanel-api	Edit Get
	cpumon	Edit Get
	ctrl_alt_del	Edit Get
	dcpumon.pl	Edit Get
	disk_cleanup.py	Edit Get
	dns-sync	Edit Get
	docroot.py	Edit Get
	du-tree	Edit Get
	envinfo.py	Edit Get
	exclude_rbl.py	Edit Get
	exclude_sender	Edit Get
	extract-vhost	Edit Get
	find_warez	Edit Get
	findbadscripts	Edit Get
	fixwpcron.py	Edit Get
	forensic.py	Edit Get
	fraudhunter.py	Edit Get
	generate_cpmove_tix	Edit Get
	hostsfilemods	Edit Get
	imap_io	Edit Get
	killall911	Edit Get
	lastcommcache.sh	Edit Get
	legal_lock_down.sh	Edit Get
	lil-cpanel	Edit Get
	limit_bots	Edit Get
	listacct	Edit Get
	mail_sources.py	Edit Get
	mailscan	Edit Get
	mass_arp_fixer.py	Edit Get
	mass_mysql_recover.py	Edit Get
	megaclisas-status	Edit Get
	modify-account	Edit Get
	modsec_disable.py	Edit Get
	move_generator.py	Edit Get
	msp.pl	Edit Get
	mysql_dstat	Edit Get
	nlp	Edit Get
	packandgo	Edit Get
	pastebin	Edit Get
	postmortem	Edit Get
	procscrape	Edit Get
	quarantine	Edit Get
	quick_post	Edit Get
	radsfunctions.sh	Edit Get
	reap_fpm_orphans.sh	Edit Get
	recent-cp	Edit Get
	reclaim_suspensions	Edit Get
	remote_dump	Edit Get
	rescp.sh	Edit Get
	reset_cpanel	Edit Get
	reset_email	Edit Get
	rotate_ip_addresses.py	Edit Get
	rrdtooldisable.sh	Edit Get
	rrdtoolenable.sh	Edit Get
	sadatarunner.sh	Edit Get
	send_customer_str	Edit Get
	send_pp_email	Edit Get
	server-load	Edit Get
	setmaxemails	Edit Get
	show-conns	Edit Get
	software_report.py	Edit Get
	sqltop	Edit Get
	strmailer	Edit Get
	suspend_domain	Edit Get
	suspend_user	Edit Get
	temp_apache_fix	Edit Get
	unsuspend_user	Edit Get
	unsusprunner.sh	Edit Get
	update_spf	Edit Get
	upgrade-check	Edit Get
	vhost_data.py	Edit Get
File Transfer

Upload files to current directory
File Editor: `sqltop`

#!/usr/lib/rads/venv/bin/python3
"""sqltop - Real-time MariaDB User Statistics monitor

Displays per-second rates of MySQL user activity (queries/sec, rows/sec, etc.)
similar to Linux `top` but for database users.

Requires MariaDB userstat plugin: SET GLOBAL userstat=1;

Usage:
    sqltop                      # Interactive curses mode (top 20 users)
    sqltop -l 50                # Show top 50 users
    sqltop -l 0                 # Show all users (no limit)
    sqltop -f 'admin%'          # Filter users matching pattern
    sqltop -b -n 5              # Batch mode, 5 iterations
"""

import argparse
import curses
import getpass
import os
import sys
import time
from dataclasses import dataclass, field

import pymysql
from pymysql.cursors import DictCursor
from prettytable import PrettyTable


# Sort key mappings for rate mode
SORT_KEYS_RATE = {
    'Q': 'QUERY/s',
    'R': 'ROWS_READ/s',
    'S': 'ROWS_SENT/s',
    'I': 'B_IN/s',
    'O': 'B_OUT/s',
    'C': 'CPU/s',
}

# Sort key mappings for cumulative mode
SORT_KEYS_CUMULATIVE = {
    'Q': 'QUERY',
    'R': 'ROWS_READ',
    'S': 'ROWS_SENT',
    'I': 'B_IN',
    'O': 'B_OUT',
    'C': 'CPU',
}


@dataclass
class Config:
    """Runtime configuration"""
    host: str = 'localhost'
    user: str | None = None
    password: str | None = None
    socket: str | None = None
    interval: float = 2.0
    batch: bool = False
    iterations: int | None = None
    sort_column: str = 'QUERY/s'
    reverse_sort: bool = True
    limit: int = 20
    filter_pattern: str | None = None


@dataclass
class StatsCache:
    """Tracks previous values for rate calculation"""
    previous: dict[str, dict] = field(default_factory=dict)
    last_time: float = 0.0

    @property
    def has_baseline(self) -> bool:
        """Returns True if we have a previous sample to compute rates from"""
        return self.last_time > 0

    def reset(self) -> None:
        """Reset cache, clearing baseline data"""
        self.previous = {}
        self.last_time = 0.0

    def compute_rates(
        self, current: list[dict], now: float
    ) -> list[dict]:
        """Compute per-second rates from cumulative counters

        Args:
            current: List of row dicts from USER_STATISTICS
            now: Current timestamp

        Returns:
            List of dicts with rate columns added
        """
        elapsed = now - self.last_time if self.last_time else 0
        rates = []

        for row in current:
            user = row['USER']
            prev = self.previous.get(user, {})

            # Cumulative totals (sum of command types for QUERY)
            total_queries = (
                row['SELECT_COMMANDS']
                + row['UPDATE_COMMANDS']
                + row['OTHER_COMMANDS']
            )

            rate_row = {
                'USER': user,
                'TOTAL_CONNECTIONS': row['TOTAL_CONNECTIONS'],
                'CONCURRENT_CONNECTIONS': row['CONCURRENT_CONNECTIONS'],
                # Cumulative columns
                'QUERY': total_queries,
                'ROWS_READ': row['ROWS_READ'],
                'ROWS_SENT': row['ROWS_SENT'],
                'B_IN': row['BYTES_RECEIVED'],
                'B_OUT': row['BYTES_SENT'],
                'CPU': row['CPU_TIME'],
            }

            # Compute rate columns
            # Clamp negative values to 0 (counters may reset externally)
            if elapsed > 0:
                # CONN/s = new connections per second
                conn_delta = (
                    row['TOTAL_CONNECTIONS']
                    - prev.get('TOTAL_CONNECTIONS', 0)
                )
                rate_row['CONN/s'] = max(0, conn_delta) / elapsed

                # QUERY/s = sum of all command types
                prev_queries = (
                    prev.get('SELECT_COMMANDS', 0)
                    + prev.get('UPDATE_COMMANDS', 0)
                    + prev.get('OTHER_COMMANDS', 0)
                )
                query_delta = total_queries - prev_queries
                rate_row['QUERY/s'] = max(0, query_delta) / elapsed

                rows_read_delta = row['ROWS_READ'] - prev.get('ROWS_READ', 0)
                rate_row['ROWS_READ/s'] = max(0, rows_read_delta) / elapsed

                rows_sent_delta = row['ROWS_SENT'] - prev.get('ROWS_SENT', 0)
                rate_row['ROWS_SENT/s'] = max(0, rows_sent_delta) / elapsed

                bytes_in_delta = (
                    row['BYTES_RECEIVED'] - prev.get('BYTES_RECEIVED', 0)
                )
                rate_row['B_IN/s'] = max(0, bytes_in_delta) / elapsed

                bytes_out_delta = row['BYTES_SENT'] - prev.get('BYTES_SENT', 0)
                rate_row['B_OUT/s'] = max(0, bytes_out_delta) / elapsed

                cpu_delta = row['CPU_TIME'] - prev.get('CPU_TIME', 0)
                rate_row['CPU/s'] = max(0, cpu_delta) / elapsed
            else:
                # First iteration - no rates available
                rate_row['CONN/s'] = 0.0
                rate_row['QUERY/s'] = 0.0
                rate_row['ROWS_READ/s'] = 0.0
                rate_row['ROWS_SENT/s'] = 0.0
                rate_row['B_IN/s'] = 0.0
                rate_row['B_OUT/s'] = 0.0
                rate_row['CPU/s'] = 0.0

            # Keep cumulative values for reference
            rate_row['_raw'] = row
            rates.append(rate_row)

        # Update cache for next iteration
        self.previous = {row['USER']: row for row in current}
        self.last_time = now

        return rates


def format_count(n: float | int) -> str:
    """Format counter with K/M/B suffixes (thousand/million/billion)"""
    if n < 0:
        return '-' + format_count(-n)
    if n >= 1_000_000_000:
        return f"{n / 1_000_000_000:.1f}B"
    if n >= 1_000_000:
        return f"{n / 1_000_000:.1f}M"
    if n >= 1_000:
        return f"{n / 1_000:.1f}K"
    if isinstance(n, float):
        if n >= 100:
            return f"{n:.0f}"
        if n >= 10:
            return f"{n:.1f}"
        return f"{n:.2f}"
    return str(int(n))


def format_bytes(n: float | int) -> str:
    """Format bytes with K/M/G suffixes (kilo/mega/giga)"""
    if n < 0:
        return '-' + format_bytes(-n)
    if n >= 1_000_000_000:
        return f"{n / 1_000_000_000:.1f}G"
    if n >= 1_000_000:
        return f"{n / 1_000_000:.1f}M"
    if n >= 1_000:
        return f"{n / 1_000:.1f}K"
    if isinstance(n, float):
        if n >= 100:
            return f"{n:.0f}"
        if n >= 10:
            return f"{n:.1f}"
        return f"{n:.2f}"
    return str(int(n))


def truncate(s: str, max_len: int) -> str:
    """Truncate string with ~ if too long"""
    if len(s) <= max_len:
        return s
    return s[: max_len - 1] + '~'


def create_connection(config: Config) -> pymysql.Connection:
    """Create database connection with appropriate credentials"""
    kwargs = {
        'host': config.host,
        'database': 'INFORMATION_SCHEMA',
        'cursorclass': DictCursor,
    }

    # Try /root/.my.cnf first if no explicit credentials
    if config.user is None and config.password is None:
        kwargs['read_default_file'] = '/root/.my.cnf'
    else:
        if config.user:
            kwargs['user'] = config.user
        if config.password:
            kwargs['password'] = config.password

    if config.socket:
        kwargs['unix_socket'] = config.socket

    return pymysql.connect(**kwargs)


def check_userstat_enabled(conn: pymysql.Connection) -> bool:
    """Check if userstat plugin is enabled"""
    with conn.cursor() as cur:
        cur.execute("SHOW GLOBAL VARIABLES LIKE 'userstat'")
        row = cur.fetchone()
        if row is None:
            return False
        value = row['Value'].upper()
        return value in ('ON', '1')


def enable_userstat(conn: pymysql.Connection) -> tuple[bool, str]:
    """Attempt to enable userstat. Returns (success, error_message)."""
    try:
        with conn.cursor() as cur:
            cur.execute("SET GLOBAL userstat = 1")
        return True, ""
    except pymysql.Error as exc:
        return False, str(exc)


def flush_user_statistics(conn: pymysql.Connection) -> tuple[bool, str]:
    """Flush USER_STATISTICS counters. Returns (success, error_message)."""
    try:
        with conn.cursor() as cur:
            cur.execute("FLUSH USER_STATISTICS")
        return True, ""
    except pymysql.Error as exc:
        return False, str(exc)


def fetch_user_statistics(
    conn: pymysql.Connection, filter_pattern: str | None = None
) -> list[dict]:
    """Fetch current USER_STATISTICS, optionally filtered by user pattern"""
    with conn.cursor() as cur:
        if filter_pattern:
            cur.execute(
                "SELECT * FROM USER_STATISTICS WHERE USER LIKE %s",
                (filter_pattern,)
            )
        else:
            cur.execute("SELECT * FROM USER_STATISTICS")
        return cur.fetchall()


def fetch_questions(conn: pymysql.Connection) -> int:
    """Fetch current Questions counter from SHOW GLOBAL STATUS"""
    with conn.cursor() as cur:
        cur.execute("SHOW GLOBAL STATUS LIKE 'Questions'")
        row = cur.fetchone()
        if row:
            return int(row['Value'])
        return 0


class SqlTop:
    """Main application class"""

    def __init__(self, conn: pymysql.Connection, config: Config):
        self.conn = conn
        self.config = config
        self.cache = StatsCache()
        self.message = ""
        self.running = True
        self.show_help = False
        self.cumulative_mode = False
        self.db_error: str | None = None
        # Server-wide QPS tracking
        self.prev_questions: int = 0
        self.prev_questions_time: float = 0.0
        self.server_qps: float = 0.0

    def update_server_qps(self) -> None:
        """Update server-wide QPS from Questions counter"""
        try:
            questions = fetch_questions(self.conn)
            now = time.time()
            if self.prev_questions_time > 0:
                elapsed = now - self.prev_questions_time
                if elapsed > 0:
                    delta = questions - self.prev_questions
                    self.server_qps = max(0, delta) / elapsed
            self.prev_questions = questions
            self.prev_questions_time = now
        except pymysql.Error:
            pass  # Keep previous QPS value on error

    def fetch_and_compute(self) -> list[dict]:
        """Fetch stats and compute rates. Returns empty list on DB error."""
        try:
            self.update_server_qps()
            raw = fetch_user_statistics(self.conn, self.config.filter_pattern)
            self.db_error = None
            now = time.time()
            return self.cache.compute_rates(raw, now)
        except pymysql.Error as exc:
            self.db_error = f"Database error: {exc}"
            return []

    def sort_rows(self, rows: list[dict]) -> list[dict]:
        """Sort rows by current sort column"""
        col = self.config.sort_column

        def sort_key(row):
            if col in row:
                return row[col]
            # Check cumulative columns
            raw = row.get('_raw', {})
            if col in raw:
                return raw[col]
            return 0

        return sorted(rows, key=sort_key, reverse=self.config.reverse_sort)

    # -------------------------------------------------------------------------
    # Batch mode (non-interactive)
    # -------------------------------------------------------------------------

    def run_batch(self) -> None:
        """Run in batch mode (no curses)"""
        iterations = self.config.iterations or 1

        # Collect baseline sample first
        print("Collecting baseline...", flush=True)
        self.fetch_and_compute()  # Establishes baseline, rates will be 0
        time.sleep(self.config.interval)

        # Now show real rate data
        for iteration in range(iterations):
            rows = self.fetch_and_compute()
            rows = self.sort_rows(rows)

            self.print_batch_output(rows, iteration)

            if iteration < iterations - 1:
                time.sleep(self.config.interval)

    def print_batch_output(self, rows: list[dict], iteration: int) -> None:
        """Print one iteration of batch output"""
        sort_dir = "desc" if self.config.reverse_sort else "asc"
        limit_info = f" (top {self.config.limit})" if self.config.limit else ""

        # Get load averages
        try:
            load1, load5, load15 = os.getloadavg()
            load_str = f"Load: {load1:.2f}, {load5:.2f}, {load15:.2f}"
        except OSError:
            load_str = "Load: N/A"

        header = (
            f"sqltop - iteration {iteration + 1}{limit_info} | "
            f"Sort: {self.config.sort_column} {sort_dir}"
        )
        stats_line = f"QPS: {format_count(self.server_qps)} | {load_str}"
        print(header)
        print(stats_line)
        print()

        table = PrettyTable()

        # Apply limit
        display_rows = rows[:self.config.limit] if self.config.limit else rows

        table.field_names = [
            'USER', 'QUERY/s', 'ROWS_READ/s',
            'ROWS_SENT/s', 'B_IN/s', 'B_OUT/s', 'CPU/s'
        ]
        table.align = 'r'
        table.align['USER'] = 'l'

        for row in display_rows:
            table.add_row([
                truncate(row['USER'], 30),
                format_count(row['QUERY/s']),
                format_count(row['ROWS_READ/s']),
                format_count(row['ROWS_SENT/s']),
                format_bytes(row['B_IN/s']),
                format_bytes(row['B_OUT/s']),
                format_count(row['CPU/s']),
            ])

        print(table)
        print()

    # -------------------------------------------------------------------------
    # Curses TUI (interactive)
    # -------------------------------------------------------------------------

    def run_interactive(self, stdscr) -> None:
        """Run interactive curses mode"""
        stdscr.nodelay(True)
        curses.curs_set(0)

        # Initialize colors if available
        if curses.has_colors():
            curses.start_color()
            curses.use_default_colors()

        # Collect baseline before entering main loop
        self._draw_baseline_message(stdscr)
        self.fetch_and_compute()
        time.sleep(self.config.interval)

        while self.running:
            try:
                height, width = stdscr.getmaxyx()
                stdscr.clear()

                if self.show_help:
                    self.draw_help(stdscr, height, width)
                else:
                    rows = self.fetch_and_compute()
                    rows = self.sort_rows(rows)

                    row_num = self.draw_header(stdscr, width)

                    # Show DB error if present
                    if self.db_error:
                        try:
                            stdscr.addstr(
                                row_num, 0,
                                self.db_error[:width - 1],
                                curses.A_BOLD
                            )
                        except curses.error:
                            pass
                        row_num += 1

                    row_num = self.draw_table(
                        stdscr, rows, row_num, height - 2, width
                    )
                    self.draw_footer(stdscr, height - 1, width)

                    if self.message:
                        try:
                            stdscr.addstr(
                                height - 2, 0,
                                self.message[:width - 1],
                                curses.A_BOLD | curses.A_REVERSE
                            )
                        except curses.error:
                            pass
                        self.message = ""

                stdscr.refresh()

                # Poll for input with short sleeps to stay responsive
                # while still honoring the refresh interval
                sleep_remaining = self.config.interval
                while sleep_remaining > 0 and self.running:
                    key = stdscr.getch()
                    if key != -1:
                        self.handle_key(key)
                        break  # Redraw immediately after keypress
                    time.sleep(0.1)
                    sleep_remaining -= 0.1

            except curses.error:
                pass  # Terminal resize or other display issue

    def draw_header(self, stdscr, width: int) -> int:
        """Draw header area, return next row number"""
        sort_dir = "desc" if self.config.reverse_sort else "asc"
        limit_info = f" (top {self.config.limit})" if self.config.limit else ""
        mode_str = "cumulative" if self.cumulative_mode else "rate"

        # Get load averages
        try:
            load1, load5, load15 = os.getloadavg()
            load_str = f"Load: {load1:.2f}, {load5:.2f}, {load15:.2f}"
        except OSError:
            load_str = "Load: N/A"

        line1 = f"sqltop - MariaDB User Statistics{limit_info}"
        line2 = (
            f"QPS: {format_count(self.server_qps)} | "
            f"{load_str} | "
            f"Refresh: {self.config.interval}s"
        )
        line3 = (
            f"Mode: {mode_str} | "
            f"Sort: {self.config.sort_column} {sort_dir}"
        )

        try:
            stdscr.addstr(0, 0, line1[:width - 1], curses.A_BOLD)
            stdscr.addstr(1, 0, line2[:width - 1])
            stdscr.addstr(2, 0, line3[:width - 1])
            stdscr.addstr(3, 0, '-' * (width - 1))
        except curses.error:
            pass

        return 4

    def draw_table(
        self, stdscr, rows: list[dict], start_row: int,
        max_rows: int, width: int
    ) -> int:
        """Draw statistics table"""
        # Formatters: None for USER (string), format_count for counters,
        # format_bytes for byte columns
        formatters = [None, format_count, format_count, format_count,
                      format_bytes, format_bytes, format_count]

        if self.cumulative_mode:
            headers = ['USER', 'QUERY', 'ROWS_READ',
                       'ROWS_SENT', 'B_IN', 'B_OUT', 'CPU']
            col_widths = [32, 12, 12, 12, 10, 10, 10]
            data_keys = ['USER', 'QUERY', 'ROWS_READ',
                         'ROWS_SENT', 'B_IN', 'B_OUT', 'CPU']
        else:
            headers = ['USER', 'QUERY/s', 'ROWS_READ/s',
                       'ROWS_SENT/s', 'B_IN/s', 'B_OUT/s', 'CPU/s']
            col_widths = [32, 12, 12, 12, 10, 10, 10]
            data_keys = ['USER', 'QUERY/s', 'ROWS_READ/s',
                         'ROWS_SENT/s', 'B_IN/s', 'B_OUT/s', 'CPU/s']

        # Draw header row
        header_line = ""
        for i, (h, w) in enumerate(zip(headers, col_widths)):
            if i == 0:
                header_line += h.ljust(w)
            else:
                header_line += h.rjust(w)
        try:
            stdscr.addstr(
                start_row, 0, header_line[:width - 1], curses.A_REVERSE
            )
        except curses.error:
            pass

        # Draw data rows
        row_num = start_row + 1
        available_rows = max_rows - start_row - 2  # Leave room for footer

        # Apply limit (use smaller of limit and available screen rows)
        max_display = min(
            self.config.limit or len(rows),
            available_rows
        )

        for row in rows[:max_display]:
            if row_num >= max_rows - 1:
                break

            # Build line using col_widths for consistent formatting
            line = ""
            for i, (key, w, fmt) in enumerate(zip(data_keys, col_widths, formatters)):
                if i == 0:
                    # USER column: left-aligned, truncated
                    line += truncate(row[key], w - 2).ljust(w)
                else:
                    # Data columns: right-aligned, formatted
                    line += fmt(row[key]).rjust(w)

            try:
                stdscr.addstr(row_num, 0, line[:width - 1])
            except curses.error:
                pass
            row_num += 1

        return row_num

    def draw_footer(self, stdscr, row: int, width: int) -> None:
        """Draw footer with key hints"""
        footer = "q:quit  m:mode  f:flush  ?:help  Q/R/S/I/O/C:sort"
        try:
            stdscr.addstr(row, 0, footer[:width - 1], curses.A_DIM)
        except curses.error:
            pass

    def _draw_baseline_message(self, stdscr) -> None:
        """Draw baseline collection message during startup"""
        try:
            _, width = stdscr.getmaxyx()
            stdscr.clear()
            self.draw_header(stdscr, width)
            stdscr.addstr(5, 0, "Collecting baseline...", curses.A_BOLD)
            stdscr.refresh()
        except curses.error:
            pass

    def draw_help(self, stdscr, height: int, width: int) -> None:
        """Draw help screen"""
        help_lines = [
            "sqltop - MariaDB User Statistics Monitor",
            "",
            "Key Bindings:",
            "  q       Quit",
            "  m       Toggle rate/cumulative mode",
            "  f       Flush statistics (resets counters)",
            "  ?       Toggle this help screen",
            "",
            "Sort Keys:",
            "  Q       Sort by QUERY (or QUERY/s in rate mode)",
            "  R       Sort by ROWS_READ (or ROWS_READ/s)",
            "  S       Sort by ROWS_SENT (or ROWS_SENT/s)",
            "  I       Sort by B_IN (bytes received)",
            "  O       Sort by B_OUT (bytes sent)",
            "  C       Sort by CPU (or CPU/s)",
            "",
            "Press any key to close help",
        ]

        try:
            stdscr.addstr(0, 0, "Help", curses.A_BOLD | curses.A_REVERSE)
            for i, line in enumerate(help_lines):
                if i + 2 >= height:
                    break
                stdscr.addstr(i + 2, 0, line[:width - 1])
        except curses.error:
            pass

    def handle_key(self, key: int) -> None:
        """Process keyboard input"""
        ch = chr(key) if 0 <= key < 256 else ''

        if ch == 'q':
            self.running = False
        elif ch == '?':
            self.show_help = not self.show_help
        elif self.show_help:
            # Any key closes help
            self.show_help = False
        elif ch == 'm':
            self.cumulative_mode = not self.cumulative_mode
            # Update sort column to match new mode
            sort_keys = (
                SORT_KEYS_CUMULATIVE if self.cumulative_mode
                else SORT_KEYS_RATE
            )
            # Find current sort key and switch to equivalent in new mode
            for key_char, col in (
                SORT_KEYS_RATE if self.cumulative_mode
                else SORT_KEYS_CUMULATIVE
            ).items():
                if col == self.config.sort_column:
                    self.config.sort_column = sort_keys[key_char]
                    break
            mode_str = "cumulative" if self.cumulative_mode else "rate"
            self.message = f"Switched to {mode_str} mode"
        elif ch == 'f':
            success, error = flush_user_statistics(self.conn)
            if success:
                self.cache.reset()
                self.message = "Statistics flushed"
            else:
                self.message = f"Flush failed: {error}"
        else:
            sort_keys = (
                SORT_KEYS_CUMULATIVE if self.cumulative_mode
                else SORT_KEYS_RATE
            )
            if ch in sort_keys:
                self.config.sort_column = sort_keys[ch]
                self.message = f"Sort by {self.config.sort_column}"


def parse_args() -> argparse.Namespace:
    """Parse command line arguments"""
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        '-b', '--batch', action='store_true',
        help='Batch mode: no interactive input, output to stdout'
    )
    parser.add_argument(
        '-n', '--iterations', type=int, default=None,
        help='Number of iterations before exiting (default: unlimited)'
    )
    parser.add_argument(
        '-i', '--interval', type=float, default=2.0,
        help='Refresh interval in seconds (default: 2)'
    )
    parser.add_argument(
        '-H', '--host', default='localhost',
        help='Database host (default: localhost)'
    )
    parser.add_argument(
        '-u', '--user',
        help='Database user (default: use /root/.my.cnf)'
    )
    parser.add_argument(
        '-p', '--password', action='store_true',
        help='Prompt for password'
    )
    parser.add_argument(
        '-S', '--socket',
        help='Unix socket path'
    )
    parser.add_argument(
        '-s', '--sort', default='QUERY/s',
        choices=['QUERY/s', 'ROWS_READ/s', 'ROWS_SENT/s', 'B_IN/s', 'B_OUT/s', 'CPU/s'],
        help='Sort column (default: QUERY/s)'
    )
    parser.add_argument(
        '-l', '--limit', type=int, default=20,
        help='Limit output to top N users (default: 20, 0 for unlimited)'
    )
    parser.add_argument(
        '-f', '--filter',
        help='Filter users by SQL LIKE pattern (e.g., "admin%%", "%%_wp")'
    )
    return parser.parse_args()


def main() -> int:
    """Main entry point"""
    args = parse_args()

    config = Config(
        host=args.host,
        user=args.user,
        socket=args.socket,
        interval=args.interval,
        batch=args.batch,
        iterations=args.iterations,
        sort_column=args.sort,
        limit=args.limit if args.limit > 0 else None,
        filter_pattern=args.filter,
    )

    if args.password:
        config.password = getpass.getpass('Password: ')

    # Connect to database
    try:
        conn = create_connection(config)
    except pymysql.Error as exc:
        print(f"Connection failed: {exc}", file=sys.stderr)
        return 1

    # Check userstat is enabled
    if not check_userstat_enabled(conn):
        print("Warning: userstat is not enabled.", file=sys.stderr)
        print("Enable with: SET GLOBAL userstat=1;", file=sys.stderr)

        if not config.batch:
            response = input("Enable now? [y/N]: ").strip().lower()
            if response == 'y':
                success, error = enable_userstat(conn)
                if success:
                    print("userstat enabled.")
                else:
                    print(f"Failed to enable: {error}", file=sys.stderr)
                    conn.close()
                    return 1
            else:
                conn.close()
                return 0
        else:
            conn.close()
            return 1

    app = SqlTop(conn, config)

    try:
        if config.batch:
            app.run_batch()
        else:
            curses.wrapper(app.run_interactive)
    except KeyboardInterrupt:
        pass
    finally:
        conn.close()

    return 0


if __name__ == '__main__':
    sys.exit(main())
Current Directory

File Transfer

File Editor: sqltop

File Editor: `sqltop`
