В итоге я собрала небольшой пайплайн, который:

1. Забирает данные из Redash по API

2. Строит аналогичные графики в matplotlib

3. Склеивает их в одну картинку

4. Отправляет в Mattermost

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

Каждый день пользователи получают вот такое сообщение:

• набор графиков по разным метрикам

• разбивка по категориям

• подписи с ключевыми цифрами

Но самое главное, эти графики можно посмотреть с утра, пока ты даже не сел за компьютер.

У Redash есть API, через который можно дергать запросы.

По сути, нам нужно:

• взять query_id

• передать параметры

• дождаться результата

Упрощенно это выглядит так:

import requests

BASE_URL = 'https://your-redash-instance'
HEADERS = {'Authorization': 'Key YOUR_API_KEY'}

def fetch_query(query_id, params):
    response = requests.post(
        f'{BASE_URL}/api/queries/{query_id}/results',
        headers=HEADERS,
        json={'parameters': params, 'max_age': 0},
    )
    return response.json()

Но есть нюанс: если результат не готов сразу, Redash вернет job, и нужно будет подождать. Поэтому в реальности добавляется polling (ожидание выполнения запроса).

Если вы дергаете не просто query, а виджет с дашборда, становится чуть сложнее.

Нужно:

• распарсить параметры из URL

• сопоставить их с параметрами запроса (это была сама сложная часть, так как нужно было учесь все типы параметров)

• корректно привести типы (даты, списки и т.д.)

Пример:

from urllib.parse import parse_qs, urlparse

def parse_dashboard_url(url):
    parsed = urlparse(url)
    params = parse_qs(parsed.query)
    return {
        key: value[0]
        for key, value in params.items()
    }

Отдельная боль это динамические даты (last 7 days, this month и т.д.).
Их приходится разворачивать руками:

from datetime import datetime, timedelta

def last_7_days():
    today = datetime.utcnow()
    return {
        'start': (today - timedelta(days=7)).strftime('%Y-%m-%d'),
        'end': today.strftime('%Y-%m-%d')
    }

Дальше используем обычный matplotlib.

Графики максимально приближенные к оригиналам ( к тому как они выглядят в redash) все признаки, цвета и типы линий тоже можно забрать по API

Пример:

import matplotlib.pyplot as plt

def build_chart(df):
    fig, ax = plt.subplots(figsize=(6, 4))

    ax.plot(df['date'], df['value'], linewidth=2)
    ax.plot(df['date'], df['plan'], linewidth=2)

    ax.set_title('Metric dynamics')

    return fig

Очень важный момент, график должен быть читаем без наведения мышки, так как это картинка, а не дажшборд

Поэтому я добавляю:

• последнее значение

• WoW изменение

• отклонение от плана

Пример:

ax.annotate(
    f"value: {last_value}",
    xy=(last_date, last_value),
    xytext=(10, 10),
    textcoords='offset points'
)

Это сильно упрощает восприятие, особенно когда график просто прислали в чат.

Отправлять 30 отдельных картинок явно плохое решение с точки зрения UX/UI, их можно объединить, например, по отделам
Для этого я собирала их в один layout с помощью PIL:

from PIL import Image

def merge_images(images, output_path):
    base = Image.new('RGB', (1200, 800), 'white')

    for img, pos in images:
        base.paste(img, pos)

    base.save(output_path)

Плюс добавляю отдельный блок с заголовком категории чтобы сразу было понятно, что за данные.

Сам код у меня стоит на кроне и проверят есть ли уже на графиках нужная мне дата, как только проверка успешна - то только тогда запускается весь скрипт выше

Пример проверки:

if df['date'].max() != target_date:
    raise Exception('Данные еще не обновились')

У Mattermost есть Python SDK, через который можно:

• загрузить файл

• отправить сообщение

• создать личный канал

Пример:

from mattermostdriver import Driver

mm = Driver({
    'url': 'your-mattermost',
    'token': 'BOT_TOKEN',
})

mm.login()

mm.posts.create_post({
    'channel_id': channel_id,
    'message': 'Отчет готов',
    'file_ids': [file_id],
})

Чтобы скрипт не отправлял один и тот же отчет несколько раз, я сохраняю дату последней отправки, и в самом начале кода проверяю существует ли уже эта запись

from pathlib import Path

STATE_FILE = Path('last_sent.txt')

def was_sent(date):
    return STATE_FILE.exists() and STATE_FILE.read_text() == str(date)

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

Схематичный пример, как это выглядит на мобильном приложении

Меня зовут Валерия Смирнова, я уже более 6 лет работаю в продуктовой и в BI-аналитике. Веду канал про BI и визуализацию данных и аналитике.