Pythonで実現！Gmailの不要なメールを指定した条件で自動削除して受信トレイをスッキリ保つ方法

google-api-python-client
google-auth-oauthlib
PyYAML

pip install -r requirements.txt

import os
import base64
from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
import pickle
import yaml
import logging
import sys

# ログ設定
logging.basicConfig(
    filename='debug.log',
    level=logging.DEBUG,
    format='%(asctime)s %(levelname)s %(message)s',
    encoding='utf-8'
)

# Gmail APIのスコープ
SCOPES = ['https://www.googleapis.com/auth/gmail.modify']

# 認証とサービスの作成
def gmail_authenticate():
    creds = None
    if os.path.exists('token.pickle'):
        with open('token.pickle', 'rb') as token:
            creds = pickle.load(token)
    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.pickle', 'wb') as token:
            pickle.dump(creds, token)
    return build('gmail', 'v1', credentials=creds)

def extract_body(payload):
    if 'parts' in payload:
        for part in payload['parts']:
            if part.get('mimeType') == 'text/html':
                data = part['body'].get('data')
                if data:
                    return base64.urlsafe_b64decode(data).decode('utf-8')
        for part in payload['parts']:
            if part.get('mimeType') == 'text/plain':
                data = part['body'].get('data')
                if data:
                    return base64.urlsafe_b64decode(data).decode('utf-8')
        for part in payload['parts']:
            body = extract_body(part)
            if body:
                return body
    else:
        if payload.get('mimeType') == 'text/html':
            data = payload['body'].get('data')
            if data:
                return base64.urlsafe_b64decode(data).decode('utf-8')
        elif payload.get('mimeType') == 'text/plain':
            data = payload['body'].get('data')
            if data:
                return base64.urlsafe_b64decode(data).decode('utf-8')
    return None

def delete_message(service, msg_id):
    service.users().messages().trash(userId='me', id=msg_id).execute()

def load_rules(yaml_path):
    with open(yaml_path, 'r', encoding='utf-8') as f:
        return yaml.safe_load(f)

def should_delete_mail(delete_rules, from_email, subject, body):
    import email.utils
    real_addr = None
    try:
        real_addr = email.utils.parseaddr(from_email or '')[1]
    except Exception:
        real_addr = from_email or ''
    for rule in delete_rules:
        emails = [e.strip().lower() for e in rule.get('email', '').split(',') if e.strip()]
        for e in emails:
            if e == real_addr.lower() or e in (from_email or '').lower():
                subjects = rule.get('Subject', [])
                bodys = rule.get('Body', [])
                if isinstance(subjects, str):
                    subjects = [subjects]
                if isinstance(bodys, str):
                    bodys = [bodys]
                if not subjects and not bodys:
                    return True
                if subjects:
                    for s in subjects:
                        if s and (s.lower().strip() in (subject or '').lower()):
                            return True
                if bodys:
                    for b in bodys:
                        if b and (b.lower().strip() in (body or '').lower()):
                            return True
    return False

def process_delete_rules(service, delete_rules):
    for rule in delete_rules:
        emails = [e.strip().lower() for e in rule.get('email', '').split(',') if e.strip()]
        for email_addr in emails:
            query = f'from:{email_addr}'
            messages = []
            page_token = None
            while True:
                params = {'userId': 'me', 'q': query, 'maxResults': 100}
                if page_token:
                    params['pageToken'] = page_token
                results = service.users().messages().list(**params).execute()
                messages.extend(results.get('messages', []))
                page_token = results.get('nextPageToken')
                if not page_token:
                    break
            for msg in messages:
                msg_detail = service.users().messages().get(userId='me', id=msg['id'], format='full').execute()
                headers = msg_detail['payload'].get('headers', [])
                from_email = subject = None
                for h in headers:
                    if h['name'].lower() == 'from':
                        from_email = h['value']
                    if h['name'].lower() == 'subject':
                        subject = h['value']
                body = extract_body(msg_detail['payload'])
                if should_delete_mail([rule], from_email, subject, body):
                    delete_message(service, msg['id'])
                    print(f'From: {from_email}, Subject: {subject} のメールを条件によりゴミ箱に移動しました')

def main():
    with open('debug.log', 'w', encoding='utf-8') as f:
        pass
    rules_path = 'rules.yml'
    rules = load_rules(rules_path) if os.path.exists(rules_path) else {}
    delete_rules = rules.get('delete_rules', [])
    
    service = gmail_authenticate()
    
    if delete_rules:
        process_delete_rules(service, delete_rules)
    
    print('メール削除処理が完了しました。')

if __name__ == '__main__':
    main()

delete_rules:
  # ルール1: newsletter@example.comからのメールをすべて削除
  - email: 'newsletter@example.com'

  # ルール2: promotions@example.comからで、件名に「期間限定」が含まれるメールを削除
  - email: 'promotions@example.com'
    Subject: '期間限定'

  # ルール3: info@example.jpからで、本文に「当選」が含まれるメールを削除
  - email: 'info@example.jp'
    Body: '当選'

python delete_mail.py