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 Role, User, utcnow
from app.runner.client import RunnerClient
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"
    RESTART = "r"


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, can_restart: bool = False
) -> tuple[str, InlineKeyboardMarkup]:
    """can_restart: an admin looking at a service deployed by tgit-runner (S-5)."""
    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]))
    if can_restart and runner_service(d):
        kb.button(text=t.restart, callback_data=StatusCb(a=StatusAct.RESTART, id=d.id[:12]))
    kb.button(text=t.to_list, callback_data=StatusCb(a=StatusAct.LIST))
    kb.adjust(2, 1, 1)
    return text, kb.as_markup()


def runner_service(d: Details) -> str | None:
    """Only services tgit-runner deployed can be restarted through it — not the bot, its DB or the proxy."""
    return d.labels.get("tgit.service") or None


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, runner: RunnerClient | None = 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, is_admin: bool = False) -> 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(), can_restart=is_admin and runner is not None)

    @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, member: User) -> 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, member.role is Role.ADMIN)
                case StatusAct.RESTART:
                    if docker is None or runner is None or member.role is not Role.ADMIN:
                        await query.answer()
                        return
                    service = runner_service(await docker.inspect(await docker.resolve(callback_data.id)))
                    if service is None:
                        await query.answer()
                        return
                    result = await runner.restart(service)
                    if not result.ok:
                        await query.answer(t.restart_failed.format(reason=result.summary)[:200], show_alert=True)
                        return
                    await query.answer(t.restarted)
                    text, kb = await card_screen(callback_data.id, True)
                    await _edit(msg, text, kb)
                    return
                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