From d7bbacb67cfbf27aa9147edadb0b35bb06cd01ba Mon Sep 17 00:00:00 2001 From: MpakobecEx Date: Tue, 3 Nov 2026 02:32:31 +0500 Subject: [PATCH] =?UTF-8?q?=D0=90=D0=BD=D0=B0=D0=BB=D0=B8=D1=82=D0=B8?= =?UTF-8?q?=D0=BA=D0=B0=20=D0=BF=D0=B5=D1=80=D0=B5=D0=B5=D1=85=D0=B0=D0=BB?= =?UTF-8?q?=D0=B0=20=D0=B2=20Postgres,=20ClickHouse=20=D0=B1=D0=BE=D0=BB?= =?UTF-8?q?=D1=8C=D1=88=D0=B5=20=D0=BD=D0=B5=20=D0=BD=D1=83=D0=B6=D0=B5?= =?UTF-8?q?=D0=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Логи сообщений больше не дублируются отдельным событием: вместо этого в схеме analytics растёт дневной счётчик по каждому игроку. События и счётчики хранятся в Postgres, партиции по дням создаются автоматически и не удаляются, а в админке появилась отдельная вкладка «Сообщения по дням». Сервис ClickHouse, скрипты его бэкапа и зависимость убраны; сбой аналитики по-прежнему не мешает боту. --- .env.example | 12 +- DEPLOY.md | 99 +++------ Makefile | 10 +- README.md | 58 +++-- admin/app.py | 7 +- admin/routes/logs.py | 81 ++++--- admin/routes/pages.py | 2 +- admin/templates/logs/list.html | 65 +++++- bot/__main__.py | 14 +- bot/middlewares/analytics.py | 12 +- bot/services/heartbeat.py | 2 +- clickhouse/config.d/disable-trace-log.xml | 8 - clickhouse/init/001_events.sql | 72 ------- common/analytics/__init__.py | 4 +- common/analytics/client.py | 252 +++++++++++----------- common/analytics/events.py | 5 +- common/analytics/partitions.py | 64 ++++++ common/analytics/read.py | 158 +++++++------- common/config.py | 11 +- common/version.py | 4 +- docker-compose.prod.yml | 39 ---- docker-compose.yml | 35 --- docs/events.md | 101 ++++----- env/backup.env.example | 32 --- env/prod.env.example | 10 +- env/test.env.example | 10 +- requirements-admin.txt | 3 - requirements.txt | 2 - scripts/check_contour.sh | 16 +- scripts/clickhouse_backup.sh | 129 ----------- scripts/clickhouse_import.sh | 58 ----- sql/V26__analytics_schema.sql | 33 +++ tests/test_admin_logs.py | 41 +++- tests/test_alerts.py | 25 +-- tests/test_analytics.py | 170 +++++++-------- tests/test_version.py | 2 +- 36 files changed, 688 insertions(+), 958 deletions(-) delete mode 100644 clickhouse/config.d/disable-trace-log.xml delete mode 100644 clickhouse/init/001_events.sql create mode 100644 common/analytics/partitions.py delete mode 100644 env/backup.env.example delete mode 100755 scripts/clickhouse_backup.sh delete mode 100755 scripts/clickhouse_import.sh create mode 100644 sql/V26__analytics_schema.sql diff --git a/.env.example b/.env.example index f0d9736..92d5c88 100644 --- a/.env.example +++ b/.env.example @@ -35,15 +35,9 @@ 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= +# Аналитика (Postgres, схема analytics). Логируются действия бота (см. docs/events.md). +# ANALYTICS_ENABLED=false — полностью выключить аналитику. +ANALYTICS_ENABLED=true # Батчинг записи событий: размер батча, интервал сброса (сек), предел буфера. ANALYTICS_BATCH_SIZE=200 ANALYTICS_FLUSH_INTERVAL=5.0 diff --git a/DEPLOY.md b/DEPLOY.md index c729a7e..97c79b0 100644 --- a/DEPLOY.md +++ b/DEPLOY.md @@ -16,16 +16,14 @@ cd DispelEvilBot ```bash cp env/test.env.example env/test.env # заполнить: BOT_TOKEN, DB_PASSWORD, CHAT_ID (основной чат), ALERT_CHAT_ID (опц.), -# ADMIN_IDS (опц.), CLICKHOUSE_PASSWORD (аналитика событий) +# ADMIN_IDS (опц.) # плюс для админки: BOT_USERNAME, ADMIN_PUBLIC_URL (HTTPS-домен), # ADMIN_SESSION_SECRET (openssl rand -hex 32) ``` -Порт ClickHouse на хосте (`CLICKHOUSE_HTTP_PORT`) у контуров должен различаться: -`prod` — `8123`, `test` — `8124` (иначе контейнеры конфликтуют на одной VPS). -Порт админки (`ADMIN_HTTP_PORT`) — так же: `prod` — `8101`, `test` — `8100` -(8000 на VPS занят Portainer). `ADMIN_BIND_IP` по умолчанию `127.0.0.1`; чтобы -nginx-proxy-manager (контейнер) видел админку, ставьте `172.17.0.1`. +Порт админки (`ADMIN_HTTP_PORT`) у контуров должен различаться: `prod` — `8101`, +`test` — `8100` (8000 на VPS занят Portainer). `ADMIN_BIND_IP` по умолчанию `127.0.0.1`; +чтобы nginx-proxy-manager (контейнер) видел админку, ставьте `172.17.0.1`. ## 3. Логин в registry (если пакет приватный) @@ -55,35 +53,36 @@ 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=...)`), аналитика и рассылки — в нём же; -5. `admin` стартует и отвечает на `GET /healthz` (порт на loopback хоста). +3. `bot` стартует (`Запуск DispelEvilBot v<версия> (APP_ENV=...)`), аналитика и рассылки — в нём же; +4. `admin` стартует и отвечает на `GET /healthz` (порт на loopback хоста). -Проверка контура (сервисы, миграции, Postgres, ClickHouse): +Проверка контура (сервисы, миграции, Postgres, схема аналитики): ```bash make check E=test ``` -## ClickHouse (аналитика событий) +## Аналитика (Postgres) -События всех действий бота пишутся в `analytics.events` (контракт — [`docs/events.md`](docs/events.md)). -ClickHouse наружу **не публикуется**: только loopback хоста (`127.0.0.1:${CLICKHOUSE_HTTP_PORT}`). +События и дневные счётчики сообщений пишутся в схему `analytics` той же БД Postgres +(контракт — [`docs/events.md`](docs/events.md)). Отдельного сервиса нет. ```bash -make ch E=test # clickhouse-client контура +make psql E=test # пример: события за сегодня по типам -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"' +SELECT event_type, count(*) FROM analytics.events +WHERE event_time >= current_date GROUP BY event_type; + +# сообщения по дням +SELECT day, sum(count) FROM analytics.messages_daily +WHERE day >= current_date - 7 GROUP BY day ORDER BY day; ``` -Если ClickHouse недоступен, бот продолжает работать: события уходят в stdout (WARNING), -полностью выключить можно флагом `CLICKHOUSE_ENABLED=false`. +Если Postgres недоступен, бот продолжает работать: батч аналитики уходит в stdout +(WARNING) и отбрасывается. Полностью выключить можно флагом `ANALYTICS_ENABLED=false`. -Изменения схемы: `clickhouse/init/*.sql` применяются лишь на пустом data-каталоге, -поэтому на существующем контуре DDL нужно применить вручную (`make ch E=...`). +Партиции создаются приложением при старте и раз в 6 часов; старые **не удаляются**. +Схема создаётся миграцией `V26__analytics_schema.sql`. ## Админка (FastAPI) @@ -119,51 +118,11 @@ reverse proxy. Сессия живёт 12 часов (`ADMIN_SESSION_MAX_AGE`), Роль администратора перечитывается из БД на каждом запросе — отключение пользователя (`enabled = false`) или удаление записи сразу закрывает доступ. -## Бэкап событий (забирает домашний сервер) +## Бэкап -Сырые события живут на VPS ограниченное время: раз в сутки домашний сервер **сам забирает** -вчерашнюю партицию (`pull`), складывает её в `Parquet.gz`, импортирует в свой ClickHouse и -сверяет количество строк — **только после сверки** на VPS удаляются партиции старше -`BACKUP_GRACE_DAYS` (по умолчанию 7). На VPS остаются агрегаты (`events_daily`, `money_daily`). - -Входящий SSH на домашний сервер не нужен: он инициирует соединение сам. - -Разовая настройка: - -```bash -# на домашнем сервере -git clone ssh://git@:2222/mpakobecex/DispelEvilBot.git -cd DispelEvilBot -cp env/backup.env.example env/backup.env # VPS_SSH, VPS_PORT, VPS_PROJECT_DIR, VPS_SSH_KEY, BACKUP_DIR, CLICKHOUSE_* -``` - -Требования: на VPS — пользователь с доступом к `docker` и каталог с репозиторием -(`VPS_PROJECT_DIR`, там запускается `docker compose`); на домашнем сервере — `clickhouse-client` -и локальный ClickHouse. Настройка SSH-доступа — `deploy/ssh-access.md`. - -Если ставить `clickhouse-client` на хост не хочется, укажите в `env/backup.env` переменную -`CLICKHOUSE_CLIENT` — путь к обёртке, которая выполняет клиент внутри контейнера: - -```bash -#!/usr/bin/env bash -set -euo pipefail -cd "$(dirname "$0")/.." -exec docker compose --env-file env/backup.env -f docker-compose.yml exec -T clickhouse clickhouse-client "$@" -``` - -Проверочный прогон (без cron): `make backup E=test` или `./scripts/clickhouse_backup.sh test`. - -Cron на домашнем сервере (вчерашний день уже закрыт, время московское): - -```cron -0 4 * * * cd /srv/DispelEvilBot && ./scripts/clickhouse_backup.sh test >> /var/log/clickhouse-backup-test.log 2>&1 -``` - -Ошибка на любом шаге оставляет данные на VPS (партиции не удаляются) и пишет понятный лог. -Импорт дня идемпотентен: повторный прогон не дублирует строки. - -Если контур на VPS не запущен (нет работающего `clickhouse` или не найден каталог проекта), -скрипт пишет об этом в лог и завершается с кодом 0 — для cron это не ошибка. +Аналитика теперь лежит в Postgres, поэтому отдельного бэкапа событий больше нет. +Бэкап БД — стандартными средствами Postgres (`pg_dump`/`pg_basebackup`); при росте +истории следите за размером `analytics` и диском. ## Обновление и откат @@ -184,13 +143,13 @@ docker compose -p <контур> --env-file env/<контур>.env -f docker-com ## Переменные окружения -См. `env/prod.env.example` / `env/test.env.example` (контуры) и `env/backup.env.example` (бэкап событий). +См. `env/prod.env.example` / `env/test.env.example` (контуры). ## Дашборд (позже) -Superset поднимается на домашнем сервере и читает **локальный** ClickHouse с полной -историей событий (сюда ежедневно приезжают архивы). Отдельный read-only пользователь -ClickHouse и сам дашборд — финальный этап, в этот контур не входят. +Superset поднимается отдельно и читает Postgres (схема `analytics`) с полной историей +событий. Отдельный read-only пользователь Postgres и сам дашборд — финальный этап, +в этот контур не входят. ## Перенос данных на прод (позже, короткое окно простоя) diff --git a/Makefile b/Makefile index 106cc8b..fb0bcf2 100644 --- a/Makefile +++ b/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 бэкап событий с VPS (запускать на домашнем сервере)" 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) diff --git a/README.md b/README.md index abc75f8..db93412 100644 --- a/README.md +++ b/README.md @@ -9,10 +9,9 @@ Telegram-бот клана Dispel Evil (aiogram 3 + Postgres). Монорепо common/ # общий пакет: config, logging, db, repositories, models, texts, analytics bot/ # сервис бота: handlers, services, keyboards, middlewares, notifications admin/ # сервис админки: FastAPI + Jinja2/HTMX (Блок 4), свой Dockerfile -sql/ # миграции Flyway (V__*.sql, R__*.sql) -clickhouse/ # init-скрипты ClickHouse (схема аналитики) +sql/ # миграции Flyway (V__*.sql, R__*.sql), включая схему analytics docs/ # документация (docs/events.md — события аналитики) -scripts/ # эксплуатационные скрипты (проверка контура, бэкап событий) +scripts/ # эксплуатационные скрипты (проверка контура) images/ # статичные картинки (приветствие) deploy/ # материалы деплоя (Caddyfile, инструкции) tests/ # тесты @@ -24,8 +23,7 @@ docker-compose.yml / docker-compose.prod.yml ## Требования - Python 3.12 -- PostgreSQL 14+ -- ClickHouse 25.3 (аналитика событий; можно выключить — `CLICKHOUSE_ENABLED=false`) +- PostgreSQL 14+ (аналитика — в схеме `analytics`) ## Настройка @@ -88,10 +86,10 @@ HttpOnly-cookie (`ADMIN_SESSION_SECRET`), все формы защищены CSR видны боту (он читает `bot.titles`). Удалить можно только звание, которое ни у кого не стоит; используемое — скрывается (`visible = false`). -Раздел «Логи» — события из ClickHouse: фильтры по типу, пользователю (ID или username) -и датам, пагинация по 50. Время показывается в `TIMEZONE`, payload раскрывается как JSON, -ID пользователя ведёт в его карточку. ClickHouse читается отдельным ленивым клиентом; -если он недоступен, страница показывает ошибку, а не падает. +Раздел «Логи» — события из Postgres и отдельная вкладка «Сообщения по дням»: фильтры +по типу (для событий), пользователю (ID или username) и датам, пагинация по 50. +Время показывается в `TIMEZONE`, payload раскрывается как JSON, ID пользователя ведёт +в его карточку. Если Postgres недоступен, страница показывает ошибку, а не падает. Раздел «Админы» — выдача и отзыв доступа к панели по Telegram-ID, роли и вкл/выкл. В карточке пользователя есть исключение из чата (бан + разбан) с удалением данных и @@ -113,16 +111,18 @@ ID пользователя ведёт в его карточку. ClickHouse ч редактором, картинка и URL-кнопки, предпросмотр и подтверждение. Каждая отправка (успешная или нет) пишется в `bot.admin_messages`. -## Аналитика (ClickHouse) +## Аналитика (Postgres) -Все значимые действия бота пишутся в ClickHouse в единую таблицу `analytics.events`. -Контракт событий — [`docs/events.md`](docs/events.md) (типы, payload, примеры запросов). +Все значимые действия бота пишутся в Postgres, схема `analytics`: +`analytics.events` (события, партиции по дням) и `analytics.messages_daily` +(количество сообщений пользователя по дням). Контракт событий — +[`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`. +- схема — `sql/V26__analytics_schema.sql`; +- клиент и батчинг — `common/analytics/` (запись неблокирующая, сбой Postgres не влияет на бота); +- сообщения не пишутся поштучно: на каждое сообщение инкрементится дневной счётчик; +- партиции создаются приложением и **не удаляются** — история хранится бессрочно; +- выключается флагом `ANALYTICS_ENABLED=false`. ## Уведомления в чат логов @@ -131,7 +131,7 @@ ID пользователя ведёт в его карточку. ClickHouse ч ## Docker -Локальный запуск (postgres + migrate + clickhouse + bot + admin): +Локальный запуск (postgres + migrate + bot + admin): ```bash cp .env.example .env # заполнить @@ -162,22 +162,17 @@ black --check . pytest ``` -## Бэкап событий +## Бэкап -Сырые события ClickHouse ежедневно забираются с VPS домашним сервером (`pull`), на VPS -остаются только агрегаты и последние дни сырых данных: - -```bash -make backup E=test # на домашнем сервере: ./scripts/clickhouse_backup.sh test -``` - -Настройка, cron и SSH-доступ — см. [`DEPLOY.md`](DEPLOY.md). +Аналитика теперь в Postgres, отдельного бэкапа событий нет. Бэкап БД — стандартными +средствами Postgres; история `analytics` не чистится, поэтому следите за размером БД +и диском. Подробности — [`DEPLOY.md`](DEPLOY.md). ## Версионирование -- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.6.9`. +- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.7.0`. - Схема — SemVer-подобная (`MAJOR.MINOR.PATCH`); git-тег = `v` + версия (`v0.1`). -- Docker-образы тегируются версией (Блок 2), события ClickHouse получают `bot_version` (Блок 3). +- Docker-образы тегируются версией (Блок 2), события аналитики получают `bot_version`. - Выпуск: изменить `common/version.py` → коммит → аннотированный тег `vX.Y` → `git push --tags`. ## Процесс работы с git @@ -231,6 +226,5 @@ Telegram не принимает кириллицу в именах команд ## Переменные окружения -- общие — `.env.example` (включая `CLICKHOUSE_*` и параметры батчинга `ANALYTICS_*`); -- контуры — `env/prod.env.example`, `env/test.env.example`; -- бэкап событий — `env/backup.env.example`. +- общие — `.env.example` (включая `ANALYTICS_*`); +- контуры — `env/prod.env.example`, `env/test.env.example`. diff --git a/admin/app.py b/admin/app.py index 6d94e3e..027d888 100644 --- a/admin/app.py +++ b/admin/app.py @@ -16,7 +16,7 @@ from admin.media import ensure_media from admin.middleware import AdminAuthMiddleware from admin.routes import admins, auth, health, logs, pages, preview, schedules, send, sequences, titles, users from common import __version__ -from common.analytics import read as analytics_read +from common.analytics.partitions import ensure_partitions from common.config import AdminSettings, get_admin_settings from common.db.pool import close_pool, init_pool from common.logging_config import setup_logging @@ -42,6 +42,10 @@ def create_app(settings: AdminSettings | None = None, *, init_db: bool = True) - if init_db: await init_pool(app.state.settings) ensure_media(app.state.settings) + try: + await ensure_partitions(app.state.settings) + except Exception: + logger.warning("Не удалось создать партиции аналитики", exc_info=True) try: inserted = await admins_repo.ensure_admins(app.state.settings.admin_ids) if inserted: @@ -56,7 +60,6 @@ def create_app(settings: AdminSettings | None = None, *, init_db: bool = True) - finally: if init_db: await close_pool() - await analytics_read.close_client() app = FastAPI( title="DispelEvilBot Admin", diff --git a/admin/routes/logs.py b/admin/routes/logs.py index 2fe0a25..4373f9a 100644 --- a/admin/routes/logs.py +++ b/admin/routes/logs.py @@ -1,4 +1,4 @@ -"""Раздел «Логи»: события из ClickHouse с фильтрами и пагинацией.""" +"""Раздел «Логи»: события и дневные счётчики сообщений из Postgres.""" from __future__ import annotations @@ -19,16 +19,18 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/logs", tags=["logs"]) -#: Сколько событий показывать на странице. +#: Сколько строк показывать на странице. PAGE_SIZE = 50 -def _pretty_payload(payload: str) -> str: +def _pretty_payload(payload: Any) -> str: """Отформатировать JSON-payload для просмотра; битый JSON — как есть.""" + if isinstance(payload, dict): + return json.dumps(payload, ensure_ascii=False, indent=2) try: - return json.dumps(json.loads(payload), ensure_ascii=False, indent=2) + return json.dumps(json.loads(str(payload or "")), ensure_ascii=False, indent=2) except (TypeError, ValueError): - return payload or "" + return str(payload or "") def _decorate(row: dict[str, Any], settings: Any) -> dict[str, Any]: @@ -44,66 +46,85 @@ def _decorate(row: dict[str, Any], settings: Any) -> dict[str, Any]: return { **row, "time_local": time_local, - "payload_pretty": _pretty_payload(str(row.get("payload") or "")), + "payload_pretty": _pretty_payload(row.get("payload")), + } + + +def _decorate_message(row: dict[str, Any]) -> dict[str, Any]: + day = row.get("day") + return { + **row, + "day_local": day.strftime("%d.%m.%Y") if hasattr(day, "strftime") else str(day or "—"), } @router.get("", response_class=HTMLResponse) async def logs_page( request: Request, + view: str = "events", event_type: str = "", user: str = "", date_from: str = "", date_to: str = "", page: int = 1, ) -> HTMLResponse: - """Показать события с фильтрами по типу, пользователю и датам.""" + """Показать события или дневные счётчики сообщений с фильтрами.""" settings = request.app.state.settings + view = view if view in {"events", "messages"} else "events" today = datetime.now(settings.tzinfo).date() date_from = date_from or (today - timedelta(days=6)).isoformat() date_to = date_to or today.isoformat() page = max(page, 1) events: list[dict[str, Any]] = [] + messages: list[dict[str, Any]] = [] total = 0 error: str | None = None try: - events, total = await analytics_read.fetch_events( - settings, - event_type=event_type, - user=user, - date_from=date_from, - date_to=date_to, - limit=PAGE_SIZE, - offset=(page - 1) * PAGE_SIZE, - ) + if view == "messages": + messages, total = await analytics_read.fetch_messages( + settings, + user=user, + date_from=date_from, + date_to=date_to, + limit=PAGE_SIZE, + offset=(page - 1) * PAGE_SIZE, + ) + else: + events, total = await analytics_read.fetch_events( + settings, + event_type=event_type, + user=user, + date_from=date_from, + date_to=date_to, + limit=PAGE_SIZE, + offset=(page - 1) * PAGE_SIZE, + ) except ValueError: error = "Даты должны быть в формате ГГГГ-ММ-ДД." except Exception: - logger.exception("Не удалось прочитать события из ClickHouse") - error = "ClickHouse недоступен — логи временно не читаются." + logger.exception("Не удалось прочитать аналитику из Postgres") + error = "Postgres недоступен — логи временно не читаются." pages = max((total + PAGE_SIZE - 1) // PAGE_SIZE, 1) - query_base = urlencode( - { - key: value - for key, value in { - "event_type": event_type, - "user": user, - "date_from": date_from, - "date_to": date_to, - }.items() - if value - } - ) + query_params = { + "view": view if view != "events" else "", + "event_type": event_type if view == "events" else "", + "user": user, + "date_from": date_from, + "date_to": date_to, + } + query_base = urlencode({key: value for key, value in query_params.items() if value}) return request.app.state.templates.TemplateResponse( request=request, name="logs/list.html", context={ "version": __version__, + "view": view, "events": [_decorate(row, settings) for row in events], + "messages": [_decorate_message(row) for row in messages], "event_types": sorted(ALL_EVENT_TYPES), "event_type": event_type, "user": user, diff --git a/admin/routes/pages.py b/admin/routes/pages.py index 8c55c26..6b6a4e9 100644 --- a/admin/routes/pages.py +++ b/admin/routes/pages.py @@ -38,7 +38,7 @@ SECTIONS: tuple[dict[str, str | None], ...] = ( }, { "title": "Логи", - "description": "События из ClickHouse с фильтрами", + "description": "События и сообщения из Postgres с фильтрами", "url": "/logs", }, { diff --git a/admin/templates/logs/list.html b/admin/templates/logs/list.html index d249f63..8c24d4c 100644 --- a/admin/templates/logs/list.html +++ b/admin/templates/logs/list.html @@ -6,11 +6,33 @@

Логи

-

События из ClickHouse · найдено: {{ total }}

+

+ {% if view == "messages" %}Сообщения по дням{% else %}События из Postgres{% endif %} + · найдено: {{ total }} +

-
+
+ + События + + + Сообщения по дням + +
+ + + {% if view == "messages" %} + + {% endif %} + {% if view == "events" %}
@@ -58,13 +81,46 @@ - Сбросить + Сбросить
{% if error %}
{{ error }}
{% endif %} +{% if view == "messages" %} +
+ + + + + + + + + + + {% for row in messages %} + + + + + + + {% else %} + + + + {% endfor %} + +
ДеньПользовательСообщений
{{ row.day_local }} + {% if row.username %}
@{{ row.username }}
{% endif %} + {{ row.user_id }} + 		{{ row.count }}
+ {% if error %}Данные не загрузились.{% else %}За выбранный период сообщений нет.{% endif %} +
+
+{% else %}
@@ -116,6 +172,7 @@
+{% endif %} {% if pages > 1 %}
diff --git a/bot/__main__.py b/bot/__main__.py index 95bdd73..1dca409 100644 --- a/bot/__main__.py +++ b/bot/__main__.py @@ -12,6 +12,7 @@ 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.analytics.partitions import ensure_partitions, run_partitions from common.config import get_settings from common.db.pool import close_pool, init_pool from common.logging_config import setup_logging @@ -26,6 +27,10 @@ async def main() -> None: await init_pool() await init_analytics() + try: + await ensure_partitions(settings) + except Exception: + logger.warning("Не удалось создать партиции аналитики", exc_info=True) record(EventType.BOT_START, payload={"app_env": settings.app_env}) bot = create_bot() @@ -36,13 +41,20 @@ async def main() -> None: notifications_task = asyncio.create_task(run_notifications(bot)) heartbeat_task = asyncio.create_task(run_heartbeat(bot)) + partitions_task = asyncio.create_task(run_partitions(settings)) try: await bot.delete_webhook(drop_pending_updates=True) await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types()) finally: notifications_task.cancel() heartbeat_task.cancel() - await asyncio.gather(notifications_task, heartbeat_task, return_exceptions=True) + partitions_task.cancel() + await asyncio.gather( + notifications_task, + heartbeat_task, + partitions_task, + return_exceptions=True, + ) record(EventType.BOT_STOP, payload={"app_env": settings.app_env, "uptime_seconds": uptime_seconds()}) await close_analytics() await close_pool() diff --git a/bot/middlewares/analytics.py b/bot/middlewares/analytics.py index c8dffa5..b65b222 100644 --- a/bot/middlewares/analytics.py +++ b/bot/middlewares/analytics.py @@ -9,8 +9,8 @@ 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.analytics import EventType, record, record_message, remember_user +from common.analytics.messages import describe_command from common.config import get_settings logger = logging.getLogger(__name__) @@ -43,13 +43,7 @@ class AnalyticsMiddleware(BaseMiddleware): 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), - ) + record_message(user_id=user.id, username=user.username, chat_id=message.chat.id) command = describe_command(message) if command is not None: record( diff --git a/bot/services/heartbeat.py b/bot/services/heartbeat.py index b621d4f..5830543 100644 --- a/bot/services/heartbeat.py +++ b/bot/services/heartbeat.py @@ -4,7 +4,7 @@ если файл свежий. Если event loop завис — heartbeat перестаёт обновляться. Раз в час в чат логов (``ALERT_CHAT_ID``) уходит короткое сообщение о работе бота, -и в ClickHouse пишется событие ``bot_alive``. +и в аналитику пишется событие ``bot_alive``. """ from __future__ import annotations diff --git a/clickhouse/config.d/disable-trace-log.xml b/clickhouse/config.d/disable-trace-log.xml deleted file mode 100644 index 620121d..0000000 --- a/clickhouse/config.d/disable-trace-log.xml +++ /dev/null @@ -1,8 +0,0 @@ - - - - diff --git a/clickhouse/init/001_events.sql b/clickhouse/init/001_events.sql deleted file mode 100644 index 1711a7c..0000000 --- a/clickhouse/init/001_events.sql +++ /dev/null @@ -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; diff --git a/common/analytics/__init__.py b/common/analytics/__init__.py index c67b1ad..5ff8547 100644 --- a/common/analytics/__init__.py +++ b/common/analytics/__init__.py @@ -1,4 +1,4 @@ -"""Аналитика: клиент ClickHouse и каталог событий. +"""Аналитика: клиент Postgres и каталог событий. Публичный интерфейс: @@ -16,6 +16,7 @@ from common.analytics.client import ( get_analytics, init_analytics, record, + record_message, remember_user, ) from common.analytics.events import ALL_EVENT_TYPES, EVENT_COLUMNS, EventType @@ -29,5 +30,6 @@ __all__ = [ "get_analytics", "init_analytics", "record", + "record_message", "remember_user", ] diff --git a/common/analytics/client.py b/common/analytics/client.py index 8c63c71..eccdd91 100644 --- a/common/analytics/client.py +++ b/common/analytics/client.py @@ -1,13 +1,10 @@ -"""Клиент ClickHouse: батч-запись событий без блокировки бота. +"""Аналитика в Postgres: батч-запись событий и дневных счётчиков сообщений. Свойства: -- запись неблокирующая: :meth:`AnalyticsClient.record` только кладёт строку в буфер; -- батчинг: отправка по размеру буфера и по таймеру; -- недоступность ClickHouse **не роняет бота**: события попадают в stdout (fallback); -- graceful-flush при остановке (:func:`close_analytics`). - -Импорт ``clickhouse_connect`` ленивый (внутри :meth:`AnalyticsClient._ensure_client`), -поэтому пакет не является обязательным для импорта модуля (нужно для тестов/админки). +- запись неблокирующая: ``record``/``record_message`` только кладут данные в буфер; +- батчинг по размеру/таймеру; +- сбой Postgres не роняет бота: батч уходит в лог (fallback) и отбрасывается; +- сообщения не пишутся поштучно: инкрементится ``analytics.messages_daily``. """ from __future__ import annotations @@ -17,11 +14,15 @@ import contextlib import json import logging from collections import deque +from collections.abc import Awaitable, Callable from datetime import UTC, datetime -from typing import Any, Protocol +from typing import Any -from common.analytics.events import EVENT_COLUMNS, EventType +from psycopg.types.json import Jsonb + +from common.analytics.events import EventType from common.config import Settings, get_settings +from common.db.pool import get_pool from common.version import __version__ logger = logging.getLogger(__name__) @@ -35,90 +36,90 @@ USERNAME_CACHE_MAX = 5000 #: Сколько строк события выводить в лог при fallback. FALLBACK_SAMPLE = 5 +#: Строка события: (time, type, user_id, username, chat_id, payload, version, prod). +EventRow = list[Any] +#: (day, user_id, chat_id) -> (username, count) +MessageCounts = dict[tuple[Any, int, int], list[Any]] -class ClickHouseClient(Protocol): - """Минимальный контракт клиента ClickHouse (используется и для подмены в тестах).""" +#: Куда сбрасывать батч (подменяется в тестах). +Writer = Callable[[list[EventRow], MessageCounts], Awaitable[None]] - async def insert(self, table: str, data: list[list[Any]], column_names: list[str]) -> Any: - """Вставить строки в таблицу.""" - async def close(self) -> None: - """Закрыть соединения.""" +async def _write_db(events: list[EventRow], messages: MessageCounts) -> None: + """Записать батч в Postgres.""" + async with get_pool().connection() as conn: + if events: + await conn.executemany( + """ + INSERT INTO analytics.events + (event_time, event_type, user_id, username, chat_id, payload, bot_version, prod) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s) + """, + [ + [ + row[0], + row[1], + row[2], + row[3], + row[4], + Jsonb(row[5]), + row[6], + row[7], + ] + for row in events + ], + ) + + for (day, user_id, chat_id), (username, count) in messages.items(): + await conn.execute( + """ + INSERT INTO analytics.messages_daily (day, user_id, username, chat_id, count) + VALUES (%s, %s, %s, %s, %s) + ON CONFLICT (day, user_id, chat_id) DO UPDATE + SET count = analytics.messages_daily.count + EXCLUDED.count, + username = EXCLUDED.username + """, + (day, user_id, username, chat_id, count), + ) class AnalyticsClient: - """Батч-запись событий в ClickHouse с деградацией в лог.""" + """Батч-запись событий и счётчиков в Postgres с деградацией в лог.""" - def __init__(self, settings: Settings | None = None, *, client: ClickHouseClient | None = None) -> None: + def __init__( + self, + settings: Settings | None = None, + *, + writer: Writer | None = None, + ) -> None: self._settings = settings or get_settings() - self._client = client - self._rows: deque[list[Any]] = deque() + self._writer = writer or _write_db + self._rows: deque[EventRow] = deque() + self._messages: MessageCounts = {} 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 + return self._settings.analytics_enabled @property def pending(self) -> int: - """Сколько событий ждёт отправки (для тестов и диагностики).""" - return len(self._rows) + return len(self._rows) + sum(int(value[1]) for value in self._messages.values()) async def start(self) -> None: - """Подключиться к ClickHouse (best-effort) и запустить фоновый сброс буфера.""" if not self.enabled: - logger.info("Аналитика ClickHouse выключена (CLICKHOUSE_ENABLED=false)") + logger.info("Аналитика выключена (ANALYTICS_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, + "Аналитика Postgres: батч %s, интервал %sс", 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: @@ -143,26 +144,41 @@ class AnalyticsClient: self._rows.append( [ - datetime.now(UTC).replace(tzinfo=None), + datetime.now(UTC), str(event_type), int(user_id), username or "", int(chat_id), - self._dump_payload(payload), + self._payload(payload), __version__, - 1 if self._settings.is_prod else 0, + bool(self._settings.is_prod), ] ) while len(self._rows) > self._settings.analytics_buffer_max: self._rows.popleft() - self._overflow += 1 - if len(self._rows) >= self._settings.analytics_batch_size: + if self.pending >= self._settings.analytics_batch_size: + self._flush_now.set() + + def record_message(self, *, user_id: int, username: str | None, chat_id: int) -> None: + """Увеличить дневной счётчик сообщений пользователя.""" + if not self.enabled or not user_id: + return + + day = datetime.now(self._settings.tzinfo).date() + key = (day, int(user_id), int(chat_id)) + entry = self._messages.get(key) + if entry is None: + self._messages[key] = [username or self._usernames.get(user_id, ""), 1] + else: + entry[0] = username or entry[0] + entry[1] = int(entry[1]) + 1 + + if self.pending >= 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): @@ -171,105 +187,74 @@ class AnalyticsClient: 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: + """Отправить накопленное. При сбое — fallback в лог (без исключений).""" + if not self._rows and not self._messages: return + events = list(self._rows) + messages = self._messages + self._rows.clear() + self._messages = {} + 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)) + await self._writer(events, messages) except Exception: - logger.exception("Не удалось записать %s событий в ClickHouse", len(rows)) - self._log_fallback(rows, "ошибка записи в ClickHouse") + self._log_fallback(events, messages) 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 + await self.flush() - 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] + def _log_fallback(self, events: list[EventRow], messages: MessageCounts) -> None: + sample = " | ".join( + f"{row[0]:%Y-%m-%d %H:%M:%S} {row[1]} user={row[2]} chat={row[4]}" for row in events[:FALLBACK_SAMPLE] ) logger.warning( - "Fallback (%s): %s событий не записаны в ClickHouse: %s%s", - reason, - len(rows), - lines, - " ..." if len(rows) > FALLBACK_SAMPLE else "", + "Fallback: не записано в Postgres событий=%s, счётчиков=%s%s", + len(events), + len(messages), + f": {sample}" if sample else "", ) @staticmethod - def _dump_payload(payload: dict[str, Any] | None) -> str: - """Сериализовать payload в JSON-строку с обрезкой по длине.""" + def _payload(payload: dict[str, Any] | None) -> dict[str, Any]: if not payload: - return "{}" + return {} try: text = json.dumps(payload, ensure_ascii=False, default=str) except (TypeError, ValueError): - text = json.dumps({"_error": "payload не сериализуется"}) - return text[:PAYLOAD_MAX_LENGTH] + return {"_error": "payload не сериализуется"} + if len(text) > PAYLOAD_MAX_LENGTH: + return {"_truncated": text[:PAYLOAD_MAX_LENGTH]} + # Пересобираем из JSON: так в jsonb не попадёт объект, который не сериализуется. + return json.loads(text) -#: Активный клиент (если аналитика не инициализирована — None). _client: AnalyticsClient | None = None async def init_analytics( settings: Settings | None = None, *, - client: ClickHouseClient | None = None, + writer: Writer | None = None, ) -> AnalyticsClient: """Создать и запустить единственный клиент аналитики. Идемпотентно.""" global _client if _client is not None: return _client - _client = AnalyticsClient(settings, client=client) + _client = AnalyticsClient(settings, writer=writer) await _client.start() return _client async def close_analytics() -> None: - """Остановить клиент аналитики и сбросить буфер при остановке приложения.""" global _client if _client is not None: await _client.close() @@ -277,7 +262,6 @@ async def close_analytics() -> None: def get_analytics() -> AnalyticsClient | None: - """Вернуть активный клиент аналитики (или None).""" return _client @@ -289,13 +273,17 @@ def record( 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 record_message(*, user_id: int, username: str | None, chat_id: int) -> None: + if _client is None: + return + _client.record_message(user_id=user_id, username=username, chat_id=chat_id) + + 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 index fd98087..74719a6 100644 --- a/common/analytics/events.py +++ b/common/analytics/events.py @@ -1,6 +1,6 @@ """Каталог событий аналитики. -Единственный источник правды по типам событий ClickHouse. Список синхронизирован +Единственный источник правды по типам событий. Список синхронизирован с документацией (``docs/events.md``) — рассинхрон ловит ``tests/test_analytics.py``. Правило: добавляешь новое событие — добавь его в ``EventType`` и в ``docs/events.md``. @@ -12,10 +12,9 @@ from enum import StrEnum class EventType(StrEnum): - """Типы событий (значение = колонка ``event_type`` в ClickHouse).""" + """Типы событий (значение = колонка ``event_type`` в Postgres).""" # --- Сообщения и команды --- - MESSAGE = "message" COMMAND = "command" # --- Экономика --- diff --git a/common/analytics/partitions.py b/common/analytics/partitions.py new file mode 100644 index 0000000..13706f8 --- /dev/null +++ b/common/analytics/partitions.py @@ -0,0 +1,64 @@ +"""Партиции аналитики по дням. + +Создаются заранее (вчера/сегодня/два дня вперёд), старые **не удаляются**: +история событий и дневных счётчиков хранится бессрочно. +""" + +from __future__ import annotations + +import asyncio +import logging +from datetime import UTC, date, datetime, timedelta + +from common.config import Settings +from common.db.pool import get_pool + +logger = logging.getLogger(__name__) + +#: Сколько дней вперёд создавать партиции. +DAYS_AHEAD = 2 +#: Сколько дней назад проверять (на случай простоя). +DAYS_BACK = 1 +#: Интервал фоновой проверки партиций. +PARTITION_INTERVAL = 6 * 60 * 60 + + +def _event_partition(day: date) -> str: + start = f"{day.isoformat()} 00:00:00+00" + end = f"{(day + timedelta(days=1)).isoformat()} 00:00:00+00" + return ( + f"CREATE TABLE IF NOT EXISTS analytics.events_{day:%Y%m%d} " + f"PARTITION OF analytics.events FOR VALUES FROM ('{start}') TO ('{end}')" + ) + + +def _messages_partition(day: date) -> str: + start = day.isoformat() + end = (day + timedelta(days=1)).isoformat() + return ( + f"CREATE TABLE IF NOT EXISTS analytics.messages_daily_{day:%Y%m%d} " + f"PARTITION OF analytics.messages_daily FOR VALUES FROM ('{start}') TO ('{end}')" + ) + + +async def ensure_partitions(settings: Settings) -> None: + """Создать недостающие партиции на несколько дней вперёд.""" + utc_today = datetime.now(UTC).date() + local_today = datetime.now(settings.tzinfo).date() + + async with get_pool().connection() as conn: + for offset in range(-DAYS_BACK, DAYS_AHEAD + 1): + await conn.execute(_event_partition(utc_today + timedelta(days=offset))) + await conn.execute(_messages_partition(local_today + timedelta(days=offset))) + + +async def run_partitions(settings: Settings) -> None: + """Периодически проверять партиции; ошибки не роняют бота.""" + while True: + try: + await ensure_partitions(settings) + except asyncio.CancelledError: + raise + except Exception: + logger.warning("Не удалось создать партиции аналитики", exc_info=True) + await asyncio.sleep(PARTITION_INTERVAL) diff --git a/common/analytics/read.py b/common/analytics/read.py index aff0d04..a4d9dca 100644 --- a/common/analytics/read.py +++ b/common/analytics/read.py @@ -1,17 +1,14 @@ -"""Чтение событий ClickHouse для админки (Блок 4.3). - -Клиент создаётся лениво и только на чтение; недоступность ClickHouse не роняет -админку — страница покажет ошибку. Запись событий живёт в ``client.py``. -""" +"""Чтение аналитики из Postgres для админки.""" from __future__ import annotations import logging -from datetime import UTC, datetime, time, timedelta +from datetime import UTC, date, datetime, time, timedelta from typing import Any from common.analytics.events import ALL_EVENT_TYPES from common.config import Settings +from common.db.pool import get_pool logger = logging.getLogger(__name__) @@ -27,62 +24,29 @@ EVENT_FIELDS = ( "prod", ) -_client: Any | None = None + +def _dates(date_from: str, date_to: str, settings: Settings) -> tuple[date, date]: + today = datetime.now(settings.tzinfo).date() + start = datetime.strptime(date_from, "%Y-%m-%d").date() if date_from else today - timedelta(days=6) + end = datetime.strptime(date_to, "%Y-%m-%d").date() if date_to else today + return start, end -async def _ensure_client(settings: Settings) -> Any: - """Создать (один раз) асинхронный клиент ClickHouse.""" - global _client - if _client is not None: - return _client - - import clickhouse_connect - - _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, - ) - return _client +def _event_bounds(start: date, end: date, settings: Settings) -> tuple[datetime, datetime]: + start_dt = datetime.combine(start, time.min, tzinfo=settings.tzinfo).astimezone(UTC) + end_dt = datetime.combine(end + timedelta(days=1), time.min, tzinfo=settings.tzinfo).astimezone(UTC) + return start_dt, end_dt -async def close_client() -> None: - """Закрыть клиент при остановке админки.""" - global _client - if _client is not None: - client, _client = _client, None - try: - await client.close() - except Exception: - logger.warning("Не удалось закрыть клиент ClickHouse", exc_info=True) - - -def _where( - *, - event_type: str, - user: str, - start: datetime, - end: datetime, -) -> tuple[str, dict[str, Any]]: - clauses = ["event_time >= {start:DateTime}", "event_time < {end:DateTime}"] - params: dict[str, Any] = {"start": start, "end": end} - - if event_type and event_type in ALL_EVENT_TYPES: - clauses.append("event_type = {event_type:String}") - params["event_type"] = event_type - +def _user_clause(user: str, params: list[Any], column: str = "user_id") -> str: user = user.strip() - if user: - if user.isdigit(): - clauses.append("user_id = {user_id:Int64}") - params["user_id"] = int(user) - else: - clauses.append("positionCaseInsensitive(username, {user_text:String}) > 0") - params["user_text"] = user - - return " WHERE " + " AND ".join(clauses), params + if not user: + return "" + if user.isdigit(): + params.append(int(user)) + return f" AND {column} = %s" + params.append(f"%{user}%") + return " AND username ILIKE %s" async def fetch_events( @@ -95,31 +59,63 @@ async def fetch_events( limit: int = 50, offset: int = 0, ) -> tuple[list[dict[str, Any]], int]: - """Вернуть события и общее количество по фильтрам. + """Вернуть события и общее количество по фильтрам.""" + start, end = _dates(date_from, date_to, settings) + start_dt, end_dt = _event_bounds(start, end, settings) - Даты — ``YYYY-MM-DD`` в таймзоне админки; внутри переводятся в UTC. - """ - today = datetime.now(settings.tzinfo).date() - start_date = datetime.strptime(date_from, "%Y-%m-%d").date() if date_from else today - timedelta(days=6) - end_date = datetime.strptime(date_to, "%Y-%m-%d").date() if date_to else today + params: list[Any] = [start_dt, end_dt] + where = " WHERE event_time >= %s AND event_time < %s" + if event_type and event_type in ALL_EVENT_TYPES: + where += " AND event_type = %s" + params.append(event_type) + where += _user_clause(user, params) - start = datetime.combine(start_date, time.min, tzinfo=settings.tzinfo).astimezone(UTC).replace(tzinfo=None) - end = ( - datetime.combine(end_date + timedelta(days=1), time.min, tzinfo=settings.tzinfo) - .astimezone(UTC) - .replace(tzinfo=None) - ) + async with get_pool().connection() as conn: + cur = await conn.execute(f"SELECT count(*) AS total FROM analytics.events{where}", tuple(params)) + row = await cur.fetchone() + total = int(row["total"]) if row else 0 - where, params = _where(event_type=event_type, user=user, start=start, end=end) - client = await _ensure_client(settings) - - count_result = await client.query(f"SELECT count() FROM analytics.events{where}", parameters=params) - total = int(count_result.result_rows[0][0]) if count_result.result_rows else 0 - - result = await client.query( - f"SELECT {', '.join(EVENT_FIELDS)} FROM analytics.events{where}" - " ORDER BY event_time DESC LIMIT {limit:UInt32} OFFSET {offset:UInt32}", - parameters={**params, "limit": limit, "offset": offset}, - ) - rows = [dict(zip(result.column_names, row, strict=False)) for row in result.result_rows] + cur = await conn.execute( + f"SELECT {', '.join(EVENT_FIELDS)} FROM analytics.events{where}" + " ORDER BY event_time DESC LIMIT %s OFFSET %s", + (*params, limit, offset), + ) + rows = await cur.fetchall() + return rows, total + + +async def fetch_messages( + settings: Settings, + *, + user: str = "", + date_from: str = "", + date_to: str = "", + limit: int = 50, + offset: int = 0, +) -> tuple[list[dict[str, Any]], int]: + """Вернуть дневные счётчики сообщений и общее количество строк.""" + start, end = _dates(date_from, date_to, settings) + + params: list[Any] = [start, end] + where = " WHERE day >= %s AND day <= %s" + where += _user_clause(user, params) + + async with get_pool().connection() as conn: + cur = await conn.execute( + f"SELECT count(*) AS total FROM analytics.messages_daily{where}", + tuple(params), + ) + row = await cur.fetchone() + total = int(row["total"]) if row else 0 + + cur = await conn.execute( + f""" + SELECT day, user_id, username, chat_id, count + FROM analytics.messages_daily{where} + ORDER BY day DESC, count DESC + LIMIT %s OFFSET %s + """, + (*params, limit, offset), + ) + rows = await cur.fetchall() return rows, total diff --git a/common/config.py b/common/config.py index a3c0099..674aa81 100644 --- a/common/config.py +++ b/common/config.py @@ -60,13 +60,8 @@ class Settings(BaseSettings): #: Публичный адрес админки (deep-link вход, ссылки из бота). admin_public_url: str = "http://localhost:8000" - #: 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 = "" + #: Аналитика событий в Postgres. ANALYTICS_ENABLED=false — полностью выключить. + analytics_enabled: bool = True #: Батчинг записи событий: размер батча, интервал сброса и предел буфера. analytics_batch_size: int = 200 @@ -135,7 +130,7 @@ def get_settings() -> Settings: class AdminSettings(Settings): """Настройки сервиса админки. - Наследует общие поля (окружение, Postgres, таймзона, ClickHouse), но не требует + Наследует общие поля (окружение, Postgres, таймзона), но не требует ``BOT_TOKEN``: на этапе каркаса он не нужен, а появится в проверке подписи Telegram-логина (Блок 4.2). """ diff --git a/common/version.py b/common/version.py index 361bd45..e2b7506 100644 --- a/common/version.py +++ b/common/version.py @@ -2,9 +2,9 @@ Единый источник правды. Git-тег совпадает с версией с префиксом ``v`` (например, версия ``0.1`` -> тег ``v0.1``). Docker-образы в Блоке 2 будут -тегироваться этой версией, а события в ClickHouse получат поле ``bot_version``. +тегироваться этой версией, а события аналитики получат поле ``bot_version``. """ from __future__ import annotations -__version__ = "0.6.9" +__version__ = "0.7.0" diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 0706478..72103ee 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -46,8 +46,6 @@ services: environment: DB_HOST: postgres DB_PORT: "5432" - CLICKHOUSE_HOST: clickhouse - CLICKHOUSE_PORT: "8123" volumes: - media:/app/media depends_on: @@ -55,9 +53,6 @@ services: condition: service_healthy migrate: condition: service_completed_successfully - # Без ожидания здоровья: аналитика некритична, бот работает и при недоступном ClickHouse. - clickhouse: - condition: service_started logging: driver: json-file options: @@ -74,8 +69,6 @@ services: environment: DB_HOST: postgres DB_PORT: "5432" - CLICKHOUSE_HOST: clickhouse - CLICKHOUSE_PORT: "8123" ADMIN_HOST: 0.0.0.0 ADMIN_PORT: "8000" volumes: @@ -95,37 +88,6 @@ services: 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 - # trace_log не нужен и на тестовом контуре съедал гигабайты в сутки. - - ./clickhouse/config.d/disable-trace-log.xml:/etc/clickhouse-server/config.d/disable-trace-log.xml: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: @@ -143,7 +105,6 @@ services: volumes: pgdata: - clickhousedata: media: # caddy_data: # caddy_config: diff --git a/docker-compose.yml b/docker-compose.yml index eb8ed8f..d48d9fb 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -51,8 +51,6 @@ services: # Внутри сети compose БД доступна по имени сервиса. DB_HOST: postgres DB_PORT: "5432" - CLICKHOUSE_HOST: clickhouse - CLICKHOUSE_PORT: "8123" volumes: - media:/app/media depends_on: @@ -60,9 +58,6 @@ services: condition: service_healthy migrate: condition: service_completed_successfully - # Аналитика некритична: бот стартует, даже если ClickHouse ещё не готов. - clickhouse: - condition: service_started logging: driver: json-file options: @@ -82,8 +77,6 @@ services: environment: DB_HOST: postgres DB_PORT: "5432" - CLICKHOUSE_HOST: clickhouse - CLICKHOUSE_PORT: "8123" ADMIN_HOST: 0.0.0.0 ADMIN_PORT: "8000" volumes: @@ -101,34 +94,6 @@ services: 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 - # trace_log не нужен: на тестовом контуре съедал гигабайты в сутки. - - ./clickhouse/config.d/disable-trace-log.xml:/etc/clickhouse-server/config.d/disable-trace-log.xml: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: media: diff --git a/docs/events.md b/docs/events.md index a7c96ed..40c761f 100644 --- a/docs/events.md +++ b/docs/events.md @@ -1,48 +1,43 @@ -# События аналитики (ClickHouse) +# События аналитики (Postgres) -Бот пишет все значимые действия в ClickHouse — единую широкую таблицу -`analytics.events`. Дашборд (Superset) и ETL — отдельный финальный этап; здесь -описан **контракт** событий: какие бывают `event_type` и что лежит в `payload`. +Бот пишет значимые действия в Postgres, схема `analytics`. Дашборд (Superset) и ETL — +отдельный финальный этап; здесь описан **контракт** событий: какие бывают `event_type` +и что лежит в `payload`. > Источник правды в коде — `common/analytics/events.py` (`EventType`). > Расхождение со списком ниже ловит `tests/test_analytics.py`. > Добавляешь событие — добавь строку в таблицу. -## Схема в ClickHouse +## Схема в Postgres -DDL — `clickhouse/init/001_events.sql` (применяется при первой инициализации контейнера). +DDL — `sql/V26__analytics_schema.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`). +- `analytics.events` — события, `PARTITION BY RANGE (event_time)` по дням. + Партиции создаёт приложение (бот и админка) на несколько дней вперёд; + **старые партиции не удаляются** — история хранится бессрочно. +- `analytics.messages_daily` — количество сообщений пользователя по дням: + `day`, `user_id`, `username`, `chat_id`, `count`; PK `(day, user_id, chat_id)`. + На каждое сообщение выполняется `INSERT ... ON CONFLICT DO UPDATE count + 1`, + поэтому вместо строки на каждое сообщение растёт только счётчик за день. -Агрегаты остаются на VPS навсегда, поэтому активность, топ команд и денежные потоки -доступны и после удаления сырых событий. Запросы к агрегатам суммируют `events`/`amount_sum` -(фон слияний `SummingMergeTree` не гарантирует полную свёртку в любой момент). +Сообщения **не** пишутся в `analytics.events` отдельным событием `message`. -## Хранение и бэкап +## Хранение | Что | Где | Сколько | | --- | --- | --- | -| Сырые события | VPS, `analytics.events` | последние `BACKUP_GRACE_DAYS` дней (по умолчанию 7) | -| Агрегаты | VPS | постоянно | -| Полная история | домашний сервер, свой ClickHouse | архивы `events_YYYY-MM-DD.parquet.gz` (ежедневно) | +| События | Postgres, `analytics.events` | бессрочно (партиции по дням) | +| Счётчики сообщений | Postgres, `analytics.messages_daily` | бессрочно (партиции по дням) | -Каждую ночь (по Москве) домашний сервер **сам забирает** вчерашнюю партицию с VPS (`pull` -по ssh), импортирует её в свой ClickHouse и **сверяет количество строк**; только после -успешной сверки он просит VPS удалить партиции старше грейса. Ошибка бэкапа данные не -удаляет. Подробности — `DEPLOY.md` и `deploy/ssh-access.md`. +Бэкап — общий бэкап Postgres. Отдельного пайплайна ClickHouse/Parquet больше нет. +При росте базы нужно следить за диском: автоматической очистки нет. ## Уведомления в чат логов -Помимо записи в ClickHouse часть событий дублируется сообщением в `ALERT_CHAT_ID` +Помимо записи в Postgres часть событий дублируется сообщением в `ALERT_CHAT_ID` (если он задан; `0` — выключено). Отправка не критична: ошибки только логируются. -| Уведомление | Когда | Событие в ClickHouse | +| Уведомление | Когда | Событие | | --- | --- | --- | | Вход участника | кто-то вошёл в основной чат | `join` | | Выход участника | кто-то вышел из основного чата | `leave` | @@ -54,20 +49,19 @@ DDL — `clickhouse/init/001_events.sql` (применяется при перв | Поле | Тип | Описание | | --- | --- | --- | -| **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 — тестовый | +| **event_time** | `timestamptz` | Момент события (UTC) | +| **event_type** | `text` | Тип события (см. каталог ниже) | +| **user_id** | `bigint` | Telegram-ID пользователя (0 — системное событие) | +| **username** | `text` | `@username` на момент события (может быть пустым) | +| **chat_id** | `bigint` | ID чата (обычно основной чат клана) | +| **payload** | `jsonb` | Дополнительные данные события | +| **bot_version** | `text` | Версия бота из `common/version.py` | +| **prod** | `boolean` | `true` — прод-контур, `false` — тестовый | ## Каталог событий | Тип события | Когда пишется | `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` | @@ -88,37 +82,44 @@ DDL — `clickhouse/init/001_events.sql` (применяется при перв | `bot_alive` | Ежечасный пинг «бот живой» | `app_env`, `uptime_seconds` | | `error` | Необработанная ошибка при обработке апдейта | `error_type`, `error`, `update_type` | -## Поведение при недоступности ClickHouse +## Поведение при недоступности Postgres Аналитика — **некритичная** подсистема: её сбой не должен влиять на бота. -- события копятся в буфере и отправляются батчами (по размеру и по таймеру); -- если ClickHouse недоступен, буфер не растёт бесконечно: при переполнении самые - старые записи отбрасываются со счётчиком в логе; -- если записать батч не удалось, события выводятся в stdout (`WARNING`, первые - несколько строк + общее количество) и отбрасываются — данные не пропадают молча; +- события и счётчики копятся в буфере и записываются батчами (по размеру и по таймеру); +- если Postgres недоступен, буфер не растёт бесконечно: при переполнении самые + старые события отбрасываются; +- если записать батч не удалось, он выводится в stdout (`WARNING`, первые несколько + строк + общее количество) и отбрасывается — данные не пропадают молча; - при остановке бота выполняется финальный сброс буфера; -- выключение целиком — `CLICKHOUSE_ENABLED=false` (тогда `record()` — no-op). +- выключение целиком — `ANALYTICS_ENABLED=false` (тогда `record()`/`record_message()` — no-op). ## Примеры запросов ```sql --- активность по дням -SELECT toDate(event_time) AS day, count() FROM analytics.events GROUP BY day ORDER BY day; +-- события по дням +SELECT date_trunc('day', event_time) AS day, count(*) +FROM analytics.events GROUP BY day ORDER BY day; -- топ команд за неделю -SELECT JSONExtractString(payload, 'command') AS cmd, count() AS uses +SELECT payload->>'command' AS cmd, count(*) AS uses FROM analytics.events -WHERE event_type = 'command' AND event_time >= now() - INTERVAL 7 DAY +WHERE event_type = 'command' AND event_time >= now() - interval '7 days' GROUP BY cmd ORDER BY uses DESC; -- покупки за месяц -SELECT event_type, count(), sum(JSONExtractInt(payload, 'price')) +SELECT event_type, count(*), sum((payload->>'price')::int) FROM analytics.events -WHERE event_type IN ('purchase_title', 'purchase_donat') AND event_time >= now() - INTERVAL 30 DAY +WHERE event_type IN ('purchase_title', 'purchase_donat') + AND event_time >= now() - interval '30 days' GROUP BY event_type; +-- сообщения по дням +SELECT day, sum(count) FROM analytics.messages_daily +WHERE day >= current_date - 30 GROUP BY day ORDER BY day; + -- ошибки -SELECT event_time, JSONExtractString(payload, 'error_type'), JSONExtractString(payload, 'error') -FROM analytics.events WHERE event_type = 'error' ORDER BY event_time DESC LIMIT 50; +SELECT event_time, payload->>'error_type', payload->>'error' +FROM analytics.events WHERE event_type = 'error' +ORDER BY event_time DESC LIMIT 50; ``` diff --git a/env/backup.env.example b/env/backup.env.example deleted file mode 100644 index 85da79e..0000000 --- a/env/backup.env.example +++ /dev/null @@ -1,32 +0,0 @@ -# Параметры бэкапа событий ClickHouse (pull: домашний сервер забирает данные с VPS). -# Скопируйте в env/backup.env (env/*.env в git не попадает) и заполните. -# -# Запускается НА ДОМАШНЕМ СЕРВЕРЕ по cron: -# scripts/clickhouse_backup.sh test|prod -# -# SSH-доступ к VPS настраивается отдельно (инструкция — deploy/ssh-access.md). - -# --- VPS --- -# user@host (хост указывайте без скобок; порт задаётся отдельно) -VPS_SSH=deploy@vps.example -VPS_PORT=27 -# Каталог с репозиторием на VPS (там запускается docker compose) -VPS_PROJECT_DIR=/opt/DispelEvilBot -# Приватный ключ для ssh на VPS (публичная часть добавлена в authorized_keys на VPS) -VPS_SSH_KEY=/root/.ssh/id_ed25519_vps - -# --- Домашний сервер --- -# Куда складывать архивы events_YYYY-MM-DD.parquet.gz -BACKUP_DIR=/srv/dispelevilbot/events - -# --- Хранение --- -# Сколько дней сырых событий остаётся на VPS (остальное удаляется после успешной сверки) -BACKUP_GRACE_DAYS=7 -# Таймзона расчёта «вчерашнего» дня -BACKUP_TZ=Europe/Moscow - -# --- ClickHouse домашнего сервера (сюда импортируются архивы) --- -CLICKHOUSE_HOST=localhost -CLICKHOUSE_PORT=9000 -CLICKHOUSE_USER=default -CLICKHOUSE_PASSWORD= diff --git a/env/prod.env.example b/env/prod.env.example index cb70602..40ce3e8 100644 --- a/env/prod.env.example +++ b/env/prod.env.example @@ -37,14 +37,8 @@ ALERT_CHAT_ID=0 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 +# Аналитика (Postgres, схема analytics; см. docs/events.md). +ANALYTICS_ENABLED=true # Админка (FastAPI). Порт на хосте; у test и prod должен различаться # (test 8100, prod 8101 — 8000 занят Portainer). diff --git a/env/test.env.example b/env/test.env.example index 3b5dca4..1768463 100644 --- a/env/test.env.example +++ b/env/test.env.example @@ -35,14 +35,8 @@ ALERT_CHAT_ID=0 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 +# Аналитика (Postgres, схема analytics; см. docs/events.md). +ANALYTICS_ENABLED=true # Админка (FastAPI). Порт на хосте; у test и prod должен различаться # (test 8100, prod 8101 — 8000 занят Portainer). diff --git a/requirements-admin.txt b/requirements-admin.txt index d5cc4f6..52651aa 100644 --- a/requirements-admin.txt +++ b/requirements-admin.txt @@ -17,9 +17,6 @@ tzdata>=2024.1 # Postgres (async psycopg v3) — понадобится разделам шаблонов и пользователей. psycopg[binary,pool]>=3.2,<4.0 -# ClickHouse — чтение логов в админке (раздел «Логи»). -clickhouse-connect[async]>=1.0,<2.0 - # Telegram Bot API для модерации (исключение из чата) и уведомлений. httpx>=0.27,<1.0 diff --git a/requirements.txt b/requirements.txt index ab2f463..d9a22f9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -10,7 +10,5 @@ python-dotenv>=1.0,<2.0 tzdata>=2024.1 psycopg[binary,pool]>=3.2,<4.0 -# ClickHouse (аналитика событий, Блок 3). Extras [async] — нативный asyncio-клиент. -clickhouse-connect[async]>=1.0,<2.0 # Блок 4 (расписания из БД) — при необходимости: # APScheduler>=3.10,<4.0 diff --git a/scripts/check_contour.sh b/scripts/check_contour.sh index b2d1a4b..53c6d1f 100755 --- a/scripts/check_contour.sh +++ b/scripts/check_contour.sh @@ -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 admin; do if echo "$running" | grep -qx "$svc"; then echo "OK ${svc}: running" else @@ -49,17 +49,11 @@ 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 +# Схема аналитики в Postgres (создаётся миграцией V26) +if $COMPOSE exec -T postgres sh -lc 'psql -U "$POSTGRES_USER" -d "$POSTGRES_DB" -tAc "SELECT to_regclass('"'"'analytics.events'"'"') IS NOT NULL AND to_regclass('"'"'analytics.messages_daily'"'"') IS NOT NULL"' | grep -qx t; then + echo "OK postgres: схема analytics доступна" else - echo "FAIL clickhouse: не отвечает" >&2 + echo "FAIL postgres: нет таблиц analytics.events/messages_daily (см. sql/V26__analytics_schema.sql)" >&2 fail=1 fi diff --git a/scripts/clickhouse_backup.sh b/scripts/clickhouse_backup.sh deleted file mode 100755 index 3903a0b..0000000 --- a/scripts/clickhouse_backup.sh +++ /dev/null @@ -1,129 +0,0 @@ -#!/usr/bin/env bash -# Бэкап сырых событий ClickHouse: домашний сервер забирает данные с VPS (pull). -# -# Запускается НА ДОМАШНЕМ СЕРВЕРЕ по cron. Входящий SSH на домашний сервер не нужен: -# скрипт сам ходит на VPS по ключу и запускает там docker compose. -# -# Что делает: -# 1) по ssh выгружает на VPS вчерашнюю партицию analytics.events в Parquet; -# 2) складывает events_YYYY-MM-DD.parquet.gz в BACKUP_DIR; -# 3) импортирует архив в локальный ClickHouse (идемпотентно, scripts/clickhouse_import.sh); -# 4) сверяет количество строк на VPS и локально; -# 5) только после успешной сверки просит VPS удалить партиции старше BACKUP_GRACE_DAYS. -# -# Ошибка на любом шаге оставляет данные на VPS (партиции не удаляются) и возвращает ненулевой код. -# -# Использование: -# ./scripts/clickhouse_backup.sh test # или prod -# make backup E=test -# -# Настройки — env/backup.env (см. env/backup.env.example). - -set -euo pipefail - -E="${1:-test}" -BACKUP_ENV="${BACKUP_ENV:-env/backup.env}" - -log() { echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*"; } -fail() { - echo "[$(date '+%Y-%m-%d %H:%M:%S')] ОШИБКА: $*" >&2 - exit 1 -} - -[ -f "$BACKUP_ENV" ] || fail "нет файла настроек ${BACKUP_ENV} (скопируйте env/backup.env.example)" -# shellcheck disable=SC1090 -set -a && . "./${BACKUP_ENV}" && set +a - -: "${VPS_SSH:?задайте VPS_SSH в env/backup.env}" -: "${BACKUP_DIR:?задайте BACKUP_DIR в env/backup.env}" -VPS_PORT="${VPS_PORT:-22}" -VPS_PROJECT_DIR="${VPS_PROJECT_DIR:?задайте VPS_PROJECT_DIR в env/backup.env}" -VPS_SSH_KEY="${VPS_SSH_KEY:-$HOME/.ssh/id_ed25519_vps}" -VPS_COMPOSE_FILE="${VPS_COMPOSE_FILE:-docker-compose.prod.yml}" -BACKUP_GRACE_DAYS="${BACKUP_GRACE_DAYS:-7}" -BACKUP_TZ="${BACKUP_TZ:-Europe/Moscow}" -IMPORT_SCRIPT="${IMPORT_SCRIPT:-$(dirname "$0")/clickhouse_import.sh}" -SCHEMA_FILE="${SCHEMA_FILE:-$(dirname "$0")/../clickhouse/init/001_events.sql}" - -SSH=(ssh -i "$VPS_SSH_KEY" -p "$VPS_PORT" -o BatchMode=yes -o ServerAliveInterval=30 "$VPS_SSH") - -# Запустить команду на VPS. Пароль ClickHouse читается из окружения контейнера, -# поэтому клиент вызывается через `sh -c` уже внутри контейнера. -# ssh склеивает аргументы в одну строку и отдаёт её удалённому shell, поэтому каждый -# аргумент экранируем: в SQL есть пробелы, скобки и кавычки. -vps_ch() { - local quoted=() arg - for arg in "$@"; do - quoted+=("$(printf '%q' "$arg")") - done - "${SSH[@]}" bash -s "$VPS_PROJECT_DIR" "$E" "$VPS_COMPOSE_FILE" "${quoted[@]}" <<'REMOTE' -set -euo pipefail -project_dir="$1"; contour="$2"; compose_file="$3"; shift 3 -cd "$project_dir" -exec docker compose -p "$contour" --env-file "env/$contour.env" -f "$compose_file" exec -T clickhouse \ - sh -c 'clickhouse-client --user "$CLICKHOUSE_USER" --password "$CLICKHOUSE_PASSWORD" "$@"' sh "$@" -REMOTE -} - -# Какие сервисы контура запущены на VPS (пусто, если проекта/контура нет). -vps_running_services() { - "${SSH[@]}" bash -s "$VPS_PROJECT_DIR" "$E" "$VPS_COMPOSE_FILE" <<'REMOTE' -set -uo pipefail -if [ ! -d "$1" ]; then - echo "каталог проекта $1 на VPS не найден" >&2 - exit 0 -fi -cd "$1" || exit 0 -docker compose -p "$2" --env-file "env/$2.env" -f "$3" ps --services --status running 2>/dev/null || true -REMOTE -} - -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')" - -log "Бэкап событий контура ${E} за ${DAY} (грейс ${BACKUP_GRACE_DAYS} дней, порог ${GRACE_CUTOFF})" - -# Бот может быть не запущен (или ещё не развёрнут) — это не ошибка, просто пропускаем прогон. -if ! RUNNING="$(vps_running_services)"; then - fail "не удалось получить состояние контура ${E} на VPS (проверьте ssh и VPS_SSH)" -fi -if ! grep -qx 'clickhouse' <<<"$RUNNING"; then - log "Контур ${E} на VPS не запущен — бэкап пропущен, на VPS ничего не меняем" - exit 0 -fi - -mkdir -p "$BACKUP_DIR" -ARCHIVE="${BACKUP_DIR}/events_${DAY}.parquet.gz" - -ROWS_VPS="$(vps_ch --param_day="$DAY" --query "SELECT count() FROM analytics.events WHERE toDate(event_time) = toDate({day:String})" | tr -d '[:space:]')" -log "На VPS строк за ${DAY}: ${ROWS_VPS}" -if [ "$ROWS_VPS" = "0" ]; then - log "ВНИМАНИЕ: за ${DAY} событий нет — партиция пустая, продолжаю (сверка 0 = 0)" -fi - -log "Выгружаю партицию в ${ARCHIVE}" -vps_ch --param_day="$DAY" --query "SELECT * FROM analytics.events WHERE toDate(event_time) = toDate({day:String}) FORMAT Parquet" | - gzip -9 >"${ARCHIVE}.part" -[ -s "${ARCHIVE}.part" ] || fail "не удалось выгрузить архив ${DAY}" -mv "${ARCHIVE}.part" "$ARCHIVE" -log "Размер архива: $(du -h "$ARCHIVE" | cut -f1)" - -log "Импортирую архив в ClickHouse домашнего сервера" -ROWS_HOME="$(BACKUP_DIR="$BACKUP_DIR" SCHEMA_FILE="$SCHEMA_FILE" "$IMPORT_SCRIPT" "$DAY" | tail -n1 | tr -d '[:space:]')" -log "На домашнем сервере строк за ${DAY}: ${ROWS_HOME}" - -[ "$ROWS_VPS" = "$ROWS_HOME" ] || fail "расхождение строк (VPS ${ROWS_VPS}, домашний ${ROWS_HOME}) — партиции на VPS не удаляю" - -log "Удаляю на VPS партиции старше ${GRACE_CUTOFF}" -DROPPED="$(vps_ch --query "SELECT DISTINCT partition FROM system.parts WHERE database='analytics' AND table='events' AND active AND partition < '${GRACE_CUTOFF}'")" -if [ -z "$DROPPED" ]; then - log "Удалять нечего" -else - for partition in $DROPPED; do - log "DROP PARTITION ID '${partition}'" - vps_ch --query "ALTER TABLE analytics.events DROP PARTITION ID '${partition}'" - done -fi - -log "Готово: события за ${DAY} в архиве ${ARCHIVE}, на VPS остались агрегаты и последние ${BACKUP_GRACE_DAYS} дней сырых событий" diff --git a/scripts/clickhouse_import.sh b/scripts/clickhouse_import.sh deleted file mode 100755 index b68bbd4..0000000 --- a/scripts/clickhouse_import.sh +++ /dev/null @@ -1,58 +0,0 @@ -#!/usr/bin/env bash -# Импорт дневного архива событий в ClickHouse домашнего сервера. -# -# Запускается на домашнем сервере; обычно вызывается из scripts/clickhouse_backup.sh, -# но можно и вручную — например, чтобы повторно залить архив после сбоя. -# -# Использование: -# ./scripts/clickhouse_import.sh 2026-01-01 -# В stdout печатается количество строк загруженного дня (для сверки), диагностика — в stderr. -# -# Настройки (окружение): -# BACKUP_DIR каталог с архивами events_YYYY-MM-DD.parquet.gz -# SCHEMA_FILE DDL схемы (по умолчанию clickhouse/init/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}/../clickhouse/init/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} (канонический DDL — 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 --param_day="$DAY" --query "SELECT count() FROM analytics.events WHERE toDate(event_time) = toDate({day:String})" diff --git a/sql/V26__analytics_schema.sql b/sql/V26__analytics_schema.sql new file mode 100644 index 0000000..b5e02af --- /dev/null +++ b/sql/V26__analytics_schema.sql @@ -0,0 +1,33 @@ +-- Аналитика в Postgres: события и дневные счётчики сообщений. +-- Партиции по дням создаёт приложение; старые партиции не удаляются. +CREATE SCHEMA IF NOT EXISTS analytics; + +CREATE TABLE IF NOT EXISTS analytics.events ( + event_time timestamptz NOT NULL, + event_type text NOT NULL, + user_id bigint NOT NULL DEFAULT 0, + username text NOT NULL DEFAULT '', + chat_id bigint NOT NULL DEFAULT 0, + payload jsonb NOT NULL DEFAULT '{}'::jsonb, + bot_version text NOT NULL DEFAULT '', + prod boolean NOT NULL DEFAULT false +) PARTITION BY RANGE (event_time); + +CREATE INDEX IF NOT EXISTS idx_events_time ON analytics.events (event_time); +CREATE INDEX IF NOT EXISTS idx_events_type_time ON analytics.events (event_type, event_time); +CREATE INDEX IF NOT EXISTS idx_events_user_time ON analytics.events (user_id, event_time); + +COMMENT ON TABLE analytics.events IS 'События бота (партиции по дням, без автоудаления)'; + +CREATE TABLE IF NOT EXISTS analytics.messages_daily ( + day date NOT NULL, + user_id bigint NOT NULL, + username text NOT NULL DEFAULT '', + chat_id bigint NOT NULL DEFAULT 0, + count bigint NOT NULL DEFAULT 0, + PRIMARY KEY (day, user_id, chat_id) +) PARTITION BY RANGE (day); + +CREATE INDEX IF NOT EXISTS idx_messages_daily_user_day ON analytics.messages_daily (user_id, day); + +COMMENT ON TABLE analytics.messages_daily IS 'Количество сообщений пользователя по дням (upsert)'; diff --git a/tests/test_admin_logs.py b/tests/test_admin_logs.py index 1aca86b..d86193e 100644 --- a/tests/test_admin_logs.py +++ b/tests/test_admin_logs.py @@ -1,8 +1,8 @@ -"""Тесты раздела «Логи» (без ClickHouse — чтение подменяется).""" +"""Тесты раздела «Логи» (Postgres не требуется — чтение подменяется).""" from __future__ import annotations -from datetime import datetime +from datetime import date, datetime from typing import Any import pytest @@ -10,26 +10,39 @@ import pytest from common.analytics import read as analytics_read EVENT: dict[str, Any] = { - "event_time": datetime(2026, 10, 18, 17, 0, 0), # UTC, без tz — как отдаёт clickhouse-connect + "event_time": datetime(2026, 10, 18, 17, 0, 0), # UTC без tz — как отдаёт psycopg "event_type": "command", "user_id": 42, "username": "test_user", "chat_id": -100123, - "payload": '{"command": "инфо", "args_count": 0}', + "payload": {"command": "инфо", "args_count": 0}, "bot_version": "0.4.8", - "prod": 0, + "prod": False, +} + +MESSAGE: dict[str, Any] = { + "day": date(2026, 10, 18), + "user_id": 42, + "username": "test_user", + "chat_id": -100123, + "count": 17, } @pytest.fixture def fake_read(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]: - state: dict[str, Any] = {"total": 1} + state: dict[str, Any] = {"total": 1, "messages_total": 1} async def fetch_events(settings: Any, **kwargs: Any) -> tuple[list[dict[str, Any]], int]: state["kwargs"] = kwargs return [EVENT], state["total"] + async def fetch_messages(settings: Any, **kwargs: Any) -> tuple[list[dict[str, Any]], int]: + state["message_kwargs"] = kwargs + return [MESSAGE], state["messages_total"] + monkeypatch.setattr(analytics_read, "fetch_events", fetch_events) + monkeypatch.setattr(analytics_read, "fetch_messages", fetch_messages) return state @@ -43,6 +56,16 @@ def test_logs_page(fake_read: dict[str, Any], client_factory) -> None: assert "20:00:00" in response.text # 17:00 UTC -> 20:00 МСК +def test_logs_messages_view(fake_read: dict[str, Any], client_factory) -> None: + response = client_factory().get("/logs", params={"view": "messages"}) + + assert response.status_code == 200 + assert "Сообщения по дням" in response.text + assert "18.10.2026" in response.text + assert "17" in response.text + assert fake_read["message_kwargs"]["limit"] == 50 + + def test_logs_filters_passed(fake_read: dict[str, Any], client_factory) -> None: client_factory().get( "/logs", @@ -73,16 +96,16 @@ def test_logs_pagination(fake_read: dict[str, Any], client_factory) -> None: assert fake_read["kwargs"]["offset"] == 50 -def test_logs_clickhouse_error(client_factory, monkeypatch: pytest.MonkeyPatch) -> None: +def test_logs_postgres_error(client_factory, monkeypatch: pytest.MonkeyPatch) -> None: async def fetch_events(settings: Any, **kwargs: Any) -> tuple[list[dict[str, Any]], int]: - raise RuntimeError("ClickHouse недоступен") + raise RuntimeError("Postgres недоступен") monkeypatch.setattr(analytics_read, "fetch_events", fetch_events) response = client_factory().get("/logs") assert response.status_code == 200 - assert "ClickHouse недоступен" in response.text + assert "Postgres недоступен" in response.text def test_logs_bad_date(client_factory, monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/tests/test_alerts.py b/tests/test_alerts.py index 332a45d..50d5d62 100644 --- a/tests/test_alerts.py +++ b/tests/test_alerts.py @@ -35,17 +35,14 @@ class FakeBot: self.forwarded.append((chat_id, from_chat_id, message_id)) -class FakeClickHouse: - """Заглушка клиента ClickHouse (см. тесты аналитики).""" +class FakeWriter: + """Заглушка записи аналитики в Postgres (см. тесты аналитики).""" def __init__(self) -> None: - self.rows: list[list[Any]] = [] + self.events: 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 + async def __call__(self, events: list[list[Any]], messages: dict[Any, Any]) -> None: + self.events.extend(events) @pytest.fixture(autouse=True) @@ -147,8 +144,8 @@ 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) + fake = FakeWriter() + await init_analytics(settings, writer=fake) bot = FakeBot() assert await heartbeat.ping_hourly(bot, now=datetime(2026, 1, 1, 10, 0, tzinfo=UTC)) is True @@ -159,15 +156,15 @@ async def test_ping_hourly_sends_once_per_hour( 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] + assert [row[1] for row in fake.events] == [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) + fake = FakeWriter() + await init_analytics(settings, writer=fake) bot = FakeBot() assert await heartbeat.ping_hourly(bot, now=datetime(2026, 1, 1, 10, 0, tzinfo=UTC)) is False @@ -176,4 +173,4 @@ async def test_ping_hourly_records_event_without_alert_chat( await close_analytics() assert bot.messages == [] - assert [row[1] for row in fake.rows] == [EventType.BOT_ALIVE] + assert [row[1] for row in fake.events] == [EventType.BOT_ALIVE] diff --git a/tests/test_analytics.py b/tests/test_analytics.py index e1acc69..4303585 100644 --- a/tests/test_analytics.py +++ b/tests/test_analytics.py @@ -1,13 +1,10 @@ -"""Тесты аналитики: каталог событий, буферизация и деградация без ClickHouse. +"""Тесты аналитики: каталог событий, буферизация и запись в Postgres. -Тесты не требуют ни ClickHouse, ни aiogram: клиент подменяется фейком, а -``clickhouse_connect`` импортируется лениво внутри клиента. +Postgres не требуется: writer подменяется фейком. """ from __future__ import annotations -import json -import logging import re from datetime import UTC, datetime from pathlib import Path @@ -16,7 +13,15 @@ from typing import Any import pytest -from common.analytics import ALL_EVENT_TYPES, EventType, close_analytics, get_analytics, init_analytics, record +from common.analytics import ( + ALL_EVENT_TYPES, + EventType, + close_analytics, + get_analytics, + init_analytics, + record, + record_message, +) from common.analytics.client import PAYLOAD_MAX_LENGTH, AnalyticsClient from common.analytics.messages import describe_command, describe_message from common.config import Settings @@ -35,25 +40,24 @@ class _FakeEnum: self.value = value -class FakeClickHouse: - """Заглушка клиента ClickHouse: пишет в список или падает.""" +class FakeWriter: + """Заглушка записи в Postgres: складывает батчи в списки.""" 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.events: list[list[Any]] = [] + self.messages: dict[tuple[Any, int, int], list[Any]] = {} self._fail = fail - async def insert(self, table: str, data: list[list[Any]], column_names: list[str]) -> None: + async def __call__(self, events: list[list[Any]], messages: dict[tuple[Any, int, int], list[Any]]) -> 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 + raise RuntimeError("Postgres недоступен") + self.events.extend(events) + for key, value in messages.items(): + if key in self.messages: + self.messages[key][1] += value[1] + self.messages[key][0] = value[0] + else: + self.messages[key] = list(value) def _settings(**overrides: Any) -> Settings: @@ -68,9 +72,9 @@ async def _clean_singleton(): await close_analytics() -def _client(**overrides: Any) -> tuple[AnalyticsClient, FakeClickHouse]: - fake = FakeClickHouse() - return AnalyticsClient(_settings(**overrides), client=fake), fake +def _client(**overrides: Any) -> tuple[AnalyticsClient, FakeWriter]: + fake = FakeWriter() + return AnalyticsClient(_settings(**overrides), writer=fake), fake # --- каталог событий --- @@ -141,41 +145,46 @@ def test_describe_command_returns_none_for_plain_message() -> 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}) # не должно бросать + record(EventType.COMMAND, user_id=1, chat_id=2, payload={"a": 1}) # не должно бросать + record_message(user_id=1, username="vasya", chat_id=2) 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": "/инфо"}) + client.record(EventType.COMMAND, user_id=11, username="vasya", chat_id=100, payload={"command": "инфо"}) + client.record(EventType.PURIFICATION, user_id=11, username="vasya", chat_id=100, payload={"success": True}) 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 + assert len(fake.events) == 2 - row = fake.rows[0] + row = fake.events[0] assert isinstance(row[0], datetime) - assert row[1] == "message" + assert row[1] == "command" assert row[2] == 11 assert row[3] == "vasya" assert row[4] == 100 - assert '"message_id": 5' in row[5] + assert row[5] == {"command": "инфо"} assert row[6] == __version__ - assert row[7] == 0 + assert row[7] is False + + +async def test_message_counter_is_aggregated_per_day() -> None: + client, fake = _client() + client.record_message(user_id=11, username="vasya", chat_id=100) + client.record_message(user_id=11, username="vasya", chat_id=100) + client.record_message(user_id=12, username=None, chat_id=100) + + assert client.pending == 3 + await client.flush() + + assert client.pending == 0 + assert len(fake.messages) == 2 + vasya = next(value for key, value in fake.messages.items() if key[1] == 11) + assert vasya == ["vasya", 2] async def test_remember_user_enriches_username() -> None: @@ -185,17 +194,17 @@ async def test_remember_user_enriches_username() -> None: await client.flush() - assert fake.rows[0][3] == "nickname" + assert fake.events[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) + client.record(EventType.COMMAND, user_id=1) + client.record(EventType.COMMAND, user_id=1) assert not client._flush_now.is_set() - client.record(EventType.MESSAGE, user_id=1) + client.record(EventType.COMMAND, user_id=1) assert client._flush_now.is_set() @@ -203,64 +212,56 @@ 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) + client.record(EventType.COMMAND, 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, fake = _client(analytics_enabled=False) - client.record(EventType.MESSAGE, user_id=1) + client.record(EventType.COMMAND, user_id=1) + client.record_message(user_id=1, username="vasya", chat_id=1) await client.flush() assert client.pending == 0 - assert fake.rows == [] + assert fake.events == [] + assert fake.messages == {} -# --- деградация без ClickHouse --- +# --- деградация без Postgres --- 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) + fake = FakeWriter(fail=True) + client = AnalyticsClient(_settings(), writer=fake) + client.record(EventType.COMMAND, user_id=1, username="vasya", chat_id=100) - with caplog.at_level(logging.WARNING): + with caplog.at_level("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)}) + client.record(EventType.COMMAND, user_id=1, payload={"text": "x" * (PAYLOAD_MAX_LENGTH * 2)}) await client.flush() - assert len(fake.rows[0][5]) == PAYLOAD_MAX_LENGTH + payload = fake.events[0][5] + assert "_truncated" in payload + assert len(payload["_truncated"]) == PAYLOAD_MAX_LENGTH # --- жизненный цикл --- async def test_init_and_close_flush_pending_events() -> None: - fake = FakeClickHouse() - client = await init_analytics(_settings(), client=fake) + fake = FakeWriter() + client = await init_analytics(_settings(), writer=fake) record(EventType.BOT_START, payload={"app_env": "dev"}) record(EventType.BOT_STOP, payload={"app_env": "dev"}) @@ -271,8 +272,7 @@ async def test_init_and_close_flush_pending_events() -> None: await close_analytics() assert get_analytics() is None - assert len(fake.rows) == 2 - assert fake.closed is True + assert len(fake.events) == 2 # --- middleware (транспортные события) --- @@ -297,22 +297,23 @@ async def _handler(event: Any, data: dict[str, Any]) -> str: return "handled" -async def test_middleware_records_message_and_command(monkeypatch: pytest.MonkeyPatch) -> None: +async def test_middleware_counts_message_and_records_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) + fake = FakeWriter() + client = await init_analytics(settings, writer=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" + assert [row[1] for row in fake.events] == ["command"] + assert fake.events[0][5] == {"command": "инфо", "args_count": 0} + assert fake.events[0][3] == "vasya" + assert len(fake.messages) == 1 + assert next(iter(fake.messages.values())) == ["vasya", 1] async def test_middleware_skips_other_chats_and_bots(monkeypatch: pytest.MonkeyPatch) -> None: @@ -321,8 +322,8 @@ async def test_middleware_skips_other_chats_and_bots(monkeypatch: pytest.MonkeyP settings = _settings(chat_id=-100) monkeypatch.setattr("bot.middlewares.analytics.get_settings", lambda: settings) - fake = FakeClickHouse() - client = await init_analytics(settings, client=fake) + fake = FakeWriter() + client = await init_analytics(settings, writer=fake) from aiogram.types import Chat, User @@ -334,7 +335,8 @@ async def test_middleware_skips_other_chats_and_bots(monkeypatch: pytest.MonkeyP ) await client.flush() - assert fake.rows == [] + assert fake.events == [] + assert fake.messages == {} async def test_middleware_never_breaks_handling(monkeypatch: pytest.MonkeyPatch) -> None: @@ -346,6 +348,6 @@ async def test_middleware_never_breaks_handling(monkeypatch: pytest.MonkeyPatch) def boom(*args: Any, **kwargs: Any) -> None: raise RuntimeError("аналитика сломалась") - monkeypatch.setattr("bot.middlewares.analytics.record", boom) + monkeypatch.setattr("bot.middlewares.analytics.record_message", boom) assert await AnalyticsMiddleware()(_handler, _message(), {}) == "handled" diff --git a/tests/test_version.py b/tests/test_version.py index e503e9f..9d434b6 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.6.9" + assert __version__ == "0.7.0"