Bot Redash v Mattermost: automatizace grafů

Automatizace dashboardů Redash: odesílání grafů do Mattermost

Python skript řeší úkol automatického doručování grafů z Redash do Mattermost. Získáme data přes API, znovu vytvoříme vizualizace v matplotlib, sloučíme je do jednotného dashboardu a odešleme do kanálu. Vhodné pro týmy, kde se data aktualizují se zpožděním a manuální prohlížení dashboardu je neefektivní.

Pipeline se spouští přes cron: kontroluje aktuálnost dat, vytváří grafy s anotacemi a publikuje hotovou zprávu před začátkem pracovního dne.

Získání dat z Redash API

Redash poskytuje API pro spouštění dotazů podle query_id s parametry. Hlavní funkce:

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

Pokud výsledek není připraven, API vrací job — implementujte polling pro čekání na dokončení. Pro dashboardy parsujte 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()
    }

Dynamická data jako 'posledních 7 dní' převeďte ručně:

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

Vytváření a anotace grafů

Vizualizace vytváříme v matplotlib, kopírujeme styly Redash: barvy, typy čar, velikosti. Základní příklad liniového grafu:

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

    return fig

Klíčový krok — anotace pro čitelnost ve statickém obrázku. Přidáme poslední hodnotu, změnu týden po týdnu, odchylku od plánu:

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

To umožňuje vnímat metriky bez interaktivity.

Sloučení grafů do dashboardu

Několik grafů (podle metrik, kategorií) slepíme do jednoho obrázku pomocí PIL pro lepší 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)

Přidejte nadpisy bloků podle oddělení nebo témat pro navigaci.

Kontroly a ochrana před duplikáty

Skript kontroluje přítomnost cílového data v datech:

if df['date'].max() != target_date:
    raise Exception('Data ještě nebyla aktualizována')

Ochrana před opakovaným odesláním přes state-soubor:

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)

Odeslání do Mattermost

Použijte mattermostdriver pro nahrání souboru a publikování:

from mattermostdriver import Driver

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

mm.login()

mm.posts.create_post({
    'channel_id': channel_id,
    'message': 'Zpráva je připravena',
    'file_ids': [file_id],
})

Co je důležité

• Automatizace snižuje manuální práci: grafy přicházejí do messengeru automaticky po aktualizaci dat.
• Anotace na grafech zajišťují plnou informační hodnotu bez najetí myší.
• Sloučení do jednoho dashboardu zlepšuje UX v chatu.
• Kontroly aktuálnosti a duplikátů zaručují spolehlivost pipeline.
• Snadno škálovatelné: přidejte nové query_id a parametry.

— Editorial Team