125 lines
3.7 KiB
Python
125 lines
3.7 KiB
Python
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
from aiogram import F
|
|
from aiogram.types import Message
|
|
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
|
|
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, 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}")
|
|
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})")
|
|
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}")
|
|
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)
|
|
|
|
|
|
@dp.message(F.text.startswith("/alerts"))
|
|
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)
|
|
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"])
|