ElebobaandClaude Opus 5.5
6c6f766abc
CI / ruff · mypy · pytest (pull_request) Successful in 4m49s
- детекторы по состоянию после каждого раунда снапшотов: краш, OOM, цикл рестартов, unhealthy, перегрузка ≥ 1 мин, мало места, Docker недоступен; пороги в config/alerts.yaml - AlertService: ключ kind+service, повтор в 10 мин — счётчик в исходном сообщении, «✅ восстановлено» ответом; разовые события (🛡) со счётчиком - рассылка всем активным (🛡 — только админам), темп ~25/с, RetryAfter, заблокировавший бота пропускается; alert_deliveries для правок и ответов - «🚨 Алерты»: активные + 7 дней по 10, карточка; кнопки «📜 Логи», «🩺 Статус» под алертом открывают новое сообщение, а не затирают алерт - 🛡 атаки через агента идут через общий сервис алертов - убраны заглушки «Раздел в разработке» — у всех разделов свои роутеры Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
111 lines
4.5 KiB
Python
111 lines
4.5 KiB
Python
"""Alerts → Telegram: one message per member (🛡 — admins only), edited on repeats, answered on recovery.
|
|
|
|
Sends are paced under Telegram's ~30 msg/s; a member who blocked the bot is skipped, not an error (AL-1, AL-3…AL-5).
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import Any
|
|
|
|
from aiogram import Bot
|
|
from aiogram.enums import ParseMode
|
|
from aiogram.exceptions import TelegramAPIError, TelegramBadRequest, TelegramForbiddenError, TelegramRetryAfter
|
|
from aiogram.types import ReplyParameters
|
|
from sqlalchemy import select
|
|
|
|
from app.access.service import AccessService
|
|
from app.bot.routers.alerts import keyboard, render, service, title
|
|
from app.bot.routers.status import duration
|
|
from app.db.models import Alert, AlertDelivery, Role, as_utc, utcnow
|
|
from app.db.session import SessionMaker
|
|
from app.texts import Texts
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
PACE = 1 / 25 # seconds between sends
|
|
|
|
|
|
class TelegramNotifier:
|
|
def __init__(self, bot: Bot, access: AccessService, sessions: SessionMaker, texts: Texts) -> None:
|
|
self._bot = bot
|
|
self._access = access
|
|
self._sessions = sessions
|
|
self._texts = texts
|
|
|
|
async def _call(self, coro_factory: Any) -> Any:
|
|
"""One Telegram call; flood control is waited out once, a blocked chat is just skipped."""
|
|
for attempt in range(2):
|
|
try:
|
|
return await coro_factory()
|
|
except TelegramRetryAfter as e:
|
|
if attempt:
|
|
raise
|
|
await asyncio.sleep(e.retry_after)
|
|
except TelegramForbiddenError:
|
|
return None
|
|
return None
|
|
|
|
async def opened(self, alert: Alert, admins_only: bool) -> None:
|
|
members = await self._access.list_members()
|
|
recipients = [m.tg_id for m in members if not admins_only or m.role is Role.ADMIN]
|
|
text = render(self._texts, alert)
|
|
kb = keyboard(self._texts, alert)
|
|
deliveries = []
|
|
for chat_id in recipients:
|
|
try:
|
|
msg = await self._call(
|
|
lambda chat_id=chat_id: self._bot.send_message(
|
|
chat_id, text, parse_mode=ParseMode.HTML, reply_markup=kb
|
|
)
|
|
)
|
|
except TelegramAPIError as e:
|
|
log.warning("alert %s to %s: %s", alert.id, chat_id, e)
|
|
continue
|
|
if msg is not None:
|
|
deliveries.append(AlertDelivery(alert_id=alert.id, chat_id=chat_id, message_id=msg.message_id))
|
|
await asyncio.sleep(PACE)
|
|
async with self._sessions.begin() as session:
|
|
session.add_all(deliveries)
|
|
|
|
async def _deliveries(self, alert: Alert) -> list[AlertDelivery]:
|
|
async with self._sessions() as session:
|
|
return list(await session.scalars(select(AlertDelivery).where(AlertDelivery.alert_id == alert.id)))
|
|
|
|
async def repeated(self, alert: Alert) -> None:
|
|
text = render(self._texts, alert)
|
|
kb = keyboard(self._texts, alert)
|
|
for d in await self._deliveries(alert):
|
|
try:
|
|
await self._call(
|
|
lambda d=d: self._bot.edit_message_text(
|
|
text, chat_id=d.chat_id, message_id=d.message_id, parse_mode=ParseMode.HTML, reply_markup=kb
|
|
)
|
|
)
|
|
except TelegramBadRequest as e: # deleted by the user, too old, not modified
|
|
log.info("alert %s edit in %s: %s", alert.id, d.chat_id, e)
|
|
except TelegramAPIError as e:
|
|
log.warning("alert %s edit in %s: %s", alert.id, d.chat_id, e)
|
|
await asyncio.sleep(PACE)
|
|
|
|
async def resolved(self, alert: Alert) -> None:
|
|
t = self._texts.alerts
|
|
took = (as_utc(alert.resolved_at or utcnow()) - as_utc(alert.opened_at)).total_seconds()
|
|
text = t.resolved.format(
|
|
title=title(self._texts, alert),
|
|
service=service(self._texts, alert),
|
|
duration=duration(self._texts.status, took),
|
|
)
|
|
for d in await self._deliveries(alert):
|
|
try:
|
|
await self._call(
|
|
lambda d=d: self._bot.send_message(
|
|
d.chat_id,
|
|
text,
|
|
parse_mode=ParseMode.HTML,
|
|
reply_parameters=ReplyParameters(message_id=d.message_id, allow_sending_without_reply=True),
|
|
)
|
|
)
|
|
except TelegramAPIError as e:
|
|
log.warning("alert %s recovery to %s: %s", alert.id, d.chat_id, e)
|
|
await asyncio.sleep(PACE)
|