🗓️

学校の課題をNotionとGoogleカレンダーで管理したら絶対に忘れなくなった

2024/11/15に公開

実現したこと

その1: Notion で課題を管理できるようになった

  • IceBox, BackLog, In progress, Done の4つのステータスで課題を管理
    • IceBox: 課題の内容をまだ見ていない
    • BackLog: 課題の内容を確認して所要時間を見積もった
    • In progress: 現在取り組んでいる
    • Done: 課題を提出した
  • IceBox -> BackLog -> In progress -> Done の順にステータスを変更していく
  • 課題の優先度と所要時間を見積もったら IceBox -> BackLog とする
  • 課題に取り組み始めたら BackLog -> In progress とする

Notion Board View
課題を管理する Notion ボードビュー

その2: 新しい課題が出されたとき Notion にタイトルと期限を登録すると、Google カレンダーにも自動で同期されるようになった

Google Calendar Register

その3: Notion で特定の課題のステータスを「Done」にすると、Google カレンダーで「✅」マークが入って課題が終わったことがカレンダー上でも分かるようになった

Google Calendar Done

なぜ作ったか

以前は学校の課題を Google ToDo で管理していました。提出期限をタスクの概要欄に記入し、取り組むタイミングをリマインダーの時間として設定していましたが、学期末には大量の課題が一度に溜まることもあり、管理が難しくなることがありました。特に短期・長期の課題が混在することもある上に、それぞれの課題の所要時間の見積もりなどを考慮するのが難しく、さらに進捗状況やどの課題に取り組んでいるかを忘れてしまうので、より効率的な管理方法が必要でした。

Google ToDo
Google ToDo で管理していた課題

そこで、普段から GitHub Projects をよく利用しているのですが、そこから着想を得て Kanban 方式で課題を管理することがいいのではないかと考えました。よく Notion を利用しているため、 Notion のデータベース機能を用いて管理してみました。

しかし Google ToDo は操作性も良いですし、カレンダーともいい感じに連携してくれるので、Notion で管理しているデータをカレンダーと連携させられる仕組みを作りたいと思いました。

システムの全体像

Architecture

AWS Lambda から Notion API と Google Calendar API を呼んでいます。
また、 AWS EventBridge と Lambda を使って定期的に以下のような処理を実行するようにしました。

  1. Notion API を呼び、提出期間が過去1年から1年後までになっている課題を全て取得する
  2. Google Calendar API で1年前から1年後の予定を取得する
  3. 1 と 2 で得たデータを比較して、未登録であったりステータスに差分があったりする場合は Google Calendar に登録する

システムの構築はそれほど難しくなく、ToDo の運用を変えることを思いついてから完成させるまで10時間くらいでした。

システムの構築手順

Google Calendar API を呼ぶための設定

Python quickstart ではユーザーが一度 OAuth 同意画面に飛んで token と refresh token を取得する前提となっていますが、今回は AWS Lambda から API を呼び出したいので、サービスアカウントと呼ばれるものを用いて簡単に Google Calendar API を利用する方法を紹介します。

サービスアカウントの設定手順

以下の記事に詳しく書かれていますが、主要な手順をメモしておきます。
https://liginc.co.jp/472637

  1. Google Cloud Platform にアクセスする
  2. プロジェクト名を適当に決めて、新しいプロジェクトを作成する。他の設定はそのままにする。
  3. ハンバーガーボタン(三本線のメニュー)から「APIとサービス」 -> 「認証情報」を選択する。
  4. 「認証情報を作成」 -> 「サービスアカウント」を選択する。
  5. サービスアカウント名を適当に入力し、他の項目は空欄のまま作成する。
  6. 作成したサービスアカウントをクリックし、上部のタブから「キー」を選択する。「鍵を追加」 -> 「新しい鍵を作成」を選択する。
  7. JSON にチェックが入っていることを確認し、そのまま作成する。JSON ファイルがダウンロードされる。

以上の手順が完了したら、サービスアカウントの詳細画面を確認します。

Service Account Summary

ここで、サービスアカウントのメールアドレスをコピーしておきます。

次にサービスアカウントと Google カレンダーの連携をします。

  1. 新しいカレンダーを作成して好きな名前をつける。
  2. 「特定のユーザーとの共有」->「ユーザーを追加」と進む。
  3. メールアドレスにサービスアカウントのメールアドレスを指定して招待する。
  4. Google Cloud Platform にアクセスし、左上のハンバーガーボタンから「APIとサービス」->「ライブラリ」を選択。
  5. Google Calendar API を見つけて有効にする。

これでサービスアカウントを用いてカレンダーにアクセスできるようになりました。

Notion のデータベースを作成する

Notion でデータベースを作成します。今回は以下のプロパティを持つデータベースを作成しました。

Notion Property
Notion データベースのプロパティ一覧

Notion Property status
「ステータス」プロパティの詳細

Notion Property tag
「タグ」プロパティの詳細

このデータベースに対して API からデータを取得するために、インテグレーションを作成します。

https://www.notion.so/profile/integrations より新しい内部インテグレーションを作成し、「内部インテグレーションシークレット」を取得します。

Notion Integration

次に作成したデータベースに対して、「コネクト」より先程作成した内部インテグレーションを追加します。

Notion Connect

AWS Lambdaにコードをデプロイする

AWS Console にサインインし、Lambda 関数を作成します。

Create lambda function

今回はランタイムを Python 3.12 として作成しました。

自分が作成したスクリプトでは google-api-python-client を利用しているので、Lambda 関数のレイヤーに追加する必要があります。

pip install google-api-python-client -t ./python/lib/python3.12/site-packages
zip -r google-api-python-layer.zip python

ここで作成した google-api-python-layer.zip を Lambda のレイヤーとして追加します。

また、Lambda の環境変数の設定より、Notion の内部インテグレーションシークレットを NOTION_TOKEN として保存しておきます。

次に、作成した Lambda を実行するトリガーを EventBridge に設定すれば完了です。
自分はAWS の無料枠に収まるように、 rate(3 minutes) で実行するように設定しました。

Lambda function summary
構築した Lambda 関数の概要

実際にデプロイしたコード
from google.oauth2 import service_account
from googleapiclient import discovery
import requests
import json
import os
from datetime import datetime, timedelta

NOTION_TOKEN = os.environ["NOTION_TOKEN"]
CREDENTIALS_FILE = "your-service-account-credential.json"

DATABASE_ID = "your-database-id"
CALENDAR_ID = "xxx@group.calendar.google.com"

NOTION_API_URL = f"https://api.notion.com/v1/databases/{DATABASE_ID}/query"

TITLE_PROPERTY_NAME = "タイトル"
DEADLINE_PROPERTY_NAME = "提出期限"
STATUS_PROPERTY_NAME = "ステータス"
TAGS_PROPERTY_NAME = "タグ"

api_call_count = 0  # Global variable to count API calls


class NotionTask:
    def __init__(self, title, deadline, status, tags, url):
        self.title = title
        self.deadline = deadline
        self.status = status
        self.tags = tags
        self.url = url

    @classmethod
    def from_notion_response(cls, result):
        properties = result.get("properties", {})
        title_property = properties.get(TITLE_PROPERTY_NAME, {}).get("title", [])
        deadline_property = (
            properties.get(DEADLINE_PROPERTY_NAME, {}).get("date", {}).get("start")
        )
        status_property = (
            properties.get(STATUS_PROPERTY_NAME, {}).get("status", {}).get("name")
        )
        tags_property = properties.get(TAGS_PROPERTY_NAME, {}).get("multi_select", [])
        url = result.get("url")

        if title_property and deadline_property and status_property:
            title = title_property[0].get("text", {}).get("content")
            tags = [tag.get("name") for tag in tags_property]
            return cls(title, deadline_property, status_property, tags, url)
        return None


def get_notion_data_by_date_range(start: datetime, end: datetime) -> dict:
    global api_call_count
    data = {
        "filter": {
            "and": [
                {
                    "property": DEADLINE_PROPERTY_NAME,
                    "date": {
                        "after": start.isoformat(),
                    },
                },
                {
                    "property": DEADLINE_PROPERTY_NAME,
                    "date": {
                        "before": end.isoformat(),
                    },
                },
            ]
        }
    }

    headers = {
        "Notion-Version": "2022-06-28",
        "Content-Type": "application/json",
        "Authorization": f"Bearer {NOTION_TOKEN}",
    }

    response = requests.post(NOTION_API_URL, headers=headers, data=json.dumps(data))
    api_call_count += 1  # Increment API call count

    if response.status_code != 200:
        raise Exception(f"Failed to fetch data from Notion: {response.text}")

    return response.json()


def create_tasks_from_notion_data(response_data):
    tasks = []
    for result in response_data.get("results", []):
        task = NotionTask.from_notion_response(result)
        if task:
            tasks.append(task)
    return tasks


def get_google_calendar_service():
    scopes = ["https://www.googleapis.com/auth/calendar"]
    credentials = service_account.Credentials.from_service_account_file(
        CREDENTIALS_FILE, scopes=scopes
    )
    return discovery.build(
        "calendar", "v3", credentials=credentials, cache_discovery=False
    )


def sync_events_to_calendar(service, tasks: list[NotionTask]):
    global api_call_count
    min_deadline = (
        min(datetime.fromisoformat(task.deadline) for task in tasks)
        - timedelta(hours=1)
    ).isoformat()

    max_deadline = (
        max(datetime.fromisoformat(task.deadline) for task in tasks)
        + timedelta(hours=1)
    ).isoformat()

    existing_events = (
        service.events()
        .list(
            calendarId=CALENDAR_ID,
            timeMin=min_deadline,
            timeMax=max_deadline,
            singleEvents=True,  # for orderBy
            orderBy="startTime",
            timeZone="Asia/Tokyo",
        )
        .execute()
    )
    api_call_count += 1

    existing_events_url = {
        event.get("location"): event for event in existing_events.get("items", [])
    }

    events_to_update = []
    events_to_insert = []

    def create_event_data(task: NotionTask):
        event_data = {
            "description": f"[{task.status}] {', '.join(task.tags)}",
            "start": {"dateTime": task.deadline, "timeZone": "Asia/Tokyo"},
            "end": {"dateTime": task.deadline, "timeZone": "Asia/Tokyo"},
            "location": task.url,
        }

        # set colorId based on status
        if task.status == "Done":
            event_data["colorId"] = "1"  # Lavender
            event_data["summary"] = f"✅ {task.title}"
        else:
            event_data["colorId"] = "3"  # Normal
            event_data["summary"] = task.title

        return event_data

    def has_event_changes(current_event_body, new_event_data):
        current_start_time = datetime.fromisoformat(
            current_event_body.get("start", {}).get("dateTime")
        )
        new_start_time = datetime.fromisoformat(new_event_data["start"]["dateTime"])

        current_end_time = datetime.fromisoformat(
            current_event_body.get("end", {}).get("dateTime")
        )
        new_end_time = datetime.fromisoformat(new_event_data["end"]["dateTime"])

        description_changed = (
            current_event_body.get("description") != new_event_data["description"]
        )
        start_time_changed = current_start_time != new_start_time
        end_time_changed = current_end_time != new_end_time
        color_id_changed = current_event_body.get("colorId") != new_event_data.get(
            "colorId"
        )

        return (
            description_changed
            or start_time_changed
            or end_time_changed
            or color_id_changed
        )

    for task in tasks:
        task_event_data = create_event_data(task)

        # check if the event already exists in the calendar
        if task.url in existing_events_url:
            current_event_body = existing_events_url[task.url]
            event_id = current_event_body["id"]

            if has_event_changes(current_event_body, task_event_data):
                current_event_body.update(task_event_data)
                events_to_update.append((event_id, current_event_body))
        else:
            events_to_insert.append(task_event_data)

    # update or insert events
    for event_id, current_event_body in events_to_update:
        service.events().update(
            calendarId=CALENDAR_ID, eventId=event_id, body=current_event_body
        ).execute()
        api_call_count += 1

    for task_event_data in events_to_insert:
        service.events().insert(calendarId=CALENDAR_ID, body=task_event_data).execute()
        api_call_count += 1


def lambda_handler(event, context):
    now = datetime.now().astimezone()
    one_year_ago = now - timedelta(days=365)
    one_year_future = now + timedelta(days=365)

    try:
        notion_data = get_notion_data_by_date_range(one_year_ago, one_year_future)
        tasks = create_tasks_from_notion_data(notion_data)
        service = get_google_calendar_service()
        sync_events_to_calendar(service, tasks)
        return {
            "statusCode": 200,
            "body": f"Events created or updated successfully. API calls made: {api_call_count}",
        }
    except Exception as e:
        return {"statusCode": 500, "body": str(e)}


if __name__ == "__main__":
    result = lambda_handler(None, None)
    print(result)

今後の課題

  • 課題を Notion に登録し忘れないように、簡単な操作で登録できるようにしたい
  • 課題のステータスによって Google カレンダーの予定のリマインダーを変更したい
    • まだ「IceBox」にいるのに提出期限が迫っている課題には1日前にリマインド
    • 「In progress」にある課題には提出期限6時間前にリマインド...みたいな?
  • API の呼び出し回数を減らすために、Google Calendar API から取得した情報を DB にキャッシュしておきたい

まとめ

  • Google Calendar API はサービスアカウントを使って簡単に利用できる
  • Notion のデータベースは内部インテグレーションを作成すれば簡単に参照できる
  • AWS Lambda の無料枠はめちゃくちゃ大きい
GitHubで編集を提案

Discussion