heartbeat/hbd/server/cli.py

"""Command line interface for hbd package."""

import argparse
import getpass
import sys

from .config import load_config
from .main import run as run_server

PUSHSRVS = ["all", "pushover", "mattermost"]


def build_parser():
    parser = argparse.ArgumentParser(
        prog="hbd",
        description="HeartBeatDaemon - Wait for heartbeat messages and act on them (or their absence)",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )

    subparsers = parser.add_subparsers(dest="command")

    # --- serve (default) ---
    serve_p = subparsers.add_parser("serve", help="Start the hbd server (default)")
    serve_p.add_argument("-c", "--config", dest="configfile", help="Config file path (YAML)")
    serve_p.add_argument("-f", "--foreground", action="store_true", help="Run in foreground")
    serve_p.add_argument("-v", "--verbose", action="store_true", help="Verbose output")
    serve_p.add_argument("-p", "--pushsrv", dest="pushsrv", choices=PUSHSRVS,
                         help="Push service to use")
    serve_p.add_argument("-x", "--debug", action="count", default=0, help="Increase debug level")

    # Legacy top-level flags (no subcommand) — kept for backward compatibility
    parser.add_argument("-c", "--config", dest="configfile", help="Config file path (YAML)")
    parser.add_argument("-f", "--foreground", action="store_true", help="Run in foreground")
    parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output")
    parser.add_argument("-p", "--pushsrv", dest="pushsrv", choices=PUSHSRVS,
                        help="Push service to use")
    parser.add_argument("-x", "--debug", action="count", default=0, help="Increase debug level")

    # --- passwd ---
    passwd_p = subparsers.add_parser(
        "passwd",
        help="Generate a password hash for use in the config file",
    )
    passwd_p.add_argument(
        "username",
        nargs="?",
        help="Username (informational only, for display)",
    )

    # --- notify ---
    notify_p = subparsers.add_parser(
        "notify",
        help="Send a test message via a configured notification channel",
    )
    notify_p.add_argument("-c", "--config", dest="configfile", help="Config file path (YAML)")
    notify_p.add_argument(
        "channel",
        help="Channel name as defined in notification_channels",
    )
    notify_p.add_argument(
        "message",
        nargs="?",
        default="Test notification from hbd",
        help="Message body (default: 'Test notification from hbd')",
    )
    notify_p.add_argument(
        "--level",
        default="WARNING",
        choices=["INFO", "WARNING", "CRITICAL", "RECOVER"],
        help="Notification level (default: WARNING)",
    )
    notify_p.add_argument(
        "--title",
        default=None,
        help="Notification title (default: '[LEVEL] test')",
    )

    # --- stop ---
    stop_p = subparsers.add_parser("stop", help="Stop the running hbd instance")
    stop_p.add_argument("-c", "--config", dest="configfile", help="Config file path (YAML)")

    # --- reload ---
    reload_p = subparsers.add_parser("reload", help="Reload configuration (SIGHUP)")
    reload_p.add_argument("-c", "--config", dest="configfile", help="Config file path (YAML)")

    # --- restart ---
    restart_p = subparsers.add_parser("restart", help="Restart the running hbd instance")
    restart_p.add_argument("-c", "--config", dest="configfile", help="Config file path (YAML)")
    restart_p.add_argument("-f", "--foreground", action="store_true", help="Run in foreground after restart")
    restart_p.add_argument("-v", "--verbose", action="store_true", help="Verbose output after restart")

    return parser


def cmd_passwd(args):
    """Interactive password hash generator."""
    from .users import hash_password

    username = args.username or ""
    prompt = f"New password for {username}: " if username else "New password: "
    while True:
        pw = getpass.getpass(prompt)
        if not pw:
            print("Password must not be empty.", file=sys.stderr)
            continue
        pw2 = getpass.getpass("Confirm password: ")
        if pw != pw2:
            print("Passwords do not match, try again.", file=sys.stderr)
            continue
        break

    hashed = hash_password(pw)
    if username:
        print(f"\nAdd the following to your config under users: -> {username}:")
    else:
        print("\nPassword hash (paste into config file under the user's 'password' key):")
    print(f"  password: {hashed}")


def cmd_notify(args):
    """Send a test message via a single notification channel."""
    from .config import load_config
    from .notify import Notification, _dispatch_to_channel, setup

    config = load_config(args.configfile)
    setup(config)

    channels = config.get("notification_channels", {})
    if args.channel not in channels:
        available = ", ".join(channels.keys()) if channels else "(none)"
        print(f"Error: channel '{args.channel}' not found in notification_channels.", file=sys.stderr)
        print(f"Available channels: {available}", file=sys.stderr)
        sys.exit(1)

    channel_cfg = channels[args.channel]
    level = args.level.upper()
    title = args.title or f"[{level}] test"
    base_url = config.get("base_url", "").rstrip("/")

    notif = Notification(
        title=title,
        body=args.message,
        level=level,
        url=f"{base_url}/plugins" if base_url else "",
    )

    import asyncio
    from .notify import _send_matrix_async, _send_sms_voipms_async, _DRIVERS
    ch_type = channel_cfg.get("type", "")
    print(f"Sending via {args.channel} ({ch_type}): {title} — {args.message}")

    if ch_type == "matrix":
        ok = asyncio.run(_send_matrix_async(channel_cfg, notif))
    elif ch_type == "sms_voipms":
        ok = asyncio.run(_send_sms_voipms_async(channel_cfg, notif))
    else:
        driver = _DRIVERS.get(ch_type)
        if driver is None:
            print(f"Error: unknown channel type '{ch_type}'", file=sys.stderr)
            sys.exit(1)
        ok = driver(channel_cfg, notif)

    if ok:
        print("OK")
    else:
        print("FAILED — check logs for details", file=sys.stderr)
        sys.exit(1)


def _read_pid(configfile) -> int | None:
    """Return the PID from the pidfile, or None if not found / not running."""
    import os
    config = load_config(configfile)
    pidfile = config.get("pidfile", "")
    if not pidfile:
        print("Error: no pidfile configured.", file=sys.stderr)
        return None
    try:
        with open(pidfile) as f:
            pid = int(f.read().strip())
        # Verify process is actually running
        os.kill(pid, 0)
        return pid
    except FileNotFoundError:
        print(f"PID file not found ({pidfile}). Is hbd running?", file=sys.stderr)
        return None
    except ProcessLookupError:
        print(f"PID file exists but process {pid} is not running.", file=sys.stderr)
        return None
    except Exception as e:
        print(f"Error reading pidfile: {e}", file=sys.stderr)
        return None


def cmd_stop(args):
    import os, signal as _signal, time
    pid = _read_pid(args.configfile)
    if pid is None:
        sys.exit(1)
    print(f"Stopping hbd (pid {pid})...")
    os.kill(pid, _signal.SIGTERM)
    # Wait up to 10 s for the process to exit
    for _ in range(20):
        time.sleep(0.5)
        try:
            os.kill(pid, 0)
        except ProcessLookupError:
            print("hbd stopped.")
            return
    print("Warning: hbd did not stop within 10 seconds.", file=sys.stderr)
    sys.exit(1)


def cmd_reload(args):
    import os, signal as _signal
    pid = _read_pid(args.configfile)
    if pid is None:
        sys.exit(1)
    print(f"Sending SIGHUP to hbd (pid {pid})...")
    os.kill(pid, _signal.SIGHUP)
    print("Reload signal sent.")


def cmd_restart(args):
    import os, signal as _signal, time, subprocess
    pid = _read_pid(args.configfile)
    if pid is not None:
        print(f"Stopping hbd (pid {pid})...")
        os.kill(pid, _signal.SIGTERM)
        for _ in range(20):
            time.sleep(0.5)
            try:
                os.kill(pid, 0)
            except ProcessLookupError:
                print("hbd stopped.")
                break
        else:
            print("Warning: hbd did not stop within 10 seconds.", file=sys.stderr)
            sys.exit(1)
    else:
        print("hbd does not appear to be running — starting fresh.")

    # Re-launch hbd with the same config
    cmd = [sys.executable, "-m", "hbd.server.cli", "serve"]
    if args.configfile:
        cmd += ["-c", args.configfile]
    if getattr(args, "foreground", False):
        cmd += ["-f"]
    if getattr(args, "verbose", False):
        cmd += ["-v"]

    if getattr(args, "foreground", False):
        # Run in foreground — replace current process
        os.execv(sys.executable, cmd)
    else:
        subprocess.Popen(cmd, start_new_session=True)
        print("hbd restarted.")


def main(argv=None):
    parser = build_parser()
    args = parser.parse_args(argv)

    if args.command == "passwd":
        cmd_passwd(args)
        return

    if args.command == "notify":
        cmd_notify(args)
        return

    if args.command == "stop":
        cmd_stop(args)
        return

    if args.command == "reload":
        cmd_reload(args)
        return

    if args.command == "restart":
        cmd_restart(args)
        return

    # Default: run the server (supports both `hbd serve ...` and `hbd ...`)
    config = load_config(args.configfile)

    # Apply CLI overrides
    if args.foreground:
        config["foreground"] = True
    if args.verbose:
        config["verbose"] = True
    if args.pushsrv:
        config["pushsrv"] = args.pushsrv
    if args.debug > 0:
        config["debug"] = args.debug

    # Pass config_path for reloading support
    run_server(config, config_path=args.configfile)


if __name__ == "__main__":
    main()