tg-admin-bot/handlers/alerts_admin.py

import time
from datetime import datetime, timedelta, timezone
from aiogram import F
from aiogram.types import Message, CallbackQuery, InlineKeyboardMarkup, InlineKeyboardButton
from app import dp, bot, cfg, ADMIN_ID
from auth import is_admin_msg
from services.alert_mute import set_mute, clear_mute, list_mutes
from services.incidents import read_recent, log_incident
from services.notify import notify


HELP_TEXT = (
    "Alerts:\n"
    "/alerts test <critical|warn|info> - send test alert\n"
    "/alerts mute <category> <minutes> - mute alerts for category\n"
    "/alerts unmute <category> - unmute category\n"
    "/alerts list - show active mutes\n"
    "/alerts recent [hours] - show incidents log (default 24h)\n"
    "Categories: load, disk, smart, raid, ssl, docker, test\n"
)


def _dispatch(msg: Message, action: str, args: list[str]):
    return {"action": action, "args": args}


async def _handle_alerts(msg: Message, action: str, args: list[str]):
    if action == "test":
        level = args[0].lower() if args else "info"
        if level not in ("critical", "warn", "info"):
            level = "info"
        key = f"test:{level}:{int(time.time())}"
        await notify(bot, msg.chat.id, f"[TEST] {level.upper()} alert", level=level, key=key, category="test")
        await msg.answer(f"Sent test alert: {level}")
        log_incident(cfg, f"alert_test level={level} by {msg.from_user.id}", category="test")
        return

    if action == "mute":
        if len(args) < 1:
            await msg.answer("Usage: /alerts mute <category> <minutes>")
            return
        category = args[0].lower()
        minutes = 60
        if len(args) >= 2:
            try:
                minutes = max(1, int(args[1]))
            except ValueError:
                minutes = 60
        until = set_mute(category, minutes * 60)
        dt = datetime.fromtimestamp(until, tz=timezone.utc).astimezone()
        await msg.answer(f"🔕 Muted {category} for {minutes}m (until {dt:%Y-%m-%d %H:%M:%S})")
        log_incident(cfg, f"alert_mute category={category} minutes={minutes} by {msg.from_user.id}", category=category)
        return

    if action == "unmute":
        if len(args) < 1:
            await msg.answer("Usage: /alerts unmute <category>")
            return
        category = args[0].lower()
        clear_mute(category)
        await msg.answer(f"🔔 Unmuted {category}")
        log_incident(cfg, f"alert_unmute category={category} by {msg.from_user.id}", category=category)
        return

    if action in ("list", "mutes"):
        mutes = list_mutes()
        if not mutes:
            await msg.answer("🔔 No active mutes")
            return
        lines = ["🔕 Active mutes:"]
        for cat, secs in mutes.items():
            mins = max(0, secs) // 60
            lines.append(f"- {cat}: {mins}m left")
        await msg.answer("\n".join(lines))
        return

    if action == "recent":
        hours = 24
        if args:
            try:
                hours = max(1, int(args[0]))
            except ValueError:
                hours = 24
        rows = read_recent(cfg, hours, limit=50)
        if not rows:
            await msg.answer(f"No incidents in last {hours}h")
            return
        await msg.answer("🧾 Incidents:\n" + "\n".join(rows))
        return

    await msg.answer(HELP_TEXT)


ALERTS_KB = InlineKeyboardMarkup(
    inline_keyboard=[
        [
            InlineKeyboardButton(text="List", callback_data="alerts:list"),
            InlineKeyboardButton(text="Recent 24h", callback_data="alerts:recent:24"),
        ],
        [
            InlineKeyboardButton(text="Mute load 60m", callback_data="alerts:mute:load:60"),
            InlineKeyboardButton(text="Unmute load", callback_data="alerts:unmute:load"),
        ],
        [
            InlineKeyboardButton(text="Test CRIT", callback_data="alerts:test:critical"),
            InlineKeyboardButton(text="Test WARN", callback_data="alerts:test:warn"),
            InlineKeyboardButton(text="Test INFO", callback_data="alerts:test:info"),
        ],
    ]
)


@dp.message(F.text.regexp(r"^/alerts(\\s|$)"))
async def alerts_cmd(msg: Message):
    if not is_admin_msg(msg):
        return

    parts = msg.text.split()
    if len(parts) < 2:
        await msg.answer(HELP_TEXT, reply_markup=ALERTS_KB)
        return

    action = parts[1].lower()
    args = parts[2:]
    await _handle_alerts(msg, action, args)


@dp.message(F.text == "/alerts_list")
async def alerts_list(msg: Message):
    if not is_admin_msg(msg):
        return
    await _handle_alerts(msg, "list", [])


@dp.message(F.text == "/alerts_recent")
async def alerts_recent(msg: Message):
    if not is_admin_msg(msg):
        return
    await _handle_alerts(msg, "recent", ["24"])


@dp.message(F.text == "/alerts_mute_load")
async def alerts_mute_load(msg: Message):
    if not is_admin_msg(msg):
        return
    await _handle_alerts(msg, "mute", ["load", "60"])


@dp.callback_query(F.data.startswith("alerts:"))
async def alerts_cb(cb: CallbackQuery):
    if cb.from_user.id != ADMIN_ID:
        await cb.answer()
        return
    parts = cb.data.split(":")
    # formats: alerts:action or alerts:action:arg1:arg2
    if len(parts) < 2:
        await cb.answer()
        return
    action = parts[1]
    args = parts[2:] if len(parts) > 2 else []
    await _handle_alerts(cb.message, action, args)
    await cb.answer()