こんにちは、請求管理ロボシステム インターン生の渡辺です。
僕が現在所属するチームでは、togglで作業時間の管理を行なっており、 毎週のミーティングではその集計情報をコピペしてslackで共有しています。
ですが、コピペがめんどくさいということで、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などを暗号化するところまでやりました。たのしい。
最後までお付き合いいただきありがとうございました🙇♂️
