Интеграция Google Analytics с Telegram

Ни для кого не секрет как важна аналитика. Она не только дает информацию о паттернах поведения пользователей, но и может помочь отследить частоту использования той или иной фичи которую вы могли недавно интегрировали. Однако, регулярное посещение сайта аналитики и протыкивание нужных эвентов может превратиться в утомительную рутину, которую можно избежать. Для того чтобы упростить задачу я набросал python сервер который бы раз в сутки забирал необходимые данные из аналитики и транслировал бы в чат. В рамках статьи я опишу базовый пример такой интеграции, однако возможности для расширения функционала довольно большие. Например, рисование графики или интеграцией с вашим сервером, если аналитика ограничивается id объекта.

Приготовления

Сперва нам конечно же нужны данные которые мы хотим отслеживать. В качестве примера я буду использовать Custom definition означающее частоту нажатия на категорию продукта. Для создания таких определений вам нужна роль как минимум редактора и не стоит ожидать что сразу после создания у вас будут данные. К сожалению, аналитика по каждому определению начинает собираться только после создания даже если по выбранному параметру уже полно данных. Подробнее о них можно почитать здесь Custom dimensions and metrics.

Вкратце как создавать такие определения:

1. Заходим в аналитику проекта.

2. Идем в панель администратора.

3. Выбираем нужный property.

4. Жмем ��а Custom definitions.

5. Создаем definition указав параметр события который хотим отслеживать.

Вторым шагом нам нужно получить credentials.json для нашего бота. Для этого нужно создать проект на Cloud Platform если у вас еще его нет и включить Google Analytics Data API v1 в нем. Чтобы максимально упростить этот процесс, можно нажать на кнопку подготовки проекта на страницы документации Google Analytics Data API (GA4).

Как описывается в выше указанной документации, нам нужно забрать email из полученного json-а и вставить его в панели администратора нашего проперти в аналитике. Viewer прав будет вполне достаточно.

С телеграмом все проще. Если у вас еще нет бота телеграм, то он создается у @BotFather который выдаст токен с доступом. Чтобы получить id чата в который вы хотите вещать, можно добавить бот в чат и написать ему что либо, потом запросить апдейты у API телеграма по токену и получить

curl https://api.telegram.org/<ваш_токен>/getUpdates

Сервер

Сервер будет написан на python и для интеграции с ними нужно две библиотеки:

• google-analytics-data

• python-telegram-bot

pip install google-analytics-data python-telegram-bot

В принципе можно было бы обойтись и без библиотеки телеграмма и обойтись REST-ом, но в перспективе хочется расширить функционал, посему почему бы и не добавить.

Чтобы изолировать конфигурацию проекта от кода, создадим папку data куда положим наш credentials.json и в последующем добавим конфигурацию самого проекта.

Google Аналитика

Сначала давайте получим данные с аналитики. Согласно примеру нам нужно выполнить что то вроде

from google.analytics.data_v1beta import BetaAnalyticsDataClient
from google.analytics.data_v1beta.types import (
    DateRange,
    Dimension,
    Metric,
    RunReportRequest,
)

def sample_run_report(property_id="YOUR-GA4-PROPERTY-ID"):
    """Runs a simple report on a Google Analytics 4 property."""
    client = BetaAnalyticsDataClient()

    request = RunReportRequest(
        property=f"properties/{property_id}",
        dimensions=[Dimension(name="city")],
        metrics=[Metric(name="activeUsers")],
        date_ranges=[DateRange(start_date="2020-03-31", end_date="today")],
    )
    response = client.run_report(request)

    print("Report result:")
    for row in response.rows:
        print(row.dimension_values[0].value, row.metric_values[0].value)

Однако, у нас есть свое dimension. Запрос на него выполняется немного по другому

dimensions=[Dimension(name="customEvent:<выш_сustom_definition>")],

Диапозон дат лучше выбрать более динамичный. Допустим за прошедшие 7 дней

date_format = "%Y-%m-%d"
week_ago = (date.today() - timedelta(days = 7)).strftime(date_format)

Ну и завернем все это в класс. Честно говоря я не так часто имею дело с python и не знаю основных практик, но что то подсказывает, что так будет лучше. Так как я планирую использовать различные property_id и не хочу таскать их толкать в функцию каждый раз. Результат выглядит так:

class Analytics:
    def __init__(self, config):
        self.property_id = config['property_id']

    def run_report(self, event_name, limit):
        """Runs a simple report on a Google Analytics 4 property."""
        client = BetaAnalyticsDataClient()
        date_format = "%Y-%m-%d"
        week_ago = (date.today() - timedelta(days = 7)).strftime(date_format)
        
        request = RunReportRequest(
            property=f"properties/{self.property_id}",
            dimensions=[Dimension(name="customEvent:" + event_name)],
            metrics=[Metric(name="activeUsers")],
            date_ranges=[DateRange(start_date=week_ago, end_date="today")], 
            limit=limit,
        )
        response = client.run_report(request)
        result = []
        for row in response.rows:
            demension = row.dimension_values[0].value
            value = row.metric_values[0].value
            if demension == "(not set)": 
                continue
            result.append((demension, value))
        return result

Здесь мы собираем в массив результаты (значение, количество) исключая такого значения которое не выбрано. Их бывает довольно много и оно не особо интересно.

Telegram

У нас уже есть id чата, токен для бота и массив данных. Теперь можно их отправить. Сперва давайте сформатирует сообщение.

def format_report(title, data): 
    message = f'*{title}:*\n'
    index = 1
    for row in data:
        message += f'{str(index)}. {row[0]} - _{row[1]}_\n'
        index += 1
    return message

Функция забирает заголовок и формирует пронумерованный список. Далее отправка

from telegram.constants import ParseMode
from telegram.ext import ApplicationBuilder, Defaults

application = (
    ApplicationBuilder()
    .token("<ваш_токен>")
    .defaults(Defaults(parse_mode=ParseMode.MARKDOWN))
    .build()
)

await application.bot.send_message(
    chat_id="<id_чата>",
    text=format_report("<заголовок>", report)
)

И так же завернем все это в класс

import logging
from telegram.constants import ParseMode
from telegram.ext import ApplicationBuilder, Defaults

logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.INFO
)

class TelegramBot:
    def __init__(self, config):
        self.chat_id = config['chat_id']
        self.application = (
            ApplicationBuilder()
            .token(config['token'])
            .defaults(Defaults(parse_mode=ParseMode.MARKDOWN))
            .build()
        )

    def format_report(self, title, data): 
        message = f'*{title}:*\n'
        index = 1
        for row in data:
            message += f'{str(index)}. {row[0]} - _{row[1]}_\n'
            index += 1
        return message

    async def send_report(self, report, title):
        message = self.format_report(title, report)
        await self.application.bot.send_message(
            chat_id=self.chat_id,
            text=message
        )

HTTP Сервер

Может показаться странным, но для того чтобы запускать те или иные события сервера я решил использовать простой HTTP сервер. Его события можно легко дергать cron-ом через curl да и в ответ на каждое действие можно выдавать какой то статус, что в каких то случаях может помочь отследить сбои по каждому частному случаю.

В качестве HTTP сервера был выбран AIOHTTP

pip install aiohttp

Сервер является связующим звеном обоих классов аналитики и бота, посему он будет держать оба объекта при себе. А так же конфигурацию проперти для выдачи.

class AnalyticsProperty:
    def __init__(self, config):
        self.title = config['title']
        self.dimension = config['dimension']
        self.limit = config['limit']
        self.endpoint = config['endpoint']

class Server: 
    def __init__(self, config, telegram: TelegramBot, analytics: Analytics):
        self.telegram = telegram
        self.analytics = analytics
        self.host = config['host']
        self.port = config['port']
        self.app = web.Application()
        self.app.add_routes([
            web.get('/analytics/{property}', self.handle_analytics)
        ])
        self.properties = {}
    
    def add_routes(self, properties):
        for property in properties:
            item = AnalyticsProperty(property)
            self.properties[item.endpoint] = item
        
    async def handle_analytics(self, request):
        property_id = request.match_info['property']
        if property_id in self.properties:
            property = self.properties[property_id]
            report = self.analytics.run_report(property.dimension, property.limit)
            await self.telegram.send_report(report, title=property.title)
            return web.Response(text='ok')
        else:
            raise web.HTTPNotFound()
        
    def run(self):
        web.run_app(self.app, host=self.host, port=self.port)

По поводу конфигурации мы поговорим позже. Основное что здесь можно выделить, это инициализация сервера и определение метода который будет отвечать за забор аналитики и их отправку. Порядок использования будет такой.

1. Мы создаем объект Server с указанием бота и аналитики.

2. Добавляем в него нужны пути с информацией о параметрах которые сервер будет запрашивать и отправлять.

В результате, будет что то вроде

bot = TelegramBot(config['telegram'])
analytics = Analytics(config['analytics'])
server = Server(config['server'], bot, analytics)
server.add_routes(config['properties'])
server.run()

Конфигурация

Для конфигурации я выбрал yaml-файл.

pip install pyyaml

Конфигурация будет резделена на 4 основные части:

• telegram - токен и id чата;

• analytics - id проперти;

• server - host и post для создания сервера;

• properties - массив из аналитических данных:

  • title - заголовок измерения;

  • dimension - наш custom difinition;

  • limit - максимальное количество результатов;

  • endpoint - компонент пути по которому будет иницирована операция.

В результате получаем такой конфиг config.yaml который положим в папку data:

server:
  host: 0.0.0.0 # Server host
  port: 8080 # Server port
analytics:
  property_id: <your_google_analytics_view_id>
telegram:
  token: <your_telegram_bot_token>
  chat_id: <your_telegram_chat_id>
properties:
  - 
    title: 'Popular categories'
    dimension: 'category_name'
    limit: 15
    endpoint: 'top_categories'

Docker

Для запуска сервера было решено запускать его в докере и дергать тот или иной endpoint кроном раз в сутки. В качестве конфига был выбран вполне шаблонный Dockerfile:

FROM python:3.9-alpine

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY ["app.py", "analytics.py", "server.py", "telegram_bot.py", "./"]
ENV GOOGLE_APPLICATION_CREDENTIALS="/app/data/credentials.json"

EXPOSE 8080
ENTRYPOINT ["python", "app.py"]

Для полноценной работы контейнер требует подключения папки data в /app/data где будет лежать файл для доступа к Google аналитке и наш конфиг.

Хочу заметить, что хорошо бы закрыть контейнер от внешнего интернета так как он не требует какой либо авторизации. Ну или же настроить cron не снаружи, а внутри контейнера и не пробрасывать порты наружу.

Cron

Стоит в кратце объяснить про запуск. Чтобы настроить запуск той или иной задачи в cron необходимо добавить его с помощью конфигурации файла который вызывается по команде

crontab -e

Такой конфигурацией раз в сутки в 00:00 будет вызывается GET запрос на локальный адрес нашего сервера:

0 0 * * * curl localhost:8060/analytics/top_categories >/dev/null 2>&1

Поиграться с различными конфигурациями можно здесь https://crontab.guru/ и найти наиболее удобный для себя

Заключение

Полный результат можно посмотреть в репозитории. Конечно же решение требует от себя доработок, например, рисование графиков или интеграцию с другими серверами для выдачи более детализированной информации. Но как прототип он показал себя очень хорошо.