"""«🩺 Статус»: 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