GCP Cloud Functions + Python + Slack Outgoing WebHooks App + Zaim APIで、SlackからZaimへ登録する

最近、 Zaim を使って家計簿をつけています。
https://zaim.net/

ただ、時々入力を忘れたり、重複入力してしまうことがありました。

そこで、

  • 日頃Slackを使っている
  • ZaimにはWeb APIがある

ということから、SlackからZaimのデータを登録する仕組みを作りましたので、メモを残します。

なお、今回の範囲では影響ありませんでしたが、Zaimの金融連携データはAPI経由では取得できないようです。プレミアムプラン契約をしている時は取得できるようになってくれるとありがたいです。

 
目次

 

環境

 

作ったもの

長いメモなので、こんな感じのものを作ったというのを書いておきます。

 

SlackからZaimへ登録する

決まったフォーマット (日付(yyyy/mm/dd or mm/dd) ジャンル名 金額 コメント) をSlackにポストすると、Zaimに反映します。

なお、登録する時に西暦を入力するのが手間だったため、西暦が省略された場合は、実行日の西暦を渡すようにしています。

Slackの様子

Zaimに登録できたら、OKなリアクションをします。

f:id:thinkAmi:20181228215056p:plain:w200

Zaimの様子

Zaimにも登録できています。

f:id:thinkAmi:20181228215250p:plain:w300

 
一方、Zaimに登録できない場合は、NGなリアクションと、NGになった理由をスレッドで返信します。

f:id:thinkAmi:20181228215652p:plain:w250

 
他にも、使い方を忘れたときのために、以下の2機能を作りました。

 

Zaimの登録可能なジャンルを知る

ジャンル とポストすると、Zaimで登録可能なジャンルの一覧をスレッドで返信します。

f:id:thinkAmi:20181228215826p:plain:w250

 

Zaimへ登録する際のポストの書式を知る

書式 もしくは フォーマット とポストすると、登録するポストの書式をスレッドで返信します。

f:id:thinkAmi:20181228220003p:plain:w250

 
以降は、これらを作った時のメモになります。

 

事前調査

SlackとZaimをつなぐ方法

SlackとZaimにはAPIがあるので、それらを使えばつなぐことができそうでした。ただ、メンテナンスの手間をかけたくないため、サーバレスで作ろうと考えました。

一般的な構成を調べると、AWS Lambda + API Gateway が多かったです。せっかくなので、今回は別の構成で作ることにしました。

 
GCPで同じ機能がないかを調べたところ、 Cloud Functions がありました。Betaながら、Python3.7.1もサポートしていました。
Google Cloud Functions documentation  |  Cloud Functions  |  Google Cloud

また、 HTTP Functions として作ることで、HTTP(S)のアクセスを直接受け付けるようでした。
HTTP Functions  |  Cloud Functions Documentation  |  Google Cloud

価格表を見ても無料枠があり、ある程度の運用ができそうでした。
Pricing  |  Cloud Functions Documentation  |  Google Cloud

 
そのため、GCP Cloud Functions を使った構成で進めることにしました。

 

Slackのポストをフックする手段

Slackのポストをフックする手段としては以下がありました。

他にも Slash Commands があったものの、Slack上で会話的にやりとりしたかったため、今回は使わないことにしました。
https://api.slack.com/slash-commands

 
まず、 Legacy Outgoing Webhooks を調べたところ、Web上にいろいろな知見がありました。

ただ、Slackのドキュメントには

You're reading this because you're looking for info on legacy custom integrations - an outdated way for teams to integrate with Slack. These integrations lack newer features and they will be deprecated and possibly removed in the future. We do not recommend their use.

https://api.slack.com/custom-integrations/outgoing-webhooks

と書かれていたため、今回は使うのをやめました。

 
次に Events API を見てみました。

その中の message.channels イベントを使えば、Slackへのポストをフックできそうでした。
https://api.slack.com/events/message.channels

しかし、実際に試してみると、全チャンネルの全ポストをフックし、Cloud Functionsにリクエストが来ていました。

Cloud Functionsに無料枠があるとはいえ、全部フックされるのはつらいので、今回は使うのをやめました。

 
最後に Outgoing WebHooks App を見てみました。

調べてみたところ、チャンネル限定でポストをフックできそうでした。

そのため、今回は Outgoing WebHooks App を使ってフックすることにしました。

 

Zaim APIについて

ZaimのAPIドキュメントを見たところ、APIでデータを登録できそうでした *1

また、APIのアクセスレベル(読込/書込/両方)も制御できました。

 

Zaim APIの category と genre について

Zaim APIでは

  • category
  • genre

の2つを指定し、APIを呼ぶ必要がありました。

 
それらが何を指すのか調べたところ、

のようでした。

 
ただ、画面上では両方の名称は確認できるものの、Zaim APIに渡すためのIDが不明でした。

 
Zaim APIの公式ドキュメントを見たところ、 GET /v2/genre を使うことで、categoryとgenreの両IDが取得できそうでした。

 

存在しないジャンルを用いた Zaim APIでの登録について

存在しないジャンルIDを用意して、Zaim APIにて登録してみたところ、エラーになることなく登録できました。

 
そのため、Zaim APIで登録する前に、ポストされたジャンルは正しいかという検証が必要そうでした。

 
以上で、事前調査が終わりました。続いて、実際のアプリケーションを作っていきます。

 

Zaimの設定

新しいアプリケーションの追加

Zaim Developers Centerへアクセスし、新しいアプリケーションを追加します。
Zaim Developers Center

 

アクセストークン系の取得

Zaimにアプリケーションを追加しただけではアクセストークン系が取得できません。

そのため、以下の記事を参考に、アクセストークンを取得するPythonスクリプトを作成します。
requestsを使ったOAuth認証 例題:Flickr - Qiita

なお、今回はターミナルで動かすスクリプトなため、 OAuth1.0aの oauth_callback は、RFC5849に従い oob (out-of-band:帯域外) を指定しました。
https://tools.ietf.org/html/rfc5849#section-2.1

class ZaimClient:
    def __init__(self):
        # 後述しますが、secret.json ファイルに、各種認証情報を設定してある前提
        with pathlib.Path(__file__).parents[1].joinpath('secret.json').open(mode='r') as f:
            secrets = json.load(f)
        self.tokens = secrets['Zaim']

    def print_access_token(self):
        """ Zaimのアクセストークンを取得・表示する

        OAuth1.0aの認証方法については、以下の記事を参考に実装
        https://qiita.com/kosystem/items/7728e57c70fa2fbfe47c
        """
        request_token = self._get_request_token()
        access_token = self._get_access_token(request_token)

        # ターミナル上にアクセストークンを表示する。形式は以下の通り
        # {'oauth_token': 'xxx', 'oauth_token_secret': 'yyy'}
        # oauth_token == ACCESS_TOKEN, oauth_token_secret == ACCESS_TOKEN_SECRET
        print(access_token)

    def _get_request_token(self):
        auth = OAuth1(
            self.tokens['CONSUMER_KEY'],
            self.tokens['CONSUMER_SECRET'],
            # CLIからの認証なので、RFC5849のsection-2.1より、 `oob` を指定しておく
            # https://tools.ietf.org/html/rfc5849#section-2.1
            callback_uri='oob')

        r = requests.post(self.tokens['REQUEST_TOKEN_URL'], auth=auth)
        return dict(urllib.parse.parse_qsl(r.text))

    def _get_access_token(self, request_token):
        # ブラウザを起動してOAuth認証確認画面を表示する
        # ユーザーが許可すると、「認証が完了」のメッセージとともにコードが表示される
        webbrowser.open(
            f'{self.tokens["AUTHORIZE_URL"]}?oauth_token={request_token["oauth_token"]}&perms=delete')

        # ターミナル上で、コードの入力を待つ(コード入力後、後続処理が行われる)
        oauth_verifier = input('コードを入力してください: ')

        auth = OAuth1(
            self.tokens['CONSUMER_KEY'],
            self.tokens['CONSUMER_SECRET'],
            request_token['oauth_token'],
            request_token['oauth_token_secret'],
            verifier=oauth_verifier)
        r = requests.post(self.tokens['ACCESS_TOKEN_URL'], auth=auth)

        access_token = dict(urllib.parse.parse_qsl(r.text))
        return access_token

 
認証情報の入った secrets.json はこんな感じです。各項目はZaimにアプリケーション登録した時に表示される値となります。

{
  "Zaim":
  {
    "REQUEST_TOKEN_URL": "https://your_url",
    "AUTHORIZE_URL": "https://your_url",
    "ACCESS_TOKEN_URL": "https://your_url",
    "CONSUMER_KEY": "xxx",
    "CONSUMER_SECRET": "xxx",
    "ACCESS_TOKEN": "",
    "ACCESS_TOKEN_SECRET": ""
  }
}

 
このスクリプトを実行し、

  • ブラウザが起動するのでログイン
  • 画面にトークンが表示
  • ターミナルにトークンを入力

とすると、ターミナルに以下の形式のアクセストークン系が表示されます。

{'oauth_token': 'xxx', 'oauth_token_secret': 'yyy'}

 
oauth_token (= ACCESS_TOKEN)、oauth_token_secret (= ACCESS_TOKEN_SECRET) を、 secrets.json に反映します。

 
ここまででZaimの設定は完了です。

 

Slackの設定

Slack Appの作成

以下のページより、今回のSlackアプリ (Slackのポストをフックして Cloud Functions を呼び出すアプリ) を作成します。
Slack API: Applications | Slack

 

OAuth Access Token の取得

Slack Appからポストできるようにするため、サイドバーの OAuth & Permissions から OAuth Access Token を取得しておきます。

この値を、前述の secrets.json にも追記しておきます。

 

Scopeの設定

今回、SlackからZaimへ登録した時の結果として、

  • 成功時:Slackポストに絵文字のリアクションを付ける
  • 失敗時:Slackポストに絵文字のリアクションを付けるとともに、ポストのスレッドにエラーメッセージを記載

を行いたいです。

 
そのため、Slack Appに以下の3つの権限を付与しました。

  • channels:history (Access user’s public channels)
  • chat:write:bot (Send messages as )
  • reactions:write (Add or remove emoji reactions for user)

 

Outgoing WebHooks Appの追加

Outgoing WebHooks Appのページから、対象のSlackスペースへアプリをインストールします。
Outgoing WebHooks | Slack App Directory

 

インテグレーションの設定

Outgoing WebHooks Appの設定を追加します。

今回、Slackのポストをフックして

  • Zaimへのデータ登録
  • Zaimへデータ登録する時のジャンルを表示
  • Zaimへデータ登録する時のフォーマットを表示

を実現したいです。

今回は Slash Commands として作成しないことから、

  • 1つの Cloud Functions 関数として作成
  • ポスト内容により、どの機能を動作させるのか振り分ける

ことにしました。

 
そのため、インテグレーションの設定は以下となりました。

項目
チャンネル Zaimへポストするためのチャンネルを専用で用意・指定
引き金となる言葉 空白 (全ての言葉に反応させるため)
URL 現時点では空白 (Cloud Functions側で設定したら、ここも設定)

 
以上がSlackの設定となります。

 

GCP Cloud Functions の設定

Googleアカウントまわりの設定

Cloud Functions を使うためにはクレジットカード情報などを登録する必要があります。

そのため、不正アクセスなどがされにくい & された時に気づきやすい点を考慮し、以下の設定を行いました。

  • 請求先アカウントのチュートリアルから作業する
  • 新しくGmailアカウントを作成
  • 2段階認証を設定
  • GCPでプロジェクトの作成
    • 任意の名前をつける
  • Cloud Functions APIを有効化
  • Cloud Functions APIのページへ移動
  • 無料トライアルに登録
    • 情報入力
      • アカウントの種類:個人
      • 名前と住所:入力
      • お支払い方法:カード情報の入力
  • 予算とアラートを追加
  • Cloud Functions に新しい関数を作成
    • リージョン: asia-northeast1
    • secret.json へ設定した内容を環境変数としても設定
    • トリガータブにエンドポイントURLがあり、これが Slack インテグレーション設定のURLとなる
      • コピペして、Slack の Outgoing WebHooks App の設定へと反映する

 

Cloud Functions の実装で悩んだところ

続いて Cloud Functions の関数を作成しますが、いくつか悩んだところがありました。  

名前について

むやみに外部からアクセスされても困るため、分かりづらい名前を付けることにしました。

そこで、Pythonuuid モジュールを使って、ランダムなuuidを生成し、名前として使うことにしました。
21.20. uuid — UUID objects according to RFC 4122 — Python 3.6.5 ドキュメント

# PythonのREPLを起動
$ python

# uuidモジュールを使って名前を出力
>>> import uuid
>>> print(str(uuid.uuid4()))

 

Slackへのレスポンス3秒ルールへの対応

Outgoing WebHooks Appでは関係ないかもしれませんが、ドキュメントに記載されていないので、念のための考慮となります。

SlackのEvent APIなどでは、Slackからの通知に対して3秒以内にレスポンスする必要があります。

Your app should respond to the event request with an HTTP 2xx within three seconds. If it does not, we'll consider the event delivery attempt failed. After a failure, we'll retry three times, backing off exponentially.

Maintain a response success rate of at least 5% of events per 60 minutes to prevent automatic disabling.

Respond to events with a HTTP 200 OK as soon as you can. Avoid actually processing and reacting to events within the same process. Implement a queue to handle inbound events after they are received.

What you do with events depends on what your application or service does.

https://api.slack.com/events-api#responding_to_events

 
しかし、今回は

  • Outgoing WebHooks Appから通知を受け取る
  • Zaim APIを使って、Zaimへ登録
  • Slack Web APIを使って、Slackへ返信

を行うため、3秒を超過する可能性があります。

 
他のSlackアプリはどうしているのか調べたところ、

  • Slackへは、すぐにHTTP200の返信を行う
  • 実際の処理は、遅延実行する

をしていました。

ただ、ほとんどの記事がAWSで実装されたものであり、Cloud Functionsのものは見当たりませんでした。

 
Cloud Functionsでできる方法を調べたところ、Pythonthreading.Thread が使えそうでした。

 
上記のstackoverflowの回答はHeroku上のものでしたが、Cloud Functionsでも問題なく動作しました。

# background()関数を用意し、そちらで Zaim APIを呼ぶなどの処理を実装する
t = Thread(target=background, kwargs={'request_data': request_data})
t.start()
return ''

 
ただ、本当に使っても大丈夫なのかは公式ドキュメントには見当たらなかったため、自己責任で...

 

Zaimのジャンル情報の保持

Zaim APIでデータを登録する際、ジャンル情報を渡す必要があります。

ただ、渡すジャンル情報はジャンル名ではなく、カテゴリIDとジャンルIDを設定する必要があります。

 
単純に考えると、Zaim APIで登録する前に、Zaim APIでジャンル情報を取得すれば良さそうでした。

ただ、1件登録するのに2回APIを呼ぶのはあまり良くない気がしました。

また、自分の運用を考えたところ、一度カテゴリやジャンルを決めてしまったら、その後は大きく変更していません。

 
そのため、都度 Zaim APIでジャンル情報を取得するのではなく、GCP側でジャンル情報を保持しておくことにしました。

もしジャンル情報を変更した場合は、保持している内容を書き換えるという運用としました。

 
次に保持する場所を考えました。ただ、これだけの目的でDBを使いたくなかったため、環境変数に設定できないかを考えました。

公式ドキュメントで環境変数のサイズを調べてみると、

Size limits

The total number of bytes used by environment variable names and values for an individual function is limited to 32KiB. However, there are no specific limits on individual keys or values within this overall capacity.

https://cloud.google.com/functions/docs/env-var#size_limits

とありました。

自分のジャンルの量からすると、トークンなどを考慮しても、32KiBで収まりそうでした。

 
以上より、環境変数JSON文字列としてジャンル情報をセットしておき、Zaim APIを呼ぶ時にジャンル名からIDへと変換するようにしました。

環境変数に設定する形式は

{"食料品": {"category_id": nnn, "genre_id": nnn}, "カフェ": {"category_id": nnn, "genre_id": nnn}, ... }

な感じとしました。

 

ロギング方法

公式ドキュメントに従い、 logging モジュールを使いました。
Writing, Viewing, and Responding to Logs  |  Cloud Functions Documentation  |  Google Cloud

なお、 logging.debug は動作しませんでした。見た感じだと無視されるようです。

 

Outgoing Webhooks Appからのアクセス判定

Slack Events API や Legacy Outgoing Webhooksでは、 X-Slack-Signature HTTPヘッダを検証することで、Slackからのリクエストかどうかが分かります。
Verifying requests from Slack | Slack

 
ただ、 Types of requests that support signed secrets を見ても、Outgoing Webhooks Appが記載されていません。 https://api.slack.com/docs/verifying-requests-from-slack#types_of_events

 
Outgoing Webhooks Appの設定を見ると、トークンという欄に

このトークンは発信ペイロードに送信されます。Slack チームから来たリクエストを確認する際にそのトークンを使用できます。

と記載されていました。

 
Cloud Functions上でSlackからのリクエストボディを確認すると

# print(request.form)の結果

ImmutableMultiDict([('token', 'xxx'), ('team_id', 'xxx'), ('team_domain', 'xxx'), ...])

でした。

 
そのため、環境変数にOutgoing Webhooks Appのトークンを設定しておくことで、

# 毎回 reqest.formと書くのが手間なので、変数に入れておく (以降の例も同様)
request_data = request.form

if request_data.get('token') != os.environ['SLACK_OUTGOING_WEBHOOKS_TOKEN']:
    # Slack以外のリクエストの処理
    ...

な形で判定できそうでした。

 
なお、Outgoing WebHooks AppからはPOSTしかされない前提のため、今回は request.form を使っています。

クエリストリングとリクエストボディの両方を取得したい場合は、 request.values になります。
http://werkzeug.pocoo.org/docs/0.14/wrappers/#werkzeug.wrappers.BaseRequest.values

 

Slack Web APIからの投稿かどうかの判定

今回、登録エラーとなった場合は、Slackのスレッドでエラーメッセージをポストします。

しかし、何も制御しないと、

  • Slack Web APIを使って、エラーメッセージを投稿
  • エラーメッセージの投稿がOutgoing WebHooks Appでフックされ、Cloud Functionsに送信
  • 同じくエラーが出るため、エラーメッセージを投稿
  • (以降繰り返し)

と、無限ループする可能性があります。

 
Events APIを使ってフックする場合は、 sub_type を見ることで Slack Web APIでの投稿かどうかが分かりそうです。

 
しかし、Outgoing WebHooks Appを使った場合、渡されてくるのは

  • token
  • team_id
  • team_domain
  • channel_id
  • channel_name
  • timestamp
  • user_id
  • user_name
  • text
  • trigger_word

だけでした。

 
そこで、自作のSlack App にはBotを作成しない状態で Slack Web APIchat.postMessage を使ってみたところ、 user_name に slackbot という値が渡されてきました。

 
そのため、Slack Web APIの投稿かどうかは、

if request_data.get('user_name') == 'slackbot':
    logging.info(f'bot access data:{request_data.get("text")}')
    return ''

と判定することにしました。

 

空データのアクセス判定

以上を実装した後にログを眺めていたところ、Outgoing WebHooks Appでは、本来のアプリからのアクセスの他に、1,2回アクセスが発生していることが分かりました。

また、この場合、リクエストデータが空っぽでした。

 
そのため、リクエストデータが空っぽの場合は、不明なアクセスと考えて、処理しないようにしました。

if request_data.get('token') != os.environ['SLACK_OUTGOING_WEBHOOKS_TOKEN']:
    logging.warning(f'not slack access, data: {request_data}')
    return ''

 

Cloud Functionsに登録した関数の内容

今までの内容をもとに作成した関数は以下の通りです。

""" GCP Cloud Functions を使って、SlackからZaimへデータをPostするためのスクリプト """

import json
import logging
import os
import unicodedata
from datetime import datetime
from threading import Thread

import zaim
from slackclient import SlackClient


def background(request_data):
    """ Cloud Functionsの別スレッドで動作する関数 """

    # 登録可能なジャンルを知りたい場合
    has_genre_response = response_all_genre(request_data)
    if has_genre_response:
        return

    # Zaimへ登録するためのフォーマットを知りたい場合
    has_format_response = response_format(request_data)
    if has_format_response:
        return

    error_msg, zaim_data = create_zaim_data(request_data)

    if zaim_data:
        error_msg = post_zaim(zaim_data)

    client = SlackClient(os.environ['SLACK_TOKEN'])
    if error_msg:
        # 念のため、ログにもエラーメッセージを出力しておく
        logging.debug(error_msg)

        # エラーの場合、NGリアクションとスレッドにエラーメッセージをポスト
        client.api_call(
            'reactions.add',
            name='man-gesturing-no',
            channel=request_data['channel_id'],
            timestamp=request_data['timestamp'],
        )
        client.api_call(
            'chat.postMessage',
            channel=request_data['channel_id'],
            thread_ts=request_data['timestamp'],
            text=error_msg,
        )
    else:
        client.api_call(
            'reactions.add',
            name='man-gesturing-ok',
            channel=request_data['channel_id'],
            timestamp=request_data['timestamp'],
        )


def response_all_genre(request_data):
    """ ジャンルを知りたい場合は、環境変数にあるジャンル一覧をスレッドとして返信する """
    text = request_data.get('text')
    if text != 'ジャンル':
        return False

    genre = load_genre()
    all_genre = ', '.join(genre.keys())

    client = SlackClient(os.environ['SLACK_TOKEN'])

    client.api_call(
        'reactions.add',
        name='book',
        channel=request_data['channel_id'],
        timestamp=request_data['timestamp'],
    )

    client.api_call(
        'chat.postMessage',
        channel=request_data['channel_id'],
        thread_ts=request_data['timestamp'],
        text=all_genre,
    )
    return True


def response_format(request_data):
    """ Zaimへ投稿するフォーマットを知りたい場合は、環境変数にあるジャンル一覧をスレッドとして返信する """
    text = request_data.get('text')
    if text not in ('書式', 'フォーマット'):
        return False

    text = '日付(yyyy/mm/dd or mm/dd) ジャンル名 金額 コメント ' \
           '(4項目は順不同、区切りはスペース(全角/半角どちらでも可))'

    client = SlackClient(os.environ['SLACK_TOKEN'])

    client.api_call(
        'reactions.add',
        name='memo',
        channel=request_data['channel_id'],
        timestamp=request_data['timestamp'],
    )

    client.api_call(
        'chat.postMessage',
        channel=request_data['channel_id'],
        thread_ts=request_data['timestamp'],
        text=text,
    )
    return True


def create_zaim_data(request_data):
    """ Zaimデータを作成する

    :param request_data: リクエストされたデータ
    :return: エラーメッセージ, Zaimデータ
    :rtype: str, dict
    """
    text = request_data.get('text')
    if not text:
        return '登録データがありません', None

    zaim_data = parse_zaim_data(text)
    if len(zaim_data.keys()) != 5:
        return f'登録するための項目が不足しています :{zaim_data}', None

    return None, zaim_data


def parse_zaim_data(text):
    """ SlackのポストをZaimデータにparseする

    :param text: Slackのポスト
    :return: Zaimデータ
    :rtype: dict
    """

    # Slackポストのフォーマット
    # 項目:日付、ジャンル名、金額、コメント (順不同、区切りはスペース(全角/半角どちらでも可))
    # 各項目や区切り文字は全角/半角のどちらでも可とするため、内部では正規化して処理する
    text_normalized = unicodedata.normalize('NFKC', text)

    # 入力項目ごとに区切る
    words = text_normalized.split()

    results = {}
    genre = load_genre()
    for word in words:
        if '/' in word:
            results['date'] = get_date(word)
        elif word.isdigit():
            results['amount'] = int(word)
        elif word in genre:
            category_id, genre_id = get_ids(word)
            if category_id and genre_id:
                results['category_id'], results['genre_id'] = category_id, genre_id
        else:
            today = datetime.today()
            results['comment'] = f'{word} (Slack登録: {today.year}/{today.month}/{today.day})'

    return results


def get_ids(word):
    """ カテゴリID、ジャンルIDを取得する

    :param word: ジャンルっぽい文字列
    :return: カテゴリID, ジャンルID (存在しない場合: None, None)
    :rtype: str, str
    """
    genres = load_genre()
    genre = genres.get(word)
    if genre:
        return genre['category_id'], genre['genre_id']
    return None, None


def load_genre():
    """ 環境変数からジャンルを取得し、Pythonオブジェクト化する """
    genre = os.environ.get('ZAIM_GENRE')
    if not genre:
        return

    return json.loads(genre)


def get_date(date_text):
    """ スラッシュ区切りで日付を指定

    01/01 -> 同年の1/1
    1/1 -> 同上
    2018/1/1 -> 年数も考慮
    それ以外 -> 判断つかないので、本日とする
    """
    date_list = date_text.split('/')
    if len(date_list) == 2:  # 月日のみ
        str_date = f'{datetime.today().year}{date_list[0].zfill(2)}{date_list[1].zfill(2)}'
        return datetime.strptime(str_date, '%Y%m%d')

    if len(date_list) == 3:  # 年月日
        str_date = f'{date_list[0]}{date_list[1].zfill(2)}{date_list[2].zfill(2)}'
        return datetime.strptime(str_date, '%Y%m%d')

    return datetime.today()


def post_zaim(zaim_data):
    """ Zaim APIにポストする

    :param zaim_data: Zaimにポストするためのデータ
    :return: 正常の場合はNone、エラーがある場合はエラーメッセージ
    :rtype: None or str
    """
    try:
        api = zaim.Api(consumer_key=os.environ['CONSUMER_KEY'],
                       consumer_secret=os.environ['CONSUMER_SECRET'],
                       access_token=os.environ['OAUTH_TOKEN'],
                       access_token_secret=os.environ['OAUTH_TOKEN_SECRET'])
        api.verify()

        # api.payment()の戻り値は以下の通り。そのため、戻り値を使って何かする、ということは無い
        # 正常:レスポンスのJSON内容が返ってくる
        # エラー:例外が出る
        api.payment(
            # 存在しないcategory_idでPOSTすると、「振替」だけれど変なデータができてしまうが、OKで通る
            # ただし、数字のところに文字列を入れると、例外が発生する
            # category_id="101xx",
            category_id=zaim_data['category_id'],
            genre_id=zaim_data['genre_id'],
            amount=zaim_data['amount'],
            date=zaim_data['date'],
            comment=zaim_data['comment']
        )
        return None
    except Exception as e:
        return str(e)


def main(request):
    """ Cloud Functions 呼ばれるメインの関数 """
    # Outgoing WebHooks App からは、POSTしかデータ送信されない前提なので、.formを使う
    # クエリストリングも同時に取得したい場合は .values を使う
    # 毎回 reqest.formと書くのが手間なので、変数に入れておく
    request_data = request.form

    # Outgoing Webhooks アプリだと、本来のアプリの他に、1,2回アクセスがある
    # この場合、request.formは空になっている
    if not request_data:
        logging.info(f'empty form data:{request_data}')
        return ''

    # Outgoing WebHooks アプリからの送信かどうかのバリデーション
    if request_data.get('token') != os.environ['SLACK_OUTGOING_WEBHOOKS_TOKEN']:
        logging.warning(f'not slack access, data: {request_data}')
        return ''

    # Botの場合に返信してしまうと、無限ループになるため除外する
    if request_data.get('user_name') == 'slackbot':
        logging.info(f'bot access data:{request_data.get("text")}')
        return ''

    # Slackの3秒ルールがあるため、リクエストが届いたということを通知するために
    # メイン処理は別スレッドに流して、ここは HTTP 200 をすぐに返す
    t = Thread(target=background, kwargs={'request_data': request_data})
    t.start()
    return ''

 
あとはこれを稼働させることで、冒頭のSlackやZaimのスクリーンショットの内容が実現できました。

 

ソースコード

GitHubに上げました。
https://github.com/thinkAmi/slack2zaim

*1:ZaimにログインしないとAPIドキュメントが読めないので、URLを貼るのはやめておきます