path: root/alfred/server.py

"""
Sopel plugin: server stats

Commands (owner only) - use !server or !srv:
  !server                       - show this usage
  !server uptime                - uptime and load averages
  !server disk                  - disk usage for /
  !server mem                   - RAM and swap usage
  !server load                  - 1/5/15 min load + process count
  !server status [service]      - rc-service status (all or one)
  !server net [iface]           - RX/TX for iface (default: first non-lo)
  !server top [n]               - top n procs by CPU (default 5)
  !server topmem [n]            - top n procs by memory (default 5)
  !server io                    - disk read/write stats for main block device
  !server conns                 - TCP connection summary
  !server who                   - logged-in users
  !server logs <service> [n]    - last n lines of service log (default 10)
"""

import os
import subprocess

from sopel import plugin
import os as _os, sys as _sys
_alfred_dir = _os.path.dirname(_os.path.abspath(__file__))
if _alfred_dir not in _sys.path:
    _sys.path.insert(0, _alfred_dir)
import helpstrings as h

SERVICES = [
    "nginx", "postfix", "dovecot", "ngircd", "soju", "websocat",
    "opendkim", "fcgiwrap", "alfred", "dcron", "0x0",
    "xandikos", "wg-quick.wg0",
]

LOG_PATHS = {
    "nginx":          ["/var/log/nginx/error.log", "/var/log/nginx/access.log"],
    "postfix":        ["/var/log/mail.log", "/var/log/messages"],
    "dovecot":        ["/var/log/dovecot.log", "/var/log/messages"],
    "ngircd":         ["/var/log/ngircd.log", "/var/log/messages"],
    "soju":           ["/var/log/soju.log"],
    "websocat":  ["/var/log/websocat/daemon.log"],
    "0x0":            ["/var/log/0x0/daemon.log", "/var/log/0x0/prune.log"],
    "xandikos":       ["/var/log/xandikos/daemon.log"],
    "alfred":         ["/var/log/alfred/sopel.log", "/var/log/alfred/daemon.log"],
    "herald":         ["/var/log/herald/sopel.log", "/var/log/herald/daemon.log"],
    "bikabot":        ["/var/log/bikabot/sopel.log", "/var/log/bikabot/daemon.log"],
    "daft":           ["/var/log/daft/sopel.log", "/var/log/daft/daemon.log"],
    "daffy":          ["/var/log/daffy/sopel.log", "/var/log/daffy/daemon.log"],
    "contessa":       ["/var/log/contessa/sopel.log", "/var/log/contessa/daemon.log"],
    "osterman":       ["/var/log/osterman/sopel.log", "/var/log/osterman/daemon.log"],
    "salvador":       ["/var/log/salvador/sopel.log", "/var/log/salvador/daemon.log"],
    "shireen":        ["/var/log/shireen/sopel.log", "/var/log/shireen/daemon.log"],
    "jeeves":         ["/var/log/jeeves/sopel.log", "/var/log/jeeves/daemon.log"],
}


def _is_owner(bot, trigger):
    return trigger.nick == bot.settings.core.owner


def _run(cmd, timeout=10):
    try:
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
        out = (r.stdout + r.stderr).strip()
        return out[:400] if out else "(no output)"
    except subprocess.TimeoutExpired:
        return "(timed out)"
    except Exception as e:
        return f"(error: {e})"


def _uptime(bot, _args):
    bot.say(_run(["uptime"]))


def _disk(bot, _args):
    for line in _run(["df", "-h", "/"]).splitlines():
        bot.say(line)


def _mem(bot, _args):
    for line in _run(["free", "-h"]).splitlines():
        bot.say(line)


def _load(bot, _args):
    try:
        with open("/proc/loadavg") as f:
            fields = f.read().split()
        load = f"{fields[0]} {fields[1]} {fields[2]}"
        procs = fields[3]
        bot.say(f"load: {load} | procs: {procs}")
    except OSError as e:
        bot.say(f"error: {e}")


def _status(bot, args):
    targets = [args[0]] if args else SERVICES
    results = []
    for svc in targets:
        r = subprocess.run(
            ["rc-service", svc, "status"],
            capture_output=True, text=True, timeout=5
        )
        out = (r.stdout + r.stderr).strip()
        state = "UP" if "started" in out else "DOWN" if "stopped" in out else out[:20]
        results.append(f"{svc}:{state}")
    bot.say(" | ".join(results))


def _net(bot, args):
    iface = args[0] if args else None
    try:
        with open("/proc/net/dev") as f:
            lines = f.readlines()[2:]
        devs = {}
        for line in lines:
            parts = line.split()
            name = parts[0].rstrip(":")
            if name == "lo":
                continue
            devs[name] = {"rx": int(parts[1]), "tx": int(parts[9])}

        if not devs:
            bot.say("net: no interfaces found")
            return

        target = iface if iface in devs else next(iter(devs))
        if iface and iface not in devs:
            bot.say(f"net: unknown iface '{iface}', using {target}")

        def fmt(n):
            for unit in ("B", "KB", "MB", "GB"):
                if n < 1024:
                    return f"{n:.1f}{unit}"
                n /= 1024
            return f"{n:.1f}TB"

        d = devs[target]
        bot.say(f"{target}: RX {fmt(d['rx'])} | TX {fmt(d['tx'])}")
    except OSError as e:
        bot.say(f"error: {e}")


def _top_procs(bot, args, sort_col, label):
    try:
        n = int(args[0]) if args else 5
        n = max(1, min(n, 15))
    except ValueError:
        bot.say("usage: !srv top [n]")
        return

    r = subprocess.run(["top", "-b", "-n", "1"], capture_output=True, text=True)
    lines = r.stdout.strip().splitlines()

    proc_lines = []
    in_procs = False
    for line in lines:
        if not in_procs:
            if "PID" in line and "PPID" in line:
                in_procs = True
            continue
        if line.strip():
            proc_lines.append(line)

    if not proc_lines:
        bot.say("no processes found")
        return

    # columns: PID PPID USER STAT VSZ %VSZ CPU %CPU COMMAND
    procs = []
    for line in proc_lines:
        parts = line.split(None, 8)
        if len(parts) < 9:
            continue
        try:
            val  = float(parts[sort_col].rstrip("%m"))
            user = parts[2]
            cmd  = parts[8].split()[0].split("/")[-1][:25]
            procs.append((val, user, cmd))
        except (ValueError, IndexError):
            continue

    procs.sort(reverse=True)
    out = " | ".join(f"{cmd}({user},{v:.1f}%)" for v, user, cmd in procs[:n])
    bot.say(f"{label}: {out}" if out else f"{label}: (no data)")


def _top(bot, args):
    _top_procs(bot, args, sort_col=7, label="cpu")


def _topmem(bot, args):
    _top_procs(bot, args, sort_col=5, label="mem")


def _io(bot, _args):
    try:
        with open("/proc/diskstats") as f:
            lines = f.readlines()
        for line in lines:
            parts = line.split()
            if len(parts) < 14:
                continue
            name = parts[2]
            if name.startswith(("sd", "vd", "nvme", "hd", "mmcblk")) and not name[-1].isdigit():
                reads = int(parts[5])
                writes = int(parts[9])
                bot.say(f"{name}: reads {reads} | writes {writes}")
                return
        bot.say("io: no block device found")
    except OSError as e:
        bot.say(f"error: {e}")


def _conns(bot, _args):
    for line in _run(["ss", "-s"]).splitlines():
        bot.say(line)


def _who(bot, _args):
    out = _run(["who"])
    if not out.strip() or out == "(no output)":
        bot.say("who: no users logged in")
        return
    for line in out.splitlines():
        bot.say(line)


def _logs(bot, args):
    if not args:
        bot.say("usage: !srv logs <service> [n]")
        return
    svc = args[0]
    try:
        n = int(args[1]) if len(args) > 1 else 10
        n = max(1, min(n, 30))
    except ValueError:
        bot.say("usage: !srv logs <service> [n]")
        return

    candidates = LOG_PATHS.get(svc, [f"/var/log/{svc}.log", f"/var/log/{svc}/error.log"])
    for path in candidates:
        if os.path.exists(path):
            out = _run(["tail", "-n", str(n), path])
            for line in out.splitlines()[:n]:
                bot.say(line)
            return
    bot.say(f"logs: no log file found for '{svc}'")


_SUBCOMMANDS = {
    "uptime": _uptime,
    "disk":   _disk,
    "mem":    _mem,
    "load":   _load,
    "status": _status,
    "net":    _net,
    "top":    _top,
    "topmem": _topmem,
    "io":     _io,
    "conns":  _conns,
    "who":    _who,
    "logs":   _logs,
}


@plugin.commands("server", "srv")
def cmd_server(bot, trigger):
    if not _is_owner(bot, trigger):
        return

    args = (trigger.group(2) or "").split()
    sub = args[0].lower() if args else ""

    if not sub:
        bot.say(h.SRV_TOPIC)
        return

    fn = _SUBCOMMANDS.get(sub)
    if fn:
        fn(bot, args[1:])
    else:
        bot.say(f"server: unknown subcommand '{sub}' - type !srv for usage")