はじめに
大学からの重要なメールを見逃してしまった経験はありませんか？授業の休講通知、課題の締切、奨学金の案内など、大学からのメールには重要な情報が多く含まれています。
そこで、Gmail APIとLINE Messaging API、さらにLLM（Gemini）を組み合わせて、大学からのメールを自動でLINEに通知し、さらにAIが内容を要約してくれるシステムを作りました。
実行ログの例
プログラムを起動すると、以下のようなログが表示され、メールの監視が開始されます。
(00:44:04) 大学からの新着メールはありません。
このメッセージは、現時点で新着メールがないことを示しています。メールが届くと、LINEに通知が送信されます。
機能概要
- 📨 指定したドメインからのGmailを定期的に監視
- 📱 新着メールをLINEにブロードキャスト通知
- 🤖 Gemini APIを使ってメール内容を自動要約
- 🔒 個人情報の匿名化機能
- ⏰ 5分間隔での自動チェック（設定変更可能）
動作環境
このプログラムは、PythonがインストールされたPCのローカル環境での実行を前提としています。
クラウドサービス（AWS, GCP, Herokuなど）へのデプロイは行っていませんが、もし継続的な運用が必要な場合は、これらのサービスを利用することで、PCを起動していなくても自動で動作させるように拡張することが可能です。
必要な準備
1. Google Cloud Console
- Gmail APIの有効化
- OAuth 2.0クライアントIDの作成
-
client_secrets.jsonのダウンロード
2. LINE Developers
- LINE Botアカウントの作成
- チャネルアクセストークンの取得
3. Google AI Studio
- Gemini APIキーの取得
環境設定
必要なライブラリのインストール
pip install google-auth google-auth-oauthlib google-auth-httplib2
pip install google-api-python-client
pip install google-generativeai
pip install requests python-dotenv
環境変数の設定（.envファイル）
UNIVERSITY_EMAIL_DOMAIN=@university.ac.jp
LINE_CHANNEL_ACCESS_TOKEN=your_line_channel_access_token
GEMINI_API_KEY=your_gemini_api_key
ソースコード
import os
import time
import json
import requests
import re
import base64
import sys
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
import google.generativeai as genai
from dotenv import load_dotenv
# .envファイルから環境変数を読み込む
load_dotenv()
# --- 設定項目 ---
# Google Cloudからダウンロードした認証情報ファイル
CLIENT_SECRETS_FILE = "client_secrets.json"
# .envファイルから設定を読み込む
UNIVERSITY_EMAIL_DOMAIN = os.environ.get("UNIVERSITY_EMAIL_DOMAIN")
LINE_CHANNEL_ACCESS_TOKEN = os.environ.get("LINE_CHANNEL_ACCESS_TOKEN")
GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY")
# APIが要求する権限の範囲 (スコープ)。今回はGmailの読み取りと変更
SCOPES = ["https://www.googleapis.com/auth/gmail.modify"]
API_SERVICE_NAME = "gmail"
API_VERSION = "v1"
# 何秒おきにメールをチェックするか (秒)
CHECK_INTERVAL_SECONDS = 300 # 300秒 = 5分
# 最後にチェックした時刻を保存するファイル名
TIMESTAMP_FILE = "last_check_timestamp.txt"
# LLMモデルの設定
LLM_MODEL_NAME = 'gemini-1.5-flash'
# --- プログラム本体 ---
def get_gmail_service():
"""Gmail APIの認証を行い、API操作用のサービスオブジェクトを返します。"""
creds = None
token_path = 'token_gmail.json'
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(CLIENT_SECRETS_FILE, SCOPES)
creds = flow.run_local_server(port=0)
with open(token_path, 'w') as token:
token.write(creds.to_json())
return build(API_SERVICE_NAME, API_VERSION, credentials=creds)
def send_line_broadcast_message(message):
"""LINE Messaging APIを使って、友だち全員にメッセージを送信（ブロードキャスト）します。"""
if not LINE_CHANNEL_ACCESS_TOKEN:
print("⚠️ LINEのチャネルアクセストークンが設定されていません。")
return
url = "https://api.line.me/v2/bot/message/broadcast"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {LINE_CHANNEL_ACCESS_TOKEN}"
}
data = {"messages": [{"type": "text", "text": message}]}
try:
response = requests.post(url, headers=headers, data=json.dumps(data))
if response.status_code == 200:
print("📱 LINEに通知を送信しました。")
else:
print(f"❌ LINE通知の送信に失敗: {response.status_code}, {response.text}")
except Exception as e:
print(f"❌ LINE通知中にエラー: {e}")
def anonymize_text(text):
"""個人情報を匿名化します。"""
# メールアドレスを匿名化
text = re.sub(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}', '[メールアドレス]', text)
# 電話番号を匿名化
text = re.sub(r'(\+?\d{1,4}[-.\s]?)?\(?\d{2,}\)?[-.\s]?\d{2,}[-.\s]?\d{2,}[-.\s]?\d{2,}', '[電話番号]', text)
# 住所を匿名化
text = re.sub(r'〒\d{3}-\d{4}\s?[^\s\n]+?市[^\s\n]+?区[^\s\n]+?町[^\s\n]+?\d+', '[住所]', text)
return text
def summarize_with_llm(email_body):
"""LLMを使ってメールの本文を要約します。"""
if not GEMINI_API_KEY:
return None, "⚠️ Gemini APIキーが設定されていません。"
try:
genai.configure(api_key=GEMINI_API_KEY)
model = genai.GenerativeModel(LLM_MODEL_NAME)
# 個人情報を匿名化してからLLMに渡す
anonymized_body = anonymize_text(email_body)
prompt = f"""以下のメール本文を日本語で簡潔に要約してください。重要なポイントやアクション項目があれば含めてください。
---
{anonymized_body}
---
要約："""
response = model.generate_content(prompt)
summary = response.text.replace('*', '') # マークダウンの*を削除
return summary, None
except Exception as e:
return None, f"❌ LLMによる要約中にエラーが発生しました: {e}"
def get_email_body(service, message_id):
"""指定されたメッセージIDのメール本文を抽出します。"""
try:
msg = service.users().messages().get(userId='me', id=message_id, format='full').execute()
if 'parts' in msg['payload']:
# multipart形式の場合
parts = msg['payload']['parts']
for part in parts:
if part['mimeType'] == 'text/plain':
data = part['body']['data']
return base64.urlsafe_b64decode(data.encode('utf-8')).decode('utf-8')
elif 'body' in msg['payload']:
# singlepart形式の場合
data = msg['payload']['body']['data']
return base64.urlsafe_b64decode(data.encode('utf-8')).decode('utf-8')
except Exception as e:
print(f"❌ メール本文の取得中にエラーが発生しました: {e}")
return ""
def check_university_emails(service, last_check_time):
"""指定された時刻以降の、大学からの未読メールをチェックして通知します。"""
try:
query = f"from:{UNIVERSITY_EMAIL_DOMAIN} is:unread after:{last_check_time}"
results = service.users().messages().list(userId='me', q=query).execute()
messages = results.get('messages', [])
if not messages:
print(f"({time.strftime('%H:%M:%S')}) 大学からの新着メールはありません。")
return
print(f"📬 {len(messages)}件の大学からの新着メールを見つけました！")
for message_info in messages:
msg = service.users().messages().get(userId='me', id=message_info['id'], format='full').execute()
headers = msg['payload']['headers']
subject = next((h['value'] for h in headers if h['name'].lower() == 'subject'), '(件名なし)')
sender = next((h['value'] for h in headers if h['name'].lower() == 'from'), '(差出人不明)')
# メール本文を取得してLLMで要約
email_body = get_email_body(service, message_info['id'])
summary, error = summarize_with_llm(email_body)
notification_message = f"""
【大学から新着メール】
━━━━━━━━━━━━
差出人: {sender}
件名: {subject}
━━━━━━━━━━━━
要約:
{summary or "要約できませんでした。"}
"""
if error:
notification_message += f"\n\nエラー: {error}"
send_line_broadcast_message(notification_message)
service.users().messages().modify(
userId='me',
id=message_info['id'],
body={'removeLabelIds': ['UNREAD']}
).execute()
print(f" - 「{subject}」を処理しました。")
except HttpError as error:
print(f"❌ Gmail APIでエラーが発生しました: {error}")
except Exception as e:
print(f"❌ 予期せぬエラーが発生しました: {e}")
def send_test_notification():
"""テスト通知をLINEに送信します。"""
test_message = """
【テスト通知】
━━━━━━━━━━━━
差出人: テスト送信
件名: プログラムの動作確認
━━━━━━━━━━━━
要約:
このメッセージは、LINE通知機能が正常に動作していることを確認するためのテスト通知です。
プログラムが正しく設定されていれば、この通知が表示されます。
"""
print("🚀 テスト通知を送信します...")
send_line_broadcast_message(test_message)
print("✅ テスト通知の送信が完了しました。LINEで確認してください。")
if __name__ == "__main__":
# コマンドライン引数からテストモードを判定
is_test_mode = "--test" in sys.argv
if is_test_mode:
send_test_notification()
sys.exit(0)
# 必須の設定項目が不足しているかチェック
if not UNIVERSITY_EMAIL_DOMAIN or not LINE_CHANNEL_ACCESS_TOKEN:
print("‼️ エラー: .envファイルに設定項目がありません。")
print(" - UNIVERSITY_EMAIL_DOMAIN")
print(" - LINE_CHANNEL_ACCESS_TOKEN")
print("を設定してください。")
else:
print("--- LLM搭載Gmail大学メールLINE通知プログラム ---")
print(f"📨 {UNIVERSITY_EMAIL_DOMAIN} からのメールを監視します。")
print(f"🕒 {CHECK_INTERVAL_SECONDS}秒おきにチェックします。")
print("プログラムを終了するには Ctrl+C を押してください。")
if not GEMINI_API_KEY:
print("⚠️ Gemini APIキーが設定されていません。要約機能は無効です。")
else:
print("🤖 LLMによるメール要約機能付き")
gmail_service = get_gmail_service()
# 起動時に、最後にチェックした時刻をファイルから読み込む
try:
with open(TIMESTAMP_FILE, 'r') as f:
last_timestamp = int(f.read())
except (FileNotFoundError, ValueError):
print("初回起動のため、現在時刻以降のメールを通知対象とします。")
last_timestamp = int(time.time())
while True:
try:
check_university_emails(gmail_service, last_timestamp)
# 現在の時刻を次のチェックのために保存
last_timestamp = int(time.time())
with open(TIMESTAMP_FILE, 'w') as f:
f.write(str(last_timestamp))
time.sleep(CHECK_INTERVAL_SECONDS)
except KeyboardInterrupt:
print("\n👋 プログラムを終了します。")
break
使い方
1. 初回設定
# テスト通知を送信して動作確認
python mail.py --test
2. メール監視開始
# 通常の監視モードで実行
python mail.py
技術的なポイント
Gmail APIの効率的な使用
Gmail APIでは、以下のクエリを使用して効率的にメールを取得しています：
query = f"from:{UNIVERSITY_EMAIL_DOMAIN} is:unread after:{last_check_time}"
これにより、指定したドメインからの未読メールのみを対象とし、前回チェック後の新着メールだけを処理できます。
個人情報の保護
LLMに送信する前に、メール内容から個人情報を自動で匿名化します：
def anonymize_text(text):
# メールアドレス、電話番号、住所を匿名化
text = re.sub(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}', '[メールアドレス]', text)
# ... その他の匿名化処理
return text
エラーハンドリング
各API呼び出しには適切なエラーハンドリングを実装し、一時的な障害でプログラムが停止しないよう配慮しています。
カスタマイズ可能な項目
-
チェック間隔:
CHECK_INTERVAL_SECONDSで調整 - 監視対象ドメイン: 複数ドメインに対応可能
- 要約プロンプト: より詳細な要約や特定の情報抽出が可能
- 通知フォーマット: LINE通知のメッセージ形式をカスタマイズ
注意事項
- Gmail APIとLINE Messaging APIには利用制限があります
- 個人情報を含むメールの取り扱いには十分注意してください
- 本格運用前には十分なテストを行ってください
まとめ
このシステムにより、重要な大学からのメールを見逃すリスクが大幅に減り、さらにAI要約により内容を素早く把握できるようになりました。学生生活をより効率的に送るためのツールとして活用していただければ幸いです。
もしこの記事が役に立ったら、いいねやコメントをお願いします！
参考リンク
コード全体
詳細なコードは、https://github.com/ishikawamasahito/LLM-equipped-Gmail-University-Email-LINE-Notification-Program で公開しています。