🤧
シフト予定自動追加アプリを作ってみた
動機
今まで，何かの予定を決める際，Googleカレンダーに入っている予定とバイトのシフトをそれぞれ見ながらどこが空いているか確認する必要がありました．バイトのシフトをカレンダーに追加しようにも，月に15回程シフトが入っているため，結構手間がかかります．
そこで，シフトの画像を読み取ることで自動でカレンダーにシフト予定を追加するアプリケーションを作りたいと考えました．
処理の流れ
- ユーザがslackのチャンネルでシフト画像を投稿する．
- 今回作成するアプリケーションがそれを受け取り，Google Cloud Visionに投げる．
- アプリケーションがGoogle Cloud Visionからの回答を受け取り，その内容を解析してシフトデータを作成する．
- 作成したシフトデータをGoogleカレンダーに予定として追加する．
- 作成したシフトデータをslackチャンネルでユーザに通知する．
設計
以下のサイトを参考にしながらやりました．
- オブジェクト指向は、こう設計しよう ( https://qiita.com/rawr/items/51af11d187aa60207ee3 )
正直，知識不足で半分くらいGeminiに設計してもらいました．
シフト画像は次の通りです．
左から順に日付，曜日，開始時刻，終了時刻を表しています．
コーディング
githubのURL：https://github.com/HayateKato/Shift-Conversion/tree/slack
API処理
以下の記事を参考にしました．
Slack
- Slackに投稿されたファイルをGoogleDriveに保存するBotをPythonで作成 ( https://sey323log.hatenablog.com/entry/20221106/1667720656 )
- Slack Python SDK でチャンネルにメッセージを投稿しよう ( https://qiita.com/seratch/items/93714b5cf3974c2f5327 )
- file_share message event ( https://api.slack.com/events/message/file_share )
Google Cloud Vision
- Google Vision APIで業務効率化！ ( https://qiita.com/yuzuyuzu_1234/items/499b43c792ce071d60cc )
- PythonでGoogle Cloud Vision APIを利用する方法 ( https://self-development.info/pythonでgoogle-cloud-vision-apiを利用する方法/ )
Googleカレンダー
- Google Calendar APIを叩いて予定を追加する ( https://qiita.com/Cayden_Madden/items/a679e0ab2a55d172fb1d )
- [初心者向け] GoogleカレンダーにPythonから予定を追加・編集してみた ( https://dev.classmethod.jp/articles/google-calendar-api-create-schedule/ )
Google Cloud Visionが抽出したデータをシフトデータに変換する処理
image_processor/vision_client.pyでGoogle Cloud Visionに画像を投げ，レスポンスをjsonファイルとして受け取ります．
image_processor/vision_client.py
"""Vision APIとアプリケーション間のやり取りを行うモジュール"""
from dotenv import load_dotenv
import os
load_dotenv("<.envファイルへの絶対パス>")
from google.cloud import vision
import json
from google.oauth2 import service_account
from unittest.mock import MagicMock, patch
class VisionClient:
"""Vision APIを使って画像からデータを抽出するクラス
Attributes:
_api_key (str): Vision APIを利用するための鍵のパス
_creds (:obj:`google.oauth2.service_account.Credentials`): 認証情報
_client (:obj:`google.cloud.vision_v1.ImageAnnotatorClient`): Vision APIとのやりとりを担うオブジェクト
"""
def __init__(self):
self._api_key = os.getenv("GOOGLE_CLOUD_API_KEY_PATH")
self._creds = service_account.Credentials.from_service_account_file(
self._api_key
)
self._client = vision.ImageAnnotatorClient(credentials=self._creds)
def extract_data_from_image(self, result_dir: str) -> None:
"""画像からデータを抽出するメソッド
Args:
result_dir (str):画像や抽出結果ファイルを格納するディレクトリへのパス
None
Examples:
(ここでは省略)
"""
with open(f"{result_dir}/shift.jpg", "rb") as f:
content = f.read()
image = vision.Image(content=content)
response = self._client.document_text_detection(image=image)
# Vision APIからのレスポンスを辞書型に変換
response_dict = vision.AnnotateImageResponse.to_dict(response)
with open(f"{result_dir}/response.json", "w", encoding="utf-8") as f:
json.dump(response_dict, f, ensure_ascii=False, indent=2)
受け取ったjsonファイルの中身は次のようになっています．
full_text_annotationキーの値に検出した文字の情報が格納されています．
verticesキーの値は各文字の左上，右上，右下，左下の座標を表しており，xは横座標，yは縦座標に対応しています．
textキーの値は検出した文字そのものです．
これら2つのデータからシフトデータを構築していきます．
response.json
{
"text_annotations": [
...
],
"full_text_annotation": {
"pages": [
{
"blocks": [
{
"paragraphs": [
{
"words": [
{
"symbols": [
{
"bounding_box": {
"vertices": [
{
"x": 35,
"y": 23
},
{
"x": 52,
"y": 23
},
{
"x": 52,
"y": 49
},
{
"x": 35,
"y": 49
}
]
},
"text": "8"
},
{
"bounding_box": {
"vertices": [
{
"x": 52,
"y": 23
},
{
"x": 67,
"y": 23
},
{
"x": 67,
"y": 49
},
{
"x": 52,
"y": 49
}
]
},
"text": "/"
},
{
"bounding_box": {
"vertices": [
{
"x": 66,
"y": 23
},
{
"x": 82,
"y": 23
},
{
"x": 82,
"y": 49
},
{
"x": 66,
"y": 49
}
]
},
"text": "1"
}
]
}
]
}
]
}
]
}
],
"text": "8/1"
},
"face_annotations": [],
"landmark_annotations": [],
"logo_annotations": [],
"label_annotations": [],
"localized_object_annotations": []
}
image_processor/shift_parser.pyでシフトデータの作成を行なっています．
image_processor/shift_parser.py
"""画像から抽出されたデータをシフトデータに変換するモジュール"""
import json
import re
from datetime import datetime
from dataclass.shift import Shift
from unittest.mock import MagicMock, patch
class ShiftParser:
"""画像から抽出されたデータをシフトデータに変換するモジュール
Attributes:
None
"""
def __init__(self):
pass
def parse_data_to_shifts(self, result_dir: str) -> list[Shift]:
"""画像から抽出されたデータを使ってシフトデータを作成するメソッド
Args:
result_dir (str):画像や抽出結果ファイルを格納するディレクトリへのパス
Returns:
list[Shift]: シフトデータ
Examples:
(ここでは省略)
"""
with open(f"{result_dir}/response.json", "r", encoding="utf-8") as f:
response = json.load(f)
# response.jsonの中身から処理に必要な部分のみを取り出す
full_text_annotation = response["full_text_annotation"]
page = full_text_annotation["pages"][0]
context = list()
for block in page["blocks"]:
for paragraph in block["paragraphs"]:
for word in paragraph["words"]:
for symbol in word["symbols"]:
text = symbol["text"]
vertices = symbol["bounding_box"]["vertices"]
context.append({"text": text, "vertices": vertices})
# "coordinate": (x座標の平均値，y座標の平均値)とする
processed_context = list()
for d in context:
processed_context.append(
{
"text": d["text"],
"coordinate": (
(
d["vertices"][0]["x"]
+ d["vertices"][1]["x"]
+ d["vertices"][2]["x"]
+ d["vertices"][3]["x"]
)
/ 4,
(
d["vertices"][0]["y"]
+ d["vertices"][1]["y"]
+ d["vertices"][2]["y"]
+ d["vertices"][3]["y"]
)
/ 4,
),
}
)
# y座標の平均値で昇順にソート
sorted_processed_context = sorted(
processed_context, key=lambda x: x["coordinate"][1]
)
# y座標の平均値が近いものでグループ化
line_context = list()
current_line = [sorted_processed_context[0]]
current_y = sorted_processed_context[0]["coordinate"][1]
for d in sorted_processed_context[1:]:
if abs(d["coordinate"][1] - current_y) < 10:
current_line.append(d)
else:
line_context.append(current_line)
current_line = [d]
current_y = d["coordinate"][1]
line_context.append(current_line) # 最後の行を追加
# グループ化した行ごとにx座標の平均値で並べ替え
sorted_line_context = list()
for line in line_context:
sorted_line = sorted(line, key=lambda x: x["coordinate"][0])
sorted_line_context.append(sorted_line)
# シフトが入っていない行(00時00分以外)を除外
removed_context = list()
for sl in sorted_line_context:
text = ""
for t in sl:
text += t["text"]
if len(text) >= 16:
removed_context.append(text)
# 誤字(Iや|)を除外
cleaned_context = list()
for rc in removed_context:
cleaned_rc = re.sub(r"[I|]", "", rc)
cleaned_context.append(cleaned_rc)
# patternに一致する文字列を探す
pattern = re.compile(
r"(\d{1,2}/\d{1,2})[月火水木金土日](\d{1,3})時(\d{2})分(翌日)?(\d{1,3})時(\d{2})分"
)
result_context = list()
for cc in cleaned_context:
match = pattern.search(cc)
if match:
# 一致したグループを抽出
date, start_hour, start_min, next_day, end_hour, end_min = (
match.groups()
)
start_time = f"{start_hour}:{start_min}"
two_days = True if next_day else False # 「翌日」が含まれる場合はフラグを立てる
end_time = f"{end_hour}:{end_min}"
result_context.append([date, two_days, start_time, end_time])
# シフトデータの作成
summary = "バイト" # summaryは固定
year = datetime.now().year # 年は現在のもの
time_difference = "+09:00:00" # 時差は+9時間
timezone = "Asia/Tokyo" # タイムゾーンは東京(Asia/Tokyo)
shifts = list()
for rc in result_context:
date = rc[0].split("/") # 月と日に分ける
two_days = rc[1] # 日を跨ぐか
start_time = rc[2].split(":") # 時と分に分ける
end_time = rc[3].split(":")
# 時間が3桁になっている場合は2桁にする(例:117->17)
if len(start_time[0]) >= 3:
start_time[0] = start_time[0][1:]
if len(end_time[0]) >= 3:
end_time[0] = end_time[0][1:]
# 開始時刻と終了時刻が00時00分になっている日はシフトデータに含めない
if start_time == ["00", "00"] and end_time == ["00", "00"]:
continue
# yyyy-mm-ddThh:mm:ss+hh:mm:ssの形式にする
start_datetime = datetime(
year=year,
month=int(date[0]),
day=int(date[1]),
hour=int(start_time[0]),
minute=int(start_time[1]),
)
splited_start_datetime = str(start_datetime).split(" ")
start_datetime_str = (
splited_start_datetime[0]
+ "T"
+ splited_start_datetime[1]
+ time_difference
)
day = int(date[1])
# 日を跨ぐ場合は終了日を(開始日+1)にする
if two_days:
day += 1
end_datetime = datetime(
year=year,
month=int(date[0]),
day=day,
hour=int(end_time[0]),
minute=int(end_time[1]),
)
splited_end_datetime = str(end_datetime).split(" ")
end_datetime_str = (
splited_end_datetime[0]
+ "T"
+ splited_end_datetime[1]
+ time_difference
)
shift = Shift(
summary=summary,
start_datetime=start_datetime_str,
end_datetime=end_datetime_str,
timezone=timezone,
)
shifts.append(shift)
return shifts
1. レスポンスファイルを読み込み，verticesとtextの値を取り出す
後の処理で扱いやすいように，textキーとverticesキーで構成される辞書のリストを作成します．
with open(f"{result_dir}/response.json", "r", encoding="utf-8") as f:
response = json.load(f)
# response.jsonの中身から処理に必要な部分のみを取り出す
full_text_annotation = response["full_text_annotation"]
page = full_text_annotation["pages"][0]
context = list()
for block in page["blocks"]:
for paragraph in block["paragraphs"]:
for word in paragraph["words"]:
for symbol in word["symbols"]:
text = symbol["text"]
vertices = symbol["bounding_box"]["vertices"]
context.append({"text": text, "vertices": vertices})
2. 文字の平均座標を計算
文字の平均座標を計算します．
また，textキーと，値として平均座標を持つcoordinateキーから成る辞書を新たに作ります．
# "coordinate": (x座標の平均値，y座標の平均値)とする
processed_context = list()
for d in context:
processed_context.append(
{
"text": d["text"],
"coordinate": (
(
d["vertices"][0]["x"]
+ d["vertices"][1]["x"]
+ d["vertices"][2]["x"]
+ d["vertices"][3]["x"]
)
/ 4,
(
d["vertices"][0]["y"]
+ d["vertices"][1]["y"]
+ d["vertices"][2]["y"]
+ d["vertices"][3]["y"]
)
/ 4,
),
}
)
3. 画像内で同じ行に属する文字をグループ化
先ほど計算した平均y座標の小さい順に辞書をソートします．
その後，平均y座標が近いものをグループ化します．このとき，平均y座標の差が10未満なら同じ行としています．
# y座標の平均値で昇順にソート
sorted_processed_context = sorted(
processed_context, key=lambda x: x["coordinate"][1]
)
# y座標の平均値が近いものでグループ化
line_context = list()
current_line = [sorted_processed_context[0]]
current_y = sorted_processed_context[0]["coordinate"][1]
for d in sorted_processed_context[1:]:
if abs(d["coordinate"][1] - current_y) < 10:
current_line.append(d)
else:
line_context.append(current_line)
current_line = [d]
current_y = d["coordinate"][1]
line_context.append(current_line) # 最後の行を追加
4. 行内の文字を正しい順番に並べ替える
平均x座標を基準にして，行内の文字を画像と同じ順番(日付，曜日，開始時刻，終了時刻の順)になるように並べ替えます．
# グループ化した行ごとにx座標の平均値で並べ替え
sorted_line_context = list()
for line in line_context:
sorted_line = sorted(line, key=lambda x: x["coordinate"][0])
sorted_line_context.append(sorted_line)
5. 不要な行や文字を削除
シフトが入っていない日(時間が- - - -となっている日)の行を削除します．
また，シフト画像内の縦線がアルファベットの「I」や記号の「|」と認識されてしまうことがあるため，これらを除外します．
# シフトが入っていない行(00時00分以外)を除外
removed_context = list()
for sl in sorted_line_context:
text = ""
for t in sl:
text += t["text"]
if len(text) >= 16:
removed_context.append(text)
# 誤字(Iや|)を除外
cleaned_context = list()
for rc in removed_context:
cleaned_rc = re.sub(r"[I|]", "", rc)
cleaned_context.append(cleaned_rc)
6. パターンにマッチする文字列を探す
シフトの日付，開始時刻，終了時刻を抽出します．この時，シフト画像内で終了時刻に「翌日」という文字が含まれている場合は，フラグを立てて日を跨ぐシフトであることを識別できるようにしています．
正規表現は以下の記事を参考にしました．
- 分かりやすいpythonの正規表現の例 (https://qiita.com/luohao0404/items/7135b2b96f9b0b196bf3 )
- Pythonの正規表現モジュールreの使い方（match, search, subなど） (https://note.nkmk.me/python-re-match-search-findall-etc/ )
# patternに一致する文字列を探す
pattern = re.compile(
r"(\d{1,2}/\d{1,2})[月火水木金土日](\d{1,3})時(\d{2})分(翌日)?(\d{1,3})時(\d{2})分"
)
result_context = list()
for cc in cleaned_context:
match = pattern.search(cc)
if match:
# 一致したグループを抽出
date, start_hour, start_min, next_day, end_hour, end_min = (
match.groups()
)
start_time = f"{start_hour}:{start_min}"
two_days = True if next_day else False # 「翌日」が含まれる場合はフラグを立てる
end_time = f"{end_hour}:{end_min}"
result_context.append([date, two_days, start_time, end_time])
7. シフトデータを作成
シフトデータをShiftオブジェクトとして作成します．この時，シフト画像内で開始時刻と終了時刻がともに「00時00分」となっている日は実際にはシフトが無いので，シフトデータに含めないようにしています．
# シフトデータの作成
summary = "バイト" # summaryは固定
year = datetime.now().year # 年は現在のもの
time_difference = "+09:00:00" # 時差は+9時間
timezone = "Asia/Tokyo" # タイムゾーンは東京(Asia/Tokyo)
shifts = list()
for rc in result_context:
date = rc[0].split("/") # 月と日に分ける
two_days = rc[1] # 日を跨ぐか
start_time = rc[2].split(":") # 時と分に分ける
end_time = rc[3].split(":")
# 時間が3桁になっている場合は2桁にする(例:117->17)
if len(start_time[0]) >= 3:
start_time[0] = start_time[0][1:]
if len(end_time[0]) >= 3:
end_time[0] = end_time[0][1:]
# 開始時刻と終了時刻が00時00分になっている日はシフトデータに含めない
if start_time == ["00", "00"] and end_time == ["00", "00"]:
continue
# yyyy-mm-ddThh:mm:ss+hh:mm:ssの形式にする
start_datetime = datetime(
year=year,
month=int(date[0]),
day=int(date[1]),
hour=int(start_time[0]),
minute=int(start_time[1]),
)
splited_start_datetime = str(start_datetime).split(" ")
start_datetime_str = (
splited_start_datetime[0]
+ "T"
+ splited_start_datetime[1]
+ time_difference
)
day = int(date[1])
# 日を跨ぐ場合は終了日を(開始日+1)にする
if two_days:
day += 1
end_datetime = datetime(
year=year,
month=int(date[0]),
day=day,
hour=int(end_time[0]),
minute=int(end_time[1]),
)
splited_end_datetime = str(end_datetime).split(" ")
end_datetime_str = (
splited_end_datetime[0]
+ "T"
+ splited_end_datetime[1]
+ time_difference
)
shift = Shift(
summary=summary,
start_datetime=start_datetime_str,
end_datetime=end_datetime_str,
timezone=timezone,
)
shifts.append(shift)
実際に使ってみた
アプリケーションコンテナを起動後，Slackで画像を送ります．
送信後，Botから画像のダウンロード完了通知と作成されたシフトデータが送られてきます．
Googleカレンダーを見てみると，シフトデータ通りに予定が追加されています．
振り返り
まず，オブジェクト指向での設計がとても大変でした．特に，Controllerクラスやデータクラスが必要になることを最初は想定できていなかったため，設計力の重要性を実感しました。今回は途中からGeminiに設計作業を頼ってしまいましたが，次回は全て自力で設計できるようにしたいと思います。
また，今回テスト駆動開発でコーディングしましたが，テストを書く際にAPIを使用する箇所をMockで置き換える部分が特に難しかったです．Mockについてあまり理解できていないので，今後勉強していきたいです．
制作期間は約2週間ぐらいかかってしまい大変でしたが，とても楽しかったです．完成したアプリは現在も日常的に活用していて，作って良かったと思えました．
今後の予定
- Djangoを使ってWebアプリケーション化する．
- バイトの同期に使ってもらい，フィードバックを受ける．
Discussion