From 84f75e7c7d83f1389d3f4af76f73d548b2b38987 Mon Sep 17 00:00:00 2001 From: MpakobecEx Date: Fri, 16 Oct 2026 15:19:58 +0500 Subject: [PATCH] =?UTF-8?q?=D0=9A=D0=BB=D0=B8=D0=B5=D0=BD=D1=82=20ClickHou?= =?UTF-8?q?se,=20=D0=BA=D0=B0=D1=82=D0=B0=D0=BB=D0=BE=D0=B3=20=D1=81=D0=BE?= =?UTF-8?q?=D0=B1=D1=8B=D1=82=D0=B8=D0=B9=20=D0=B8=20=D0=B1=D0=B0=D1=82?= =?UTF-8?q?=D1=87=D0=B8=D0=BD=D0=B3=20=D0=B7=D0=B0=D0=BF=D0=B8=D1=81=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - common/analytics — AnalyticsClient с буфером, flush по размеру и таймеру, graceful-flush при остановке - при недоступности ClickHouse события уходят в stdout, бот не падает - каталог EventType и docs/events.md проверяются тестом на синхронность - настройки CLICKHOUSE_* и ANALYTICS_* с дефолтами - версия 0.3.0 --- .env.example | 14 ++ README.md | 2 +- common/analytics/__init__.py | 33 ++++ common/analytics/client.py | 301 +++++++++++++++++++++++++++++++++++ common/analytics/events.py | 63 ++++++++ common/config.py | 13 ++ common/version.py | 2 +- docs/events.md | 82 ++++++++++ requirements.txt | 4 +- tests/test_analytics.py | 215 +++++++++++++++++++++++++ tests/test_version.py | 2 +- 11 files changed, 726 insertions(+), 5 deletions(-) create mode 100644 common/analytics/__init__.py create mode 100644 common/analytics/client.py create mode 100644 common/analytics/events.py create mode 100644 docs/events.md create mode 100644 tests/test_analytics.py diff --git a/.env.example b/.env.example index 0f157f5..cbecffc 100644 --- a/.env.example +++ b/.env.example @@ -28,3 +28,17 @@ 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 diff --git a/README.md b/README.md index ad37202..7358dae 100644 --- a/README.md +++ b/README.md @@ -73,7 +73,7 @@ pytest ## Версионирование -- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.2.5`. +- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.3.0`. - Схема — SemVer-подобная (`MAJOR.MINOR.PATCH`); git-тег = `v` + версия (`v0.1`). - Docker-образы тегируются версией (Блок 2), события ClickHouse получают `bot_version` (Блок 3). - Выпуск: изменить `common/version.py` → коммит → аннотированный тег `vX.Y` → `git push --tags`. diff --git a/common/analytics/__init__.py b/common/analytics/__init__.py new file mode 100644 index 0000000..c67b1ad --- /dev/null +++ b/common/analytics/__init__.py @@ -0,0 +1,33 @@ +"""Аналитика: клиент 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", +] diff --git a/common/analytics/client.py b/common/analytics/client.py new file mode 100644 index 0000000..8c63c71 --- /dev/null +++ b/common/analytics/client.py @@ -0,0 +1,301 @@ +"""Клиент 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) diff --git a/common/analytics/events.py b/common/analytics/events.py new file mode 100644 index 0000000..fd98087 --- /dev/null +++ b/common/analytics/events.py @@ -0,0 +1,63 @@ +"""Каталог событий аналитики. + +Единственный источник правды по типам событий 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"] diff --git a/common/config.py b/common/config.py index b1ddd30..d8f56ca 100644 --- a/common/config.py +++ b/common/config.py @@ -53,6 +53,19 @@ 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: diff --git a/common/version.py b/common/version.py index 2003d97..3d5b690 100644 --- a/common/version.py +++ b/common/version.py @@ -7,4 +7,4 @@ from __future__ import annotations -__version__ = "0.2.5" +__version__ = "0.3.0" diff --git a/docs/events.md b/docs/events.md new file mode 100644 index 0000000..7477a37 --- /dev/null +++ b/docs/events.md @@ -0,0 +1,82 @@ +# События аналитики (ClickHouse) + +Бот пишет все значимые действия в ClickHouse — единую широкую таблицу +`analytics.events`. Дашборд (Superset) и ETL — отдельный финальный этап; здесь +описан **контракт** событий: какие бывают `event_type` и что лежит в `payload`. + +> Источник правды в коде — `common/analytics/events.py` (`EventType`). +> Расхождение со списком ниже ловит `tests/test_analytics.py`. +> Добавляешь событие — добавь строку в таблицу. + +## Поля события + +| Поле | Тип | Описание | +| --- | --- | --- | +| **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; +``` diff --git a/requirements.txt b/requirements.txt index 8948810..dfd864f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,7 +7,7 @@ pydantic-settings>=2.3,<3.0 python-dotenv>=1.0,<2.0 psycopg[binary,pool]>=3.2,<4.0 -# Блок 3 (ClickHouse) — подключим позже: -# clickhouse-connect>=0.7,<1.0 +# ClickHouse (аналитика событий, Блок 3). Extras [async] — нативный asyncio-клиент. +clickhouse-connect[async]>=1.0,<2.0 # Блок 4 (расписания из БД) — при необходимости: # APScheduler>=3.10,<4.0 diff --git a/tests/test_analytics.py b/tests/test_analytics.py new file mode 100644 index 0000000..9e9d364 --- /dev/null +++ b/tests/test_analytics.py @@ -0,0 +1,215 @@ +"""Тесты аналитики: каталог событий, буферизация и деградация без ClickHouse. + +Тесты не требуют ни ClickHouse, ни aiogram: клиент подменяется фейком, а +``clickhouse_connect`` импортируется лениво внутри клиента. +""" + +from __future__ import annotations + +import logging +import re +from datetime import datetime +from pathlib import Path +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.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 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 разошёлся" + + +# --- буферизация и запись --- + + +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 diff --git a/tests/test_version.py b/tests/test_version.py index 1015e50..9a3d7c9 100644 --- a/tests/test_version.py +++ b/tests/test_version.py @@ -7,4 +7,4 @@ from common import __version__ def test_version_is_set(): assert __version__ - assert __version__ == "0.2.5" + assert __version__ == "0.3.0"