Bot Redash w Mattermost: automatyzacja wykresów

Automatyzacja dashboardów Redash: wysyłanie wykresów do Mattermost

Skrypt w Pythonie rozwiązuje problem automatycznego dostarczania wykresów z Redash do Mattermost. Pobieramy dane przez API, odtwarzamy wizualizacje w matplotlib, łączymy je w jeden dashboard i wysyłamy do kanału. Rozwiązanie idealne dla zespołów, w których dane aktualizują się z opóźnieniem, a ręczne przeglądanie dashboardu jest nieefektywne.

Pipeline uruchamiany jest przez cron: sprawdza świeżość danych, tworzy wykresy z adnotacjami i publikuje gotowy raport przed rozpoczęciem dnia pracy.

Pobieranie danych z Redash API

Redash udostępnia API do wykonywania zapytań za pomocą query_id z parametrami. Główna funkcja:

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()

Jeśli wynik nie jest gotowy, API zwraca job — zaimplementuj polling, aby czekać na zakończenie. Dla dashboardów parsuj parametry z 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()
    }

Dynamiczne daty, takie jak 'ostatnie 7 dni', przekształć ręcznie:

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')
    }

Tworzenie i adnotacja wykresów

Wizualizacje tworzymy w matplotlib, kopiując style Redash: kolory, typy linii, rozmiary. Podstawowy przykład wykresu liniowego:

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('Dynamika metryki')

    return fig

Kluczowy krok — adnotacje dla czytelności w statycznym obrazie. Dodajemy ostatnią wartość, zmianę tydzień do tygodnia, odchylenie od planu:

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

To pozwala na interpretację metryk bez interakcji.

Łączenie wykresów w dashboard

Kilka wykresów (według metryk, kategorii) łączymy w jeden obraz za pomocą PIL dla lepszego UX:

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)

Dodaj nagłówki bloków według działów lub tematów dla łatwiejszej nawigacji.

Kontrole i ochrona przed duplikatami

Skrypt sprawdza obecność docelowej daty w danych:

if df['date'].max() != target_date:
    raise Exception('Dane jeszcze się nie zaktualizowały')

Ochrona przed ponownym wysłaniem przez plik stanu:

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)

Wysyłanie do Mattermost

Użyj mattermostdriver do przesłania pliku i publikacji:

from mattermostdriver import Driver

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

mm.login()

mm.posts.create_post({
    'channel_id': channel_id,
    'message': 'Raport gotowy',
    'file_ids': [file_id],
})

Co jest ważne

• Automatyzacja redukuje pracę ręczną: wykresy trafiają do komunikatora automatycznie po aktualizacji danych.
• Adnotacje na wykresach zapewniają pełną informację bez hovera.
• Łączenie w jeden dashboard poprawia UX w czacie.
• Kontrole świeżości i duplikatów gwarantują niezawodność pipeline'u.
• Łatwe skalowanie: dodaj nowe query_id i parametry.

— Editorial Team