使用 Cloud Run Functions 和 Cloud Scheduler 透過圖形自動發送 Slack 通知

我最近建立了一個系統來自動執行 Slack 通知，並透過圖表視覺化過去 7 天的會話計數。這是透過結合用於資料處理和圖形生成的 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"
</dataset></project>

依賴關係

建立一個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 的服務帳戶： serviceAccount:scheduler-account@example.iam.gserviceaccount.com
• 對於公眾訪問（不建議）： 所有用戶

設定雲端調度程序

使用 Cloud Scheduler 在每週一上午 10:00 (JST) 自動執行函數。這是一個範例配置：

Slack API 配置

要讓您的 Cloud Run 功能能夠發送 Slack 通知，請依照下列步驟操作：

1. 前往 Slack API 並建立一個新應用程式。
2. 在 OAuth 和權限 下分配以下機器人令牌範圍：
   • 頻道：閱讀
   • 聊天：寫
   • 檔案：寫入

1. 將應用程式安裝到您的 Slack 工作區並複製 機器人用戶 OAuth 令牌。

1. 將應用程式新增至您要發佈通知的 Slack 頻道。

1. 複製通道 ID 並將其與 Bot 令牌一起貼上到 Cloud Run 函數的 SLACK_CHANNEL_ID 和 SLACK_API_TOKEN 環境變數中。

最終結果

一切設定完畢後，您的 Slack 頻道將收到每週通知，其中包含以下圖表：

以上是使用 Cloud Run Functions 和 Cloud Scheduler 透過圖形自動發送 Slack 通知的詳細內容。更多資訊請關注PHP中文網其他相關文章！