| 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) |
|
|
