tgit/app/bot/routers/alerts.py

"""«🚨 Алерты»: 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<pre>{html.escape(tail[-LOG_CHARS:])}</pre>" 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