"""«🚨 Алерты»: the alert message itself, the list (active first, then 7 days, 10 per page) and cards (SPEC AL-3, AL-6). Buttons under an alert open «📜 Логи» / «🩺 Статус» as a new message: editing would wipe the alert out of the chat. """ import html import logging from collections import defaultdict from enum import StrEnum from typing import Any from aiogram import F, Router from aiogram.enums import ParseMode from aiogram.exceptions import TelegramBadRequest from aiogram.filters.callback_data import CallbackData from aiogram.fsm.context import FSMContext from aiogram.types import CallbackQuery, InlineKeyboardMarkup, Message from aiogram.utils.keyboard import InlineKeyboardBuilder from app.alerts.service import AlertService from app.bot.menu import Nav, NavAction from app.bot.routers.logs import container_keyboard from app.bot.routers.status import render_card as status_card from app.db.models import Alert, AlertKind, as_utc, utcnow from app.runtime.docker import DockerClient, DockerError from app.texts import Texts log = logging.getLogger(__name__) PER_PAGE = 10 LOG_CHARS = 3000 class AlertsAct(StrEnum): LIST = "l" CARD = "c" LOGS = "g" STATUS = "s" class AlertsCb(CallbackData, prefix="al"): a: AlertsAct p: int = 0 # page id: int = 0 # alert id c: str = "" # 12-char container id prefix for LOGS / STATUS def title(texts: Texts, alert: Alert) -> str: return texts.alerts.kinds[alert.kind] def service(texts: Texts, alert: Alert) -> str: return html.escape(alert.service or texts.alerts.host) def detail(texts: Texts, alert: Alert) -> str: values: defaultdict[str, Any] = defaultdict(str, alert.payload) return html.escape(texts.alerts.details[alert.kind].format_map(values)) def _with_log(text: str, alert: Alert) -> str: tail = str(alert.payload.get("log", "")) return f"{text}\n
{html.escape(tail[-LOG_CHARS:])}
" if tail else text def render(texts: Texts, alert: Alert) -> str: """The alert message (AL-3); re-rendered with the counter on repeats (AL-5).""" t = texts.alerts text = t.message.format( title=title(texts, alert), service=service(texts, alert), detail=detail(texts, alert), opened=as_utc(alert.opened_at).strftime("%d.%m %H:%M"), repeats=t.repeats.format(count=alert.count) if alert.count > 1 else "", ) return _with_log(text, alert) def keyboard(texts: Texts, alert: Alert, page: int | None = None) -> InlineKeyboardMarkup: """page=None: under the alert message («📋 Список алертов»); a number: on a card («⬅️ К списку»).""" t = texts.alerts kb = InlineKeyboardBuilder() cid = str(alert.payload.get("container_id", ""))[:12] if cid: kb.button(text=t.logs_button, callback_data=AlertsCb(a=AlertsAct.LOGS, c=cid)) kb.button(text=t.status_button, callback_data=AlertsCb(a=AlertsAct.STATUS, c=cid)) back = t.list_button if page is None else t.to_list kb.button(text=back, callback_data=AlertsCb(a=AlertsAct.LIST, p=page or 0)) kb.adjust(2, 1) return kb.as_markup() def render_list( texts: Texts, alerts: list[Alert], total: int, active: int, page: int ) -> tuple[str, InlineKeyboardMarkup]: t = texts.alerts pages = max((total + PER_PAGE - 1) // PER_PAGE, 1) lines = [t.list_title.format(active=active, total=total)] if not alerts: lines += ["", t.list_empty] kb = InlineKeyboardBuilder() for a in alerts: line = t.list_line.format( mark=t.active_mark if a.resolved_at is None else t.resolved_mark, title=title(texts, a), service=a.service or t.host, time=as_utc(a.last_at).strftime("%d.%m %H:%M"), repeats=t.repeats.format(count=a.count) if a.count > 1 else "", ) kb.button(text=line[:64], callback_data=AlertsCb(a=AlertsAct.CARD, p=page, id=a.id)) nav = [] if page > 0: kb.button(text=t.prev, callback_data=AlertsCb(a=AlertsAct.LIST, p=page - 1)) nav.append(1) if page + 1 < pages: kb.button(text=t.next, callback_data=AlertsCb(a=AlertsAct.LIST, p=page + 1)) nav.append(1) kb.button(text=texts.menu.back, callback_data=Nav(action=NavAction.BACK)) kb.adjust(*([1] * len(alerts)), *([len(nav)] if nav else []), 1) if pages > 1: lines.append(t.page.format(page=page + 1, pages=pages)) return "\n".join(lines), kb.as_markup() def render_alert_card(texts: Texts, a: Alert, page: int) -> tuple[str, InlineKeyboardMarkup]: t = texts.alerts text = t.card.format( title=title(texts, a), service=service(texts, a), detail=detail(texts, a), opened=as_utc(a.opened_at).strftime("%d.%m %H:%M"), resolved=as_utc(a.resolved_at).strftime("%d.%m %H:%M UTC") if a.resolved_at else t.ongoing, count=a.count, ) return _with_log(text, a), keyboard(texts, a, page) def build_router(texts: Texts, alerts: AlertService | None, docker: DockerClient | None = None) -> Router: router = Router(name="alerts") t = texts.alerts async def list_screen(page: int) -> tuple[str, InlineKeyboardMarkup]: if alerts is None: return render_list(texts, [], 0, 0, 0) rows, total = await alerts.page(page, PER_PAGE) if not rows and page > 0: # the history shrank since the button was drawn page = 0 rows, total = await alerts.page(0, PER_PAGE) return render_list(texts, rows, total, await alerts.active_count(), page) @router.message(F.text == texts.menu.alerts) async def open_alerts(message: Message, state: FSMContext) -> None: await state.clear() text, kb = await list_screen(0) await message.answer(text, reply_markup=kb) @router.callback_query(AlertsCb.filter()) async def on_action(query: CallbackQuery, callback_data: AlertsCb) -> None: if not isinstance(query.message, Message): await query.answer() return msg = query.message # Pressed under an alert: never overwrite it — answer with a new message rows = msg.reply_markup.inline_keyboard if msg.reply_markup else [] under_alert = any(b.text == t.list_button for row in rows for b in row) match callback_data.a: case AlertsAct.LOGS | AlertsAct.STATUS: try: if docker is None: raise DockerError("docker is not configured (DOCKER_URL)") c = await docker.get(callback_data.c) if callback_data.a is AlertsAct.LOGS: await msg.answer( texts.logs.container.format(name=c.name), reply_markup=container_keyboard(texts, c.short_id) ) else: details = await docker.inspect(c.id) stats = await docker.stats(c.id) if details.state == "running" else None text, kb = status_card(texts, details, stats, utcnow()) await msg.answer(text, reply_markup=kb) except DockerError as e: log.warning("alerts: %s", e) await query.answer(texts.status.unavailable, show_alert=True) return await query.answer() return case AlertsAct.CARD: alert = await alerts.get(callback_data.id) if alerts else None if alert is None or alert.kind is AlertKind.AGENT_ATTACK: # 🛡 lives only in the admins' chats await query.answer(t.gone, show_alert=True) return text, kb = render_alert_card(texts, alert, callback_data.p) case AlertsAct.LIST: text, kb = await list_screen(max(callback_data.p, 0)) if under_alert: await msg.answer(text, parse_mode=ParseMode.HTML, reply_markup=kb) else: try: await msg.edit_text(text, parse_mode=ParseMode.HTML, reply_markup=kb) except TelegramBadRequest as e: if "message is not modified" not in str(e): raise await query.answer() return router