| | from __future__ import annotations |
| | from functools import wraps |
| | from typing import TYPE_CHECKING, Any, Callable, TypeVar |
| |
|
| | from aiogram.types import CallbackQuery, Message |
| |
|
| | from bot.analytics.amplitude import AmplitudeTelegramLogger |
| | from bot.analytics.types import AbstractAnalyticsLogger, BaseEvent, EventProperties, EventType, UserProperties |
| | from bot.core.config import settings |
| | from bot.utils.singleton import SingletonMeta |
| |
|
| | if TYPE_CHECKING: |
| | from collections.abc import Awaitable |
| |
|
| | _Func = TypeVar("_Func") |
| |
|
| |
|
| | class AnalyticsService(metaclass=SingletonMeta): |
| | def __init__(self, logger: AbstractAnalyticsLogger | None) -> None: |
| | self.logger = logger |
| |
|
| | async def _track_error(self, user_id: int, error_text: str) -> None: |
| | if not self.logger: |
| | return |
| |
|
| | await self.logger.log_event( |
| | BaseEvent( |
| | user_id=user_id, |
| | event_type="Error", |
| | event_properties=EventProperties(text=error_text), |
| | ), |
| | ) |
| |
|
| | def track_event( |
| | self, |
| | event_name: EventType, |
| | ) -> Callable[[Callable[..., Awaitable[_Func]]], Callable[..., Awaitable[_Func]]]: |
| | """Decorator for tracking events in Amplitude, Google Analytics or Posthog.""" |
| |
|
| | def decorator( |
| | handler: Callable[[Message | CallbackQuery, dict[str, Any]], Awaitable[_Func]], |
| | ) -> Callable[..., Awaitable[_Func]]: |
| | @wraps(handler) |
| | async def wrapper(update: Message | CallbackQuery, *args: Any) -> Any: |
| | if not self.logger: |
| | return await handler(update, *args) |
| |
|
| | if (isinstance(update, (Message, CallbackQuery))) and update.from_user: |
| | user_id = update.from_user.id |
| | first_name = update.from_user.first_name |
| | last_name = update.from_user.last_name |
| | username = update.from_user.username |
| | url = update.from_user.url |
| | language = update.from_user.language_code |
| | else: |
| | return None |
| |
|
| | chat_id: int | None |
| | chat_type: str | None |
| | if isinstance(update, Message): |
| | chat_id = update.chat.id |
| | chat_type = update.chat.type |
| | text = update.text |
| | command = update.text if update.text and update.text.startswith("/") else None |
| | elif isinstance(update, CallbackQuery): |
| | chat_id = update.message.chat.id if update.message else None |
| | chat_type = update.message.chat.type if update.message else None |
| | text = update.data |
| | command = None |
| |
|
| | await self.logger.log_event( |
| | BaseEvent( |
| | user_id=user_id, |
| | event_type=event_name, |
| | user_properties=UserProperties( |
| | first_name=first_name, |
| | last_name=last_name, |
| | username=username, |
| | url=url, |
| | ), |
| | event_properties=EventProperties( |
| | chat_id=chat_id, |
| | chat_type=chat_type, |
| | text=text, |
| | command=command, |
| | ), |
| | language=language, |
| | ), |
| | ) |
| | try: |
| | result = await handler(update, *args) |
| | except Exception as e: |
| | await self._track_error(user_id, str(e)) |
| | raise |
| | return result |
| |
|
| | return wrapper |
| |
|
| | return decorator |
| |
|
| |
|
| | logger = AmplitudeTelegramLogger(api_token=settings.AMPLITUDE_API_KEY) if settings.AMPLITUDE_API_KEY else None |
| |
|
| | analytics = AnalyticsService(logger) |
| |
|
