複数Googleアカウントのカレンダーを一元管理する方法【OpenClaw連携】

はじめに

仕事用とプライベート用で複数のGoogleアカウントを使い分けている人は多いと思います。しかし、カレンダーを複数開いて確認するのは面倒ですよね。

そこで今回、複数のGoogleカレンダーを自動監視して、新しい予定をリアルタイムで通知するツールを開発しました。OpenClawという個人用AIアシスタントと連携して、Telegramに自動通知します。

開発の背景

私は以下の2つのGoogleアカウントを使い分けています：

1. クロスリンクアカウント（仕事用）

– 会社の会議・MTG

– 顧客との打ち合わせ

2. プログラミングスクールアカウント（副業用）

– 体験会・研修

– リーダーMTG

これらを常に2つのカレンダーを開いて確認するのが非効率だったため、一元管理できるシステムを構築しました。

システム構成

アーキテクチャ

Google Calendar API
    ↓
Python監視スクリプト（30分ごとに実行）
    ↓
OpenClaw（AIアシスタント）
    ↓
Telegram通知

技術スタック

実装のポイント

1. Google Calendar API の認証

Google Calendar APIを使うには、OAuth 2.0認証が必要です。

credentials.json の取得

1. [Google Cloud Console](https://console.cloud.google.com/)にアクセス

2. プロジェクトを作成

3. 「APIとサービス」→「OAuth同意画面」で設定

4. 「認証情報」→「OAuth 2.0 クライアントID」を作成

5. `credentials.json`をダウンロード

認証フロー

from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from googleapiclient.discovery import build

SCOPES = [
    'https://www.googleapis.com/auth/calendar.readonly',  # カレンダー読み取り
    'https://www.googleapis.com/auth/calendar.events'     # イベント作成・編集
]

def authenticate(account: str):
    """
    Googleアカウントを認証してトークンを保存
    
    Args:
        account: 'crosslink' or 'programming_school'
    """
    token_path = f'token_{account}.json'
    creds = None
    
    # 既存のトークンを読み込み
    if os.path.exists(token_path):
        creds = Credentials.from_authorized_user_file(token_path, SCOPES)
    
    # トークンが無効または存在しない場合、再認証
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            # トークンをリフレッシュ
            creds.refresh(Request())
        else:
            # 新規認証（ブラウザが開く）
            flow = InstalledAppFlow.from_client_secrets_file(
                'credentials.json', SCOPES)
            creds = flow.run_local_server(port=0)
        
        # トークンを保存
        with open(token_path, 'w') as token:
            token.write(creds.to_json())
    
    return build('calendar', 'v3', credentials=creds)

ポイント：

2. カレンダーイベントの監視

30分ごとに新しいイベントをチェックして、通知します。

import json
from datetime import datetime, timedelta

STATE_FILE = 'calendar_monitor_state.json'

def load_state():
    """前回のチェック状態を読み込む"""
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE, 'r') as f:
            return json.load(f)
    
    return {
        'last_check': None,
        'seen_events': {}  # {account: [event_id, ...]}
    }

def check_new_events(account: str, state: dict):
    """
    新しいイベントをチェック
    
    Returns:
        新しいイベントのリスト
    """
    # カレンダーサービスを取得
    service = authenticate(account)
    
    # 前回チェック時刻以降のイベントを取得
    last_check = state.get('last_check')
    if last_check:
        time_min = datetime.fromisoformat(last_check).isoformat() + 'Z'
    else:
        # 初回は現在時刻から
        time_min = datetime.utcnow().isoformat() + 'Z'
    
    time_max = (datetime.utcnow() + timedelta(days=30)).isoformat() + 'Z'
    
    # イベント取得
    events_result = service.events().list(
        calendarId='primary',
        timeMin=time_min,
        timeMax=time_max,
        singleEvents=True,
        orderBy='startTime'
    ).execute()
    
    events = events_result.get('items', [])
    
    # 既に見たイベントを除外
    seen_ids = set(state.get('seen_events', {}).get(account, []))
    new_events = [
        event for event in events
        if event['id'] not in seen_ids
    ]
    
    return new_events

def monitor_calendars():
    """全カレンダーを監視"""
    state = load_state()
    
    accounts = ['crosslink', 'programming_school']
    all_new_events = []
    
    for account in accounts:
        print(f"📅 Checking {account}...")
        
        try:
            new_events = check_new_events(account, state)
            
            if new_events:
                print(f"  ✨ {len(new_events)} new event(s)")
                
                # 見たイベントIDを記録
                if account not in state['seen_events']:
                    state['seen_events'][account] = []
                
                for event in new_events:
                    state['seen_events'][account].append(event['id'])
                    all_new_events.append({
                        'account': account,
                        'event': event
                    })
            else:
                print(f"  ℹ️ No new events")
        
        except Exception as e:
            print(f"  ❌ Error: {e}")
    
    # 状態を更新
    state['last_check'] = datetime.utcnow().isoformat()
    save_state(state)
    
    # 通知
    if all_new_events:
        notify_new_events(all_new_events)
    else:
        print("\n✅ No new events to notify")

def save_state(state: dict):
    """状態を保存"""
    with open(STATE_FILE, 'w') as f:
        json.dump(state, f, indent=2)

ポイント：

3. OpenClawへの通知

OpenClawのcron機能を使って、Telegramに自動通知します。

def notify_new_events(events: list):
    """
    新しいイベントをOpenClaw経由でTelegram通知
    
    Args:
        events: [{account: str, event: dict}, ...]
    """
    message = "🆕 **新しいカレンダーイベントが追加されました！**\n\n"
    
    for item in events:
        account = item['account']
        event = item['event']
        
        # 日本語のアカウント名
        account_name = {
            'crosslink': 'クロスリンク',
            'programming_school': 'プログラミングスクール'
        }.get(account, account)
        
        # イベント情報を整形
        summary = event.get('summary', '(タイトルなし)')
        start = event['start'].get('dateTime', event['start'].get('date'))
        
        # 日時をパース
        if 'T' in start:
            # 時刻あり
            dt = datetime.fromisoformat(start.replace('Z', '+00:00'))
            time_str = dt.strftime('%m/%d %H:%M')
        else:
            # 終日イベント
            dt = datetime.fromisoformat(start)
            time_str = dt.strftime('%m/%d')
        
        message += f"  {time_str} - {summary}\n"
        message += f"    ({account_name})\n\n"
    
    # OpenClawのwake APIを呼び出し
    import requests
    
    # OpenClawのgateway URLを環境変数から取得
    gateway_url = os.getenv('OPENCLAW_GATEWAY_URL', 'http://localhost:8765')
    
    response = requests.post(
        f'{gateway_url}/api/wake',
        json={'text': message},
        headers={'Content-Type': 'application/json'}
    )
    
    if response.status_code == 200:
        print("✅ Notification sent to OpenClaw")
    else:
        print(f"❌ Failed to notify: {response.status_code}")

4. Cronによる定期実行

30分ごとに監視スクリプトを実行：

#!/bin/bash
# run_monitor.sh

cd "$(dirname "$0")"
python3 calendar_monitor.py

# crontab -e で設定
*/30 * * * * /path/to/run_monitor.sh >> /path/to/logs/calendar_monitor.log 2>&1

より高度な実装：OpenClaw cronを使う

# OpenClawのcron機能を使えば、より柔軟な設定が可能
openclaw cron add \
  --name "Calendar Monitor" \
  --schedule "*/30 * * * *" \
  --command "python3 /path/to/calendar_monitor.py"

実際の通知例

Telegramに以下のような通知が届きます：

🆕 新しいカレンダーイベントが追加されました！

  03/10 09:30 - 新ラーニング朝会（Google Meet）
    (クロスリンク)

  03/12 14:00 - リーダーMTG
    (プログラミングスクール)

  03/15 13:00 - プロシーズさん定例会
    (クロスリンク)

追加機能

1. 会議前リマインダー

30分前、1時間前に自動リマインド：

def check_upcoming_events():
    """30分以内に始まる予定をチェック"""
    now = datetime.now()
    threshold = now + timedelta(minutes=30)
    
    for account in ['crosslink', 'programming_school']:
        service = authenticate(account)
        
        events_result = service.events().list(
            calendarId='primary',
            timeMin=now.isoformat() + 'Z',
            timeMax=threshold.isoformat() + 'Z',
            singleEvents=True,
            orderBy='startTime'
        ).execute()
        
        events = events_result.get('items', [])
        
        for event in events:
            start = event['start'].get('dateTime')
            if start:
                start_dt = datetime.fromisoformat(start.replace('Z', '+00:00'))
                minutes_until = (start_dt - now).total_seconds() / 60
                
                if 25 <= minutes_until <= 30:
                    notify_reminder(event, int(minutes_until))

def notify_reminder(event: dict, minutes: int):
    """リマインダー通知"""
    summary = event.get('summary', '(タイトルなし)')
    message = f"🔔 **まもなく予定があるよ！（{minutes}分後）**\n\n"
    message += f"📝 {summary}"
    
    # OpenClawに通知
    # ... (省略)

2. 予定のコンフリクト検出

同じ時間に複数の予定が入っている場合に警告：

def detect_conflicts():
    """予定の重複を検出"""
    all_events = []
    
    # 全アカウントのイベントを取得
    for account in ['crosslink', 'programming_school']:
        service = authenticate(account)
        events = get_events(service)
        all_events.extend([(account, e) for e in events])
    
    # 時間でソート
    all_events.sort(key=lambda x: x[1]['start'].get('dateTime', ''))
    
    conflicts = []
    for i in range(len(all_events) - 1):
        event1 = all_events[i][1]
        event2 = all_events[i + 1][1]
        
        end1 = event1['end'].get('dateTime')
        start2 = event2['start'].get('dateTime')
        
        if end1 and start2:
            end1_dt = datetime.fromisoformat(end1)
            start2_dt = datetime.fromisoformat(start2)
            
            if end1_dt > start2_dt:
                conflicts.append((event1, event2))
    
    if conflicts:
        notify_conflicts(conflicts)

セキュリティ対策

1. トークンの安全な保存

# トークンファイルのパーミッションを制限
chmod 600 token_*.json

2. 環境変数での管理

# .env ファイルを使用
OPENCLAW_GATEWAY_URL=http://localhost:8765
OPENCLAW_GATEWAY_TOKEN=your_secret_token_here

from dotenv import load_dotenv

load_dotenv()

gateway_url = os.getenv('OPENCLAW_GATEWAY_URL')
gateway_token = os.getenv('OPENCLAW_GATEWAY_TOKEN')

3. .gitignoreで秘密情報を除外

# Google Calendar認証
credentials.json
token_*.json

# 環境変数
.env

# 状態ファイル
calendar_monitor_state.json

トラブルシューティング

エラー: `invalid_grant`

原因：トークンの有効期限切れ

解決策：

# トークンを削除して再認証
rm token_*.json
python server.py
# → ブラウザで認証

エラー: `API has not been used`

原因：Google Calendar APIが有効化されていない

解決策：

1. Google Cloud Consoleで「APIとサービス」→「有効なAPIとサービス」

2. 「+ APIとサービスの有効化」をクリック

3. 「Google Calendar API」を検索して有効化

まとめ

複数Googleアカウントのカレンダー統合で得られた効果：

→ 全ての予定を一箇所で確認

→ 自動通知で即座に気づく

→ 30分前通知で準備できる

→ ダブルブッキングを防ぐ

Google Calendar APIは非常に強力で、今回紹介した機能以外にも：

など、様々なことが可能です。業務効率化の第一歩として、ぜひ試してみてください！

---

技術スタック: Python, Google Calendar API, OAuth 2.0, OpenClaw

開発期間: 約1週間

コード行数: 約1,000行

対応サービス: Google Calendar