Cloud FunctionsからSlackへmatplotlibで生成したグラフを投稿する

こんにちは、Miotavaです。

直近で引越しをすることとなり、どんな書斎にしようかなとイメージをStable Diffusionで生成しながら固めているところです。

今まではPinterestでインテリアの参考画像を探していたのですが、生成AI使えば割とドンピシャ（死語？）のイメージを作ってくれるので、チョベリグ（死語）です。

本記事について

さて、以前BigQueryから抽出したデータを表形式でSlack投稿する方法をまとめてみました。
https://zenn.dev/spacemarket/articles/c6f99ca8a36b8a

本記事ではさらにリッチなデータ集計としてグラフデータをCloud Functionsで生成し、Slack通知する方法についてまとめたいと思います。
以前の記事内容と組み合わせれば、BigQueryのデータをいい感じに集計・グラフ化してSlackに定期投稿する、なんてことができます。

課題に感じていたこと

Cloud Functions上でグラフを生成するとなると、一時的にでもファイルとしてどこかに保存する必要があるから何らかのストレージサービス使わないといけないのかなー、ちょっとめんどくさいなー、と感じていました。

また、Slackへメッセージではなくファイルを投稿する、といったことについてもちょっとしたハードルでした。

課題の解決

生成したグラフデータはメモリ上に保存しておく

Pythonの標準ライブラリであるioの BytesIO　を用いてバイナリストリームを生成し、そこにグラフデータを保存するようにします。

# グラフにしたいデータを用意
df = pd.DataFrame(
        {
            "date": [
                "2023-05-01",
                "2023-05-02",
                "2023-05-03",
                "2023-05-04",
                "2023-05-05",
                "2023-05-06",
            ],
            "value": [100, 105, 110, 103, 115, 125],
        }
    )

# データをグラフにする
plt.plot(df["date"], df["value"])

# バイナリストリームを用意
buff = io.BytesIO()

# バイナリストリーム（メモリ）上にグラフデータを保存
plt.savefig(buff, format="png")

# グラフ表示は不要なのでcloseする
plt.close()

https://docs.python.org/ja/3/library/io.html#binary-i-o

Slack投稿する際は、バイナリストリームからバイナリデータを取得し、Slack APIにPOSTパラメータとしてバイナリデータをセットすればOKでした。

この方法であればグラフデータをいちいちストレージに保存したりする必要がなくCloud Functions上のメモリだけでデータが完結するので簡単です。

[ただの雑談]
調べてみたら、BytesIOを利用して標準入出力を高速化する手法なんてものが紹介されていました。面白い！

ファイル投稿もSlack APIを叩けばOK

files.uploadのSlack APIが用意されているので、その仕様に従って叩けば投稿可能です。

import requests

requests.post(
    url="https://slack.com/api/files.upload",
    data={
        "title": title,
        "token": <SLACK_ACCESS_TOKEN>,
        "channels": <SLACK_CHANNEL_ID>,
    },
    files={"file": file},
)

https://api.slack.com/methods/files.upload

上記の file には前述のバイナリストリーム buff から getvalue メソッドでバイナリデータを取得しセットするようにします。（詳細は最終形の実装を参照ください）

最終形

実装

main.py

import io
import os

import matplotlib.pyplot as plt
import pandas as pd
from lib.slack import Slack


def main(events, context) -> None:
    # Cloud Functionsの環境変数として入れているSlack認証用情報読み込み
    SLACK_CHANNEL_ID = os.environ.get("SLACK_CHANNEL_ID")
    SLACK_ACCESS_TOKEN = os.environ.get("SLACK_ACCESS_TOKEN")

    if SLACK_CHANNEL_ID is None or SLACK_ACCESS_TOKEN is None:
        raise Exception(
            "SLACK_CHANNEL_ID と SLACK_ACCESS_TOKEN が環境変数としてセットされていることを確認してください。"
        )

    # グラフにしたいデータを作る
    df = pd.DataFrame(
        {
            "date": [
                "2023-05-01",
                "2023-05-02",
                "2023-05-03",
                "2023-05-04",
                "2023-05-05",
                "2023-05-06",
            ],
            "value": [100, 105, 110, 103, 115, 125],
        }
    )

    # データをグラフにする
    plt.plot(df["date"], df["value"])

    # バイナリストリームを用意
    buff = io.BytesIO()

    # バイナリストリーム（メモリ）上にグラフデータを保存
    plt.savefig(buff, format="png")

    # グラフ表示は不要なのでcloseする
    plt.close()

    # buffに保存したデータをbytesデータに変換し、Slackに送信する
    slack = Slack(SLACK_CHANNEL_ID, SLACK_ACCESS_TOKEN)
    slack.post_file(buff.getvalue(), "グラフファイルのタイトル")

requirements.txt

pandas
matplotlib
requests

lib/slack.py

import requests


class Slack:
    def __init__(self, channel_id: str, access_token: str) -> None:
        self.channel_id = channel_id
        self.access_token = access_token

    def post_file(self, file: bytes, title: str) -> None:
        try:
            requests.post(
                url="https://slack.com/api/files.upload",
                data={
                    "title": title,
                    "token": self.access_token,
                    "channels": self.channel_id,
                },
                files={"file": file},
            )
        except requests.exceptions.RequestException as e:
            print("Slack投稿リクエスト時にエラー発生しました。詳細： ", e)

投稿してみた様子