使用 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
对于公众访问（不推荐）：所有用户