Compare commits
No commits in common. "a0fa223eaafed93e2869a8f6fd75f7e327fff1ac" and "aafb9c9e3b1e74e54c0cfe1441fe92a7fa3f8de3" have entirely different histories.
a0fa223eaa
...
aafb9c9e3b
14
.env.example
14
.env.example
|
|
@ -28,17 +28,3 @@ ADMIN_IDS=
|
|||
|
||||
# Часовой пояс для расписаний
|
||||
TIMEZONE=Europe/Moscow
|
||||
|
||||
# Аналитика (ClickHouse). Логируются все действия бота (см. docs/events.md).
|
||||
# CLICKHOUSE_ENABLED=false — полностью выключить аналитику.
|
||||
CLICKHOUSE_ENABLED=true
|
||||
# Для локального запуска — localhost; в compose бот ходит по имени сервиса.
|
||||
CLICKHOUSE_HOST=localhost
|
||||
CLICKHOUSE_PORT=8123
|
||||
CLICKHOUSE_DB=analytics
|
||||
CLICKHOUSE_USER=default
|
||||
CLICKHOUSE_PASSWORD=
|
||||
# Батчинг записи событий: размер батча, интервал сброса (сек), предел буфера.
|
||||
ANALYTICS_BATCH_SIZE=200
|
||||
ANALYTICS_FLUSH_INTERVAL=5.0
|
||||
ANALYTICS_BUFFER_MAX=5000
|
||||
|
|
|
|||
71
DEPLOY.md
71
DEPLOY.md
|
|
@ -15,13 +15,9 @@ cd DispelEvilBot
|
|||
|
||||
```bash
|
||||
cp env/test.env.example env/test.env
|
||||
# заполнить: BOT_TOKEN, DB_PASSWORD, CHAT_ID (основной чат), ALERT_CHAT_ID (опц.),
|
||||
# ADMIN_IDS (опц.), CLICKHOUSE_PASSWORD (аналитика событий)
|
||||
# заполнить: BOT_TOKEN, DB_PASSWORD, CHAT_ID (основной чат), ALERT_CHAT_ID (опц.), ADMIN_IDS (опц.)
|
||||
```
|
||||
|
||||
Порт ClickHouse на хосте (`CLICKHOUSE_HTTP_PORT`) у контуров должен различаться:
|
||||
`prod` — `8123`, `test` — `8124` (иначе контейнеры конфликтуют на одной VPS).
|
||||
|
||||
## 3. Логин в registry (если пакет приватный)
|
||||
|
||||
```bash
|
||||
|
|
@ -50,62 +46,7 @@ docker compose -p test --env-file env/test.env -f docker-compose.prod.yml logs -
|
|||
|
||||
1. `postgres` поднимается и проходит healthcheck;
|
||||
2. `migrate` (**Flyway**) накатывает миграции из `sql/` — для пустой БД с нуля;
|
||||
3. `clickhouse` поднимается и применяет `clickhouse/init/001_events.sql` —
|
||||
**только при первой инициализации** пустого data-каталога;
|
||||
4. `bot` стартует (`Запуск DispelEvilBot v<версия> (APP_ENV=...)`), аналитика и рассылки — в нём же.
|
||||
|
||||
Проверка контура (сервисы, миграции, Postgres, ClickHouse):
|
||||
|
||||
```bash
|
||||
make check E=test
|
||||
```
|
||||
|
||||
## ClickHouse (аналитика событий)
|
||||
|
||||
События всех действий бота пишутся в `analytics.events` (контракт — [`docs/events.md`](docs/events.md)).
|
||||
ClickHouse наружу **не публикуется**: только loopback хоста (`127.0.0.1:${CLICKHOUSE_HTTP_PORT}`).
|
||||
|
||||
```bash
|
||||
make ch E=test # clickhouse-client контура
|
||||
# пример: события за сегодня по типам
|
||||
docker compose -p test --env-file env/test.env -f docker-compose.prod.yml exec -T clickhouse \
|
||||
sh -lc 'clickhouse-client --user "$CLICKHOUSE_USER" --password "$CLICKHOUSE_PASSWORD" \
|
||||
--query "SELECT event_type, count() FROM analytics.events WHERE toDate(event_time) = today() GROUP BY event_type"'
|
||||
```
|
||||
|
||||
Если ClickHouse недоступен, бот продолжает работать: события уходят в stdout (WARNING),
|
||||
полностью выключить можно флагом `CLICKHOUSE_ENABLED=false`.
|
||||
|
||||
Изменения схемы: `clickhouse/init/*.sql` применяются лишь на пустом data-каталоге,
|
||||
поэтому на существующем контуре DDL нужно применить вручную (`make ch E=...`).
|
||||
|
||||
## Бэкап событий на домашний сервер
|
||||
|
||||
Сырые события живут на VPS ограниченное время: раз в сутки вчерашняя партиция
|
||||
выгружается в `Parquet.gz`, передаётся на домашний сервер, импортируется в тамошний
|
||||
ClickHouse и сверяется по количеству строк — **только после сверки** партиции старше
|
||||
`BACKUP_GRACE_DAYS` (по умолчанию 7) удаляются на VPS. На VPS остаются агрегаты
|
||||
(`events_daily`, `money_daily`).
|
||||
|
||||
Разовая настройка:
|
||||
|
||||
```bash
|
||||
cp env/backup.env.example env/backup.env # BACKUP_REMOTE, BACKUP_DIR, BACKUP_SSH_KEY, ...
|
||||
# на домашнем сервере (в каталоге, куда кладутся архивы) разместить:
|
||||
# scripts/clickhouse_import.sh
|
||||
# clickhouse/init/001_events.sql -> как 001_events.sql рядом со скриптом
|
||||
```
|
||||
|
||||
Проверочный прогон (без cron): `make backup E=test`.
|
||||
|
||||
Cron на VPS (вчерашний день уже закрыт, время московское):
|
||||
|
||||
```cron
|
||||
0 4 * * * cd /opt/DispelEvilBot && ./scripts/clickhouse_backup.sh test >> /var/log/clickhouse-backup-test.log 2>&1
|
||||
```
|
||||
|
||||
Ошибка на любом шаге оставляет данные на VPS (партиции не удаляются) и пишет понятный лог.
|
||||
Импорт дня идемпотентен: повторный прогон не дублирует строки.
|
||||
3. `bot` стартует (`Запуск DispelEvilBot v<версия> (APP_ENV=...)`).
|
||||
|
||||
## Обновление и откат
|
||||
|
||||
|
|
@ -126,13 +67,7 @@ docker compose -p <контур> --env-file env/<контур>.env -f docker-com
|
|||
|
||||
## Переменные окружения
|
||||
|
||||
См. `env/prod.env.example` / `env/test.env.example` (контуры) и `env/backup.env.example` (бэкап событий).
|
||||
|
||||
## Дашборд (позже)
|
||||
|
||||
Superset поднимается на домашнем сервере и читает **локальный** ClickHouse с полной
|
||||
историей событий (сюда ежедневно приезжают архивы). Отдельный read-only пользователь
|
||||
ClickHouse и сам дашборд — финальный этап, в этот контур не входят.
|
||||
См. `env/prod.env.example` / `env/test.env.example`.
|
||||
|
||||
## Перенос данных на прод (позже, короткое окно простоя)
|
||||
|
||||
|
|
|
|||
10
Makefile
10
Makefile
|
|
@ -9,7 +9,7 @@ COMPOSE := docker compose -p $(E) --env-file $(FILE) -f docker-compose.prod.yml
|
|||
.DEFAULT_GOAL := help
|
||||
.RECIPEPREFIX := >
|
||||
|
||||
.PHONY: help up pull down restart logs ps images check psql ch backup
|
||||
.PHONY: help up pull down restart logs ps images check psql
|
||||
|
||||
help:
|
||||
> @echo "Контур задаётся через E=test|prod (по умолчанию test):"
|
||||
|
|
@ -22,8 +22,6 @@ help:
|
|||
> @echo " make images E=test образы контура"
|
||||
> @echo " make check E=test проверка контура (скрипт)"
|
||||
> @echo " make psql E=test psql в БД контура"
|
||||
> @echo " make ch E=test clickhouse-client контура"
|
||||
> @echo " make backup E=test бэкап вчерашних событий на домашний сервер"
|
||||
|
||||
up:
|
||||
> $(COMPOSE) up -d
|
||||
|
|
@ -51,9 +49,3 @@ check:
|
|||
|
||||
psql:
|
||||
> $(COMPOSE) exec postgres sh -lc 'psql -U "$$POSTGRES_USER" -d "$$POSTGRES_DB"'
|
||||
|
||||
ch:
|
||||
> $(COMPOSE) exec clickhouse sh -lc 'clickhouse-client --user "$$CLICKHOUSE_USER" --password "$$CLICKHOUSE_PASSWORD" --database analytics'
|
||||
|
||||
backup:
|
||||
> ./scripts/clickhouse_backup.sh $(E)
|
||||
|
|
|
|||
49
README.md
49
README.md
|
|
@ -6,12 +6,9 @@ Telegram-бот клана Dispel Evil (aiogram 3 + Postgres). Монорепо
|
|||
## Структура
|
||||
|
||||
```
|
||||
common/ # общий пакет: config, logging, db, repositories, models, texts, analytics
|
||||
bot/ # сервис бота: handlers, services, keyboards, middlewares, notifications
|
||||
common/ # общий пакет: config, logging, db, repositories, models, texts
|
||||
bot/ # сервис бота: handlers, services, keyboards, notifications
|
||||
sql/ # миграции Flyway (V__*.sql, R__*.sql)
|
||||
clickhouse/ # init-скрипты ClickHouse (схема аналитики)
|
||||
docs/ # документация (docs/events.md — события аналитики)
|
||||
scripts/ # эксплуатационные скрипты (проверка контура, бэкап событий)
|
||||
images/ # статичные картинки (приветствие)
|
||||
deploy/ # материалы деплоя (Caddyfile, инструкции)
|
||||
tests/ # тесты
|
||||
|
|
@ -24,7 +21,6 @@ docker-compose.yml / docker-compose.prod.yml
|
|||
|
||||
- Python 3.12
|
||||
- PostgreSQL 14+
|
||||
- ClickHouse 25.3 (аналитика событий; можно выключить — `CLICKHOUSE_ENABLED=false`)
|
||||
|
||||
## Настройка
|
||||
|
||||
|
|
@ -48,25 +44,9 @@ APP_ENV=prod python -m bot # prod
|
|||
Схема управляется миграциями **Flyway** в `sql/` (`V__*.sql` — versioned, `R__*.sql` — repeatable).
|
||||
При Docker-развёртывании накат выполняется автоматически сервисом `migrate` перед ботом.
|
||||
|
||||
## Аналитика (ClickHouse)
|
||||
|
||||
Все значимые действия бота пишутся в ClickHouse в единую таблицу `analytics.events`.
|
||||
Контракт событий — [`docs/events.md`](docs/events.md) (типы, payload, примеры запросов).
|
||||
|
||||
- схема — `clickhouse/init/001_events.sql` (применяется при первой инициализации контейнера);
|
||||
- клиент и батчинг — `common/analytics/` (запись неблокирующая, сбой ClickHouse не влияет на бота);
|
||||
- сырые события живут на VPS 7 дней и ежедневно уезжают архивом на домашний сервер,
|
||||
на VPS остаются агрегаты (`events_daily`, `money_daily`) — см. [DEPLOY.md](DEPLOY.md);
|
||||
- выключается флагом `CLICKHOUSE_ENABLED=false`.
|
||||
|
||||
## Уведомления в чат логов
|
||||
|
||||
В `ALERT_CHAT_ID` (если задан) уходят: вход и выход участников, успешные очищения и
|
||||
ежечасный пинг «бот живой». Отправка не критична: ошибки только пишутся в лог.
|
||||
|
||||
## Docker
|
||||
|
||||
Локальный запуск (postgres + migrate + clickhouse + bot):
|
||||
Локальный запуск (postgres + migrate + bot):
|
||||
|
||||
```bash
|
||||
cp .env.example .env # заполнить
|
||||
|
|
@ -91,20 +71,9 @@ black --check .
|
|||
pytest
|
||||
```
|
||||
|
||||
## Бэкап событий
|
||||
|
||||
Сырые события ClickHouse ежедневно выгружаются архивом на домашний сервер (по cron),
|
||||
на VPS остаются только агрегаты и последние дни сырых данных:
|
||||
|
||||
```bash
|
||||
make backup E=test # ./scripts/clickhouse_backup.sh test
|
||||
```
|
||||
|
||||
Настройка и расписание — см. [`DEPLOY.md`](DEPLOY.md).
|
||||
|
||||
## Версионирование
|
||||
|
||||
- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.3.6`.
|
||||
- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.2.5`.
|
||||
- Схема — SemVer-подобная (`MAJOR.MINOR.PATCH`); git-тег = `v` + версия (`v0.1`).
|
||||
- Docker-образы тегируются версией (Блок 2), события ClickHouse получают `bot_version` (Блок 3).
|
||||
- Выпуск: изменить `common/version.py` → коммит → аннотированный тег `vX.Y` → `git push --tags`.
|
||||
|
|
@ -121,14 +90,14 @@ make backup E=test # ./scripts/clickhouse_backup.sh test
|
|||
- Переход на «ветки + PR» и защиту ветки — когда появится второй разработчик.
|
||||
|
||||
```bash
|
||||
# обычная работа: сообщение без префиксов, по существу изменения
|
||||
# обычная работа
|
||||
git add -A
|
||||
git commit -m "Краткое описание изменения"
|
||||
git commit -m "feat: ..."
|
||||
git push origin master
|
||||
|
||||
# выпуск релиза
|
||||
# 1) поднять __version__ в common/version.py
|
||||
git commit -am "Версия X.Y"
|
||||
git commit -am "chore: bump version to X.Y"
|
||||
git tag -a vX.Y -m "Release vX.Y"
|
||||
git push origin master --tags
|
||||
```
|
||||
|
|
@ -140,6 +109,4 @@ git push origin master --tags
|
|||
|
||||
## Переменные окружения
|
||||
|
||||
- общие — `.env.example` (включая `CLICKHOUSE_*` и параметры батчинга `ANALYTICS_*`);
|
||||
- контуры — `env/prod.env.example`, `env/test.env.example`;
|
||||
- бэкап событий — `env/backup.env.example`.
|
||||
См. `.env.example`.
|
||||
|
|
|
|||
|
|
@ -8,9 +8,7 @@ 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
|
||||
|
|
@ -24,14 +22,12 @@ 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()
|
||||
|
||||
notifications_task = asyncio.create_task(run_notifications(bot))
|
||||
heartbeat_task = asyncio.create_task(run_heartbeat(bot))
|
||||
heartbeat_task = asyncio.create_task(run_heartbeat())
|
||||
try:
|
||||
await bot.delete_webhook(drop_pending_updates=True)
|
||||
await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types())
|
||||
|
|
@ -39,8 +35,6 @@ 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()
|
||||
|
||||
|
|
|
|||
|
|
@ -27,11 +27,8 @@ 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)
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ from aiogram import Router
|
|||
from aiogram.filters import Command
|
||||
from aiogram.types import Message
|
||||
|
||||
from bot.services import access, alerts
|
||||
from bot.services import access
|
||||
from bot.services import economy as economy_service
|
||||
from common.config import get_settings
|
||||
from common.texts import economy, system
|
||||
|
|
@ -37,12 +37,6 @@ async def purification_handler(message: Message) -> None:
|
|||
else:
|
||||
context = {"money_add": result.money_add, "exp_add": result.exp_add}
|
||||
await message.answer(render(economy.MSG_PURIFICATION, context))
|
||||
await alerts.notify_purification(
|
||||
message.bot,
|
||||
username=message.from_user.username,
|
||||
user_id=message.from_user.id,
|
||||
result=result,
|
||||
)
|
||||
|
||||
|
||||
@router.message(Command("яички", ignore_case=True))
|
||||
|
|
|
|||
|
|
@ -7,13 +7,8 @@ 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:
|
||||
"""Зарегистрировать обработчик, чтобы одна ошибка не роняла поллинг."""
|
||||
|
|
@ -25,15 +20,4 @@ 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
|
||||
|
|
|
|||
|
|
@ -9,9 +9,7 @@ from aiogram.filters import JOIN_TRANSITION, LEAVE_TRANSITION, ChatMemberUpdated
|
|||
from aiogram.types import ChatMemberUpdated, FSInputFile
|
||||
from aiogram.utils.keyboard import InlineKeyboardBuilder
|
||||
|
||||
from bot.services import alerts
|
||||
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
|
||||
|
|
@ -41,12 +39,6 @@ async def welcome_handler(event: ChatMemberUpdated) -> None:
|
|||
return
|
||||
|
||||
await users_service.register_from_event(event)
|
||||
await alerts.notify_join(
|
||||
event.bot,
|
||||
username=user.username,
|
||||
user_id=user.id,
|
||||
chat_title=event.chat.title or "",
|
||||
)
|
||||
|
||||
builder = InlineKeyboardBuilder()
|
||||
builder.button(text=welcome.MSG_WELCOME_BUTTON_CAPTION, url=welcome.MSG_WELCOME_BUTTON_URL)
|
||||
|
|
@ -67,23 +59,20 @@ async def welcome_handler(event: ChatMemberUpdated) -> None:
|
|||
@router.chat_member(ChatMemberUpdatedFilter(member_status_changed=LEAVE_TRANSITION))
|
||||
async def goodbye_handler(event: ChatMemberUpdated) -> None:
|
||||
"""Пользователь вышел из чата — удалить запись."""
|
||||
settings = get_settings()
|
||||
user = event.from_user
|
||||
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)
|
||||
|
||||
await alerts.notify_leave(
|
||||
event.bot,
|
||||
username=user.username,
|
||||
user_id=user.id,
|
||||
chat_title=event.chat.title or "",
|
||||
if not settings.alert_chat_id:
|
||||
return
|
||||
try:
|
||||
await event.bot.send_message(
|
||||
chat_id=settings.alert_chat_id,
|
||||
text=f"[goodbye] {user.username or user.id} покинул чат {event.chat.title}",
|
||||
)
|
||||
except Exception:
|
||||
logger.exception("Не удалось отправить уведомление о выходе пользователя")
|
||||
|
|
|
|||
|
|
@ -1,7 +0,0 @@
|
|||
"""Middleware бота (транспортный слой)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from bot.middlewares.analytics import AnalyticsMiddleware
|
||||
|
||||
__all__ = ["AnalyticsMiddleware"]
|
||||
|
|
@ -1,66 +0,0 @@
|
|||
"""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"]
|
||||
|
|
@ -9,12 +9,8 @@ from aiogram import Bot
|
|||
from aiogram.enums import ChatMemberStatus
|
||||
from aiogram.types import Message
|
||||
|
||||
from bot.services import alerts
|
||||
from bot.services import users as users_service
|
||||
from common.analytics import EventType, record, remember_user
|
||||
from common.config import get_settings
|
||||
from common.texts import alerts as alerts_text
|
||||
from common.texts.render import render
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -52,16 +48,8 @@ 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)
|
||||
|
|
@ -70,8 +58,20 @@ async def check_user(message: Message) -> bool:
|
|||
|
||||
async def _notify_denied(message: Message) -> None:
|
||||
"""Сообщить в чат логов (если задан) о попытке доступа извне."""
|
||||
user = message.from_user
|
||||
if user is None:
|
||||
settings = get_settings()
|
||||
if not settings.alert_chat_id:
|
||||
return
|
||||
text = render(alerts_text.MSG_ACCESS_DENIED, {"user": alerts.user_tag(user.username, user.id)})
|
||||
await alerts.notify_alert(message.bot, text, forward=message)
|
||||
|
||||
username = message.from_user.username if message.from_user else None
|
||||
try:
|
||||
await message.bot.send_message(
|
||||
chat_id=settings.alert_chat_id,
|
||||
text=f"[check_user]\nСтучится в чат {username}!",
|
||||
)
|
||||
await message.bot.forward_message(
|
||||
chat_id=settings.alert_chat_id,
|
||||
from_chat_id=message.chat.id,
|
||||
message_id=message.message_id,
|
||||
)
|
||||
except Exception: # уведомление не должно ломать обработку
|
||||
logger.exception("Не удалось отправить уведомление о запрете доступа")
|
||||
|
|
|
|||
|
|
@ -1,74 +0,0 @@
|
|||
"""Уведомления в чат логов (``ALERT_CHAT_ID``).
|
||||
|
||||
Единая точка отправки служебных сообщений бота: вход/выход участников,
|
||||
успешные очищения, пинг «бот живой». Уведомления **необязательные**:
|
||||
если чат не задан (``ALERT_CHAT_ID=0``) или отправка не удалась — это только
|
||||
запись в лог, обработка не прерывается.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from common.config import get_settings
|
||||
from common.texts import alerts as alerts_text
|
||||
from common.texts.render import render
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from aiogram import Bot
|
||||
from aiogram.types import Message
|
||||
|
||||
from common.models import PurificationResult
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def user_tag(username: str | None, user_id: int) -> str:
|
||||
"""Представление пользователя для служебного сообщения (``@username`` или id)."""
|
||||
return f"@{username}" if username else str(user_id)
|
||||
|
||||
|
||||
async def notify_alert(bot: Bot, text: str, *, forward: Message | None = None) -> bool:
|
||||
"""Отправить текст в чат логов (при необходимости — с пересылкой сообщения).
|
||||
|
||||
Возвращает ``False``, если чат логов не задан или отправка не удалась.
|
||||
"""
|
||||
settings = get_settings()
|
||||
if not settings.alert_chat_id:
|
||||
return False
|
||||
|
||||
try:
|
||||
await bot.send_message(chat_id=settings.alert_chat_id, text=text)
|
||||
if forward is not None:
|
||||
await bot.forward_message(
|
||||
chat_id=settings.alert_chat_id,
|
||||
from_chat_id=forward.chat.id,
|
||||
message_id=forward.message_id,
|
||||
)
|
||||
except Exception: # уведомление не должно ломать обработку
|
||||
logger.exception("Не удалось отправить уведомление в чат логов")
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
async def notify_join(bot: Bot, *, username: str | None, user_id: int, chat_title: str) -> bool:
|
||||
"""Уведомление о входе участника в чат."""
|
||||
context = {"username": user_tag(username, user_id), "chat_title": chat_title}
|
||||
return await notify_alert(bot, render(alerts_text.MSG_JOIN, context))
|
||||
|
||||
|
||||
async def notify_leave(bot: Bot, *, username: str | None, user_id: int, chat_title: str) -> bool:
|
||||
"""Уведомление о выходе участника из чата."""
|
||||
context = {"username": user_tag(username, user_id), "chat_title": chat_title}
|
||||
return await notify_alert(bot, render(alerts_text.MSG_LEAVE, context))
|
||||
|
||||
|
||||
async def notify_purification(bot: Bot, *, username: str | None, user_id: int, result: PurificationResult) -> bool:
|
||||
"""Уведомление об успешном очищении."""
|
||||
context = {
|
||||
"username": user_tag(username, user_id),
|
||||
"exp_add": result.exp_add,
|
||||
"money_add": result.money_add,
|
||||
}
|
||||
return await notify_alert(bot, render(alerts_text.MSG_PURIFICATION, context))
|
||||
|
|
@ -2,84 +2,42 @@
|
|||
|
||||
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)
|
||||
|
||||
result = PurificationResult(
|
||||
return 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)
|
||||
|
||||
result = EggsResult(
|
||||
return 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},
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,72 +1,28 @@
|
|||
"""Heartbeat для Docker healthcheck и ежечасный пинг «бот живой».
|
||||
"""Heartbeat для Docker healthcheck.
|
||||
|
||||
Бот периодически обновляет mtime файла; healthcheck считает контейнер живым,
|
||||
если файл свежий. Если event loop завис — heartbeat перестаёт обновляться.
|
||||
|
||||
Раз в час в чат логов (``ALERT_CHAT_ID``) уходит короткое сообщение о работе бота,
|
||||
и в ClickHouse пишется событие ``bot_alive``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from bot.services.alerts import notify_alert
|
||||
from bot.services.runtime import uptime_human, uptime_seconds
|
||||
from common import __version__
|
||||
from common.analytics import EventType, record
|
||||
from common.config import get_settings
|
||||
from common.texts import alerts as alerts_text
|
||||
from common.texts.render import render
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from aiogram import Bot
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
#: Как часто обновлять heartbeat-файл.
|
||||
HEARTBEAT_INTERVAL = 30
|
||||
|
||||
#: Час (в таймзоне бота), за который пинг уже отправлен.
|
||||
_last_ping_hour: datetime | None = None
|
||||
|
||||
|
||||
async def ping_hourly(bot: Bot, now: datetime | None = None) -> bool:
|
||||
"""Записать событие ``bot_alive`` и отправить пинг в чат логов раз в час.
|
||||
|
||||
Возвращает ``True``, если в этом часу пинг отправлен (или предпринята попытка).
|
||||
Повторные вызовы в течение того же часа ничего не делают.
|
||||
"""
|
||||
global _last_ping_hour
|
||||
|
||||
settings = get_settings()
|
||||
current_hour = (now or datetime.now(settings.tzinfo)).replace(minute=0, second=0, microsecond=0)
|
||||
if current_hour == _last_ping_hour:
|
||||
return False
|
||||
|
||||
_last_ping_hour = current_hour
|
||||
record(
|
||||
EventType.BOT_ALIVE,
|
||||
payload={"app_env": settings.app_env, "uptime_seconds": uptime_seconds()},
|
||||
)
|
||||
text = render(
|
||||
alerts_text.MSG_ALIVE,
|
||||
{"version": __version__, "uptime": uptime_human()},
|
||||
)
|
||||
return await notify_alert(bot, text)
|
||||
|
||||
|
||||
async def run_heartbeat(bot: Bot) -> None:
|
||||
"""Обновлять heartbeat-файл каждые ``HEARTBEAT_INTERVAL`` секунд и пинговать раз в час."""
|
||||
async def run_heartbeat() -> None:
|
||||
"""Обновлять файл-heartbeat каждые ``HEARTBEAT_INTERVAL`` секунд."""
|
||||
path = Path(get_settings().heartbeat_path)
|
||||
while True:
|
||||
try:
|
||||
path.touch()
|
||||
except OSError:
|
||||
logger.exception("Не удалось обновить heartbeat-файл %s", path)
|
||||
await ping_hourly(bot)
|
||||
await asyncio.sleep(HEARTBEAT_INTERVAL)
|
||||
|
|
|
|||
|
|
@ -61,6 +61,11 @@ async def _send_broadcast(bot: Bot, template: str, subscribers: list[dict]) -> N
|
|||
|
||||
|
||||
async def _tick(bot: Bot, now: datetime) -> None:
|
||||
settings = get_settings()
|
||||
if now.minute == 0 and settings.alert_chat_id:
|
||||
# Пинг «жив» в чат логов (если задан).
|
||||
await bot.send_message(chat_id=settings.alert_chat_id, text="Я ЖЫФ!")
|
||||
|
||||
for schedule in SCHEDULES:
|
||||
if schedule.matches(now):
|
||||
subscribers = await users_service.list_subscribers()
|
||||
|
|
|
|||
|
|
@ -1,27 +0,0 @@
|
|||
"""Общие 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)
|
||||
|
||||
|
||||
def uptime_human() -> str:
|
||||
"""Аптайм в виде ``3ч 05м`` или ``12м`` (для уведомлений)."""
|
||||
hours, remainder = divmod(uptime_seconds(), 3600)
|
||||
minutes = remainder // 60
|
||||
return f"{hours}ч {minutes:02d}м" if hours else f"{minutes}м"
|
||||
|
||||
|
||||
__all__ = ["STARTED_AT", "uptime_human", "uptime_seconds"]
|
||||
|
|
@ -2,7 +2,6 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
from common.analytics import EventType, record
|
||||
from common.models import Donat, Title
|
||||
from common.repositories import shop_repo
|
||||
|
||||
|
|
@ -19,48 +18,12 @@ async def list_donats() -> list[Donat]:
|
|||
|
||||
async def buy_title(id_user: int, id_title: int) -> bool:
|
||||
"""Купить звание. Возвращает True при успехе."""
|
||||
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
|
||||
return await shop_repo.buy_title(id_user, id_title) == 1
|
||||
|
||||
|
||||
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)
|
||||
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)
|
||||
if result == "-1":
|
||||
return None
|
||||
return result
|
||||
|
|
|
|||
|
|
@ -2,15 +2,12 @@
|
|||
|
||||
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)
|
||||
|
|
|
|||
|
|
@ -4,10 +4,8 @@ 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 economy_repo, users_repo
|
||||
from common.repositories import users_repo
|
||||
|
||||
|
||||
def _user_fields(user: User) -> tuple[int, str, str, str]:
|
||||
|
|
@ -33,34 +31,14 @@ 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]:
|
||||
|
|
|
|||
|
|
@ -1,72 +0,0 @@
|
|||
-- Схема аналитики ClickHouse.
|
||||
--
|
||||
-- Этот каталог монтируется в контейнер как /docker-entrypoint-initdb.d,
|
||||
-- поэтому скрипт применяется ТОЛЬКО при первой инициализации пустого data-каталога.
|
||||
-- Flyway у нас только для Postgres; изменения схемы здесь — новым файлом
|
||||
-- 00N_*.sql (и применением вручную на существующем контуре) плюс раздел в DEPLOY.md.
|
||||
|
||||
CREATE DATABASE IF NOT EXISTS analytics;
|
||||
|
||||
-- Единая широкая таблица событий (см. docs/events.md).
|
||||
-- Партиционирование по дню: сырые партиции удаляются после ежедневного бэкапа
|
||||
-- на домашний сервер (scripts/clickhouse_backup.sh), поэтому удаление должно
|
||||
-- быть дешёвым (DROP PARTITION, без mutation).
|
||||
CREATE TABLE IF NOT EXISTS analytics.events
|
||||
(
|
||||
event_time DateTime('UTC'),
|
||||
event_type LowCardinality(String),
|
||||
user_id Int64,
|
||||
username String,
|
||||
chat_id Int64,
|
||||
payload String,
|
||||
bot_version String,
|
||||
prod UInt8
|
||||
)
|
||||
ENGINE = MergeTree
|
||||
PARTITION BY toYYYYMMDD(event_time)
|
||||
ORDER BY (event_time, event_type, user_id);
|
||||
|
||||
-- Агрегат: события по дням и типам. Остаётся на VPS, когда сырые удалены.
|
||||
CREATE TABLE IF NOT EXISTS analytics.events_daily
|
||||
(
|
||||
day Date,
|
||||
event_type LowCardinality(String),
|
||||
user_id Int64,
|
||||
chat_id Int64,
|
||||
events UInt64
|
||||
)
|
||||
ENGINE = SummingMergeTree
|
||||
ORDER BY (day, event_type, user_id, chat_id);
|
||||
|
||||
CREATE MATERIALIZED VIEW IF NOT EXISTS analytics.events_daily_mv TO analytics.events_daily AS
|
||||
SELECT
|
||||
toDate(event_time) AS day,
|
||||
event_type,
|
||||
user_id,
|
||||
chat_id,
|
||||
count() AS events
|
||||
FROM analytics.events
|
||||
GROUP BY day, event_type, user_id, chat_id;
|
||||
|
||||
-- Агрегат: денежные потоки по дням (сумма amount из payload событий экономики).
|
||||
CREATE TABLE IF NOT EXISTS analytics.money_daily
|
||||
(
|
||||
day Date,
|
||||
event_type LowCardinality(String),
|
||||
user_id Int64,
|
||||
amount_sum Int64,
|
||||
events UInt64
|
||||
)
|
||||
ENGINE = SummingMergeTree
|
||||
ORDER BY (day, event_type, user_id);
|
||||
|
||||
CREATE MATERIALIZED VIEW IF NOT EXISTS analytics.money_daily_mv TO analytics.money_daily AS
|
||||
SELECT
|
||||
toDate(event_time) AS day,
|
||||
event_type,
|
||||
user_id,
|
||||
sum(JSONExtractInt(payload, 'amount')) AS amount_sum,
|
||||
count() AS events
|
||||
FROM analytics.events
|
||||
WHERE event_type IN ('money_accrual', 'money_spend')
|
||||
GROUP BY day, event_type, user_id;
|
||||
|
|
@ -1,33 +0,0 @@
|
|||
"""Аналитика: клиент ClickHouse и каталог событий.
|
||||
|
||||
Публичный интерфейс:
|
||||
|
||||
- :func:`init_analytics` / :func:`close_analytics` — жизненный цикл (вызывает бот при старте/остановке);
|
||||
- :func:`record` — записать событие из сервисного слоя (неблокирующе);
|
||||
- :func:`remember_user` — запомнить ``username`` для обогащения событий;
|
||||
- :class:`EventType` / :data:`ALL_EVENT_TYPES` — каталог типов событий.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from common.analytics.client import (
|
||||
AnalyticsClient,
|
||||
close_analytics,
|
||||
get_analytics,
|
||||
init_analytics,
|
||||
record,
|
||||
remember_user,
|
||||
)
|
||||
from common.analytics.events import ALL_EVENT_TYPES, EVENT_COLUMNS, EventType
|
||||
|
||||
__all__ = [
|
||||
"ALL_EVENT_TYPES",
|
||||
"EVENT_COLUMNS",
|
||||
"AnalyticsClient",
|
||||
"EventType",
|
||||
"close_analytics",
|
||||
"get_analytics",
|
||||
"init_analytics",
|
||||
"record",
|
||||
"remember_user",
|
||||
]
|
||||
|
|
@ -1,301 +0,0 @@
|
|||
"""Клиент ClickHouse: батч-запись событий без блокировки бота.
|
||||
|
||||
Свойства:
|
||||
- запись неблокирующая: :meth:`AnalyticsClient.record` только кладёт строку в буфер;
|
||||
- батчинг: отправка по размеру буфера и по таймеру;
|
||||
- недоступность ClickHouse **не роняет бота**: события попадают в stdout (fallback);
|
||||
- graceful-flush при остановке (:func:`close_analytics`).
|
||||
|
||||
Импорт ``clickhouse_connect`` ленивый (внутри :meth:`AnalyticsClient._ensure_client`),
|
||||
поэтому пакет не является обязательным для импорта модуля (нужно для тестов/админки).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import json
|
||||
import logging
|
||||
from collections import deque
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any, Protocol
|
||||
|
||||
from common.analytics.events import EVENT_COLUMNS, EventType
|
||||
from common.config import Settings, get_settings
|
||||
from common.version import __version__
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
#: Ограничение длины JSON-payload (защита от «толстых» событий).
|
||||
PAYLOAD_MAX_LENGTH = 4000
|
||||
|
||||
#: Ограничение кэша ``id_user -> username``.
|
||||
USERNAME_CACHE_MAX = 5000
|
||||
|
||||
#: Сколько строк события выводить в лог при fallback.
|
||||
FALLBACK_SAMPLE = 5
|
||||
|
||||
|
||||
class ClickHouseClient(Protocol):
|
||||
"""Минимальный контракт клиента ClickHouse (используется и для подмены в тестах)."""
|
||||
|
||||
async def insert(self, table: str, data: list[list[Any]], column_names: list[str]) -> Any:
|
||||
"""Вставить строки в таблицу."""
|
||||
|
||||
async def close(self) -> None:
|
||||
"""Закрыть соединения."""
|
||||
|
||||
|
||||
class AnalyticsClient:
|
||||
"""Батч-запись событий в ClickHouse с деградацией в лог."""
|
||||
|
||||
def __init__(self, settings: Settings | None = None, *, client: ClickHouseClient | None = None) -> None:
|
||||
self._settings = settings or get_settings()
|
||||
self._client = client
|
||||
self._rows: deque[list[Any]] = deque()
|
||||
self._flush_now = asyncio.Event()
|
||||
self._task: asyncio.Task[None] | None = None
|
||||
self._usernames: dict[int, str] = {}
|
||||
self._unavailable_logged = False
|
||||
self._overflow = 0
|
||||
self._table = f"{self._settings.clickhouse_db}.events"
|
||||
|
||||
@property
|
||||
def enabled(self) -> bool:
|
||||
"""Включена ли аналитика (``CLICKHOUSE_ENABLED``)."""
|
||||
return self._settings.clickhouse_enabled
|
||||
|
||||
@property
|
||||
def pending(self) -> int:
|
||||
"""Сколько событий ждёт отправки (для тестов и диагностики)."""
|
||||
return len(self._rows)
|
||||
|
||||
async def start(self) -> None:
|
||||
"""Подключиться к ClickHouse (best-effort) и запустить фоновый сброс буфера."""
|
||||
if not self.enabled:
|
||||
logger.info("Аналитика ClickHouse выключена (CLICKHOUSE_ENABLED=false)")
|
||||
return
|
||||
|
||||
await self._ensure_client()
|
||||
self._task = asyncio.create_task(self._run(), name="analytics-flush")
|
||||
logger.info(
|
||||
"Аналитика ClickHouse: %s:%s/%s (батч %s, интервал %sс)",
|
||||
self._settings.clickhouse_host,
|
||||
self._settings.clickhouse_port,
|
||||
self._settings.clickhouse_db,
|
||||
self._settings.analytics_batch_size,
|
||||
self._settings.analytics_flush_interval,
|
||||
)
|
||||
|
||||
async def _ensure_client(self) -> bool:
|
||||
"""Вернуть True, если клиент ClickHouse доступен."""
|
||||
if self._client is not None:
|
||||
return True
|
||||
|
||||
settings = self._settings
|
||||
try:
|
||||
import clickhouse_connect
|
||||
|
||||
self._client = await clickhouse_connect.get_async_client(
|
||||
host=settings.clickhouse_host,
|
||||
port=settings.clickhouse_port,
|
||||
username=settings.clickhouse_user,
|
||||
password=settings.clickhouse_password,
|
||||
database=settings.clickhouse_db,
|
||||
)
|
||||
except Exception:
|
||||
if not self._unavailable_logged:
|
||||
self._unavailable_logged = True
|
||||
logger.warning(
|
||||
"ClickHouse %s:%s недоступен: события пишутся в stdout (fallback)",
|
||||
settings.clickhouse_host,
|
||||
settings.clickhouse_port,
|
||||
exc_info=True,
|
||||
)
|
||||
return False
|
||||
|
||||
self._unavailable_logged = False
|
||||
return True
|
||||
|
||||
def remember_user(self, user_id: int, username: str | None) -> None:
|
||||
"""Запомнить связку ``id_user -> username`` для обогащения будущих событий."""
|
||||
if not user_id or not username:
|
||||
return
|
||||
if user_id not in self._usernames and len(self._usernames) >= USERNAME_CACHE_MAX:
|
||||
self._usernames.pop(next(iter(self._usernames)))
|
||||
self._usernames[user_id] = username
|
||||
|
||||
def record(
|
||||
self,
|
||||
event_type: EventType | str,
|
||||
*,
|
||||
user_id: int = 0,
|
||||
username: str | None = None,
|
||||
chat_id: int = 0,
|
||||
payload: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
"""Положить событие в буфер (никогда не бросает исключений)."""
|
||||
if not self.enabled:
|
||||
return
|
||||
|
||||
if username is None:
|
||||
username = self._usernames.get(user_id, "")
|
||||
|
||||
self._rows.append(
|
||||
[
|
||||
datetime.now(UTC).replace(tzinfo=None),
|
||||
str(event_type),
|
||||
int(user_id),
|
||||
username or "",
|
||||
int(chat_id),
|
||||
self._dump_payload(payload),
|
||||
__version__,
|
||||
1 if self._settings.is_prod else 0,
|
||||
]
|
||||
)
|
||||
|
||||
while len(self._rows) > self._settings.analytics_buffer_max:
|
||||
self._rows.popleft()
|
||||
self._overflow += 1
|
||||
|
||||
if len(self._rows) >= self._settings.analytics_batch_size:
|
||||
self._flush_now.set()
|
||||
|
||||
async def _run(self) -> None:
|
||||
"""Периодически (или по достижении батча) сбрасывать буфер в ClickHouse."""
|
||||
interval = self._settings.analytics_flush_interval
|
||||
while True:
|
||||
with contextlib.suppress(TimeoutError):
|
||||
await asyncio.wait_for(self._flush_now.wait(), timeout=interval)
|
||||
self._flush_now.clear()
|
||||
await self.flush()
|
||||
|
||||
async def flush(self) -> None:
|
||||
"""Отправить накопленные события. При сбое — fallback в лог (без исключений)."""
|
||||
overflow, self._overflow = self._overflow, 0
|
||||
if overflow:
|
||||
logger.warning("Буфер событий переполнен, потеряно %s событий (не записаны в ClickHouse)", overflow)
|
||||
|
||||
if not self._rows:
|
||||
return
|
||||
|
||||
if not self.enabled:
|
||||
self._rows.clear()
|
||||
return
|
||||
|
||||
rows = list(self._rows)
|
||||
self._rows.clear()
|
||||
|
||||
if not await self._ensure_client():
|
||||
self._log_fallback(rows, "ClickHouse недоступен")
|
||||
return
|
||||
|
||||
client = self._client
|
||||
if client is None:
|
||||
self._log_fallback(rows, "клиент ClickHouse не создан")
|
||||
return
|
||||
|
||||
try:
|
||||
await client.insert(self._table, rows, list(EVENT_COLUMNS))
|
||||
except Exception:
|
||||
logger.exception("Не удалось записать %s событий в ClickHouse", len(rows))
|
||||
self._log_fallback(rows, "ошибка записи в ClickHouse")
|
||||
|
||||
async def close(self) -> None:
|
||||
"""Остановить фоновый таск, сбросить остаток и закрыть клиент."""
|
||||
if self._task is not None:
|
||||
self._task.cancel()
|
||||
with contextlib.suppress(asyncio.CancelledError):
|
||||
await self._task
|
||||
self._task = None
|
||||
|
||||
if self._client is not None:
|
||||
await self.flush()
|
||||
client, self._client = self._client, None
|
||||
try:
|
||||
await client.close()
|
||||
except Exception:
|
||||
logger.warning("Не удалось закрыть клиент ClickHouse", exc_info=True)
|
||||
elif self._rows:
|
||||
# Клиента нет (ClickHouse недоступен) — не пробуем подключиться на выходе.
|
||||
rows = list(self._rows)
|
||||
self._rows.clear()
|
||||
self._log_fallback(rows, "остановка бота")
|
||||
|
||||
def _log_fallback(self, rows: list[list[Any]], reason: str) -> None:
|
||||
"""Вывести потерянные события в лог, чтобы данные не пропадали молча."""
|
||||
if not rows:
|
||||
return
|
||||
lines = " | ".join(
|
||||
f"{row[0]:%Y-%m-%d %H:%M:%S} {row[1]} user={row[2]} chat={row[4]} {row[5]}"
|
||||
for row in rows[:FALLBACK_SAMPLE]
|
||||
)
|
||||
logger.warning(
|
||||
"Fallback (%s): %s событий не записаны в ClickHouse: %s%s",
|
||||
reason,
|
||||
len(rows),
|
||||
lines,
|
||||
" ..." if len(rows) > FALLBACK_SAMPLE else "",
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _dump_payload(payload: dict[str, Any] | None) -> str:
|
||||
"""Сериализовать payload в JSON-строку с обрезкой по длине."""
|
||||
if not payload:
|
||||
return "{}"
|
||||
try:
|
||||
text = json.dumps(payload, ensure_ascii=False, default=str)
|
||||
except (TypeError, ValueError):
|
||||
text = json.dumps({"_error": "payload не сериализуется"})
|
||||
return text[:PAYLOAD_MAX_LENGTH]
|
||||
|
||||
|
||||
#: Активный клиент (если аналитика не инициализирована — None).
|
||||
_client: AnalyticsClient | None = None
|
||||
|
||||
|
||||
async def init_analytics(
|
||||
settings: Settings | None = None,
|
||||
*,
|
||||
client: ClickHouseClient | None = None,
|
||||
) -> AnalyticsClient:
|
||||
"""Создать и запустить единственный клиент аналитики. Идемпотентно."""
|
||||
global _client
|
||||
if _client is not None:
|
||||
return _client
|
||||
_client = AnalyticsClient(settings, client=client)
|
||||
await _client.start()
|
||||
return _client
|
||||
|
||||
|
||||
async def close_analytics() -> None:
|
||||
"""Остановить клиент аналитики и сбросить буфер при остановке приложения."""
|
||||
global _client
|
||||
if _client is not None:
|
||||
await _client.close()
|
||||
_client = None
|
||||
|
||||
|
||||
def get_analytics() -> AnalyticsClient | None:
|
||||
"""Вернуть активный клиент аналитики (или None)."""
|
||||
return _client
|
||||
|
||||
|
||||
def record(
|
||||
event_type: EventType | str,
|
||||
*,
|
||||
user_id: int = 0,
|
||||
username: str | None = None,
|
||||
chat_id: int = 0,
|
||||
payload: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
"""Записать событие. Если аналитика не инициализирована — no-op."""
|
||||
if _client is None:
|
||||
return
|
||||
_client.record(event_type, user_id=user_id, username=username, chat_id=chat_id, payload=payload)
|
||||
|
||||
|
||||
def remember_user(user_id: int, username: str | None) -> None:
|
||||
"""Запомнить username пользователя (для обогащения событий из сервисов)."""
|
||||
if _client is not None:
|
||||
_client.remember_user(user_id, username)
|
||||
|
|
@ -1,63 +0,0 @@
|
|||
"""Каталог событий аналитики.
|
||||
|
||||
Единственный источник правды по типам событий ClickHouse. Список синхронизирован
|
||||
с документацией (``docs/events.md``) — рассинхрон ловит ``tests/test_analytics.py``.
|
||||
|
||||
Правило: добавляешь новое событие — добавь его в ``EventType`` и в ``docs/events.md``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from enum import StrEnum
|
||||
|
||||
|
||||
class EventType(StrEnum):
|
||||
"""Типы событий (значение = колонка ``event_type`` в ClickHouse)."""
|
||||
|
||||
# --- Сообщения и команды ---
|
||||
MESSAGE = "message"
|
||||
COMMAND = "command"
|
||||
|
||||
# --- Экономика ---
|
||||
PURIFICATION = "purification"
|
||||
EXP_ACCRUAL = "exp_accrual"
|
||||
LEVEL_UP = "level_up"
|
||||
MONEY_ACCRUAL = "money_accrual"
|
||||
MONEY_SPEND = "money_spend"
|
||||
EGGS = "eggs"
|
||||
PURCHASE_TITLE = "purchase_title"
|
||||
PURCHASE_DONAT = "purchase_donat"
|
||||
|
||||
# --- Участники ---
|
||||
JOIN = "join"
|
||||
LEAVE = "leave"
|
||||
NICK_SET = "nick_set"
|
||||
|
||||
# --- Подписки и доступ ---
|
||||
SUBSCRIBE = "subscribe"
|
||||
UNSUBSCRIBE = "unsubscribe"
|
||||
ACCESS_DENIED = "access_denied"
|
||||
|
||||
# --- Жизненный цикл бота и ошибки ---
|
||||
BOT_START = "bot_start"
|
||||
BOT_STOP = "bot_stop"
|
||||
BOT_ALIVE = "bot_alive"
|
||||
ERROR = "error"
|
||||
|
||||
|
||||
#: Полный список типов событий (для проверки синхронности с документацией).
|
||||
ALL_EVENT_TYPES: frozenset[str] = frozenset(member.value for member in EventType)
|
||||
|
||||
#: Колонки таблицы ``analytics.events`` в порядке вставки.
|
||||
EVENT_COLUMNS: tuple[str, ...] = (
|
||||
"event_time",
|
||||
"event_type",
|
||||
"user_id",
|
||||
"username",
|
||||
"chat_id",
|
||||
"payload",
|
||||
"bot_version",
|
||||
"prod",
|
||||
)
|
||||
|
||||
__all__ = ["ALL_EVENT_TYPES", "EVENT_COLUMNS", "EventType"]
|
||||
|
|
@ -1,54 +0,0 @@
|
|||
"""Метаданные сообщения для событий аналитики.
|
||||
|
||||
Модуль намеренно **не зависит от 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"]
|
||||
|
|
@ -53,19 +53,6 @@ class Settings(BaseSettings):
|
|||
#: Файл-heartbeat для Docker healthcheck (обновляется ботом периодически).
|
||||
heartbeat_path: str = "/tmp/dispelevilbot.heartbeat"
|
||||
|
||||
#: ClickHouse (аналитика событий). CLICKHOUSE_ENABLED=false — полностью выключить.
|
||||
clickhouse_enabled: bool = True
|
||||
clickhouse_host: str = "localhost"
|
||||
clickhouse_port: int = 8123
|
||||
clickhouse_db: str = "analytics"
|
||||
clickhouse_user: str = "default"
|
||||
clickhouse_password: str = ""
|
||||
|
||||
#: Батчинг записи событий: размер батча, интервал сброса и предел буфера.
|
||||
analytics_batch_size: int = 200
|
||||
analytics_flush_interval: float = 5.0
|
||||
analytics_buffer_max: int = 5000
|
||||
|
||||
@field_validator("chat_id", "alert_chat_id", mode="before")
|
||||
@classmethod
|
||||
def _empty_int_to_zero(cls, value: object) -> object:
|
||||
|
|
|
|||
|
|
@ -15,16 +15,6 @@ 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)
|
||||
|
|
|
|||
|
|
@ -1,13 +0,0 @@
|
|||
"""Тексты уведомлений в чат логов (``ALERT_CHAT_ID``).
|
||||
|
||||
Служебные сообщения для админов (не для участников), поэтому в ``TEMPLATES``
|
||||
(редактируемые шаблоны Блока 4) они не входят.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
MSG_JOIN = "[вход] {username} вошёл в чат «{chat_title}»"
|
||||
MSG_LEAVE = "[выход] {username} покинул чат «{chat_title}»"
|
||||
MSG_PURIFICATION = "[очищение] {username} — опыт +{exp_add}, монеты +{money_add}"
|
||||
MSG_ALIVE = "[бот живой] версия {version}, аптайм {uptime}"
|
||||
MSG_ACCESS_DENIED = "[доступ] {username} стучится в чат"
|
||||
|
|
@ -42,10 +42,6 @@ TAG_REGISTRY: dict[str, str] = {
|
|||
"users_list_count": "Количество игроков в списке",
|
||||
"tops": "Топ игроков по сообщениям",
|
||||
"promo": "Промокод доната",
|
||||
# --- Служебные теги уведомлений в чат логов (ALERT_CHAT_ID) ---
|
||||
"chat_title": "Название чата",
|
||||
"version": "Версия бота",
|
||||
"uptime": "Время работы бота",
|
||||
}
|
||||
|
||||
SUPPORTED_TAGS: frozenset[str] = frozenset(TAG_REGISTRY)
|
||||
|
|
|
|||
|
|
@ -7,4 +7,4 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
__version__ = "0.3.6"
|
||||
__version__ = "0.2.5"
|
||||
|
|
|
|||
|
|
@ -46,51 +46,17 @@ services:
|
|||
environment:
|
||||
DB_HOST: postgres
|
||||
DB_PORT: "5432"
|
||||
CLICKHOUSE_HOST: clickhouse
|
||||
CLICKHOUSE_PORT: "8123"
|
||||
depends_on:
|
||||
postgres:
|
||||
condition: service_healthy
|
||||
migrate:
|
||||
condition: service_completed_successfully
|
||||
# Без ожидания здоровья: аналитика некритична, бот работает и при недоступном ClickHouse.
|
||||
clickhouse:
|
||||
condition: service_started
|
||||
logging:
|
||||
driver: json-file
|
||||
options:
|
||||
max-size: "10m"
|
||||
max-file: "3"
|
||||
|
||||
# Аналитика событий (docs/events.md). Наружу не публикуется:
|
||||
# только loopback хоста для бэкапа/туннеля; бот ходит внутрь сети compose.
|
||||
clickhouse:
|
||||
image: clickhouse/clickhouse-server:25.3
|
||||
restart: unless-stopped
|
||||
environment:
|
||||
CLICKHOUSE_DB: ${CLICKHOUSE_DB:-analytics}
|
||||
CLICKHOUSE_USER: ${CLICKHOUSE_USER:-de_bot}
|
||||
CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:?задайте CLICKHOUSE_PASSWORD}
|
||||
# Нужно для создания пользователей/грантов (в финале — read-only пользователь для Superset).
|
||||
CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: "1"
|
||||
volumes:
|
||||
- clickhousedata:/var/lib/clickhouse
|
||||
# DDL применяется один раз при первой инициализации (пустой data-каталог).
|
||||
- ./clickhouse/init:/docker-entrypoint-initdb.d:ro
|
||||
ports:
|
||||
# Порт контура задаётся в env (prod 8123, test 8124) — иначе конфликт на одной VPS.
|
||||
- "127.0.0.1:${CLICKHOUSE_HTTP_PORT:-8123}:8123"
|
||||
ulimits:
|
||||
nofile:
|
||||
soft: 262144
|
||||
hard: 262144
|
||||
healthcheck:
|
||||
test: ["CMD-SHELL", "wget --spider -q -T 5 http://127.0.0.1:8123/ping || exit 1"]
|
||||
interval: 15s
|
||||
timeout: 10s
|
||||
retries: 10
|
||||
start_period: 20s
|
||||
|
||||
# --- Reverse proxy и TLS (Блок 2.4) ---
|
||||
# Включается, когда появится админка/Mini App (Блоки 4–5).
|
||||
# Раскомментировать и положить deploy/Caddyfile:
|
||||
|
|
@ -108,6 +74,5 @@ services:
|
|||
|
||||
volumes:
|
||||
pgdata:
|
||||
clickhousedata:
|
||||
# caddy_data:
|
||||
# caddy_config:
|
||||
|
|
|
|||
|
|
@ -51,47 +51,16 @@ services:
|
|||
# Внутри сети compose БД доступна по имени сервиса.
|
||||
DB_HOST: postgres
|
||||
DB_PORT: "5432"
|
||||
CLICKHOUSE_HOST: clickhouse
|
||||
CLICKHOUSE_PORT: "8123"
|
||||
depends_on:
|
||||
postgres:
|
||||
condition: service_healthy
|
||||
migrate:
|
||||
condition: service_completed_successfully
|
||||
# Аналитика некритична: бот стартует, даже если ClickHouse ещё не готов.
|
||||
clickhouse:
|
||||
condition: service_started
|
||||
logging:
|
||||
driver: json-file
|
||||
options:
|
||||
max-size: "10m"
|
||||
max-file: "3"
|
||||
|
||||
# Аналитика событий (Блок 3). DDL — в clickhouse/init/.
|
||||
clickhouse:
|
||||
image: clickhouse/clickhouse-server:25.3
|
||||
restart: unless-stopped
|
||||
environment:
|
||||
CLICKHOUSE_DB: ${CLICKHOUSE_DB:-analytics}
|
||||
CLICKHOUSE_USER: ${CLICKHOUSE_USER:-de_bot}
|
||||
CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:?задайте CLICKHOUSE_PASSWORD в .env}
|
||||
CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: "1"
|
||||
volumes:
|
||||
- clickhousedata:/var/lib/clickhouse
|
||||
- ./clickhouse/init:/docker-entrypoint-initdb.d:ro
|
||||
ports:
|
||||
- "127.0.0.1:8123:8123"
|
||||
ulimits:
|
||||
nofile:
|
||||
soft: 262144
|
||||
hard: 262144
|
||||
healthcheck:
|
||||
test: ["CMD-SHELL", "wget --spider -q -T 5 http://127.0.0.1:8123/ping || exit 1"]
|
||||
interval: 15s
|
||||
timeout: 10s
|
||||
retries: 10
|
||||
start_period: 20s
|
||||
|
||||
volumes:
|
||||
pgdata:
|
||||
clickhousedata:
|
||||
|
|
|
|||
124
docs/events.md
124
docs/events.md
|
|
@ -1,124 +0,0 @@
|
|||
# События аналитики (ClickHouse)
|
||||
|
||||
Бот пишет все значимые действия в ClickHouse — единую широкую таблицу
|
||||
`analytics.events`. Дашборд (Superset) и ETL — отдельный финальный этап; здесь
|
||||
описан **контракт** событий: какие бывают `event_type` и что лежит в `payload`.
|
||||
|
||||
> Источник правды в коде — `common/analytics/events.py` (`EventType`).
|
||||
> Расхождение со списком ниже ловит `tests/test_analytics.py`.
|
||||
> Добавляешь событие — добавь строку в таблицу.
|
||||
|
||||
## Схема в ClickHouse
|
||||
|
||||
DDL — `clickhouse/init/001_events.sql` (применяется при первой инициализации контейнера).
|
||||
|
||||
- `analytics.events` — сырые события, `MergeTree`, `PARTITION BY toYYYYMMDD(event_time)`,
|
||||
`ORDER BY (event_time, event_type, user_id)`. Партиционирование по дню выбрано специально:
|
||||
сырые партиции ежедневно удаляются после бэкапа (дешёвый `DROP PARTITION`, без mutation).
|
||||
- `analytics.events_daily` — агрегат (`SummingMergeTree`): `day`, `event_type`, `user_id`,
|
||||
`chat_id`, `events`. Строится материализованным представлением `events_daily_mv`.
|
||||
- `analytics.money_daily` — агрегат по деньгам: `day`, `event_type`, `user_id`,
|
||||
`amount_sum`, `events` (из `payload.amount` событий `money_accrual`/`money_spend`).
|
||||
|
||||
Агрегаты остаются на VPS навсегда, поэтому активность, топ команд и денежные потоки
|
||||
доступны и после удаления сырых событий. Запросы к агрегатам суммируют `events`/`amount_sum`
|
||||
(фон слияний `SummingMergeTree` не гарантирует полную свёртку в любой момент).
|
||||
|
||||
## Хранение и бэкап
|
||||
|
||||
| Что | Где | Сколько |
|
||||
| --- | --- | --- |
|
||||
| Сырые события | VPS, `analytics.events` | последние `BACKUP_GRACE_DAYS` дней (по умолчанию 7) |
|
||||
| Агрегаты | VPS | постоянно |
|
||||
| Полная история | домашний сервер, свой ClickHouse | архивы `events_YYYY-MM-DD.parquet.gz` (ежедневно) |
|
||||
|
||||
Каждую ночь (по Москве) `scripts/clickhouse_backup.sh` выгружает вчерашнюю партицию,
|
||||
передаёт её на домашний сервер, импортирует и **сверяет количество строк**; только после
|
||||
успешной сверки на VPS удаляются партиции старше грейса. Ошибка бэкапа данные не удаляет.
|
||||
Подробности — `DEPLOY.md`.
|
||||
|
||||
## Уведомления в чат логов
|
||||
|
||||
Помимо записи в ClickHouse часть событий дублируется сообщением в `ALERT_CHAT_ID`
|
||||
(если он задан; `0` — выключено). Отправка не критична: ошибки только логируются.
|
||||
|
||||
| Уведомление | Когда | Событие в ClickHouse |
|
||||
| --- | --- | --- |
|
||||
| Вход участника | кто-то вошёл в основной чат | `join` |
|
||||
| Выход участника | кто-то вышел из основного чата | `leave` |
|
||||
| Очищение | успешное `/очищение` (`+опыт`, `+монеты`) | `purification` |
|
||||
| Пинг «бот живой» | раз в час | `bot_alive` |
|
||||
| Попытка доступа извне | не участник чата стучится в бота | `access_denied` |
|
||||
|
||||
## Поля события
|
||||
|
||||
| Поле | Тип | Описание |
|
||||
| --- | --- | --- |
|
||||
| **event_time** | `DateTime('UTC')` | Момент события (UTC, без таймзоны) |
|
||||
| **event_type** | `LowCardinality(String)` | Тип события (см. каталог ниже) |
|
||||
| **user_id** | `Int64` | Telegram-ID пользователя (0 — системное событие) |
|
||||
| **username** | `String` | `@username` на момент события (может быть пустым) |
|
||||
| **chat_id** | `Int64` | ID чата (обычно основной чат клана) |
|
||||
| **payload** | `String` | Дополнительные данные события в виде JSON |
|
||||
| **bot_version** | `String` | Версия бота из `common/version.py` |
|
||||
| **prod** | `UInt8` | 1 — прод-контур, 0 — тестовый |
|
||||
|
||||
## Каталог событий
|
||||
|
||||
| Тип события | Когда пишется | `payload` |
|
||||
| --- | --- | --- |
|
||||
| `message` | Любое сообщение в основном чате | `message_id`, `chat_type`, `content_type`, `text_len`, `is_command` |
|
||||
| `command` | Использование команды (`/инфо`, `/очищение`, …) | `command`, `args_count` |
|
||||
| `purification` | Ежедневное очищение (`/очищение`) | `success`, `exp_add`, `money_add` |
|
||||
| `exp_accrual` | Начисление опыта за очищение | `amount`, `streak` |
|
||||
| `level_up` | Повышение уровня по итогам очищения | `level_from`, `level_to` |
|
||||
| `money_accrual` | Начисление валюты (монетка за активность, награда за очищение) | `amount`, `reason`, `action_id` |
|
||||
| `money_spend` | Трата валюты (покупка, обнуление баланса при удалении) | `amount`, `reason`, `product_id` |
|
||||
| `eggs` | Получение яичек (`/яички`) | `amount` |
|
||||
| `purchase_title` | Покупка звания | `title_id`, `price`, `success` |
|
||||
| `purchase_donat` | Покупка доната | `donat_id`, `price`, `success` |
|
||||
| `join` | Вход участника в основной чат | `chat_title` |
|
||||
| `leave` | Выход участника из основного чата | `chat_title` |
|
||||
| `nick_set` | Смена ника (`/ник`) | `nick` |
|
||||
| `subscribe` | Подписка на рассылки (`/подписка`) | — |
|
||||
| `unsubscribe` | Отписка от рассылок (`/отписка`) | — |
|
||||
| `access_denied` | Отказ в доступе (не участник чата и не админ) | `reason` |
|
||||
| `bot_start` | Старт бота | `app_env` |
|
||||
| `bot_stop` | Остановка бота | `app_env`, `uptime_seconds` |
|
||||
| `bot_alive` | Ежечасный пинг «бот живой» | `app_env`, `uptime_seconds` |
|
||||
| `error` | Необработанная ошибка при обработке апдейта | `error_type`, `error`, `update_type` |
|
||||
|
||||
## Поведение при недоступности ClickHouse
|
||||
|
||||
Аналитика — **некритичная** подсистема: её сбой не должен влиять на бота.
|
||||
|
||||
- события копятся в буфере и отправляются батчами (по размеру и по таймеру);
|
||||
- если ClickHouse недоступен, буфер не растёт бесконечно: при переполнении самые
|
||||
старые записи отбрасываются со счётчиком в логе;
|
||||
- если записать батч не удалось, события выводятся в stdout (`WARNING`, первые
|
||||
несколько строк + общее количество) и отбрасываются — данные не пропадают молча;
|
||||
- при остановке бота выполняется финальный сброс буфера;
|
||||
- выключение целиком — `CLICKHOUSE_ENABLED=false` (тогда `record()` — no-op).
|
||||
|
||||
## Примеры запросов
|
||||
|
||||
```sql
|
||||
-- активность по дням
|
||||
SELECT toDate(event_time) AS day, count() FROM analytics.events GROUP BY day ORDER BY day;
|
||||
|
||||
-- топ команд за неделю
|
||||
SELECT JSONExtractString(payload, 'command') AS cmd, count() AS uses
|
||||
FROM analytics.events
|
||||
WHERE event_type = 'command' AND event_time >= now() - INTERVAL 7 DAY
|
||||
GROUP BY cmd ORDER BY uses DESC;
|
||||
|
||||
-- покупки за месяц
|
||||
SELECT event_type, count(), sum(JSONExtractInt(payload, 'price'))
|
||||
FROM analytics.events
|
||||
WHERE event_type IN ('purchase_title', 'purchase_donat') AND event_time >= now() - INTERVAL 30 DAY
|
||||
GROUP BY event_type;
|
||||
|
||||
-- ошибки
|
||||
SELECT event_time, JSONExtractString(payload, 'error_type'), JSONExtractString(payload, 'error')
|
||||
FROM analytics.events WHERE event_type = 'error' ORDER BY event_time DESC LIMIT 50;
|
||||
```
|
||||
|
|
@ -1,28 +0,0 @@
|
|||
# Параметры ежедневного бэкапа событий ClickHouse с VPS на домашний сервер.
|
||||
# Скопируйте в env/backup.env (env/*.env в git не попадает) и заполните.
|
||||
#
|
||||
# Используется скриптами:
|
||||
# scripts/clickhouse_backup.sh (запускается на VPS по cron, см. DEPLOY.md)
|
||||
# scripts/clickhouse_import.sh (лежит на домашнем сервере)
|
||||
|
||||
# Куда передавать архивы: пользователь@хост и каталог на домашнем сервере.
|
||||
BACKUP_REMOTE=user@home.example
|
||||
BACKUP_DIR=/srv/dispelevilbot/events
|
||||
|
||||
# Приватный ключ для ssh/rsync (без пароля; на домашнем сервере — соответствующий public key).
|
||||
BACKUP_SSH_KEY=/root/.ssh/dispelevilbot_backup
|
||||
|
||||
# Сколько дней сырых событий оставлять на VPS (остальное удаляется после успешной сверки).
|
||||
BACKUP_GRACE_DAYS=7
|
||||
|
||||
# Таймзона расчёта «вчерашнего» дня.
|
||||
BACKUP_TZ=Europe/Moscow
|
||||
|
||||
# Как запускать импорт на домашнем сервере (путь к scripts/clickhouse_import.sh).
|
||||
REMOTE_IMPORT_SCRIPT=./clickhouse_import.sh
|
||||
|
||||
# ClickHouse домашнего сервера. Пароль лучше без кавычек и пробелов — он передаётся по ssh.
|
||||
REMOTE_CLICKHOUSE_HOST=localhost
|
||||
REMOTE_CLICKHOUSE_PORT=9000
|
||||
REMOTE_CLICKHOUSE_USER=default
|
||||
REMOTE_CLICKHOUSE_PASSWORD=
|
||||
|
|
@ -7,7 +7,7 @@ ENV_FILE=env/prod.env
|
|||
|
||||
APP_ENV=prod
|
||||
|
||||
# Образ в реестре (latest = последняя сборка из master; можно закрепить <sha> или версию)
|
||||
# Образ в реестре (latest = последняя сборка из master; можно закрепить <sha> или версию, напр. 0.2.5)
|
||||
REGISTRY=git.mpakosrv.ru
|
||||
IMAGE_TAG=latest
|
||||
|
||||
|
|
@ -30,12 +30,3 @@ ALERT_CHAT_ID=0
|
|||
# Telegram-ID админов (доступ в обход проверки членства), через запятую; можно пусто
|
||||
ADMIN_IDS=
|
||||
TIMEZONE=Europe/Moscow
|
||||
|
||||
# ClickHouse (аналитика событий, см. docs/events.md).
|
||||
# Наружу не публикуется: только loopback хоста для бэкапа/туннеля;
|
||||
# бот ходит внутрь сети compose по имени сервиса clickhouse:8123.
|
||||
CLICKHOUSE_DB=analytics
|
||||
CLICKHOUSE_USER=de_bot
|
||||
CLICKHOUSE_PASSWORD=
|
||||
# Порт на хосте (для бэкапа). У test и prod он должен различаться.
|
||||
CLICKHOUSE_HTTP_PORT=8123
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ ENV_FILE=env/test.env
|
|||
|
||||
APP_ENV=dev
|
||||
|
||||
# Образ в реестре (latest = последняя сборка из master; можно закрепить <sha> или версию)
|
||||
# Образ в реестре (latest = последняя сборка из master; можно закрепить <sha> или версию, напр. 0.2.5)
|
||||
REGISTRY=git.mpakosrv.ru
|
||||
IMAGE_TAG=latest
|
||||
|
||||
|
|
@ -28,12 +28,3 @@ ALERT_CHAT_ID=0
|
|||
# Telegram-ID админов (доступ в обход проверки членства), через запятую; можно пусто
|
||||
ADMIN_IDS=
|
||||
TIMEZONE=Europe/Moscow
|
||||
|
||||
# ClickHouse (аналитика событий, см. docs/events.md).
|
||||
# Наружу не публикуется: только loopback хоста для бэкапа/туннеля;
|
||||
# бот ходит внутрь сети compose по имени сервиса clickhouse:8123.
|
||||
CLICKHOUSE_DB=analytics
|
||||
CLICKHOUSE_USER=de_bot
|
||||
CLICKHOUSE_PASSWORD=
|
||||
# Порт на хосте: у test он отличается от prod (8123), иначе конфликт на одной VPS.
|
||||
CLICKHOUSE_HTTP_PORT=8124
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ pydantic-settings>=2.3,<3.0
|
|||
python-dotenv>=1.0,<2.0
|
||||
psycopg[binary,pool]>=3.2,<4.0
|
||||
|
||||
# ClickHouse (аналитика событий, Блок 3). Extras [async] — нативный asyncio-клиент.
|
||||
clickhouse-connect[async]>=1.0,<2.0
|
||||
# Блок 3 (ClickHouse) — подключим позже:
|
||||
# clickhouse-connect>=0.7,<1.0
|
||||
# Блок 4 (расписания из БД) — при необходимости:
|
||||
# APScheduler>=3.10,<4.0
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@ echo "== проверки =="
|
|||
|
||||
running="$($COMPOSE ps --services --status running || true)"
|
||||
fail=0
|
||||
for svc in postgres clickhouse bot; do
|
||||
for svc in postgres bot; do
|
||||
if echo "$running" | grep -qx "$svc"; then
|
||||
echo "OK ${svc}: running"
|
||||
else
|
||||
|
|
@ -49,18 +49,4 @@ else
|
|||
fail=1
|
||||
fi
|
||||
|
||||
# ClickHouse отвечает и схема аналитики применена (initdb применяется один раз,
|
||||
# при первой инициализации пустого data-каталога)
|
||||
if $COMPOSE exec -T clickhouse sh -lc 'clickhouse-client --user "$CLICKHOUSE_USER" --password "$CLICKHOUSE_PASSWORD" --query "SELECT 1"' >/dev/null 2>&1; then
|
||||
echo "OK clickhouse: принимает запросы"
|
||||
if $COMPOSE exec -T clickhouse sh -lc 'clickhouse-client --user "$CLICKHOUSE_USER" --password "$CLICKHOUSE_PASSWORD" --query "SELECT count() FROM analytics.events"' >/dev/null 2>&1; then
|
||||
echo "OK clickhouse: таблица analytics.events доступна"
|
||||
else
|
||||
echo "WARN clickhouse: нет доступа к analytics.events (см. clickhouse/init/001_events.sql)" >&2
|
||||
fi
|
||||
else
|
||||
echo "FAIL clickhouse: не отвечает" >&2
|
||||
fail=1
|
||||
fi
|
||||
|
||||
exit $fail
|
||||
|
|
|
|||
|
|
@ -1,109 +0,0 @@
|
|||
#!/usr/bin/env bash
|
||||
# Ежедневный бэкап сырых событий ClickHouse с VPS на домашний сервер.
|
||||
#
|
||||
# Что делает:
|
||||
# 1) выгружает партицию «вчерашнего» дня из analytics.events в Parquet.gz;
|
||||
# 2) передаёт файл на домашний сервер (rsync) и импортирует его в тамошний ClickHouse;
|
||||
# 3) сверяет количество строк на VPS и на домашнем сервере;
|
||||
# 4) только после успешной сверки удаляет на VPS партиции старше BACKUP_GRACE_DAYS.
|
||||
#
|
||||
# Ошибка на любом шаге оставляет данные на VPS и возвращает ненулевой код.
|
||||
#
|
||||
# Запуск (на VPS, из каталога проекта):
|
||||
# ./scripts/clickhouse_backup.sh test # или prod
|
||||
# make backup E=test
|
||||
#
|
||||
# Настройки (env/<контур>.env + env/backup.env, см. env/backup.env.example):
|
||||
# CLICKHOUSE_* доступ к ClickHouse контура
|
||||
# BACKUP_REMOTE user@host домашнего сервера
|
||||
# BACKUP_DIR каталог архива на домашнем сервере
|
||||
# BACKUP_SSH_KEY приватный ключ для ssh/rsync
|
||||
# BACKUP_GRACE_DAYS сколько дней сырых событий оставлять на VPS (по умолчанию 7)
|
||||
# BACKUP_TZ таймзона расчёта «вчера» (по умолчанию Europe/Moscow)
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
E="${1:-test}"
|
||||
ENV_FILE="env/${E}.env"
|
||||
BACKUP_ENV="env/backup.env"
|
||||
COMPOSE="docker compose -p ${E} --env-file ${ENV_FILE} -f docker-compose.prod.yml"
|
||||
|
||||
log() { echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*"; }
|
||||
fail() {
|
||||
echo "[$(date '+%Y-%m-%d %H:%M:%S')] ОШИБКА: $*" >&2
|
||||
exit 1
|
||||
}
|
||||
|
||||
[ -f "$ENV_FILE" ] || fail "нет файла окружения ${ENV_FILE}"
|
||||
# shellcheck disable=SC1090
|
||||
set -a && . "./${ENV_FILE}" && set +a
|
||||
if [ -f "$BACKUP_ENV" ]; then
|
||||
# shellcheck disable=SC1090
|
||||
set -a && . "./${BACKUP_ENV}" && set +a
|
||||
fi
|
||||
|
||||
: "${BACKUP_REMOTE:?задайте BACKUP_REMOTE в env/backup.env}"
|
||||
: "${BACKUP_DIR:?задайте BACKUP_DIR в env/backup.env}"
|
||||
BACKUP_SSH_KEY="${BACKUP_SSH_KEY:-$HOME/.ssh/id_rsa}"
|
||||
BACKUP_GRACE_DAYS="${BACKUP_GRACE_DAYS:-7}"
|
||||
BACKUP_TZ="${BACKUP_TZ:-Europe/Moscow}"
|
||||
REMOTE_IMPORT_SCRIPT="${REMOTE_IMPORT_SCRIPT:-./clickhouse_import.sh}"
|
||||
REMOTE_CLICKHOUSE_HOST="${REMOTE_CLICKHOUSE_HOST:-localhost}"
|
||||
REMOTE_CLICKHOUSE_PORT="${REMOTE_CLICKHOUSE_PORT:-9000}"
|
||||
REMOTE_CLICKHOUSE_USER="${REMOTE_CLICKHOUSE_USER:-default}"
|
||||
REMOTE_CLICKHOUSE_PASSWORD="${REMOTE_CLICKHOUSE_PASSWORD:-}"
|
||||
SSH=(ssh -i "$BACKUP_SSH_KEY" -o BatchMode=yes "$BACKUP_REMOTE")
|
||||
|
||||
ch() {
|
||||
$COMPOSE exec -T clickhouse clickhouse-client \
|
||||
--user "$CLICKHOUSE_USER" --password "$CLICKHOUSE_PASSWORD" "$@"
|
||||
}
|
||||
|
||||
DAY="$(TZ="$BACKUP_TZ" date -d 'yesterday' '+%Y-%m-%d')"
|
||||
PARTITION="${DAY//-/}"
|
||||
GRACE_CUTOFF="$(TZ="$BACKUP_TZ" date -d "${BACKUP_GRACE_DAYS} days ago" '+%Y%m%d')"
|
||||
TMP_DIR="$(mktemp -d)"
|
||||
trap 'rm -rf "$TMP_DIR"' EXIT
|
||||
|
||||
log "Контур ${E}: бэкап событий за ${DAY} (грейс ${BACKUP_GRACE_DAYS} дней, порог ${GRACE_CUTOFF})"
|
||||
|
||||
if ! $COMPOSE ps --services --status running 2>/dev/null | grep -qx clickhouse; then
|
||||
fail "контейнер clickhouse контура ${E} не запущен"
|
||||
fi
|
||||
|
||||
ROWS_LOCAL="$(ch --query "SELECT count() FROM analytics.events WHERE toDate(event_time) = toDate('${DAY}')" | tr -d '[:space:]')"
|
||||
log "На VPS строк за ${DAY}: ${ROWS_LOCAL}"
|
||||
if [ "$ROWS_LOCAL" = "0" ]; then
|
||||
log "ВНИМАНИЕ: за ${DAY} событий нет — партиция пустая, продолжаю (сверка 0 = 0)"
|
||||
fi
|
||||
|
||||
ARCHIVE="${TMP_DIR}/events_${DAY}.parquet.gz"
|
||||
log "Выгружаю партицию в ${ARCHIVE}"
|
||||
ch --query "SELECT * FROM analytics.events WHERE toDate(event_time) = toDate('${DAY}') FORMAT Parquet" | gzip -9 >"$ARCHIVE"
|
||||
[ -s "$ARCHIVE" ] || fail "не удалось выгрузить архив ${DAY}"
|
||||
log "Размер архива: $(du -h "$ARCHIVE" | cut -f1)"
|
||||
|
||||
log "Передаю архив на ${BACKUP_REMOTE}:${BACKUP_DIR}"
|
||||
rsync -az -e "ssh -i ${BACKUP_SSH_KEY} -o BatchMode=yes" "$ARCHIVE" "${BACKUP_REMOTE}:${BACKUP_DIR}/"
|
||||
|
||||
log "Импортирую архив в ClickHouse домашнего сервера"
|
||||
ROWS_REMOTE="$("${SSH[@]}" \
|
||||
"BACKUP_DIR='${BACKUP_DIR}' CLICKHOUSE_HOST='${REMOTE_CLICKHOUSE_HOST}' CLICKHOUSE_PORT='${REMOTE_CLICKHOUSE_PORT}' \
|
||||
CLICKHOUSE_USER='${REMOTE_CLICKHOUSE_USER}' CLICKHOUSE_PASSWORD='${REMOTE_CLICKHOUSE_PASSWORD}' \
|
||||
'${REMOTE_IMPORT_SCRIPT}' '${DAY}'" | tail -n1 | tr -d '[:space:]')"
|
||||
log "На домашнем сервере строк за ${DAY}: ${ROWS_REMOTE}"
|
||||
|
||||
[ "$ROWS_LOCAL" = "$ROWS_REMOTE" ] || fail "расхождение строк (VPS ${ROWS_LOCAL}, домашний ${ROWS_REMOTE}) — партиции на VPS не удаляю"
|
||||
|
||||
log "Удаляю на VPS партиции старше ${GRACE_CUTOFF}"
|
||||
OLDER="$(ch --query "SELECT DISTINCT partition FROM system.parts WHERE database='analytics' AND table='events' AND active AND partition < '${GRACE_CUTOFF}'")"
|
||||
if [ -z "$OLDER" ]; then
|
||||
log "Удалять нечего"
|
||||
else
|
||||
for partition in $OLDER; do
|
||||
log "DROP PARTITION ID '${partition}'"
|
||||
ch --query "ALTER TABLE analytics.events DROP PARTITION ID '${partition}'"
|
||||
done
|
||||
fi
|
||||
|
||||
log "Готово: события за ${DAY} в архиве, на VPS остались агрегаты и последние ${BACKUP_GRACE_DAYS} дней сырых событий"
|
||||
|
|
@ -1,58 +0,0 @@
|
|||
#!/usr/bin/env bash
|
||||
# Импорт дневного архива событий в ClickHouse домашнего сервера.
|
||||
#
|
||||
# Запускается на домашнем сервере (обычно по ssh из scripts/clickhouse_backup.sh).
|
||||
# Рядом должны лежать этот скрипт и канонический DDL кликхаус-схемы
|
||||
# (clickhouse/init/001_events.sql из репозитория).
|
||||
#
|
||||
# Использование: ./clickhouse_import.sh 2026-01-01
|
||||
# В stdout печатается количество строк загруженного дня (для сверки), диагностика — в stderr.
|
||||
#
|
||||
# Настройки (окружение):
|
||||
# BACKUP_DIR каталог, куда складываются архивы events_YYYY-MM-DD.parquet.gz
|
||||
# SCHEMA_FILE путь к DDL схемы (по умолчанию ./001_events.sql рядом со скриптом)
|
||||
# CLICKHOUSE_CLIENT бинарь клиента (по умолчанию clickhouse-client)
|
||||
# CLICKHOUSE_HOST/PORT/USER/PASSWORD доступ к ClickHouse домашнего сервера
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
DAY="${1:?укажите дату в формате YYYY-MM-DD}"
|
||||
PARTITION="${DAY//-/}"
|
||||
ARCHIVE_DIR="${BACKUP_DIR:-.}"
|
||||
ARCHIVE="${ARCHIVE_DIR}/events_${DAY}.parquet.gz"
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
SCHEMA_FILE="${SCHEMA_FILE:-${SCRIPT_DIR}/001_events.sql}"
|
||||
|
||||
CLIENT="${CLICKHOUSE_CLIENT:-clickhouse-client}"
|
||||
|
||||
log() { echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*" >&2; }
|
||||
|
||||
ch() {
|
||||
"$CLIENT" \
|
||||
--host "${CLICKHOUSE_HOST:-localhost}" \
|
||||
--port "${CLICKHOUSE_PORT:-9000}" \
|
||||
--user "${CLICKHOUSE_USER:-default}" \
|
||||
--password "${CLICKHOUSE_PASSWORD:-}" "$@"
|
||||
}
|
||||
|
||||
[ -f "$ARCHIVE" ] || {
|
||||
log "нет файла архива ${ARCHIVE}"
|
||||
exit 1
|
||||
}
|
||||
[ -f "$SCHEMA_FILE" ] || {
|
||||
log "нет файла схемы ${SCHEMA_FILE} (скопируйте clickhouse/init/001_events.sql на домашний сервер)"
|
||||
exit 1
|
||||
}
|
||||
|
||||
log "Применяю схему из ${SCHEMA_FILE}"
|
||||
ch --multiquery <"$SCHEMA_FILE"
|
||||
|
||||
# Повторный импорт того же дня не должен дублировать строки.
|
||||
log "Очищаю партицию ${PARTITION} перед импортом"
|
||||
ch --query "ALTER TABLE analytics.events DROP PARTITION ID '${PARTITION}'" 2>/dev/null || true
|
||||
|
||||
log "Импортирую ${ARCHIVE} в analytics.events"
|
||||
gzip -dc "$ARCHIVE" | ch --query "INSERT INTO analytics.events FORMAT Parquet"
|
||||
|
||||
log "Импорт завершён"
|
||||
ch --query "SELECT count() FROM analytics.events WHERE toDate(event_time) = toDate('${DAY}')"
|
||||
|
|
@ -35,28 +35,12 @@ 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,
|
||||
lvl_before bigint,
|
||||
lvl_after bigint,
|
||||
streak bigint
|
||||
)
|
||||
RETURNS TABLE(res integer, expr_add bigint, mny_add bigint, h integer, m integer, s integer)
|
||||
LANGUAGE plpgsql
|
||||
AS $$
|
||||
DECLARE
|
||||
v_lvl bigint;
|
||||
v_lvl_before bigint;
|
||||
v_expr bigint;
|
||||
v_expr_add bigint;
|
||||
v_streak bigint;
|
||||
|
|
@ -73,8 +57,6 @@ BEGIN
|
|||
FROM bot.de_users
|
||||
WHERE id_user = p_id_user;
|
||||
|
||||
v_lvl_before := v_lvl;
|
||||
|
||||
BEGIN
|
||||
SELECT dt INTO v_dt_pur
|
||||
FROM (
|
||||
|
|
@ -131,7 +113,7 @@ BEGIN
|
|||
v_result := 1;
|
||||
END IF;
|
||||
|
||||
RETURN QUERY SELECT v_result, v_expr_add, v_mny_add, v_h, v_m, v_s, v_lvl_before, v_lvl, v_streak;
|
||||
RETURN QUERY SELECT v_result, v_expr_add, v_mny_add, v_h, v_m, v_s;
|
||||
END;
|
||||
$$;
|
||||
|
||||
|
|
|
|||
|
|
@ -1,20 +0,0 @@
|
|||
"""Общие фикстуры тестов."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from common.config import Settings
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def settings_factory() -> Callable[..., Settings]:
|
||||
"""Фабрика ``Settings`` с обязательным ``BOT_TOKEN`` и без чтения ``.env``."""
|
||||
|
||||
def _factory(**overrides: Any) -> Settings:
|
||||
return Settings(_env_file=None, bot_token="1:a", **overrides)
|
||||
|
||||
return _factory
|
||||
|
|
@ -1,179 +0,0 @@
|
|||
"""Тесты уведомлений в чат логов и ежечасного пинга «бот живой»."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from collections.abc import Callable
|
||||
from datetime import UTC, datetime
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from bot.services import alerts, heartbeat
|
||||
from common.analytics import EventType, close_analytics, init_analytics
|
||||
from common.config import Settings
|
||||
from common.models import PurificationResult
|
||||
|
||||
|
||||
class FakeBot:
|
||||
"""Заглушка aiogram.Bot: записывает отправленные и пересланные сообщения."""
|
||||
|
||||
def __init__(self, *, fail: bool = False) -> None:
|
||||
self.messages: list[tuple[int, str]] = []
|
||||
self.forwarded: list[tuple[int, int, int]] = []
|
||||
self._fail = fail
|
||||
|
||||
async def send_message(self, chat_id: int, text: str) -> None:
|
||||
if self._fail:
|
||||
raise RuntimeError("Telegram недоступен")
|
||||
self.messages.append((chat_id, text))
|
||||
|
||||
async def forward_message(self, chat_id: int, from_chat_id: int, message_id: int) -> None:
|
||||
if self._fail:
|
||||
raise RuntimeError("Telegram недоступен")
|
||||
self.forwarded.append((chat_id, from_chat_id, message_id))
|
||||
|
||||
|
||||
class FakeClickHouse:
|
||||
"""Заглушка клиента ClickHouse (см. тесты аналитики)."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.rows: list[list[Any]] = []
|
||||
|
||||
async def insert(self, table: str, data: list[list[Any]], column_names: list[str]) -> None:
|
||||
self.rows.extend(data)
|
||||
|
||||
async def close(self) -> None:
|
||||
pass
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def _clean_singleton():
|
||||
await close_analytics()
|
||||
yield
|
||||
await close_analytics()
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_ping_hour(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Пинг не должен «протекать» между тестами."""
|
||||
monkeypatch.setattr(heartbeat, "_last_ping_hour", None)
|
||||
|
||||
|
||||
def _patch_settings(monkeypatch: pytest.MonkeyPatch, settings: Settings) -> Settings:
|
||||
"""Подменить настройки во всех модулях, которые их читают."""
|
||||
monkeypatch.setattr("bot.services.alerts.get_settings", lambda: settings)
|
||||
monkeypatch.setattr("bot.services.heartbeat.get_settings", lambda: settings)
|
||||
return settings
|
||||
|
||||
|
||||
# --- отправка в чат логов ---
|
||||
|
||||
|
||||
async def test_notify_alert_skips_when_chat_disabled(
|
||||
settings_factory: Callable[..., Settings], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
_patch_settings(monkeypatch, settings_factory(alert_chat_id=0))
|
||||
bot = FakeBot()
|
||||
|
||||
assert await alerts.notify_alert(bot, "текст") is False
|
||||
assert bot.messages == []
|
||||
|
||||
|
||||
async def test_notify_alert_sends_to_alert_chat(
|
||||
settings_factory: Callable[..., Settings], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
_patch_settings(monkeypatch, settings_factory(alert_chat_id=-100))
|
||||
bot = FakeBot()
|
||||
|
||||
assert await alerts.notify_alert(bot, "текст") is True
|
||||
assert bot.messages == [(-100, "текст")]
|
||||
|
||||
|
||||
async def test_notify_alert_failure_is_logged(
|
||||
settings_factory: Callable[..., Settings],
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
_patch_settings(monkeypatch, settings_factory(alert_chat_id=-100))
|
||||
|
||||
with caplog.at_level(logging.ERROR):
|
||||
assert await alerts.notify_alert(FakeBot(fail=True), "текст") is False
|
||||
|
||||
assert any("не удалось отправить уведомление" in message.lower() for message in caplog.messages)
|
||||
|
||||
|
||||
async def test_notify_alert_forwards_message(
|
||||
settings_factory: Callable[..., Settings], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
_patch_settings(monkeypatch, settings_factory(alert_chat_id=-100))
|
||||
bot = FakeBot()
|
||||
message = SimpleNamespace(chat=SimpleNamespace(id=555), message_id=42)
|
||||
|
||||
assert await alerts.notify_alert(bot, "текст", forward=message) is True
|
||||
assert bot.forwarded == [(-100, 555, 42)]
|
||||
|
||||
|
||||
async def test_notify_join_and_leave_text(
|
||||
settings_factory: Callable[..., Settings], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
_patch_settings(monkeypatch, settings_factory(alert_chat_id=-100))
|
||||
bot = FakeBot()
|
||||
|
||||
await alerts.notify_join(bot, username="vasya", user_id=7, chat_title="Клан")
|
||||
await alerts.notify_leave(bot, username=None, user_id=7, chat_title="Клан")
|
||||
|
||||
assert bot.messages[0] == (-100, "[вход] @vasya вошёл в чат «Клан»")
|
||||
assert bot.messages[1] == (-100, "[выход] 7 покинул чат «Клан»")
|
||||
|
||||
|
||||
async def test_notify_purification_text(
|
||||
settings_factory: Callable[..., Settings], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
_patch_settings(monkeypatch, settings_factory(alert_chat_id=-100))
|
||||
bot = FakeBot()
|
||||
|
||||
result = PurificationResult(success=True, exp_add=3, money_add=41)
|
||||
await alerts.notify_purification(bot, username="vasya", user_id=7, result=result)
|
||||
|
||||
assert bot.messages == [(-100, "[очищение] @vasya — опыт +3, монеты +41")]
|
||||
|
||||
|
||||
# --- ежечасный пинг «бот живой» ---
|
||||
|
||||
|
||||
async def test_ping_hourly_sends_once_per_hour(
|
||||
settings_factory: Callable[..., Settings], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
settings = _patch_settings(monkeypatch, settings_factory(alert_chat_id=-100, app_env="prod"))
|
||||
fake = FakeClickHouse()
|
||||
await init_analytics(settings, client=fake)
|
||||
bot = FakeBot()
|
||||
|
||||
assert await heartbeat.ping_hourly(bot, now=datetime(2026, 1, 1, 10, 0, tzinfo=UTC)) is True
|
||||
assert await heartbeat.ping_hourly(bot, now=datetime(2026, 1, 1, 10, 30, tzinfo=UTC)) is False
|
||||
assert await heartbeat.ping_hourly(bot, now=datetime(2026, 1, 1, 11, 5, tzinfo=UTC)) is True
|
||||
|
||||
await close_analytics()
|
||||
|
||||
assert len(bot.messages) == 2
|
||||
assert "бот живой" in bot.messages[0][1]
|
||||
assert [row[1] for row in fake.rows] == [EventType.BOT_ALIVE, EventType.BOT_ALIVE]
|
||||
|
||||
|
||||
async def test_ping_hourly_records_event_without_alert_chat(
|
||||
settings_factory: Callable[..., Settings], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
settings = _patch_settings(monkeypatch, settings_factory(alert_chat_id=0))
|
||||
fake = FakeClickHouse()
|
||||
await init_analytics(settings, client=fake)
|
||||
bot = FakeBot()
|
||||
|
||||
assert await heartbeat.ping_hourly(bot, now=datetime(2026, 1, 1, 10, 0, tzinfo=UTC)) is False
|
||||
assert await heartbeat.ping_hourly(bot, now=datetime(2026, 1, 1, 10, 45, tzinfo=UTC)) is False
|
||||
|
||||
await close_analytics()
|
||||
|
||||
assert bot.messages == []
|
||||
assert [row[1] for row in fake.rows] == [EventType.BOT_ALIVE]
|
||||
|
|
@ -1,351 +0,0 @@
|
|||
"""Тесты аналитики: каталог событий, буферизация и деградация без ClickHouse.
|
||||
|
||||
Тесты не требуют ни ClickHouse, ни aiogram: клиент подменяется фейком, а
|
||||
``clickhouse_connect`` импортируется лениво внутри клиента.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
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__
|
||||
|
||||
_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: пишет в список или падает."""
|
||||
|
||||
def __init__(self, *, fail: bool = False) -> None:
|
||||
self.rows: list[list[Any]] = []
|
||||
self.table: str | None = None
|
||||
self.columns: list[str] | None = None
|
||||
self.closed = False
|
||||
self._fail = fail
|
||||
|
||||
async def insert(self, table: str, data: list[list[Any]], column_names: list[str]) -> None:
|
||||
if self._fail:
|
||||
raise RuntimeError("ClickHouse недоступен")
|
||||
self.table = table
|
||||
self.columns = list(column_names)
|
||||
self.rows.extend(data)
|
||||
|
||||
async def close(self) -> None:
|
||||
self.closed = True
|
||||
|
||||
|
||||
def _settings(**overrides: Any) -> Settings:
|
||||
return Settings(_env_file=None, bot_token="1:a", **overrides)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def _clean_singleton():
|
||||
"""Тесты не должны оставлять активный singleton аналитики."""
|
||||
await close_analytics()
|
||||
yield
|
||||
await close_analytics()
|
||||
|
||||
|
||||
def _client(**overrides: Any) -> tuple[AnalyticsClient, FakeClickHouse]:
|
||||
fake = FakeClickHouse()
|
||||
return AnalyticsClient(_settings(**overrides), client=fake), fake
|
||||
|
||||
|
||||
# --- каталог событий ---
|
||||
|
||||
|
||||
def test_event_types_are_unique_and_nonempty() -> None:
|
||||
assert len(ALL_EVENT_TYPES) == len(EventType)
|
||||
assert ALL_EVENT_TYPES
|
||||
|
||||
|
||||
def test_event_types_match_documentation() -> None:
|
||||
documented = set(_EVENT_ROW.findall(_DOCS_PATH.read_text(encoding="utf-8")))
|
||||
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
|
||||
|
||||
|
||||
# --- буферизация и запись ---
|
||||
|
||||
|
||||
async def test_record_without_init_is_noop() -> None:
|
||||
assert get_analytics() is None
|
||||
record(EventType.MESSAGE, user_id=1, chat_id=2, payload={"a": 1}) # не должно бросать
|
||||
|
||||
|
||||
async def test_client_buffers_and_flushes_rows() -> None:
|
||||
client, fake = _client()
|
||||
client.record(EventType.MESSAGE, user_id=11, username="vasya", chat_id=100, payload={"message_id": 5})
|
||||
client.record(EventType.COMMAND, user_id=11, username="vasya", chat_id=100, payload={"command": "/инфо"})
|
||||
|
||||
assert client.pending == 2
|
||||
|
||||
await client.flush()
|
||||
|
||||
assert client.pending == 0
|
||||
assert fake.table == "analytics.events"
|
||||
assert fake.columns == [
|
||||
"event_time",
|
||||
"event_type",
|
||||
"user_id",
|
||||
"username",
|
||||
"chat_id",
|
||||
"payload",
|
||||
"bot_version",
|
||||
"prod",
|
||||
]
|
||||
assert len(fake.rows) == 2
|
||||
|
||||
row = fake.rows[0]
|
||||
assert isinstance(row[0], datetime)
|
||||
assert row[1] == "message"
|
||||
assert row[2] == 11
|
||||
assert row[3] == "vasya"
|
||||
assert row[4] == 100
|
||||
assert '"message_id": 5' in row[5]
|
||||
assert row[6] == __version__
|
||||
assert row[7] == 0
|
||||
|
||||
|
||||
async def test_remember_user_enriches_username() -> None:
|
||||
client, fake = _client()
|
||||
client.remember_user(42, "nickname")
|
||||
client.record(EventType.PURIFICATION, user_id=42, chat_id=1)
|
||||
|
||||
await client.flush()
|
||||
|
||||
assert fake.rows[0][3] == "nickname"
|
||||
|
||||
|
||||
async def test_record_triggers_flush_at_batch_size() -> None:
|
||||
client, _ = _client(analytics_batch_size=3)
|
||||
|
||||
client.record(EventType.MESSAGE, user_id=1)
|
||||
client.record(EventType.MESSAGE, user_id=1)
|
||||
assert not client._flush_now.is_set()
|
||||
|
||||
client.record(EventType.MESSAGE, user_id=1)
|
||||
assert client._flush_now.is_set()
|
||||
|
||||
|
||||
async def test_overflow_drops_oldest_events() -> None:
|
||||
client, _ = _client(analytics_buffer_max=2, analytics_batch_size=100)
|
||||
|
||||
for user_id in range(1, 5):
|
||||
client.record(EventType.MESSAGE, user_id=user_id)
|
||||
|
||||
assert client.pending == 2
|
||||
assert [row[2] for row in client._rows] == [3, 4]
|
||||
|
||||
|
||||
async def test_disabled_analytics_ignores_events() -> None:
|
||||
client, fake = _client(clickhouse_enabled=False)
|
||||
|
||||
client.record(EventType.MESSAGE, user_id=1)
|
||||
await client.flush()
|
||||
|
||||
assert client.pending == 0
|
||||
assert fake.rows == []
|
||||
|
||||
|
||||
# --- деградация без ClickHouse ---
|
||||
|
||||
|
||||
async def test_flush_failure_logs_fallback_and_drops(caplog: pytest.LogCaptureFixture) -> None:
|
||||
fake = FakeClickHouse(fail=True)
|
||||
client = AnalyticsClient(_settings(), client=fake)
|
||||
client.record(EventType.MESSAGE, user_id=1, username="vasya", chat_id=100)
|
||||
|
||||
with caplog.at_level(logging.WARNING):
|
||||
await client.flush() # не должно бросать
|
||||
|
||||
assert client.pending == 0
|
||||
assert any("Fallback" in message for message in caplog.messages)
|
||||
|
||||
|
||||
async def test_unavailable_client_falls_back_to_log(caplog: pytest.LogCaptureFixture) -> None:
|
||||
settings = _settings(clickhouse_host="127.0.0.1", clickhouse_port=1, clickhouse_password="x")
|
||||
client = AnalyticsClient(settings)
|
||||
client.record(EventType.BOT_START, payload={"app_env": "dev"})
|
||||
|
||||
with caplog.at_level(logging.WARNING):
|
||||
await client.flush()
|
||||
|
||||
assert client.pending == 0
|
||||
assert any("Fallback" in message for message in caplog.messages)
|
||||
|
||||
|
||||
async def test_payload_is_truncated() -> None:
|
||||
client, fake = _client()
|
||||
client.record(EventType.MESSAGE, user_id=1, payload={"text": "x" * (PAYLOAD_MAX_LENGTH * 2)})
|
||||
|
||||
await client.flush()
|
||||
|
||||
assert len(fake.rows[0][5]) == PAYLOAD_MAX_LENGTH
|
||||
|
||||
|
||||
# --- жизненный цикл ---
|
||||
|
||||
|
||||
async def test_init_and_close_flush_pending_events() -> None:
|
||||
fake = FakeClickHouse()
|
||||
client = await init_analytics(_settings(), client=fake)
|
||||
|
||||
record(EventType.BOT_START, payload={"app_env": "dev"})
|
||||
record(EventType.BOT_STOP, payload={"app_env": "dev"})
|
||||
|
||||
assert get_analytics() is client
|
||||
assert client.pending == 2
|
||||
|
||||
await close_analytics()
|
||||
|
||||
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"
|
||||
|
|
@ -7,4 +7,4 @@ from common import __version__
|
|||
|
||||
def test_version_is_set():
|
||||
assert __version__
|
||||
assert __version__ == "0.3.6"
|
||||
assert __version__ == "0.2.5"
|
||||
|
|
|
|||
Loading…
Reference in New Issue