使用 Cloud Run Functions 和 Cloud Scheduler 透過圖形自動發送 Slack 通知-Python教學-PHP中文網

首頁

後端開發

Python教學

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

Mary-Kate Olsen

Dec 11, 2024 am 04:47 AM

我最近建立了一個系統來自動執行 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"

登入後複製

依賴關係

建立一個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
對於公眾訪問（不建議）：所有用戶