More than 3 years have passed since last update.

togglからslackにレポートを通知するwith Serverless Framework

Last updated at 2019-11-13Posted at 2019-11-13

はじめに

こんにちは、早稲田大学創造理工学部総合機械工学化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などを暗号化するところまでやりました。たのしい。
最後までお付き合いいただきありがとうございました🙇‍♂️

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up