Zenn
🗓️

Google Calendarの予定を操作するDiscord Botを作る

2022/02/09に公開
Python
Bot
Discord
tech

概要

Discord上で友人と話していたときにゲーム内の予定をチャットに流せたら便利だと言う話になったので、カレンダー情報を流すBotを作成しました。

言語

Python

使用ライブラリ

Pycord
schedule
google-api-python-client
google-auth-httplib2
google-auth-oauthlib

API

Google Calendar API
Discord Webhook

本篇

  • Bot準備
  • コマンド
  • DiscordWebhookについて
  • 予定の挿入
  • 予定の確認
  • 通知

プログラム中のsetterとgetterは一部省略しています。

Bot準備

公式ページのBotアカウント作成に従ってBotを作ります。ここで取得したBot用のトークンはconfigに保存しました。

続いて、以下のコードを実行します。これで最低限のBotが稼働します。

main.py
import discord

with open('config.json', 'r') as f:
    config = json.load(f)
    discordToken = config['discord_token']
    
client = discord.Client(ws = int(os.environ.get('PORT', 5000)))

@client.event
async def on_ready():
    print('READY')
    
client.run(discordToken)

コマンド

Botを導入しているサーバー内で指定のコマンドとメッセージが送られたときにメッセージのコマンド部分を削除して処理に回しています。

main.py
import os
import json
import discord
from logic.calendarLogic import 

with open('config.json', 'r') as f:
    config = json.load(f)
    discordToken = config['discord_token']
+    calendarText = config['calendar'].get('calendar_text')
+    listText = config['calendar'].get('list_text')
+    addEventText = config['calendar'].get('add_event_text')
+    addLongEventText = config['calendar'].get('add_long_event_text')
+    deleteEventText = config['calendar'].get('delete_text')

client = discord.Client(ws = int(os.environ.get('PORT', 5000)))

@client.event
async def on_ready():
    print('READY')
    
+ @client.event
+ async def on_message(message):
+    content = message.content
+
+    if content == calendarText:
+        logic = CalendarLogic()
+        result = logic.calendarUrl()
+
+    if content == listText:
+        logic = CalendarLogic()
+        result = logic.get()
+        
+    elif re.search(addEventText, content):
+        event = content.replace(addEventText, '')
+        logic = CalendarLogic()
+        logic.insert(event)
+
+    elif re.search(addLongEventText, content):
+        event = content.replace(addLongEventText, '')
+        logic = CalendarLogic()
+        logic.insertLongEvent(event)
+
+    elif re.search(deleteEventText, content):
+        eventId = content.replace(deleteEventText, '')
+        logic = CalendarLogic()
+        logic.delete(eventId)

client.run(discordToken)

同様に取得・削除のコマンドも追加します。
最終的に以下のようになりました。

DiscordWebhookについて

今回Botからのメッセージの送信はDiscord Webhookを使用します。
Webhookを使用することでメッセージごとのアバターの名前や送信者名の変更、embedsの整形が容易になります。

webhookData.py
import json

class WebhookContent():
    def __init__(self):
        with open('config.json', 'r') as f:
            config = json.load(f)
            self.__config = config
            self.__userName = config['webhook'].get('message').get('user_name')
            self.__avatarUrl = config['webhook'].get('message').get('avatar_url')
            self.__color = config['webhook'].get('message').get('color')

    def createMessage(self, message):
        body = {
            "username": self.userName,
            "avatar_url": self.avatarUrl,
            "content" : message,
            "embeds": []
        }
        return  body

    def createEmbeds(self, data):
        embeds = {
            "title": data.summary,
            "description": f"{data.date} {data.time}",
            "color": int(self.color, 16),
            "footer": {
                "icon_url": self.avatarUrl,
                "text": data.eventId
            },
        }
        if data.description:
            fields = {
                "fields": [
                    {
                        "name": "詳細",
                        "value": data.description,
                    }
                ]
            }
            embeds.update(fields)
        return embeds

    def createLongEventEmbeds(self, data):
        embeds = {
            "title": data.summary,
            "description": f"{data.date} 〜 {data.endDate}",
            "color": int(self.color, 16),
            "footer": {
                "icon_url": self.avatarUrl,
                "text": data.eventId
            },
        }
        if data.description:
            fields = {
                "fields": [
                    {
                        "name": "詳細",
                        "value": data.description,
                    }
                ]
            }
            embeds.update(fields)
        return embeds
webhook.py
import json
import requests

class Webhook():
    def __init__(self, body = None):
        with open('config.json', 'r') as f:
            config = json.load(f)
        self.__url = config["webhook_url"]

    @property
    def body(self):
        return self.__body

    @body.setter
    def body(self, val):
        self.__body = val

    @property
    def url(self):
        return self.__url

    def send(self, body):
        requests.post(self.url,json.dumps(body),headers={'Content-Type': 'application/json'})

カレンダーコマンド実行時にWebhookに必要な情報を乗せ、指定のURLにリクエストを行いメッセージを送信します。

予定の挿入

botを導入しているDiscordサーバーで所定のフォーマットのメッセージを受け取った時にカレンダーに予定を挿入します。
リクエスト操作については別記事にまとめましたので、ここでは実装部分を記載します。
https://zenn.dev/kahra/articles/8cb29a55ed392b
受け取ったメッセージを各項目に分解してインスタンスを生成し、リクエストを行います。

calendarLogic.py
import　json
from api.calendarApi import CalendarApi
from data.calendarData import CalendarData
from data.calendarData import CalendarContent

class CalendarLogic():
    def __init__(self):
        with open('config.json', 'r') as f:
            config = json.load(f)
            self.__config = config

    def insert(self, event):
        try:
            # 予定挿入
            calendar = CalendarApi()
            calData = CalendarData()
            calContent = CalendarContent()

            content = event.split(' ')
            calData.date = content[0]
            date = calData.date.split('/')
            calData.year = int(date[0])
            calData.month = int(date[1]) 
            calData.day = int(date[2])

            calData.time = content[1]
            time = calData.time.split(':')
            calData.hour = int(time[0])
            calData.minute = int(time[1])

            calData.summary = content[2]
            if len(content) == 4:
                calData.description = content[3]
            else:
                calData.description = ''

            body = calContent.createInsertData(calData)
            result = calendar.insert(body)
            calData.eventId = result["id"]

            # Webhookメッセージ送信
            webhook = Webhook()
            webhookData = WebhookData()
            webhookContent = WebhookContent()

            webhookData.summary = calData.summary
            webhookData.description = calData.description
            webhookData.eventId = calData.eventId
            webhookData.date = calData.date
            webhookData.time = calData.time
            
            message = self.config['webhook'].get('message').get('insert_message')
            body = webhookContent.createMessage(message)
            embeds = webhookContent.createEmbeds(webhookData)
            body["embeds"].append(embeds)

            webhook.send(body)

        except Exception as e:
            print(e)
	    # エラーメッセージ送信
            webhook = Webhook()
            webhookContent = WebhookContent()
            message = self.config['webhook'].get('message').get('insert_error_message')
            body = webhookContent.createMessage(message)

            webhook.send(body)

Botを導入しているサーバーに上で記したコマンドを送ることで予定を挿入します。

予定の取得

同様に予定の取得コマンドも実装します。
予定取得時にタイムゾーンとしてJSTを生成し指定しているところがポイントです。

calendarLogic.py
    # 一週間分の予定を返す
    def get(self):
        try:
	    result = []
            calendar = CalendarApi()
            result = calendar.get()

            # メッセージ送信処理
            webhook = Webhook()
            webhookData = WebhookData()
            webhookContent = WebhookContent()

            if result == []:
                message = self.config['webhook'].get('message')].get('event_none_message')
                body = webhookContent.createMessage(message)
            else:
                message = self.config['webhook'].get('message').get('event_message')
                body = webhookContent.createMessage(message)
                
                for event in result:
                    webhookData.summary = event['summary']
                    webhookData.description = event.get('description', '')
                    webhookData.eventId = event['id']

                    # 終日予定と時間予定を区別しフォーマットを設定
                    start = event['start'].get('dateTime','').split('T')
                    if start != ['']:
                        webhookData.date = start[0].replace('-','/')
                        webhookData.time = start[1].replace(':00+09:00','')
                        embeds = webhookContent.createEmbeds(webhookData)
                    else:
                        webhookData.date = event["start"].get("date").replace('-','/')
                        webhookData.endDate = event["end"].get("date").replace('-','/')
                        embeds = webhookContent.createLongEventEmbeds(webhookData)
                
                    body["embeds"].append(embeds)

            webhook.send(body)

        except Exception as e:
            print(e)
            webhook = Webhook()
            webhookData = WebhookData()
            webhookContent = WebhookContent()
            message = self.config['webhook'].get('message').get('event_error_message')
            body = webhookContent.createMessage(message)

            webhook.send(body)

コマンドを送ったときのプレビューです。

予定の削除

予定の削除コマンドも実装します。
予定のメッセージembedsのfooterに記載しているイベントIDを受け取り、リクエストを送ります。

calendarLogic.py
    def delete(self, event):
        try:
            calendar = CalendarApi()
            calendar.delete(event)

            webhook = Webhook()
            webhookContent = WebhookContent()

            message = self.config['webhook'].get('message').get('delete_message')
            body = webhookContent.createMessage(message)
            webhook.send(body)
        except Exception as e:
            print(e)
            webhook = Webhook()
            webhookContent = WebhookContent()
            message = self.config['webhook'].get('message').get('delete_error_message')
            body = webhookContent.createMessage(message)

            webhook.send(body)

通知

個人で利用する時はGoogle Calendarに通知を送って貰えば良いとも思ったのですが、小規模のコミュニティで通知をしたい時は便利なのでDiscord上に予定の通知メッセージを送ります。

スケジューラにはscheduleライブラリを使用し、毎週月曜日と毎日の指定時刻に予定を取得して通知します。

main.py
import schedule
import time
from logic.schedulerLogic import SchedulerLogic


def EventSchedule():
    logic = SchedulerLogic()
    schedule.every().monday.at("08:45").do(logic.getWeeklyEvents)
    schedule.every().day.at("09:00").do(logic.getDailyEvents)

EventSchedule()

while True:
    schedule.run_pending()
    time.sleep(1)
calendarApi.py
import json
import datetime
from google.auth import load_credentials_from_file
from googleapiclient.discovery import build

class CalendarApi():
    def __init__(self):
        with open('config.json', 'r') as f:
            config = json.load(f)
            self.__calendarId = config["calendar"].get('id')

        SCOPES = [config["calendar"].get('calendar')]
        gapi_creds = load_credentials_from_file('credentials.json', SCOPES)[0]
        self.__service = build('calendar', 'v3', credentials=gapi_creds)

    def get(self, day=0, routine=False):
        now = datetime.datetime.now(datetime.timezone.utc)

        period = now + datetime.timedelta(days=day)
        if routine:
            timeMax = period + datetime.timedelta(minutes=59)
            timeMax = datetime.datetime(timeMax.year, timeMax.month, timeMax.day, timeMax.hour, timeMax.minutes, tzinfo=datetime.timezone.utc)
        else:
            timeMax = datetime.datetime(period.year, period.month, period.day, 23, 59, tzinfo=datetime.timezone.utc)

        events_result = self.service.events().list(
            calendarId=self.calendarId, 
            timeMin=now.isoformat(),
            timeMax=timeMax.isoformat(),
            maxResults=5,
            singleEvents=True,
            orderBy='startTime'
        ).execute()
        events = events_result.get('items', [])

        return events

引数daysで予定を取得したい日数を指定します。

schedulerLogic.py
import  json
import datetime
from api.calendarApi import CalendarApi
from data.calendarData import CalendarData
from api.webhook import Webhook
from data.webhookData import WebhookData
from data.webhookData import WebhookContent

class SchedulerLogic():
    def __init__(self):
        with open('config.json', 'r') as f:
            config = json.load(f)
            self.__config = config
        
    @property
    def config(self):
        return self.__config
                
    def getWeeklyEvents(self):
        try:
	    result = []
            calendar = CalendarApi()
            result = calendar.getEventsByDays(days=7)

            webhook = Webhook()
            webhookData = WebhookData()
            webhookContent = WebhookContent()

            if result != []:
                message = self.config['webhook'].get('message').get('weekly_event_message')
                body = webhookContent.create(message)

                for event in result:
                    webhookData.summary = event['summary']
                    webhookData.description = event.get('description', '')
                    webhookData.eventId = event['id']

                    start = event['start'].get('dateTime','').split('T')
                    if start != ['']:
                        webhookData.date = start[0].replace('-','/')
                        webhookData.time = start[1].replace(':00+09:00','')
                        embeds = webhookContent.createEmbeds(webhookData)
                    else:
                        webhookData.date = event["start"].get("date").replace('-','/')
                        webhookData.endDate = event["end"].get("date").replace('-','/')
                        embeds = webhookContent.createLongEventEmbeds(webhookData)

                    body["embeds"].append(embeds)
                webhook.send(body)

            else:
                message = self.config['webhook'].get('message').get('event_none_message')
                body = webhookContent.create(message)
                webhook.send(body)

        except Exception as e:
            print(e)

    def getDailyEvents(self):
        try:
	    result = []
            calendar = CalendarApi()
            result = calendar.getEventsByDays(days=0)

            webhook = Webhook()
            webhookData = WebhookData()
            webhookContent = WebhookContent()

            if result != []:
                message = self.config['webhook'].get('message').get('daily_event_message')
                body = webhookContent.create(message)
                eventFlag = False

                for event in result:
                    webhookData.summary = event['summary']
                    webhookData.description = event.get('description', '')
                    webhookData.eventId = event['id']

                    start = event['start'].get('dateTime','').split('T')
                    if start != ['']:
                        eventFlag = True

                        webhookData.date = start[0].replace('-','/')
                        webhookData.time = start[1].replace(':00+09:00','')
                        embeds = webhookContent.createEmbeds(webhookData)
                        body["embeds"].append(embeds)
                if(eventFlag):
                    webhook.send(body)

        except Exception as e:
            print(e)

処理の多くはBotの予定確認と変わりません。
日毎の確認の場合、終日開催のイベント通知を省いています。


以上です。

おわりに

以前の記事ではBotからメッセージを送信していましたが、今回はWebhookを使用する方法に変更しました。Webhookの方が使用感がよく、見栄えも整えやすいので今後は優先的に使用したいと思います。

記事を書くことにも慣れていなく、コードのベタ貼りが多くなったと反省しています。簡潔に作ったものを伝えられるよう努力します。

Discussion