ホームページ >バックエンド開発 >Python チュートリアル >Cloud Run 機能と Cloud Scheduler を使用してグラフによる Slack 通知を自動化する

Cloud Run 機能と Cloud Scheduler を使用してグラフによる Slack 通知を自動化する

Mary-Kate Olsenオリジナル: 2024-12-11 04:47:09798ブラウズ

私は最近、過去 7 日間のセッション数を視覚化するグラフを使用して Slack 通知を自動化するシステムを構築しました。これは、データ処理とグラフ生成のための Cloud Run 機能と、実行のスケジュール設定のための Cloud Scheduler の組み合わせを使用して実現されました。

実装の概要

クラウドラン機能

Cloud Run 関数は、BigQuery にクエリを実行してセッションデータを取得し、Matplotlib を使用して折れ線グラフを作成し、そのグラフを Slack API 経由で Slack に送信します。次の手順では、セットアッププロセスの概要を説明します。

これが main.py のコードです。実行する前に、SLACK_API_TOKEN と SLACK_CHANNEL_ID を環境変数として設定する必要があります。後で設定するため、今は空のままにしておいても問題ありません。

import os
import matplotlib.pyplot as plt
from google.cloud import bigquery
from datetime import datetime, timedelta
import io
import pytz
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError

def create_weekly_total_sessions_chart(_):
    SLACK_TOKEN = os.environ.get('SLACK_API_TOKEN')
    SLACK_CHANNEL_ID = os.environ.get('SLACK_CHANNEL_ID')

    client = bigquery.Client()

    # Calculate the date range for the last 7 days
    jst = pytz.timezone('Asia/Tokyo')
    today = datetime.now(jst)
    start_date = (today - timedelta(days=7)).strftime('%Y-%m-%d')
    end_date = (today - timedelta(days=1)).strftime('%Y-%m-%d')

    query = f"""
        SELECT 
            DATE(created_at) AS date,
            COUNT(DISTINCT session_id) AS unique_sessions
        FROM `<project>.<dataset>.summary_all`
        WHERE created_at BETWEEN '{start_date} 00:00:00' AND '{end_date} 23:59:59'
        GROUP BY date
        ORDER BY date;
    """

    query_job = client.query(query)
    results = query_job.result()

    # Prepare data for the graph
    dates = []
    session_counts = []
    for row in results:
        dates.append(row['date'].strftime('%Y-%m-%d'))
        session_counts.append(row['unique_sessions'])

    # Generate the graph
    plt.figure()
    plt.plot(dates, session_counts, marker='o')
    plt.title('Unique Session Counts (Last 7 Days)')
    plt.xlabel('Date')
    plt.ylabel('Unique Sessions')
    plt.xticks(rotation=45)
    plt.tight_layout()

    # Save the graph as an image
    image_binary = io.BytesIO()
    plt.savefig(image_binary, format='png')
    image_binary.seek(0)

    # Send the graph to Slack
    client = WebClient(token=SLACK_TOKEN)
    try:
        response = client.files_upload_v2(
            channel=SLACK_CHANNEL_ID,
            file_uploads=[{
                "file": image_binary,
                "filename": "unique_sessions.png",
                "title": "Unique Session Counts (Last 7 Days)"
            }],
            initial_comment="Here are the session counts for the last 7 days!"
        )
    except SlackApiError as e:
        return f"Error uploading file: {e.response['error']}"

    return "Success"

依存関係

requirements.txt ファイルを作成し、次の依存関係を含めます:

functions-framework==3.*
google-cloud-bigquery
matplotlib
slack_sdk
pytz

Cloud Run 機能へのアクセス権の付与

Cloud Scheduler または他のサービスが Cloud Run 関数を呼び出せるようにするには、roles/run.invoker ロールを適切なエンティティに割り当てる必要があります。これを行うには、次のコマンドを使用します:

gcloud functions add-invoker-policy-binding create-weekly-total-sessions-chart \
      --region="asia-northeast1" \
      --member="MEMBER_NAME"

MEMBER_NAME を次のいずれかに置き換えます:

Cloud Scheduler のサービスアカウント: サービスアカウント:scheduler-account@example.iam.gserviceaccount.com
パブリックアクセスの場合 (推奨されません): すべてのユーザー