126 lines
4.1 KiB
Python
126 lines
4.1 KiB
Python
"""Чтение событий ClickHouse для админки (Блок 4.3).
|
||
|
||
Клиент создаётся лениво и только на чтение; недоступность ClickHouse не роняет
|
||
админку — страница покажет ошибку. Запись событий живёт в ``client.py``.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from datetime import UTC, datetime, time, timedelta
|
||
from typing import Any
|
||
|
||
from common.analytics.events import ALL_EVENT_TYPES
|
||
from common.config import Settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
#: Колонки, которые показываем в логах.
|
||
EVENT_FIELDS = (
|
||
"event_time",
|
||
"event_type",
|
||
"user_id",
|
||
"username",
|
||
"chat_id",
|
||
"payload",
|
||
"bot_version",
|
||
"prod",
|
||
)
|
||
|
||
_client: Any | None = None
|
||
|
||
|
||
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
|
||
|
||
|
||
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
|
||
|
||
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
|
||
|
||
|
||
async def fetch_events(
|
||
settings: Settings,
|
||
*,
|
||
event_type: str = "",
|
||
user: str = "",
|
||
date_from: str = "",
|
||
date_to: str = "",
|
||
limit: int = 50,
|
||
offset: int = 0,
|
||
) -> tuple[list[dict[str, Any]], int]:
|
||
"""Вернуть события и общее количество по фильтрам.
|
||
|
||
Даты — ``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
|
||
|
||
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)
|
||
)
|
||
|
||
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]
|
||
return rows, total
|