この記事は何？
Google Calendarの予定をMessagingAPIを使用して、とあるメンズコーチ風に通知する方法を記した記事。
概要
最近筋トレさぼってない？厳しいって。
LINEグループに招待して、メンズコーチをメンションすると、GoogleCalendarの予定からグループのメンバーにご指導をいただけるよ。
Google Calendar から予定を取得する
- GCPのコンソールに移動し、Google CalendarのAPIを有効化
- 自分のカレンダーに移動し、1で作成したbotのメールアドレスにカレンダーを共有
詳しくは以下の記事を参照。
https://www.coppla-note.net/posts/tutorial/google-calendar-api/
Google Calendar から予定を取得し、ジムの回数を取得する関数は以下。
import datetime
import pytz
from googleapiclient.discovery import build
from google.auth import load_credentials_from_file
# 認証情報ファイルのパス
CREDENTIALS_PATH = '../credentials_service.json'
TOKEN_PATH = '../token.json'
# 認証範囲
SCOPES = ['https://www.googleapis.com/auth/calendar.readonly']
def get_schedule(calendar_id):
creds = load_credentials_from_file(
CREDENTIALS_PATH, SCOPES
)[0]
service = build('calendar', 'v3', credentials=creds)
JST = pytz.timezone('Asia/Tokyo')
# プログラム実行時間から1週間後までに期間を設定
now = datetime.datetime.now(JST)
start_time = now.isoformat()
end_time = (now + datetime.timedelta(days=7)).isoformat()
events_result = service.events().list(calendarId=calendar_id, timeMin=start_time, timeMax=end_time,
maxResults=50, singleEvents=True,
orderBy='startTime').execute()
events = events_result.get('items', [])
if not events:
return None
schedule = []
for event in events:
event_info = {
'summary': event.get('summary'),
'start': event['start'].get('dateTime', event['start'].get('date')),
'attendees': event.get('attendees', [])
}
schedule.append(event_info)
return schedule
def count_workout_events(schedule):
workout_count = {}
for event in schedule:
if 'ジム' in event['summary']:
for attendee in event['attendees']:
email = attendee.get('email')
response_status = attendee.get('responseStatus')
if email and response_status == 'accepted':
if email in workout_count:
workout_count[email] += 1
else:
workout_count[email] = 1
return workout_count
if __name__ == '__main__':
schedule = get_schedule()
count = count_workout_events(schedule)
print(count)
Messaging API によるBOTの作成
LINE BOT の設定
LINE Deveropersに登録し、コンソールにてプロバイダを作成する。
プロフィール情報を入力し、トークンなどの認証情報を取得する。
取得した認証情報は環境変数に書き込む or 指定したパス上に配置する。
webhook の設定
以下のタイミングでBOTがアクションを取れるように設定する。
- グループに参加した時
- ユーザーにメンションされた時
1. グループに参加した時
BOTがグループに招待された時、グループIDをcsvに保存する。
また、挨拶メッセージを送るようにする。
以下はapp.pyに記述する。
from flask import Flask, request, abort, jsonify
from linebot import WebhookHandler, LineBotApi
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, TextMessage, TextSendMessage, JoinEvent
from dotenv import load_dotenv
import requests
import os
import csv
from main import main
app = Flask(__name__)
load_dotenv()
CHANNEL_SECRET = os.getenv('CHANNEL_SECRET')
CHANNEL_ACCESS_TOKEN = os.getenv('CHANNEL_ACCESS_TOKEN')
if not CHANNEL_SECRET or not CHANNEL_ACCESS_TOKEN:
raise ValueError("CHANNEL_SECRET and CHANNEL_ACCESS_TOKEN must be set in environment variables.")
handler = WebhookHandler(CHANNEL_SECRET)
line_bot_api = LineBotApi(CHANNEL_ACCESS_TOKEN)
@handler.add(JoinEvent)
def handle_join(event):
group_id = event.source.group_id
add_group_id_to_csv(group_id)
line_bot_api.push_message(group_id, TextSendMessage(text="厳しいって"))
def add_group_id_to_csv(group_id):
csv_file_path = '../data/group_id.csv'
file_exists = os.path.isfile(csv_file_path)
with open(csv_file_path, mode='a', newline='') as csv_file:
writer = csv.writer(csv_file)
if not file_exists:
writer.writerow(['group_id'])
writer.writerow([group_id])
return
2. メンションされた時
メンションされた時に、main.py のmain関数を実行するようにする。(main.pyは後述)
ユーザーIDは送りたいグループIDを指定する。
グループIDはグループに参加した際csvに保存されるので、csvを読み込む必要がある。
def load_group_ids(file_path):
""" ファイルからグループIDを読み込む """
with open(file_path, 'r') as file:
return [line.strip() for line in file]
@handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
user_ids = load_group_ids('../data/group_id.csv')
if event.message.text == '@ジョージ' and event.source.group_id == user_ids[1]:
main()
APIの作成
メッセージを送るため、main関数を実行するための処理をapp.pyに書いていく。
def send_push_message(token, user_id, message_text):
url = 'https://api.line.me/v2/bot/message/push'
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {token}'
}
data = {
'to': user_id,
'messages': [{
'type': 'text',
'text': message_text
}]
}
response = requests.post(url, headers=headers, json=data)
return response.status_code, response.text
@app.route("/send_message")
def send_message():
token = os.getenv('CHANNEL_ACCESS_TOKEN')
user_ids = load_group_ids('../data/group_id.csv')
message_text = request.args.get('message', 'こんにちは、これはプッシュメッセージです！')
results = []
for user_id in user_ids:
status_code, response_text = send_push_message(token, user_id, message_text)
results.append({'user_id': user_id, 'Status Code': status_code, 'Response': response_text})
return jsonify(results)
ngrok の使用方法
webhookを使用するために、ローカルのFlaskサーバーのポートを外部に公開する必要がある。
そのためにngrokを用いた。
if __name__ == '__main__':
app.run(port=8080)
8080番ポートを外部に共有するためには、以下コマンドを実行。
ngrok http 8080
上記のコマンドを打つと、
Forwarding https://hogehoge.ngrok-free.app
のように表示されるので、リンクをコピー。
MessagingAPIの設定ページに移動し、webhookURLを設定。
https://hogehoge.ngrok-free.app/callback と入力。
これにて設定完了。
BOTをグループに招待し、メンションすると、カレンダーの予定をもとにご指導をいただける形となった。
実装できなかったこと
- グループのユーザーをメンションすること
LINE Developersの有料会員にならないとグループにいるユーザーのユーザーIDを取得することができなかったため - メンションイベントをwebhookとして受け取ること
そもそも存在しなかった。Slack等ではできるらしい - ユーザーごとに予定を取得
本来であれば招待されたグループごとにカレンダーのIDを参照できるのがのぞましい。今回は個人利用のため、自分のカレンダーのみを取得し、グループメンバーには筋トレに行く時には自分を招待してもらうようにした。
コード