はじめに
こんにちは、早稲田大学創造理工学部総合機械工学化2年、現在休学中の渡辺です。
この度、インターン先でのタスクとして、
時間管理ツールTogglからレポートを取得してきてslackで共有する作業を自動化して欲しいと頼まれたので、
Serverless FrameworkとAWS Lambdaを使って作ったそんな感じのものをご紹介します。
仕様
- python, AWS Lambda, Serverless Frameworkを使用する
- CloudWatch Eventsで定期的に発火(今回は毎週水曜日朝10時)させる
- togglから特定のプロジェクトの過去一週間の詳細(detail)レポートを取得する
- 取得したレポートをいい感じに整形してslackに投げる
参考
- Toggl Reports API v2 : toggl report apiの仕様書です。
- Serverless Frameworkの使い方まとめ(@horike37) : Serverless Frameworkの使い方を教えてくれます。ありがとうございます。
- Route 53ヘルスチェック結果を元に稼働率を計算してslackにpostする with Serverless Framework(@j-un) : githubにのせたくないapi tokenなどを暗号化してaws上に保持しておく方法を教えてくれます。ありがとうございますじゅんさん。
手順
Serverless Frameworkの使い方は参考記事2を読むとわかります。
ざっくり言うと、Serverless Frameworkをインストールして、handler.pyに実行する処理を書き、serverless.ymlに諸設定を書くだけです。
Node.jsのインストール
Node.jsの公式サイトよりインストールできます。
Serverless Frameworkのインストール
パッケージ管理ツールnpmでインストールする
npm install -g serverless
インストールされているかバージョン確認する
serverless --version
IAMユーザーの設定
serverless config credentials --provider aws --key [Access key ID] --secret [Secret access key]
サービスの作成
今回はpython３で書くのでpython3のテンプレートを選択します。
一つ目のmy-special-serviceにはサービス名を、
二つ目のmy-special-serviceにはパスを指定します。
serverless create --template aws-python3 --name my-special-service --path my-special-service
一旦デプロイ
正常にデプロイできるか確認します。
serverless deploy -v
AWS Lambdaのページに行って、関数が作成されていたら成功です。
-vオプションを付けるとverboseというモードでデプロイが実施され、途中経過がターミナル上で確認できます。
設定ファイルの編集
Serverless Frameworkではserverless.ymlというファイルでいろいろな設定ができます。
serverless.yml
service: toggl-to-slack
frameworkVersion: ">=1.2.0 <2.0.0"
provider:
name: aws
runtime: python3.7
timeout: 300
plugins:
- serverless-python-requirements
functions:
cron:
handler: handler.run
events:
- schedule: cron(0 1 ? * 4 *)
environment:
toggl_api_token: ${ssm:toggl_api_token~true}
toggl_user_agent: ${ssm:toggl_user_agent~true}
toggl_workspace_id: '0000000'
toggl_survey_id: '00000000'
slack_url: ${ssm:slack_url~true}
slack_channel_name: '#チャンネル名'
- service: サービス名
- provider: timeout: タイムアウトするまでの時間(秒)の設定(今回はmaxの300秒を指定)
- plugins: プラグイン(今回はpythonの外部モジュールを管理するプラグインserverless-python-requirementsを設定します)
- functions: cron(名称を任意に指定): handler: handler.run(呼び出す関数名。今回はrun。)
- functions: cron: events: 今回はクロン式を指定して、毎週水曜日朝10時に実行されるようにします。
- functions: cron: environment: 使用する環境変数を設定します。(ssm:...~trueとかっていう表記がありますが、githubとかで公開したくないapi keyなどをawsのkmsとssmパラメータストアで暗号化して使用できるようにしています。使い方は、参考記事3を見るとわかります。)
外部モジュールの設定
Lambdaで外部モジュール(requestsなど)を使用するには、Lambda上に外部モジュールごとアップロードする必要があります。
まず、requirements.txtというファイルを新規作成して、その中に使用したい外部のモジュール名を書きます。
requirements.txt
requests
次に外部モジュールの管理用プラグインをインストールします
npm install --save serverless-python-requirements
実行ファイルの編集
handler.py(クリックするとコード全文が表示されます)
handler.py
# -*- coding: utf-8 -*-
"""
toggl report api より過去一週間の調査projectの情報を取得し、slackに送信します。
toggleではなくtogglなので注意。
"""
import logging
import traceback
import os
import json
import requests
import datetime
import time
from decimal import Decimal, ROUND_HALF_UP
# ログの設定
logger = logging.getLogger(__name__)
logger.setLevel(logging.ERROR)
# 定数
TOGGL_API_TOKEN = os.environ['toggl_api_token']
TOGGL_DETAILS_URL = 'https://toggl.com/reports/api/v2/details'
TOGGL_USER_AGENT = os.environ['toggl_user_agent']
TOGGL_WORKSPACE_ID = os.environ['toggl_workspace_id']
TOGGL_SURVEY_ID = os.environ['toggl_survey_id'] # 調査
SLACK_URL = os.environ['slack_url']
SLACK_CHANNEL_NAME = os.environ['slack_channel_name']
SLACK_USER_NAME = 'toggl'
def run(event, lambda_context):
""" メイン関数
"""
# 調査レポートを取得して、slackに送るtextを作成
title = '【過去一週間の調査】'
try:
# 一回のリクエストでは取得仕切れないので分けて取得する
total_count = 100 # 取得する全てのdataの個数(current_countより大きい必要があるのでとりあえず100にした
current_count = 0 # 現在取得したdataの個数
current_page = 1 # 現在のpage数
responses = [] # 各レスポンスをここに格納する
while current_count < total_count:
# データ取得
options = {'project_ids': TOGGL_SURVEY_ID, 'order_field': 'user', 'page': current_page}
response = get_toggl_reports(TOGGL_API_TOKEN, TOGGL_DETAILS_URL, TOGGL_USER_AGENT, TOGGL_WORKSPACE_ID, options)
# 各レスポンスに対してエラーかどうか判定
if is_error_response(response):
text = make_error_text(title, response)
break
else:
# エラーでなければresponsesにresponseを追加して都度text作成
responses.append(response)
text = make_survey_text(title, responses)
# 変数を更新する
total_count = response['total_count']
current_page += 1
current_count += len(response['data'])
except:
logger.error(traceback.format_exc())
text = make_exception_text(title, traceback.format_exc())
# slackに送信
try:
send_to_slack(SLACK_URL, SLACK_CHANNEL_NAME, SLACK_USER_NAME, text)
except:
logger.error(traceback.format_exc())
def get_toggl_reports(api_token, url, user_agent, workspace_id, options):
""" togglAPIからデータを取得し、json形式でresponseを返す
"""
time.sleep(1) # api仕様書のRate limitingに則り、1秒間待機
today = datetime.date.today()
yesterday = today - datetime.timedelta(days=1)
a_week_ago = today - datetime.timedelta(days=7)
headers = {'content-type': 'application/json'}
params = {
'user_agent': user_agent,
'workspace_id': workspace_id,
'since': a_week_ago,
'until': yesterday,
}
params.update(options)
auth = requests.auth.HTTPBasicAuth(api_token, 'api_token')
response = requests.get(url, auth=auth, headers=headers, params=params)
return json.loads(response.text)
def is_error_response(response):
""" レスポンスがエラーかどうか判定する
"""
return 'error' in response.keys()
def make_error_text(title, response):
""" エラー文を作成する
"""
error = response['error']
text = title + '\n'
text += 'message: ' + error['message'] + '\n'
text += 'tip: ' + error['tip'] + '\n'
text += 'code: ' + str(error['code'])
return text
def make_survey_text(title, responses):
""" responsesを整形して変数textに入れて、返す
"""
# user, description, durをキーとする辞書をreportsリストに格納する
reports = []
for response in responses:
for datum in response['data']:
reports.append({'user': datum['user'], 'description': datum['description'], 'dur': datum['dur']})
# user, descriptionごとにdurを集計
reports = sum_dur(reports)
# 1行目
text = title + '\n'
# 2行目
text += '合計時間: ' + str(milliseconds_to_hours(sum_all_dur(reports))) + 'h\n' # 全合計時間
# 3行目以降
text += '```\n' # 枠始まり
user = ''
for report in reports:
# 各userの始まりをわかりやすくするため(toggl_reportはuserでソートしてある)
if user == '': # 1回目のループ
text += report['user'] + '\n' # 名前
elif not report['user'] == user: # 2回目以降のループ
text += '```\n' # 枠終わり
text += '```\n' # 枠始まり
text += report['user'] + '\n' # 名前
text += '・' + report['description'] + ' ' # 業務タイトル
text += '[' + str(milliseconds_to_hours(report['dur'])) + 'h]\n' # 時間
user = report['user']
text += '```' # 枠終わり
return text
def make_exception_text(title, e):
""" 例外時のtextを作成する
"""
text = title + '\n'
text += '予期せぬエラーです。\n'
text += e
return text
def sum_dur(reports):
""" user, descriptionが同じもののdurを合計する
"""
report_dict = {}
for report in reports:
key = report['user'] + '_' + report['description']
if key in report_dict.keys():
report_dict[key]['dur'] += report['dur']
else:
report_dict[key] = report
result = [report for report in report_dict.values()]
return result
def sum_all_dur(reports):
""" reports内の全てのdurを合計した値を返す
"""
return sum([report['dur'] for report in reports])
def milliseconds_to_hours(milliseconds):
""" ミリ秒を小数第二位で四捨五入された時間に変換する
"""
hours = milliseconds / 1000 / 3600
# 四捨五入する
hours = Decimal(str(hours)).quantize(Decimal('0.1'), rounding=ROUND_HALF_UP)
return hours
def send_to_slack(slack_url, channel_name, username, text):
""" textをslackに送信する
"""
payload = {
'channel': channel_name,
'username': username,
'text': text
}
data = json.dumps(payload)
requests.post(slack_url, data)
return
諸々書いてますが、やってることは、①toggl report apiからトグルのレポートを集計してきて、②テキストとして整形し、③slackに送信することです。
①toggl report apiでレポートを集計して取得する
toggl report apiの公式ドキュメントを参照します。
今回は特定のプロジェクトのdetailレポートを取得したいので、toggl detailed reportも参照します。
handler.py
def get_toggl_reports(api_token, url, user_agent, workspace_id, options):
""" togglAPIからデータを取得し、json形式でresponseを返す
"""
time.sleep(1) # api仕様書のRate limitingに則り、1秒間待機
today = datetime.date.today()
yesterday = today - datetime.timedelta(days=1)
a_week_ago = today - datetime.timedelta(days=7)
headers = {'content-type': 'application/json'}
params = {
'user_agent': user_agent,
'workspace_id': workspace_id,
'since': a_week_ago,
'until': yesterday,
}
params.update(options)
auth = requests.auth.HTTPBasicAuth(api_token, 'api_token')
response = requests.get(url, auth=auth, headers=headers, params=params)
return json.loads(response.text)
- api_token
- user_agent: The name of your application or your email address so we can get in touch in case you're doing something wrong.
- workspace_id: The workspace whose data you want to access.
が必須なので、指定します。
また、今回は、過去一週間、特定のプロジェクト、という条件も必要なため、
- since: ISO 8601 date (YYYY-MM-DD) format. Defaults to today - 6 days.
- until: ISO 8601 date (YYYY-MM-DD) format. Note: Maximum date span (until - since) is one year. Defaults to today, unless since is in future or more than year ago, in this case until is since + 6 days.
- project_ids: A list of project IDs separated by a comma. Use "0" if you want to filter out time entries without a project.
も、パラメータに加えます。
正しく取得できれば、ドキュメントにもあるように、
{
"total_grand":null,
"total_billable":null,
"total_currencies":[{"currency":null,"amount":null}],
"data":[
{
"total_grand":23045000,
"total_billable":23045000,
"total_count":2,
"per_page":50,
"total_currencies":[{"currency":"EUR","amount":128.07}],
"data":[
{
"id":43669578,
"pid":1930589,
"tid":null,
"uid":777,
"description":"tegin tööd",
"start":"2013-05-20T06:55:04",
"end":"2013-05-20T10:55:04",
"updated":"2013-05-20T13:56:04",
"dur":14400000,
"user":"John Swift",
"use_stop":true,
"client":"Avies",
"project":"Toggl Desktop",
"task":null,
"billable":8.00,
"is_billable":true,
"cur":"EUR",
"tags":["paid"]
},{
"id":43669579,
"pid":1930625,
"tid":1334973,
"uid":7776,
"description":"agee",
"start":"2013-05-20T09:37:00",
"end":"2013-05-20T12:01:41",
"updated":"2013-05-20T15:01:41",
"dur":8645000,
"user":"John Swift",
"use_stop":true,
"client":"Apprise",
"project":"Development project",
"task":"Work hard",
"billable":120.07,
"is_billable":true,
"cur":"EUR",
"tags":[]
}
]
}
]
}
のようにレスポンスが帰ってくるはずです。
ちなみに、エラーレスポンスは以下のようになります。
{
"error": {
"message":"We are sorry, this Error should never happen to you",
"tip":"Please contact support@toggl.com with information on your request",
"code":500
}
}
注意点としては、detailed reportは一回のリクエストで取得できるデータが50件までとなります。
50件以上取得したい場合は、pageというパラメータを指定すると取得できます。(詳しくはドキュメント参照)
②整形する
次に、レスポンスをslackに送信できるように整形します。
handler.py
def make_survey_text(title, responses):
""" responsesを整形して変数textに入れて、返す
"""
# user, description, durをキーとする辞書をreportsリストに格納する
reports = []
for response in responses:
for datum in response['data']:
reports.append({'user': datum['user'], 'description': datum['description'], 'dur': datum['dur']})
# user, descriptionごとにdurを集計
reports = sum_dur(reports)
# 1行目
text = title + '\n'
# 2行目
text += '合計時間: ' + str(milliseconds_to_hours(sum_all_dur(reports))) + 'h\n' # 全合計時間
# 3行目以降
text += '```\n' # 枠始まり
user = ''
for report in reports:
# 各userの始まりをわかりやすくするため(toggl_reportはuserでソートしてある)
if user == '': # 1回目のループ
text += report['user'] + '\n' # 名前
elif not report['user'] == user: # 2回目以降のループ
text += '```\n' # 枠終わり
text += '```\n' # 枠始まり
text += report['user'] + '\n' # 名前
text += '・' + report['description'] + ' ' # 業務タイトル
text += '[' + str(milliseconds_to_hours(report['dur'])) + 'h]\n' # 時間
user = report['user']
text += '```' # 枠終わり
return text
ここら辺は、使用用途に合わせて整形します。
③slackに送信する
最後にslackに送信します。
handler.py
def send_to_slack(slack_url, channel_name, username, text):
""" textをslackに送信する
"""
payload = {
'channel': channel_name,
'username': username,
'text': text
}
data = json.dumps(payload)
requests.post(slack_url, data)
return
- slack_url: slackのwebhookurl
- channel: 送信したいチャンネル名
- usernam: 通知の表示名
- text: 送信するメッセージ
を指定し、slackに送信します。
結果
serverless deploy -v
デプロイして、水曜日の午前10時になると、
いい感じに通知がきました。
まとめ
今回は、もともとtogglのレポートを手動でslackに共有していたものを自動化したいということで、Serverless Frameworkを使ってAWS Lambdaで作ってみました。
AWS LambdaってなんぞやってところからServerless Frameworkを使ってgit管理できるようにし、かつaws ssmとkmsを用いてapi tokenなどを暗号化するところまでやりました。たのしい。
最後までお付き合いいただきありがとうございました🙇♂️