"""Чтение аналитики из Postgres для админки.""" from __future__ import annotations import logging 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__) #: Колонки, которые показываем в логах. EVENT_FIELDS = ( "event_time", "event_type", "user_id", "username", "chat_id", "payload", "bot_version", "prod", ) 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 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 def _user_clause(user: str, params: list[Any], column: str = "user_id") -> str: user = user.strip() 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( 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]: """Вернуть события и общее количество по фильтрам.""" start, end = _dates(date_from, date_to, settings) start_dt, end_dt = _event_bounds(start, end, settings) 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) 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 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