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')",
    )

    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 "",
    )

    # Bypass min_level for explicit test sends; run async channels directly
    import asyncio
    ch_type = channel_cfg.get("type", "")
    print(f"Sending via {args.channel} ({ch_type}): {title} — {args.message}")

    if ch_type in ("matrix", "sms_voipms"):
        from .notify import _send_matrix_async, _send_sms_voipms_async
        driver_async = _send_matrix_async if ch_type == "matrix" else _send_sms_voipms_async
        ok = asyncio.run(driver_async(channel_cfg, notif))
    else:
        from .notify import _DRIVERS
        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 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

    # 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()