SQLiteでメール重複防止：永続化とクエリ効率化のテクニック

はじめに

メール監視システムを15分間隔で自動実行すると、同じメールを何度も通知してしまう問題が発生します。この記事では、SQLiteを使った重複防止の実装を紹介します。

重複問題の本質

Gmail APIでメールを取得すると、過去50件などの範囲で取得します。Cronジョブで15分ごとに実行すると：

前回チェック時のメールが再度取得される
同じメールを繰り返し「重要」と判定してしまう
通知がスパム化する

解決策：SQLite永続化

データベース設計

メールごとにユニークIDを記録し、通知履歴を保存します。

CREATE TABLE IF NOT EXISTS emails (
    id TEXT PRIMARY KEY,
    account TEXT NOT NULL,
    subject TEXT,
    sender TEXT,
    received_at TIMESTAMP,
    importance_level TEXT,
    score INTEGER,
    first_detected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    last_notified_at TIMESTAMP,
    notification_count INTEGER DEFAULT 0
);

CREATE INDEX idx_account_received ON emails(account, received_at);
CREATE INDEX idx_importance ON emails(importance_level);

初期化

import sqlite3
from datetime import datetime

def init_database(db_path='emails.db'):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    cursor.execute('''
        CREATE TABLE IF NOT EXISTS emails (
            id TEXT PRIMARY KEY,
            account TEXT NOT NULL,
            subject TEXT,
            sender TEXT,
            received_at TIMESTAMP,
            importance_level TEXT,
            score INTEGER,
            first_detected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            last_notified_at TIMESTAMP,
            notification_count INTEGER DEFAULT 0
        )
    ''')

    cursor.execute('CREATE INDEX IF NOT EXISTS idx_account_received ON emails(account, received_at)')
    cursor.execute('CREATE INDEX IF NOT EXISTS idx_importance ON emails(importance_level)')

    conn.commit()
    conn.close()

重複チェック

新規メールかどうかを判定し、重複なら既存レコードを返します。

def is_duplicate(email_id, db_path='emails.db'):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    cursor.execute('SELECT * FROM emails WHERE id = ?', (email_id,))
    result = cursor.fetchone()

    conn.close()
    return result is not None

メール登録

新規メールをデータベースに保存します。

def save_email(email_data, db_path='emails.db'):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    cursor.execute('''
        INSERT OR REPLACE INTO emails 
        (id, account, subject, sender, received_at, importance_level, score, last_notified_at, notification_count)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, 
                COALESCE((SELECT notification_count FROM emails WHERE id = ?), 0) + 1)
    ''', (
        email_data['id'],
        email_data['account'],
        email_data['subject'],
        email_data['from'],
        email_data['received_at'],
        email_data['priority'],
        email_data['score'],
        datetime.now().isoformat(),
        email_data['id']  # サブクエリ用
    ))

    conn.commit()
    conn.close()

再通知ロジック

重要メールは一定時間経過後に再通知します。

再通知判定

from datetime import datetime, timedelta

def should_renotify(email_id, priority, db_path='emails.db'):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    cursor.execute('SELECT last_notified_at FROM emails WHERE id = ?', (email_id,))
    result = cursor.fetchone()
    conn.close()

    if not result or not result[0]:
        return True  # 初回通知

    last_notified = datetime.fromisoformat(result[0])
    now = datetime.now()

    # 優先度別の再通知間隔
    intervals = {
        'HIGH': timedelta(hours=2),
        'MEDIUM': timedelta(hours=24)
    }

    interval = intervals.get(priority, timedelta(days=999))  # LOWは再通知なし

    return (now - last_notified) >= interval

統合フロー

def process_emails(emails, account, db_path='emails.db'):
    new_important = []

    for email in emails:
        email_id = email['id']
        priority = email['priority']

        if is_duplicate(email_id, db_path):
            if should_renotify(email_id, priority, db_path):
                save_email(email, db_path)  # 通知カウント更新
                new_important.append(email)
        else:
            save_email(email, db_path)
            new_important.append(email)

    return new_important

パフォーマンス最適化

インデックス活用

頻繁に使うクエリにインデックスを設定します。

-- アカウント + 受信日時での検索
CREATE INDEX idx_account_received ON emails(account, received_at);

-- 重要度での絞り込み
CREATE INDEX idx_importance ON emails(importance_level);

-- 再通知判定
CREATE INDEX idx_last_notified ON emails(last_notified_at);

バッチ処理

複数メールを1トランザクションで処理します。

def save_emails_batch(emails, db_path='emails.db'):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    data = [
        (e['id'], e['account'], e['subject'], e['from'], 
         e['received_at'], e['priority'], e['score'])
        for e in emails
    ]

    cursor.executemany('''
        INSERT OR IGNORE INTO emails 
        (id, account, subject, sender, received_at, importance_level, score)
        VALUES (?, ?, ?, ?, ?, ?, ?)
    ''', data)

    conn.commit()
    conn.close()

運用結果

初回チェック: 64通 → 7通の重要メール検出
2回目以降: 重複除外により通知0件（正常動作）
2時間後: High優先度メール1件を再通知

まとめ

SQLiteを使った永続化により、重複通知を完全に防止できます。再通知ロジックと組み合わせることで、重要メールを適切なタイミングで再度知らせる仕組みも実現できます。

関連記事:
– Gmail重要メール監視システム構築
– 複数Googleアカウント統合管理
– OpenClaw Cronジョブ活用術