| | from aiohttp import ClientSession |
| | from loguru import logger |
| |
|
| | from bot.analytics.types import AbstractAnalyticsLogger, BaseEvent |
| |
|
| | GOOGLE_ANALYTICS_ENDPOINT = "https://www.google-analytics.com" |
| |
|
| |
|
| | class GoogleAnalyticsTelegramLogger(AbstractAnalyticsLogger): |
| | def __init__(self, api_secret: str, measurement_id: str, base_url: str = GOOGLE_ANALYTICS_ENDPOINT) -> None: |
| | self._api_secret: str = api_secret |
| | self._measurement_id: str = measurement_id |
| | self._base_url: str = base_url |
| | self._headers = {"Content-Type": "application/json", "Accept": "*/*"} |
| |
|
| | async def _send_request( |
| | self, |
| | event: BaseEvent, |
| | ) -> dict: |
| | url = f"{self._base_url}/mp/collect?measurement_id={self._measurement_id}&api_secret={self._api_secret}" |
| | params = dict(event) |
| |
|
| | async with ClientSession() as session, session.post(url, headers=self._headers, json=params) as response: |
| | json_response = await response.json(content_type="application/json") |
| |
|
| | logger.info("Send record to Google Analytics") |
| | logger.info(f"{json_response=}") |
| |
|
| | return self._validate_response(json_response) |
| |
|
| | @staticmethod |
| | def _validate_response(response: dict) -> dict: |
| | """Validate response.""" |
| | if not response.get("ok"): |
| | name = response["error"]["name"] |
| | code = response["error"]["code"] |
| |
|
| | logger.error(f"get error from cryptopay api | name: {name} | code: {code}") |
| | msg = f"Error in CryptoPay API call | name: {name} | code: {code}" |
| | raise ValueError(msg) |
| |
|
| | logger.info(f"got response | ok: {response['ok']} | result: {response['result']}") |
| | return response |
| |
|
| | async def log_event( |
| | self, |
| | event: BaseEvent, |
| | ) -> None: |
| | """Use this method to sends event to Google Analytics.""" |
| | await self._send_request(event) |
| |
|
