tgit/app/bot/routers/status.py

"""«🩺 Статус»: managed containers, a card per container, on-demand checks (SPEC S-1…S-4)."""

import logging
from datetime import datetime
from enum import StrEnum

from aiogram import F, Router
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.bot.menu import Nav, NavAction
from app.bot.routers.commit import human_size
from app.db.models import utcnow
from app.runtime.checks import Checker, CheckName, ContainerReport
from app.runtime.docker import Container, Details, DockerClient, DockerError, Stats
from app.texts import StatusTexts, Texts

log = logging.getLogger(__name__)

MAX_TEXT = 4000  # Telegram's limit is 4096


class StatusAct(StrEnum):
    LIST = "l"
    CARD = "c"
    CHECK = "k"


class StatusCb(CallbackData, prefix="s"):
    a: StatusAct
    id: str = ""  # 12-char container id prefix: callback data is limited to 64 bytes


def duration(t: StatusTexts, seconds: float) -> str:
    s = int(max(seconds, 0))
    d, s = divmod(s, 86400)
    h, s = divmod(s, 3600)
    m, s = divmod(s, 60)
    if d:
        return f"{d} {t.days} {h} {t.hours}"
    if h:
        return f"{h} {t.hours} {m} {t.minutes}"
    if m:
        return f"{m} {t.minutes}"
    return f"{s} {t.seconds}"


def _clip(text: str) -> str:
    return text if len(text) <= MAX_TEXT else text[: MAX_TEXT - 1] + "…"


def _fmt_time(dt: datetime | None, none: str) -> str:
    return dt.strftime("%d.%m %H:%M") if dt else none


def render_list(texts: Texts, containers: list[Container], now: datetime) -> tuple[str, InlineKeyboardMarkup]:
    t = texts.status
    lines = [t.title.format(count=len(containers), time=now.strftime("%H:%M:%S")), ""]
    if not containers:
        lines.append(t.empty)
    for c in containers:
        state = c.state if c.health is None else f"{c.state}, {c.health}"
        lines.append(t.line.format(light=c.light.value, name=c.name, state=state))
    kb = InlineKeyboardBuilder()
    for c in containers:
        kb.button(text=f"{c.light.value} {c.name}", callback_data=StatusCb(a=StatusAct.CARD, id=c.short_id))
    if containers:
        kb.button(text=t.check_button, callback_data=StatusCb(a=StatusAct.CHECK))
    kb.button(text=t.refresh, callback_data=StatusCb(a=StatusAct.LIST))
    kb.button(text=texts.menu.back, callback_data=Nav(action=NavAction.BACK))
    kb.adjust(1)
    return _clip("\n".join(lines)), kb.as_markup()


def render_card(texts: Texts, d: Details, stats: Stats | None, now: datetime) -> tuple[str, InlineKeyboardMarkup]:
    t = texts.status
    running = d.state == "running"
    uptime = duration(t, (now - d.started_at).total_seconds()) if running and d.started_at else t.none
    if stats is None:
        cpu = mem = limit = mem_pct = t.none
    else:
        cpu = f"{stats.cpu_pct:.1f}%"
        mem = human_size(stats.mem_bytes)
        limit = human_size(stats.mem_limit) if stats.mem_limit else t.no_limit
        mem_pct = f"{stats.mem_pct:.0f}%"
    text = t.card.format(
        light=d.light.value,
        name=d.name,
        service=d.service or t.none,
        image=d.image or t.none,
        state=d.state if running else f"{d.state} (exit {d.exit_code}{', OOM' if d.oom_killed else ''})",
        health=d.health or t.none,
        uptime=uptime,
        restarts=d.restart_count,
        cpu=cpu,
        mem=mem,
        limit=limit,
        mem_pct=mem_pct,
        sha=(d.sha or t.none)[:12],
        deployed=_fmt_time(d.deployed_at, t.none),
    )
    kb = InlineKeyboardBuilder()
    kb.button(text=t.check_button, callback_data=StatusCb(a=StatusAct.CHECK, id=d.id[:12]))
    kb.button(text=t.refresh, callback_data=StatusCb(a=StatusAct.CARD, id=d.id[:12]))
    kb.button(text=t.to_list, callback_data=StatusCb(a=StatusAct.LIST))
    kb.adjust(2, 1)
    return text, kb.as_markup()


def render_report(
    texts: Texts, reports: list[ContainerReport], now: datetime, back_id: str = ""
) -> tuple[str, InlineKeyboardMarkup]:
    """back_id: the report was run from a card — offer the way back to it."""
    t = texts.status
    titles = {
        CheckName.RUNNING: t.check_running,
        CheckName.HEALTH: t.check_health,
        CheckName.RESTARTS: t.check_restarts,
        CheckName.CPU: t.check_cpu,
        CheckName.MEM: t.check_mem,
        CheckName.LOG_ERRORS: t.check_log_errors,
        CheckName.HTTP: t.check_http,
    }
    ok = sum(r.ok for r in reports)
    lines = [t.report_title.format(ok=ok, total=len(reports), time=now.strftime("%H:%M:%S"))]
    for report in reports:
        lines.append("")
        lines.append(t.report_container.format(light="✅" if report.ok else "❌", name=report.details.name))
        for r in report.results:
            detail = f" · {r.detail}" if r.detail else ""
            lines.append(t.report_line.format(verdict=r.verdict.value, title=titles[r.name], detail=detail))
    kb = InlineKeyboardBuilder()
    kb.button(text=t.refresh, callback_data=StatusCb(a=StatusAct.CHECK, id=back_id))
    if back_id:
        kb.button(text=t.to_card, callback_data=StatusCb(a=StatusAct.CARD, id=back_id))
    kb.button(text=t.to_list, callback_data=StatusCb(a=StatusAct.LIST))
    kb.adjust(1)
    return _clip("\n".join(lines)), kb.as_markup()


def build_router(texts: Texts, docker: DockerClient | None, checker: Checker | None) -> Router:
    router = Router(name="status")
    t = texts.status

    async def list_screen() -> tuple[str, InlineKeyboardMarkup]:
        if docker is None:
            raise DockerError("docker is not configured (DOCKER_URL)")
        return render_list(texts, await docker.containers(), utcnow())

    async def card_screen(short_id: str) -> tuple[str, InlineKeyboardMarkup]:
        if docker is None:
            raise DockerError("docker is not configured (DOCKER_URL)")
        cid = await docker.resolve(short_id)
        details = await docker.inspect(cid)
        stats = None
        if details.state == "running":
            try:
                stats = await docker.stats(cid)
            except DockerError:
                log.warning("stats %s unavailable", short_id)
        return render_card(texts, details, stats, utcnow())

    @router.message(F.text == texts.menu.status)
    async def open_status(message: Message, state: FSMContext) -> None:
        await state.clear()
        try:
            text, kb = await list_screen()
        except DockerError as e:
            log.warning("status: %s", e)
            await message.answer(t.unavailable)
            return
        await message.answer(text, reply_markup=kb)

    @router.callback_query(StatusCb.filter())
    async def on_action(query: CallbackQuery, callback_data: StatusCb) -> None:
        if not isinstance(query.message, Message):
            await query.answer()
            return
        msg = query.message
        try:
            match callback_data.a:
                case StatusAct.LIST:
                    text, kb = await list_screen()
                case StatusAct.CARD:
                    text, kb = await card_screen(callback_data.id)
                case StatusAct.CHECK:
                    if docker is None or checker is None:
                        raise DockerError("docker is not configured (DOCKER_URL)")
                    await query.answer(t.checking)
                    if callback_data.id:
                        reports = [await checker.check(await docker.resolve(callback_data.id))]
                    else:
                        reports = await checker.check_all()
                    text, kb = render_report(texts, reports, utcnow(), callback_data.id)
                    await _edit(msg, text, kb)
                    return
        except DockerError as e:
            log.warning("status: %s", e)
            await query.answer(t.unavailable, show_alert=True)
            return
        await _edit(msg, text, kb)
        await query.answer()

    return router


async def _edit(msg: Message, text: str, kb: InlineKeyboardMarkup) -> None:
    # «🔄 Обновить» with nothing changed: Telegram rejects an identical edit — not an error for the user
    try:
        await msg.edit_text(text, reply_markup=kb)
    except TelegramBadRequest as e:
        if "message is not modified" not in str(e):
            raise