背景
普段の業務において、どのコミュニケーションツールを使っているかは人それぞれだと思います。
私の場合、基本的はSlackがメインではあるのですが、そうは言ってもやはり一部においてGoogleのGmailなどに頼らざるを得ない事などがあります。（他社の人たちとのやりとり時などは特にそう）
ただ、頻度としてはそこまで高くはないため、必然的にSlackと比べてチェックの回数が少なくなり、レスポンスが大幅に遅れてしまうなんて事もしばしば...。
そこで、Googleが公開しているGmail APIを使って定期的に未読メッセージを取得し、Slackへ通知する事ができればその問題も解消できるかなと考えました。
完成イメージ
未読メッセージがある場合、その概要（いつ、誰から、どんな内容で、どのメールなのか）を通知してくれます。
仕様
- Python 3.7
- Gmail API
- AWS
- Lambda
- EventBridge
Gmail APIを利用するためのGCP（Google Cloud Platform）やAWSへの登録はすでに済んでいるものとして話を進めるので、あらかじめご了承ください。
事前準備
具体的なコードを書いていく前にいくつか準備しておかなければいけない事があります。
- Gmail APIの有効化およびOAuth同意画面・認証情報の作成
- Slackアプリの作成
先にそちらを片付けてしまいましょう。
Gmail APIの有効化およびOAuth同意画面・認証情報の作成
https://console.cloud.google.com/
まず、GCPにアクセスして任意のプロジェクトを選択。
左上のナビゲーションメニュー（ハンバーガーメニュー）から「APIとサービス」→「ライブラリ」へと進みます。
するとこんな感じでGoogleが公開している様々なAPI一覧が出てくるので、検索フォームの部分に「Gmail」とGmail APIを探してください。
あとは「有効にする」ボタンを押して有効化してあげればOKです。
次にOAuth同意画面の作成を行います。先ほどと同じく左上のナビゲーションメニューから「APIとサービス」→「OAuth同意画面」へと進みましょう。
User Typeは「外部」を選択し、「作成」をクリック。
- アプリ名
- 任意の文字列
- ユーザーサポートメール
- 任意のメールアドレス
- デベロッパーの連絡先情報
- 任意のメールアドレス
- テストユーザー
- 任意のメールアドレス
OAuth同意画面の作成が完了したら、認証情報の作成を行います。例の如く左上のナビゲーションメニューから「APIとサービス」→「認証情報」へと進みましょう。
上部の「認証情報を作成」から「OAuthクライアントID」を選択。
- アプリケーションの種類
- デスクトップアプリ
- 名前
- 任意の文字列
それぞれ入力できたら「作成」をクリックしてください。
作成が済んだら「JSONをダウンロード」をクリックし、JSONファイルを保存しておきましょう。後ほど使います。
Slackアプリの作成
Slackアプリの作成に関しては以下の記事を参考にしてください。
参照:【Slack】無料版でも簡単に作れるSlackアプリで通知Botを作る方法
- スコープ（権限）の付与
- 認証トークン（xoxb-*****）の発行
最終的にこの2つができていればOKです。
実装
それぞれの外部サービスと連携するための事前準備ができたので、いよいよコードを書いていきます。
作業ディレクトリの作成 & 各種ファイルの作成
$ mkdir unread-gmail-watcher && cd unread-gmail-watcher
$ touch main.py gmail_service.py slack_notifer.py .env requirements.txt
あとは先ほどダウンロードしたJSONファイルを「credentials.json」とリネームし、ルートディレクトリに配置してください。
最終的に次のような構成になっていればOKです。
unread-gmail-watcher
├── .env
├── credentials.json
├── gmail_service.py
├── main.py
├── requirements.txt
└── slack_notifer.py
コード
import pickle
import os.path
from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
import dateutil.parser as parser
import base64
import json
CREDENTIALS_PATH = './credentials.json' # 認証情報が記載されたjsonファイルへのパス
SCOPES = ['https://www.googleapis.com/auth/gmail.readonly'] # スコープ（権限）
class GmailService:
def __init__(self):
self.credentials = self.get_credentials()
self.service = build('gmail', 'v1', credentials = self.credentials)
# 認証情報を取得
def get_credentials(self):
credentials = None
# トークンを含んだpickleファイルが存在する場合はそれを使用
if os.path.exists('./token.pickle'):
with open('./token.pickle', 'rb') as token:
credentials = pickle.load(token)
# トークンを含んだpickleファイルが存在しない（有効でない）場合は作成（更新）
if not credentials or not credentials.valid:
if credentials and credentials.expired and credentials.refresh_token:
credentials.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file(CREDENTIALS_PATH, SCOPES)
credentials = flow.run_local_server()
with open('./token.pickle', 'wb') as token:
pickle.dump(credentials, token)
return credentials
# Gmail APIを叩いてメッセージを取得
def get_messages(self, max_results = 5, q = ''):
print('Gmailから未読メッセージを取得中...')
messages = self.service.users().messages()
messages_list = messages.list(
userId = 'me',
labelIds = ['UNREAD'], # ラベル（今回は「未読」を指定）
maxResults = max_results, # 最大取得件数
q = q # クエリ（from、to、after、beforeなど）
).execute()
# メッセージが1件も無かった場合はNoneを返す
if 'messages' not in messages_list: return None
dicts = []
for message in messages_list['messages']:
# idをもとにメッセージの詳細を取得
message_detail = messages.get(userId = 'me', id = message['id']).execute()
dict = {}
dict['id'] = message['id']
# ヘッダー部分（日付、差出人、宛先、件名）
for header in message_detail['payload']['headers']:
if header['name'] == 'Date':
dict['date'] = parser.parse(header['value']).strftime('%Y/%m/%d %H:%M')
elif header['name'] == 'From':
dict['from'] = header['value']
elif header['name'] == 'To':
dict['to'] = header['value']
elif header['name'] == 'Subject':
if header['value'] != '':
dict['subject'] = header['value']
else:
dict['subject'] = '件名なし'
# 本文の概要
if message_detail['snippet'] != '':
dict['snippet'] = message_detail['snippet']
else:
dict['snippet'] = '本文なし'
dicts.append(dict)
return json.dumps(dicts, ensure_ascii = False)
import os
from slack_sdk import WebClient
from dotenv import load_dotenv
load_dotenv()
SLACK_BOT_TOKEN = os.getenv('SLACK_BOT_TOKEN') # トークン
SLACK_CHANNEL = os.getenv('SLACK_CHANNEL') # 送信先のチャンネル
class SlackNotifer:
def __init__(self):
self.slack_bot_token = SLACK_BOT_TOKEN
self.slack_channel = SLACK_CHANNEL
self.client = WebClient(token = self.slack_bot_token)
def send(self, text):
self.client.chat_postMessage(
channel = self.slack_channel,
text = text,
as_user = True
)
from gmail_service import GmailService
from slack_notifer import SlackNotifer
import json
import textwrap
def run():
messages = GmailService().get_messages()
# 未読メッセージが無い場合は終了
if messages is None:
print('未読のメッセージはありません。')
return
# 未読メッセージがある場合はSlack通知を送信
for message in json.loads(messages):
heredoc = textwrap.dedent('''
```
☆ 未読のメッセージがあります。☆
日付: {date}
差出人: {_from}
件名: {subject}
本文: {snippet}
続きはこちら → https://mail.google.com/mail/u/0/#label/unread/{id}
```
''').format(
date = message['date'],
_from = message['from'],
subject = message['subject'],
snippet = message['snippet'],
id = message['id']
)
print('Slack通知を送信中...')
SlackNotifer().send(heredoc)
if __name__ == '__main__':
run()
SLACK_BOT_TOKEN=xoxb-****************
SLACK_CHANNEL=#***********
google-api-python-client
google-auth-httplib2
google-auth-oauthlib
python-dateutil
python-dotenv
slack-sdk
動作確認
実際に動くかどうか確認しましょう。
$ pip install -r requirements.txt
$ python main.py
Please visit this URL to authorize this application: https://accounts.google.com/o/oauth2/auth?response_type=code&client_id=***************.apps.googleusercontent.com~
初回実行時のみ、ブラウザが立ち上がりGoogleアカウントへのリクエスト許可が求められるので、画面の指示に従い許可してください。
成功するとブラウザ上に「The authentication flow has completed. You may close this window.」という文字が表示されるとともに、Slackに通知が届いているはず。
※事前に未読メールを最低一つは準備しておきましょう。
デプロイ
開発環境での動作確認は済んだので、次は本番環境でも動くようにデプロイしていきます。先述のように、今回はAWS LambdaとEventBridgeを利用します。
Lambda関数の作成
$ touch lambda_function.py
from gmail_service import GmailService
from slack_notifer import SlackNotifer
import json
import textwrap
def lambda_handler(event, context):
messages = GmailService().get_messages()
# 未読メッセージが無い場合は終了
if messages is None:
print('未読のメッセージはありません。')
return
# 未読メッセージがある場合はSlack通知を送信
for message in json.loads(messages):
heredoc = textwrap.dedent('''
```
☆ 未読のメッセージがあります。☆
日付: {date}
差出人: {_from}
件名: {subject}
本文: {snippet}
続きはこちら → https://mail.google.com/mail/u/0/#label/unread/{id}
```
''').format(
date = message['date'],
_from = message['from'],
subject = message['subject'],
snippet = message['snippet'],
id = message['id']
)
print('Slack通知を送信中...')
SlackNotifer().send(heredoc)
基本的には「main.py」の内容と同じですが、引数に
- event
- context
を渡していたりするのが相違点ですね。
$ zip -r codes.zip lambda_function.py gmail_service.py slack_notifer.py credentials.json token.pickle
次に実行に必要な各種コードをzip形式でパッケージングします。上手くいくとルートディレクトリに「codes.zip」が吐き出されているはず。
あとはAWSのコンソール画面から「Lambda」→「関数」→「関数の作成」へと進み、
- 関数名
- 任意
- ランタイム
- Python 3.7
とそれぞれ入力し、関数を作成してください。
作成できたら、右にある「アップロード元」から「zipファイル」を選択し、先ほどルートディレクトリに吐き出されていたzipファイルをアップロードしましょう。
こんな感じで無事コードがアップロードできていればOKです。
Lambdaレイヤーの作成
参照記事: AWS Lambda Layersでライブラリを共通化
注意点として、今のままだと先ほどアップロードしたコードは動きません。なぜなら、デフォルトのLambda実行環境には外部ライブラリ（今回で言えば「google-api-python-client」や「slack-sdk」など）が参照できないからです。
外部ライブラリを使用するためには、あらかじめ自前で準備したものを一緒にアップロードしたりする必要があります。
そこで役に立つのがLambdaレイヤーというわけですね。
一度作ってAWSにアップロードしておけば後ほど他のLambda関数で同じような外部ライブラリが必要になった時に使い回せたりしますし、せっかくなので作っておきましょう。
主な手順については以下の記事を参考にします。
参照記事: DockerでAWS LambdaのPython用Layerを作成する
$ touch Dockerfile entrypoint.sh
FROM amazonlinux:2
ARG PYTHON_VERSION=3.7.0
RUN yum update -y && yum install -y tar gzip make gcc openssl-devel bzip2-devel libffi-devel \
&& curl https://www.python.org/ftp/python/${PYTHON_VERSION}/Python-${PYTHON_VERSION}.tgz | tar xz \
&& cd Python-${PYTHON_VERSION} && ./configure && make && make install \
&& cd - && rm -rf Python-${PYTHON_VERSION}
ADD entrypoint.sh /
RUN yum install -y zip \
&& mkdir /python \
&& chmod +x /entrypoint.sh
ENTRYPOINT ["/entrypoint.sh"]
#!/bin/bash -eu
SRC=/python
DIST=/dist/layer.zip
pip3 install -t ${SRC} $@
rm -f ${DIST}
zip -q -r ${DIST} ${SRC}
上記2ファイルの準備ができたら、次のコマンドを実行してください。
$ docker build . -t bundle-pip-modules-for-aws-lambda-layers:python3.7
$ docker run --rm \
-v $(pwd)/requirements.txt:/requirements.txt \
-v $(pwd):/dist \
bundle-pip-modules-for-aws-lambda-layers:python3.7 \
-r requirements.txt
すると、ルートディレクトリに「layer.zip」が吐き出されていると思います。
あとはAWSコンソール画面から「Lambda」→「レイヤー」→「レイヤーの作成」と進み、
- 名前
- 任意の文字列
- 説明
- 任意の文字列
- ランタイム
- Python 3.7
それぞれの項目を入力したら「作成」をクリックしてください。
作成が済んだら、「レイヤーの追加」をクリックし、
先ほど作ったものを追加しましょう。これでLambda関数から外部ライブラリの参照ができるようになります。
その他設定
あと一息です。その他の細かい設定（メモリ・タイムアウトや環境変数）を片付けます。
- メモリ
- 256MB
- タイムアウト
- 1分0秒
この辺は各々の環境（どれくらいの件数のメールをチェックするのかなど）によっても違うと思うので、お好みの数値を入れてください。ちょっとした件数であれば↑の設定で十分だと思います。
- SLACK_BOT_TOKEN
- SLACK_CHANNEL
それぞれ自分のものを入力してください。
動作確認
さて、諸々の準備が終わったので、正常に動作するかの確認に入ります。
「Test」というオレンジ色のボタンをクリックしてください。
するとこんな感じの設定画面が出てくるので、スクショを参考に設定を行いましょう。
設定を済ませた後、もう一度「Test」ボタンを押すと処理が開始されるので、ちゃんと成功するかどうか確かめてください。
特に問題無ければ上記のようなログが表示され、Slackに通知が飛んでいるはず。
定期実行の設定
最後の仕上げとして、EventBrideで定期実行（〇〇分に1回など）できるようにしておきましょう。
「トリガー」を追加をクリックし、
- トリガー
- EventBridge
- ルール
- 新規ルールの作成
- ルール名
- 任意の文字列
- ルールの説明
- 任意の文字列
- ルールタイプ
- スケジュール式
- スケジュール式
- cron(0 * * * ? *)
こんな感じで設定してください。
今回は1時間に1回のルールにしましたが、cronの書き方については世の中に腐るほど記事があるので各自お好みにしていただければと思います。
これで定期実行が可能になりました。
もし時間になっても動いていなさそうだったら、CloudWatchにログが残っているはずなので、そちらから適宜デバッグしましょう。
番外編（AWS CDKを使う場合）
先ほど作成したAWSの各種リソースをコードで管理（IaC）するとしたら、こんな感じになります。（今回はAWS CDKの使用を想定）
from aws_cdk import aws_lambda, aws_ssm, aws_events, aws_events_targets, core
class UnreadGmailWatcherStack(core.Stack):
def __init__(self, scope: core.Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# 認証トークン（事前にパラメータストアに値を格納しておく事）
slack_bot_token = aws_ssm.StringParameter.from_string_parameter_attributes(self, 'slack-bot-token',
parameter_name = '/unread-gmail-watcher/SLACK_BOT_TOKEN'
).string_value
# 送信先のチャンネル名（事前にパラメータストアに値を格納しておく事）
slack_channel = aws_ssm.StringParameter.from_string_parameter_attributes(self, 'slack-channel',
parameter_name = '/unread-gmail-watcher/SLACK_CHANNEL'
).string_value
# Lambdaレイヤー
lambda_layer = aws_lambda.LayerVersion(
self,
'lambda-layer',
compatible_runtimes = [aws_lambda.Runtime.PYTHON_3_7],
code = aws_lambda.AssetCode('lambda/layers')
)
# 各種設定値
params = dict(
runtime = aws_lambda.Runtime.PYTHON_3_7,
handler = 'lambda_function.lambda_handler',
timeout = core.Duration.seconds(60),
memory_size = 256,
)
# Lambda関数
lambda_function = aws_lambda.Function(
self,
'lambda-function',
code = aws_lambda.Code.asset('lambda/functions'),
layers = [lambda_layer],
environment = {
'SLACK_BOT_TOKEN': slack_bot_token,
'SLACK_CHANNEL': slack_channel
},
**params
)
# EventBrdge
event_bridge = aws_events.Rule(
self,
'event-bridge',
schedule = aws_events.Schedule.cron(
minute = '0',
hour = '*',
month = '*',
year = '*'
)
)
event_bridge.add_target(aws_events_targets.LambdaFunction(lambda_function))
番外編と言っているように本筋とはズレるため、詳細についてはここでは割愛しますが、興味のある方はコードを追ってみてください。
あとがき
以上、Pythonを使ってGmailの未読メッセージをSlackに通知する仕組みを作ってみました。
少しでも業務効率化の参考になれば幸いです。