最近、 Zaim を使って家計簿をつけています。
https://zaim.net/
ただ、時々入力を忘れたり、重複入力してしまうことがありました。
そこで、
- 日頃Slackを使っている
- ZaimにはWeb APIがある
ということから、SlackからZaimのデータを登録する仕組みを作りましたので、メモを残します。
なお、今回の範囲では影響ありませんでしたが、Zaimの金融連携データはAPI経由では取得できないようです。プレミアムプラン契約をしている時は取得できるようになってくれるとありがたいです。
目次
環境
- GCP Cloud Functions
- Cloud Functionsの
requirements.txt に追加したライブラリ
- zaim 0.2.2
- slackclient 1.3.0
作ったもの
長いメモなので、こんな感じのものを作ったというのを書いておきます。
SlackからZaimへ登録する
決まったフォーマット (日付(yyyy/mm/dd or mm/dd) ジャンル名 金額 コメント) をSlackにポストすると、Zaimに反映します。
なお、登録する時に西暦を入力するのが手間だったため、西暦が省略された場合は、実行日の西暦を渡すようにしています。
Slackの様子
Zaimに登録できたら、OKなリアクションをします。
Zaimの様子
Zaimにも登録できています。
一方、Zaimに登録できない場合は、NGなリアクションと、NGになった理由をスレッドで返信します。
他にも、使い方を忘れたときのために、以下の2機能を作りました。
Zaimの登録可能なジャンルを知る
ジャンル とポストすると、Zaimで登録可能なジャンルの一覧をスレッドで返信します。
Zaimへ登録する際のポストの書式を知る
書式 もしくは フォーマット とポストすると、登録するポストの書式をスレッドで返信します。
以降は、これらを作った時のメモになります。
事前調査
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のポストをフックする手段としては以下がありました。
- Legacy Outgoing Webhooks
- Events API
- Outgoing WebHooks App
他にも 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では
の2つを指定し、APIを呼ぶ必要がありました。
それらが何を指すのか調べたところ、
- category
- genre
- 上記のURLで
内訳 をクリックすると表示されるもの
のようでした。
ただ、画面上では両方の名称は確認できるものの、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):
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)
print(access_token)
def _get_request_token(self):
auth = OAuth1(
self.tokens['CONSUMER_KEY'],
self.tokens['CONSUMER_SECRET'],
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):
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 の設定
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 の関数を作成しますが、いくつか悩んだところがありました。
名前について
むやみに外部からアクセスされても困るため、分かりづらい名前を付けることにしました。
そこで、Pythonの uuid モジュールを使って、ランダムな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でできる方法を調べたところ、Pythonの threading.Thread が使えそうでした。
上記のstackoverflowの回答はHeroku上のものでしたが、Cloud Functionsでも問題なく動作しました。
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のトークンを設定しておくことで、
request_data = request.form
if request_data.get('token') != os.environ['SLACK_OUTGOING_WEBHOOKS_TOKEN']:
...
な形で判定できそうでした。
なお、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 APIの chat.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
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)
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
"""
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(
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 呼ばれるメインの関数 """
request_data = request.form
if not request_data:
logging.info(f'empty form data:{request_data}')
return ''
if request_data.get('token') != os.environ['SLACK_OUTGOING_WEBHOOKS_TOKEN']:
logging.warning(f'not slack access, data: {request_data}')
return ''
if request_data.get('user_name') == 'slackbot':
logging.info(f'bot access data:{request_data.get("text")}')
return ''
t = Thread(target=background, kwargs={'request_data': request_data})
t.start()
return ''
あとはこれを稼働させることで、冒頭のSlackやZaimのスクリーンショットの内容が実現できました。
GitHubに上げました。
https://github.com/thinkAmi/slack2zaim
