|
|
from __future__ import annotations |
|
|
|
|
|
from aiohttp import ClientSession |
|
|
from loguru import logger |
|
|
|
|
|
from bot.analytics.types import AbstractAnalyticsLogger, BaseEvent |
|
|
|
|
|
POSTHOG_ENDPOINT = "https://app.posthog.com" |
|
|
|
|
|
|
|
|
class PosthogTelegramLogger(AbstractAnalyticsLogger): |
|
|
def __init__(self, api_token: str, base_url: str = POSTHOG_ENDPOINT) -> None: |
|
|
self._api_token: str = api_token |
|
|
self._base_url: str = base_url |
|
|
self._headers = { |
|
|
"Authorization": "Bearer ${POSTHOG_PERSONAL_API_KEY}", |
|
|
"Content-Type": "application/json", |
|
|
"Accept": "*/*", |
|
|
} |
|
|
self._timeout = 15 |
|
|
|
|
|
async def _send_request( |
|
|
self, |
|
|
event: BaseEvent, |
|
|
) -> dict: |
|
|
url = f"{self._base_url}/api/event/?personal_api_key={self._api_token}" |
|
|
params = dict(event) |
|
|
|
|
|
async with ( |
|
|
ClientSession() as session, |
|
|
session.post( |
|
|
url, |
|
|
headers=self._headers, |
|
|
json=params, |
|
|
timeout=self._timeout, |
|
|
) as response, |
|
|
): |
|
|
json_response = await response.json(content_type="application/json") |
|
|
|
|
|
logger.info("Send record to Posthog") |
|
|
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 Posthog.""" |
|
|
await self._send_request(event) |
|
|
|
