Логирование действий бота в ClickHouse

- middleware пишет события message и command по основному чату, текст сообщений не сохраняется
- сервисы экономики, магазина, подписок и участников пишут события своих операций
- bot.purification возвращает уровень до и после, чтобы отличать повышение уровня
- добавлены события bot_start, bot_stop, access_denied и error
- аналитика инициализируется и закрывается вместе с ботом, её сбой не влияет на работу
- версия 0.3.2
This commit is contained in:
MpakobecEx 2026-10-16 15:24:24 +05:00
parent f85f876a7f
commit 6ec8de88b4
19 changed files with 472 additions and 15 deletions

View File

@ -73,7 +73,7 @@ pytest
## Версионирование
- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.3.1`.
- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.3.2`.
- Схема — SemVer-подобная (`MAJOR.MINOR.PATCH`); git-тег = `v` + версия (`v0.1`).
- Docker-образы тегируются версией (Блок 2), события ClickHouse получают `bot_version` (Блок 3).
- Выпуск: изменить `common/version.py` → коммит → аннотированный тег `vX.Y` → `git push --tags`.

View File

@ -8,7 +8,9 @@ import logging
from bot.bot_factory import create_bot, create_dispatcher
from bot.services.heartbeat import run_heartbeat
from bot.services.notifications import run_notifications
from bot.services.runtime import uptime_seconds
from common import __version__
from common.analytics import EventType, close_analytics, init_analytics, record
from common.config import get_settings
from common.db.pool import close_pool, init_pool
from common.logging_config import setup_logging
@ -22,6 +24,8 @@ async def main() -> None:
logger.info("Запуск DispelEvilBot v%s (APP_ENV=%s)", __version__, settings.app_env)
await init_pool()
await init_analytics()
record(EventType.BOT_START, payload={"app_env": settings.app_env})
bot = create_bot()
dp = create_dispatcher()
@ -35,6 +39,8 @@ async def main() -> None:
notifications_task.cancel()
heartbeat_task.cancel()
await asyncio.gather(notifications_task, heartbeat_task, return_exceptions=True)
record(EventType.BOT_STOP, payload={"app_env": settings.app_env, "uptime_seconds": uptime_seconds()})
await close_analytics()
await close_pool()
await bot.session.close()

View File

@ -27,8 +27,11 @@ def create_dispatcher() -> Dispatcher:
"""
from bot.handlers import commands, economy, member_events, shop
from bot.handlers.errors import register_errors
from bot.middlewares import AnalyticsMiddleware
dp = Dispatcher(storage=MemoryStorage())
# Внешний middleware: ловит все сообщения до фильтров (события `message`/`command`).
dp.message.outer_middleware(AnalyticsMiddleware())
dp.include_router(member_events.router)
dp.include_router(commands.router)
dp.include_router(shop.router)

View File

@ -7,8 +7,13 @@ import logging
from aiogram import Dispatcher
from aiogram.types import ErrorEvent
from common.analytics import EventType, record
logger = logging.getLogger(__name__)
#: Максимальная длина текста ошибки в payload события.
ERROR_TEXT_MAX = 500
def register_errors(dp: Dispatcher) -> None:
"""Зарегистрировать обработчик, чтобы одна ошибка не роняла поллинг."""
@ -20,4 +25,15 @@ def register_errors(dp: Dispatcher) -> None:
"Необработанная ошибка при обработке апдейта",
exc_info=(type(exc), exc, exc.__traceback__),
)
user = getattr(event.update, "event_from_user", None)
record(
EventType.ERROR,
user_id=getattr(user, "id", 0) or 0,
username=getattr(user, "username", None),
payload={
"error_type": type(exc).__name__,
"error": str(exc)[:ERROR_TEXT_MAX],
"update_type": str(getattr(event.update, "event_type", "") or ""),
},
)
return True

View File

@ -10,6 +10,7 @@ from aiogram.types import ChatMemberUpdated, FSInputFile
from aiogram.utils.keyboard import InlineKeyboardBuilder
from bot.services import users as users_service
from common.analytics import EventType, record
from common.config import get_settings
from common.texts import system, welcome
from common.texts.render import render
@ -64,6 +65,13 @@ async def goodbye_handler(event: ChatMemberUpdated) -> None:
if user is None or user.is_bot:
return
record(
EventType.LEAVE,
user_id=user.id,
username=user.username,
chat_id=event.chat.id,
payload={"chat_title": event.chat.title or ""},
)
await users_service.delete_user(user.id)
logger.info("Удалён пользователь %s после выхода из чата", user.id)

View File

@ -0,0 +1,7 @@
"""Middleware бота (транспортный слой)."""
from __future__ import annotations
from bot.middlewares.analytics import AnalyticsMiddleware
__all__ = ["AnalyticsMiddleware"]

View File

@ -0,0 +1,66 @@
"""Middleware аналитики: события ``message`` и ``command`` приходят с транспорта."""
from __future__ import annotations
import logging
from collections.abc import Awaitable, Callable
from typing import Any
from aiogram import BaseMiddleware
from aiogram.types import Message, TelegramObject
from common.analytics import EventType, record, remember_user
from common.analytics.messages import describe_command, describe_message
from common.config import get_settings
logger = logging.getLogger(__name__)
class AnalyticsMiddleware(BaseMiddleware):
"""Логирует сообщения и команды основного чата.
Текст сообщения не сохраняется — только метаданные (см. ``docs/events.md``).
Аналитика не должна мешать обработке: любые сбои логируются и глушатся.
"""
async def __call__(
self,
handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
event: TelegramObject,
data: dict[str, Any],
) -> Any:
if isinstance(event, Message):
self._log_message(event)
return await handler(event, data)
@staticmethod
def _log_message(message: Message) -> None:
user = message.from_user
if user is None or user.is_bot:
return
if message.chat.id != get_settings().chat_id:
return
try:
remember_user(user.id, user.username)
record(
EventType.MESSAGE,
user_id=user.id,
username=user.username,
chat_id=message.chat.id,
payload=describe_message(message),
)
command = describe_command(message)
if command is not None:
record(
EventType.COMMAND,
user_id=user.id,
username=user.username,
chat_id=message.chat.id,
payload=command,
)
except Exception: # аналитика не должна ломать обработку сообщения
logger.exception("Не удалось записать события сообщения в аналитику")
__all__ = ["AnalyticsMiddleware"]

View File

@ -10,6 +10,7 @@ from aiogram.enums import ChatMemberStatus
from aiogram.types import Message
from bot.services import users as users_service
from common.analytics import EventType, record, remember_user
from common.config import get_settings
logger = logging.getLogger(__name__)
@ -48,8 +49,16 @@ async def check_user(message: Message) -> bool:
if user.id in settings.admin_ids or await _is_member(message.bot, user.id):
await users_service.register_from_message(message)
remember_user(user.id, user.username)
return True
record(
EventType.ACCESS_DENIED,
user_id=user.id,
username=user.username,
chat_id=message.chat.id,
payload={"reason": "not_member"},
)
await users_service.delete_user(user.id)
logger.warning("Пользователь %s стучится в чат, доступ запрещён", user.id)
await _notify_denied(message)

View File

@ -2,42 +2,84 @@
from __future__ import annotations
from common.analytics import EventType, record
from common.models import EggsResult, PurificationResult
from common.repositories import economy_repo
#: id действия для начисления монеток.
ACTION_COIN = 1
#: id действия для награды за очищение.
ACTION_PURIFICATION = 4
#: id действия для обнуления баланса при удалении пользователя.
ACTION_USER_DELETED = 5
async def purify(id_user: int) -> PurificationResult:
"""Выполнить ежедневное очищение."""
"""Выполнить ежедневное очищение и записать события аналитики."""
row = await economy_repo.purification(id_user)
if not row:
return PurificationResult(success=False)
return PurificationResult(
result = PurificationResult(
success=int(row.get("res") or 0) != 0,
exp_add=int(row.get("expr_add") or 0),
money_add=int(row.get("mny_add") or 0),
hours=int(row.get("h") or 0),
minutes=int(row.get("m") or 0),
seconds=int(row.get("s") or 0),
level_before=int(row.get("lvl_before") or 0),
level=int(row.get("lvl_after") or 0),
streak=int(row.get("streak") or 0),
)
record(
EventType.PURIFICATION,
user_id=id_user,
payload={"success": result.success, "exp_add": result.exp_add, "money_add": result.money_add},
)
if not result.success:
return result
if result.exp_add:
record(EventType.EXP_ACCRUAL, user_id=id_user, payload={"amount": result.exp_add, "streak": result.streak})
if result.level_up:
record(
EventType.LEVEL_UP,
user_id=id_user,
payload={"level_from": result.level_before, "level_to": result.level},
)
if result.money_add:
record(
EventType.MONEY_ACCRUAL,
user_id=id_user,
payload={"amount": result.money_add, "reason": "purification", "action_id": ACTION_PURIFICATION},
)
return result
async def collect_eggs(id_user: int) -> EggsResult:
"""Получить ежедневные яички."""
row = await economy_repo.add_eggs(id_user)
if not row:
return EggsResult(success=False)
return EggsResult(
result = EggsResult(
success=int(row.get("res") or 0) != 0,
eggs=int(row.get("eggs_add") or 0),
hours=int(row.get("h") or 0),
minutes=int(row.get("m") or 0),
seconds=int(row.get("s") or 0),
)
if result.success:
record(EventType.EGGS, user_id=id_user, payload={"amount": result.eggs})
return result
async def add_coins(id_user: int, amount: int) -> None:
"""Начислить монеты."""
"""Начислить монеты (монетка за активность)."""
await economy_repo.money_action(id_user, amount, ACTION_COIN)
record(
EventType.MONEY_ACCRUAL,
user_id=id_user,
payload={"amount": amount, "reason": "coin", "action_id": ACTION_COIN},
)

20
bot/services/runtime.py Normal file
View File

@ -0,0 +1,20 @@
"""Общие runtime-данные процесса бота.
Используется для событий аналитики (``bot_start``/``bot_stop``/``bot_alive``)
и ежечасного пинга «бот живой» в чат логов.
"""
from __future__ import annotations
import time
#: Момент запуска процесса (монотонные секунды).
STARTED_AT = time.monotonic()
def uptime_seconds() -> int:
"""Сколько секунд работает процесс бота."""
return int(time.monotonic() - STARTED_AT)
__all__ = ["STARTED_AT", "uptime_seconds"]

View File

@ -2,6 +2,7 @@
from __future__ import annotations
from common.analytics import EventType, record
from common.models import Donat, Title
from common.repositories import shop_repo
@ -18,12 +19,48 @@ async def list_donats() -> list[Donat]:
async def buy_title(id_user: int, id_title: int) -> bool:
"""Купить звание. Возвращает True при успехе."""
return await shop_repo.buy_title(id_user, id_title) == 1
price = await _title_price(id_title)
success = await shop_repo.buy_title(id_user, id_title) == 1
record(
EventType.PURCHASE_TITLE,
user_id=id_user,
payload={"title_id": id_title, "price": price, "success": success},
)
if success:
record(
EventType.MONEY_SPEND,
user_id=id_user,
payload={"amount": price, "reason": "title", "product_id": id_title},
)
return success
async def buy_donat(id_user: int, id_donat: int) -> str | None:
"""Купить донат. Возвращает промокод или None при неуспехе."""
price = await _donat_price(id_donat)
result = await shop_repo.buy_donat(id_user, id_donat)
if result == "-1":
return None
return result
promo = None if result == "-1" else result
record(
EventType.PURCHASE_DONAT,
user_id=id_user,
payload={"donat_id": id_donat, "price": price, "success": promo is not None},
)
if promo is not None:
record(
EventType.MONEY_SPEND,
user_id=id_user,
payload={"amount": price, "reason": "donat", "product_id": id_donat},
)
return promo
async def _title_price(id_title: int) -> int:
"""Цена звания для события покупки (0, если позиция недоступна)."""
return next((title.price for title in await list_titles() if title.id == id_title), 0)
async def _donat_price(id_donat: int) -> int:
"""Цена доната для события покупки (0, если позиция недоступна)."""
return next((donat.price for donat in await list_donats() if donat.id == id_donat), 0)

View File

@ -2,12 +2,15 @@
from __future__ import annotations
from common.analytics import EventType, record
from common.repositories import subscriptions_repo
async def subscribe(id_user: int) -> None:
await subscriptions_repo.subscribe(id_user)
record(EventType.SUBSCRIBE, user_id=id_user)
async def unsubscribe(id_user: int) -> None:
await subscriptions_repo.unsubscribe(id_user)
record(EventType.UNSUBSCRIBE, user_id=id_user)

View File

@ -4,8 +4,10 @@ from __future__ import annotations
from aiogram.types import ChatMemberUpdated, Message, User
from bot.services.economy import ACTION_USER_DELETED
from common.analytics import EventType, record
from common.models import NickEntry, UserInfo
from common.repositories import users_repo
from common.repositories import economy_repo, users_repo
def _user_fields(user: User) -> tuple[int, str, str, str]:
@ -31,14 +33,34 @@ async def register_from_event(event: ChatMemberUpdated) -> None:
if user.is_bot:
return
await users_repo.add_user(*_user_fields(user))
record(
EventType.JOIN,
user_id=user.id,
username=user.username,
chat_id=event.chat.id,
payload={"chat_title": event.chat.title or ""},
)
async def set_nick(id_user: int, nick: str) -> None:
await users_repo.add_nick(id_user, nick)
record(EventType.NICK_SET, user_id=id_user, payload={"nick": nick})
async def delete_user(id_user: int) -> None:
"""Удалить пользователя и зафиксировать обнуление баланса.
Баланс читается **до** удаления: ``bot.del_user_by_id`` обнуляет его записью
в журнал ``money``.
"""
balance = await economy_repo.get_money(id_user)
await users_repo.del_user_by_id(id_user)
if balance:
record(
EventType.MONEY_SPEND,
user_id=id_user,
payload={"amount": abs(balance), "reason": "user_deleted", "action_id": ACTION_USER_DELETED},
)
async def list_nicks() -> list[NickEntry]:

View File

@ -0,0 +1,54 @@
"""Метаданные сообщения для событий аналитики.
Модуль намеренно **не зависит от aiogram**: работает с любым объектом, у которого
есть нужные атрибуты (``Message`` в боте или заглушка в тестах). Текст сообщения
не сохраняется — только метаданные (см. ``docs/events.md``).
"""
from __future__ import annotations
from typing import Any
#: Максимальная длина имени команды в payload.
COMMAND_MAX_LENGTH = 64
#: Префикс, разделяющий команду и бота в групповом чате (``/команда@bot``).
BOT_MENTION_SEPARATOR = "@"
def _text(message: Any) -> str:
"""Текст сообщения (или подпись) без лишних пробелов."""
text = getattr(message, "text", None) or getattr(message, "caption", None) or ""
return str(text).strip()
def _enum_value(value: Any) -> str:
"""Строковое значение enum-полей aiogram (``ChatType.PRIVATE`` -> ``private``)."""
return str(getattr(value, "value", value) or "")
def describe_message(message: Any) -> dict[str, Any]:
"""Метаданные сообщения для события ``message`` (без текста сообщения)."""
text = _text(message)
chat = getattr(message, "chat", None)
return {
"message_id": int(getattr(message, "message_id", 0) or 0),
"chat_type": _enum_value(getattr(chat, "type", "")) if chat is not None else "",
"content_type": _enum_value(getattr(message, "content_type", "")),
"text_len": len(text),
"is_command": text.startswith("/"),
}
def describe_command(message: Any) -> dict[str, Any] | None:
"""Метаданные команды для события ``command`` (``None``, если это не команда)."""
text = _text(message)
if not text.startswith("/"):
return None
parts = text.split()
name = parts[0].split(BOT_MENTION_SEPARATOR, 1)[0].lstrip("/").lower()[:COMMAND_MAX_LENGTH]
return {"command": name, "args_count": max(len(parts) - 1, 0)}
__all__ = ["describe_command", "describe_message"]

View File

@ -15,6 +15,16 @@ class PurificationResult:
hours: int = 0
minutes: int = 0
seconds: int = 0
#: Уровень до и после очищения (для события `level_up`).
level_before: int = 0
level: int = 0
#: Серия ежедневных очищений (для события `exp_accrual`).
streak: int = 0
@property
def level_up(self) -> bool:
"""Повысился ли уровень по итогам очищения."""
return self.level > self.level_before
@dataclass(slots=True)

View File

@ -7,4 +7,4 @@
from __future__ import annotations
__version__ = "0.3.1"
__version__ = "0.3.2"

View File

@ -35,12 +35,28 @@ BEGIN
END;
$$;
-- Ежедневное очищение: опыт, монеты, уровень.
-- Колонки lvl_before/lvl/streak нужны аналитике (события exp_accrual и level_up),
-- поэтому сигнатуру пришлось изменить — сначала DROP (repeatable-миграция).
DROP FUNCTION IF EXISTS bot.purification(bigint);
CREATE OR REPLACE FUNCTION bot.purification(p_id_user bigint)
RETURNS TABLE(res integer, expr_add bigint, mny_add bigint, h integer, m integer, s integer)
RETURNS TABLE(
res integer,
expr_add bigint,
mny_add bigint,
h integer,
m integer,
s integer,
lvl_before bigint,
lvl_after bigint,
streak bigint
)
LANGUAGE plpgsql
AS $$
DECLARE
v_lvl bigint;
v_lvl_before bigint;
v_expr bigint;
v_expr_add bigint;
v_streak bigint;
@ -57,6 +73,8 @@ BEGIN
FROM bot.de_users
WHERE id_user = p_id_user;
v_lvl_before := v_lvl;
BEGIN
SELECT dt INTO v_dt_pur
FROM (
@ -113,7 +131,7 @@ BEGIN
v_result := 1;
END IF;
RETURN QUERY SELECT v_result, v_expr_add, v_mny_add, v_h, v_m, v_s;
RETURN QUERY SELECT v_result, v_expr_add, v_mny_add, v_h, v_m, v_s, v_lvl_before, v_lvl, v_streak;
END;
$$;

View File

@ -6,16 +6,19 @@
from __future__ import annotations
import json
import logging
import re
from datetime import datetime
from datetime import UTC, datetime
from pathlib import Path
from types import SimpleNamespace
from typing import Any
import pytest
from common.analytics import ALL_EVENT_TYPES, EventType, close_analytics, get_analytics, init_analytics, record
from common.analytics.client import PAYLOAD_MAX_LENGTH, AnalyticsClient
from common.analytics.messages import describe_command, describe_message
from common.config import Settings
from common.version import __version__
@ -25,6 +28,13 @@ _DOCS_PATH = Path(__file__).resolve().parents[1] / "docs" / "events.md"
_EVENT_ROW = re.compile(r"^\|\s*`([a-z_]+)`\s*\|", re.MULTILINE)
class _FakeEnum:
"""Заглушка enum-полей aiogram (у значений есть ``.value``)."""
def __init__(self, value: str) -> None:
self.value = value
class FakeClickHouse:
"""Заглушка клиента ClickHouse: пишет в список или падает."""
@ -76,6 +86,56 @@ def test_event_types_match_documentation() -> None:
assert documented == set(ALL_EVENT_TYPES), "Список событий в common/analytics/events.py и docs/events.md разошёлся"
# --- метаданные сообщения (без aiogram) ---
def _fake_message(**overrides: Any) -> Any:
"""Минимальная заглушка aiogram Message для describe_message/describe_command."""
fields: dict[str, Any] = {
"message_id": 77,
"text": "привет",
"caption": None,
"content_type": _FakeEnum("text"),
"chat": SimpleNamespace(type=_FakeEnum("supergroup")),
}
fields.update(overrides)
return SimpleNamespace(**fields)
def test_describe_message_without_text() -> None:
payload = describe_message(_fake_message(text="привет"))
assert payload == {
"message_id": 77,
"chat_type": "supergroup",
"content_type": "text",
"text_len": 6,
"is_command": False,
}
def test_describe_message_uses_caption_for_media() -> None:
payload = describe_message(_fake_message(text=None, caption="подпись", content_type=_FakeEnum("photo")))
assert payload["content_type"] == "photo"
assert payload["text_len"] == len("подпись")
assert payload["is_command"] is False
def test_describe_command_bare_and_with_args() -> None:
assert describe_command(_fake_message(text="/очищение")) == {"command": "очищение", "args_count": 0}
assert describe_command(_fake_message(text="/ник Вася")) == {"command": "ник", "args_count": 1}
def test_describe_command_strips_bot_mention() -> None:
assert describe_command(_fake_message(text="/Ник@DispelEvilBot Вася")) == {"command": "ник", "args_count": 1}
def test_describe_command_returns_none_for_plain_message() -> None:
assert describe_command(_fake_message(text="просто текст")) is None
assert describe_command(_fake_message(text=None, caption="подпись")) is None
# --- буферизация и запись ---
@ -213,3 +273,79 @@ async def test_init_and_close_flush_pending_events() -> None:
assert get_analytics() is None
assert len(fake.rows) == 2
assert fake.closed is True
# --- middleware (транспортные события) ---
def _message(**overrides: Any) -> Any:
"""Реальное сообщение aiogram для проверки middleware."""
from aiogram.types import Chat, Message, User
fields: dict[str, Any] = {
"message_id": 5,
"date": datetime.now(UTC),
"chat": Chat(id=-100, type="supergroup", title="Клан"),
"from_user": User(id=7, is_bot=False, first_name="Вася", username="vasya"),
"text": "/инфо",
}
fields.update(overrides)
return Message(**fields)
async def _handler(event: Any, data: dict[str, Any]) -> str:
return "handled"
async def test_middleware_records_message_and_command(monkeypatch: pytest.MonkeyPatch) -> None:
from bot.middlewares.analytics import AnalyticsMiddleware
settings = _settings(chat_id=-100)
monkeypatch.setattr("bot.middlewares.analytics.get_settings", lambda: settings)
fake = FakeClickHouse()
client = await init_analytics(settings, client=fake)
assert await AnalyticsMiddleware()(_handler, _message(), {}) == "handled"
await client.flush()
assert [row[1] for row in fake.rows] == ["message", "command"]
assert json.loads(fake.rows[0][5])["is_command"] is True
assert json.loads(fake.rows[1][5]) == {"command": "инфо", "args_count": 0}
assert fake.rows[0][3] == "vasya"
async def test_middleware_skips_other_chats_and_bots(monkeypatch: pytest.MonkeyPatch) -> None:
from bot.middlewares.analytics import AnalyticsMiddleware
settings = _settings(chat_id=-100)
monkeypatch.setattr("bot.middlewares.analytics.get_settings", lambda: settings)
fake = FakeClickHouse()
client = await init_analytics(settings, client=fake)
from aiogram.types import Chat, User
await AnalyticsMiddleware()(_handler, _message(chat=Chat(id=-999, type="supergroup")), {})
await AnalyticsMiddleware()(
_handler,
_message(from_user=User(id=7, is_bot=True, first_name="Bot")),
{},
)
await client.flush()
assert fake.rows == []
async def test_middleware_never_breaks_handling(monkeypatch: pytest.MonkeyPatch) -> None:
from bot.middlewares.analytics import AnalyticsMiddleware
settings = _settings(chat_id=-100)
monkeypatch.setattr("bot.middlewares.analytics.get_settings", lambda: settings)
def boom(*args: Any, **kwargs: Any) -> None:
raise RuntimeError("аналитика сломалась")
monkeypatch.setattr("bot.middlewares.analytics.record", boom)
assert await AnalyticsMiddleware()(_handler, _message(), {}) == "handled"

View File

@ -7,4 +7,4 @@ from common import __version__
def test_version_is_set():
assert __version__
assert __version__ == "0.3.1"
assert __version__ == "0.3.2"