【Python】Zoom会議をお知らせしてくれるサーバーレスアプリ【#StayHome】

テレワークが基本となった近頃、Zoomを使ったWebミーティングも多くなったのではないでしょうか？
ぶっちゃけ私は別のコミュニケーションツールを使うことが多いのですが、Zoomの利用が急激に増加している昨今、これを手動で毎日操作して運用していくのもそれなりに不都合や煩わしさがあるんだろうなあ〜みたいに想像しています。

そこで、Zoom会議を自動で作成してくれて、URLを勝手にSlackに通知してくれる機能なんてあったら嬉しくない？と思い、簡単ではありますがそれらしいものを作ってみました。

アプリの全体構成

今回のアプリはAWS上で構築しています。基本方針としては、全力でサーバーレスを活用し、疎結合で変更に強いアーキテクチャを意識すること。自分の学習的目的が強いですが。

アーキテクチャ

アーキテクチャは↓のような形で設計しました。中途半端で申し訳ないですが、今回は赤枠の範囲を実際に構築しています。やり切れてない範囲は時間を見つけて、改めてやりたいところですが。。。

ポイントは以下のような感じになります。

1. 処理は全てLambdaを使って作成
2. データベースはDynamoDBを使ってマネージドサービスを活用
3. ファイル管理するストレージにはS3を活用
4. 各処理はStep Functionsで統合し、処理間の依存性を低減
5. エラー通知はSNSのトピックにPush

処理の流れ

次に処理の流れをざっくりと説明します。先述の通り、今回は赤枠の部分までしか実際には作れていませんが。。。

1.Cloudwatch Eventで定期的にStep Functionsを実行

実行タイミングはcronで定義します。※ちなみに今回は日次で実行することを想定しています。

2.Step Functionsが起動

ステートマシーンを定義したJSONは割愛しますが、フローは以下の図になります。
各処理については詳細を後述します。

3.JWT Tokenの生成

ZoomのAPIを実行するにあたり、APIキーとシークレットキーを用いてJWT Tokenを生成します。
ZoomがサポートしているJWTの仕様についてはこちらをご覧ください。

4.Zoom会議の作成

3で生成したJWTを使ってZoomの Create a Meeting API を実行します。レスポンスで受け取った会議情報（IDやURL）はDynamoにPUTします。このとき、BodyのJSONは基本的に変更がなく常に固定なので、S3に保管しておき、実行ごとにS3から取得する方式にしています。

5.会議作成結果の判定

Step Functionsのchoiceで4の処理結果に応じて後続処理を決定します。成功であれば6のSlack通知へ、失敗であれば7のSNS通知へといった感じです。

6.会議情報のSlack通知

Dynamoに格納されている会議情報をGETし、slackの Incoming Webhook を使ってSlackのチャンネルに通知します。

7.エラー通知のSNS通知

Zoom APIに失敗したときはSNSのトピックに通知します。通知を受けたトピックはサブスクライバにしているLambdaを経由してメッセージをSlackに通知します。

だいたいこんな感じの流れになります。では、コードも交えて各種もう少し詳しく書いていきます！

Zoom会議の作成

Lambdaで以下のようなコードを実行します。リクエストボディに使うJSONをS3から取得、Zoomから得た情報はDynamoに格納みたいなAWSリソースを使った処理はboto3と使って記述します。Zoom APIの実行が成功すれば{"Result" : True}、失敗すれば{"Result" : False}をStep Functionsに返します。Step FunctionはResultをキーに処理の成否を判定するというワケです。

import json
import urllib.request
import os
import boto3
from datetime import datetime as dt

def lambda_handler(event, context):

    return create_meeting(event)

def create_meeting(event):
    url = "https://api.zoom.us/v2/users/"+os.environ['user_id']+"/meetings"

    s3_get = boto3.client('s3')
    bucket_name = 'XXXXXXXXX'
    objkey = 'body.json'
    obj = s3_get.get_object(Bucket=bucket_name, Key=objkey)
    body = obj['Body'].read()
    bodystr = body.decode('utf-8')
    jwt = event['jwt']
    headers = {
    'content-type': "application/json",
    'authorization': "Bearer " + jwt
    }
    print(headers)
    request = urllib.request.Request(
        url, 
        data=bodystr.encode('utf-8'),
        headers = headers,
        method="POST"
    )
    try:
        with urllib.request.urlopen(request) as response:
            response_body = response.read().decode('utf-8')
            res = json.loads(response_body)
            meeting_url = res['join_url']
            meeting_id = res['id']
            tdatetime = dt.now()
            tstr = tdatetime.strftime('%Y/%m/%d')

            dynamodb = boto3.resource('dynamodb')
            table    = dynamodb.Table('meetings')
            table.put_item(
                Item = {
                    "date" : tstr,
                    "meeting_id" : meeting_id,
                    "meeting_url" : meeting_url
                }
            )
            return {"Result" : True}
    except urllib.error.HTTPError as error:
        print(str(error.code) + error.reason)
        return {"Result" : False}
    else:
        return {"Result" : False}

会議情報のSlack通知

こちらもLambdaで以下のコードを実行します。以前作ったSlack通知Lambdaを使いまわしています。

import json
import urllib.request
import os
import boto3
from datetime import datetime as dt

def lambda_handler(event, context):
    result = post_slack()
    if 'ok' in json.dumps(result):
        body = json.dumps(result)
    else:
        body = 'NG!'

    return {
        'statusCode': 200,
        'body': "slack result is " + body
    }

def post_slack():
    dynamodb = boto3.resource('dynamodb')
    table    = dynamodb.Table('meetings')

    tdatetime = dt.now()
    tstr = tdatetime.strftime('%Y/%m/%d')

    items = table.get_item(
        Key={
            "date":tstr
        }
    )
    zoom_url = items['Item']['meeting_url']

    send_data = {
    "username": "Zoom Notification BOT",
    "icon_emoji": ":laughing:",
    "attachments": [
        {
            "color": "#36a64f",
            "pretext": "Daily Zoom Meeting is created!",
            "author_name": "Owner",
            "title": "Zoom Meeting URL is here!",
            "title_link": zoom_url,
            "text": "Let' join us!"
        }
    ]
}

    send_text = "payload=" + json.dumps(send_data)
    headers = {"Content-Type" : "application/json"}
    request = urllib.request.Request(
        os.environ["slack_url"], 
        data=send_text.encode('utf-8'),
        method="POST"
    )
    try:
        with urllib.request.urlopen(request) as response:
            response_body = response.read().decode('utf-8')
    except urllib.error.HTTPError as error:
        print(str(error.code) + error.reason)
    else:
        return response_body    

Slackには↓みたいに通知されます。Zoom Meetin URL is here!をクリックすれば作られたZoomが開きます。

エラー通知のSNS通知

Zoom APIに失敗すれば、Step Functionsはエラーを知らせるメッセージをSNSトピックに通知します。通知を受けたトピックはサブスクライバのLambdaを呼び出します。そのLambdaコードは以下のようになります。SNSからのメッセージ取得なんかはブループリントからピピっと引っ張ってきました😅

from __future__ import print_function

import json
import os
import urllib.request

print('Loading function')


def lambda_handler(event, context):

    message = event['Records'][0]['Sns']['Message']
    print("From SNS: " + message)
    result = post_slack(message)
    if 'ok' in json.dumps(result):
        body = json.dumps(result)
    else:
        body = 'NG!'

    return {
        'statusCode': 200,
        'body': "slack result is " + body
    }

def post_slack(message):

    send_data = {
    "username": "Zoom Notification BOT",
    "icon_emoji": ":cry:",
    "attachments": [
        {
            "color": "#36a64f",
            "pretext": "Failure!!",
            "author_name": "Owner",
            "text": message
        }
    ]
}

    send_text = "payload=" + json.dumps(send_data)
    headers = {"Content-Type" : "application/json"}
    request = urllib.request.Request(
        os.environ["slack_url"], 
        data=send_text.encode('utf-8'),
        method="POST"
    )
    try:
        with urllib.request.urlopen(request) as response:
            response_body = response.read().decode('utf-8')
    except urllib.error.HTTPError as error:
        print(str(error.code) + error.reason)
    else:
        return response_body

Slackには↓みたいに通知されます。単純に失敗したことを伝えているだけですが。。。

AWSリソースについて

基本CLIを使ってリソースは操作しました。

S3

• バケットの作成

aws s3 mb s3://{bucket name}/{object key}

• ローカルのファイル（今回だとJSONファイル）の転送

aws s3 cp {file path} s3://{bucket name}/{object key}

DynamoDB

日次の実行を想定しているので、作成日を意味するdateをキーとしたmeetingsテーブルを作成しています。

• テーブルの作成

aws dynamodb create-table --table-name 'meetings'
--attribute-definitions '[{"AttributeName":"date","AttributeType": "S"}]'
--key-schema '[{"AttributeName":"date","KeyType": "HASH"}]'
--provisioned-throughput '{"ReadCapacityUnits": 5,"WriteCapacityUnits": 5}'

やってみて

AWSをもっと勉強したいなーと思っていたので、ハンズオンがてらこんなことをしてみました。かなりいい練習になったと思ってます。
あと、サーバーレスが現代ビジネスにいかに効果的かというところも改めて考えました。今回の構築は一日程度でできています。これだけのスピード感で保守性の高いアーキテクチャでアプリケーションが作れるのはサーバーレスの強さかなと思います。アジリティの要求が高まる一方のビジネス環境においてこうしたアーキテクチャ設計がより求められるのだろうなと思った次第です。