Redash Bot 在 Mattermost 中：图表自动化

自动化 Redash 仪表板：将图表发送至 Mattermost

一个 Python 脚本解决了自动将 Redash 图表发送到 Mattermost 的任务。它通过 API 获取数据，在 matplotlib 中重新创建可视化图表，将它们合并到一个仪表板中，并发送到频道。对于数据更新有延迟且手动查看仪表板效率低下的团队来说，这是理想选择。

该流程通过 cron 作业运行：检查数据新鲜度，构建带注释的图表，并在工作日开始前发布最终报告。

从 Redash API 获取数据

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

如果结果未就绪，API 会返回一个作业——实现轮询以等待完成。对于仪表板，从 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()
    }

手动转换动态日期，如“最近 7 天”：

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 风格：颜色、线条类型、大小。一个基本的折线图示例：

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('指标动态')

    return fig

关键步骤是为静态图像添加注释以提高可读性。添加最新值、周环比变化、与计划的偏差：

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

这使得指标无需交互即可理解。

将图表合并到仪表板中

多个图表（按指标、类别）使用 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('数据尚未更新')

通过状态文件防止重复发送：

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)

发送到 Mattermost

使用 mattermostdriver 上传文件并发布：

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],
})

关键要点

• 自动化减少手动工作：图表在数据更新时自动到达消息应用。
• 图表上的注释确保无需悬停交互即可完全理解信息。
• 合并到一个仪表板中改善了聊天中的用户体验。
• 新鲜度和重复检查保证了流程的可靠性。
• 易于扩展：添加新的 query_id 和参数。

— Editorial Team