Интеграция 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/ и найти наиболее удобный для себя

Заключение

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