#!/usr/bin/env python3 """ Email to Discord Webhook Forwarder メールサーバーを監視してDiscordに転送するアプリケーション """ import os import sys import time import imaplib import email import json import logging import requests from datetime import datetime from typing import Dict, List, Optional import signal import threading from dataclasses import dataclass # ログ設定 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(sys.stdout) ] ) logger = logging.getLogger(__name__) @dataclass class EmailMessage: """メールメッセージのデータクラス""" subject: str sender: str date: str body: str uid: str class EmailMonitor: """メールサーバー監視クラス""" def __init__(self): self.running = False self.mail_client = None self.last_processed_uid = None # 環境変数から設定を取得 self.imap_server = os.getenv('IMAP_SERVER', 'imap.gmail.com') self.imap_port = int(os.getenv('IMAP_PORT', '993')) self.email_user = os.getenv('EMAIL_USER') self.email_password = os.getenv('EMAIL_PASSWORD') self.discord_webhook_url = os.getenv('DISCORD_WEBHOOK_URL') self.check_interval = int(os.getenv('CHECK_INTERVAL', '60')) # 秒 self.mailbox = os.getenv('MAILBOX', 'INBOX') self.use_ssl = os.getenv('USE_SSL', 'true').lower() == 'true' # 設定の検証 self._validate_config() def _validate_config(self): """設定の検証""" required_vars = ['EMAIL_USER', 'EMAIL_PASSWORD', 'DISCORD_WEBHOOK_URL'] missing_vars = [var for var in required_vars if not os.getenv(var)] if missing_vars: logger.error(f"必要な環境変数が設定されていません: {', '.join(missing_vars)}") sys.exit(1) logger.info("設定の検証が完了しました") def connect_to_email(self) -> bool: """メールサーバーに接続""" try: if self.use_ssl: self.mail_client = imaplib.IMAP4_SSL(self.imap_server, self.imap_port) else: self.mail_client = imaplib.IMAP4(self.imap_server, self.imap_port) self.mail_client.login(self.email_user, self.email_password) self.mail_client.select(self.mailbox) logger.info(f"メールサーバーに接続しました: {self.imap_server}:{self.imap_port}") return True except Exception as e: logger.error(f"メールサーバーへの接続に失敗しました: {str(e)}") return False def disconnect_from_email(self): """メールサーバーから切断""" if self.mail_client: try: self.mail_client.close() self.mail_client.logout() logger.info("メールサーバーから切断しました") except Exception as e: logger.warning(f"メールサーバーからの切断でエラーが発生しました: {str(e)}") def get_latest_email_uid(self) -> Optional[str]: """最新のメールUIDを取得""" try: typ, data = self.mail_client.search(None, 'ALL') if typ != 'OK' or not data[0]: logger.warning("メールボックスが空です") return None email_ids = data[0].split() return email_ids[-1].decode() # 最新のUIDを取得 except Exception as e: logger.error(f"最新メールUIDの取得に失敗しました: {str(e)}") return None def get_new_emails_since_uid(self, last_uid: str) -> List[EmailMessage]: """指定したUID以降の新しいメールを取得""" try: typ, data = self.mail_client.search(None, f'UID {last_uid}:*') if typ != 'OK': logger.warning("メール検索に失敗しました") return [] email_ids = data[0].split()[1:] # 最初のUIDは除外 new_emails = [] for email_id in email_ids: try: typ, msg_data = self.mail_client.fetch(email_id, '(RFC822)') if typ != 'OK': continue email_body = msg_data[0][1] email_message = email.message_from_bytes(email_body) parsed_email = self._parse_email(email_message, email_id.decode()) if parsed_email: new_emails.append(parsed_email) except Exception as e: logger.error(f"メール解析エラー (ID: {email_id}): {str(e)}") continue return new_emails except Exception as e: logger.error(f"新しいメールの取得に失敗しました: {str(e)}") return [] def get_new_emails(self) -> List[EmailMessage]: """新しいメールを取得""" try: # UNSEENフラグのメールを検索 typ, data = self.mail_client.search(None, 'UNSEEN') if typ != 'OK': logger.warning("メール検索に失敗しました") return [] email_ids = data[0].split() new_emails = [] for email_id in email_ids: try: # メールを取得 typ, msg_data = self.mail_client.fetch(email_id, '(RFC822)') if typ != 'OK': continue email_body = msg_data[0][1] email_message = email.message_from_bytes(email_body) # メールデータを解析 parsed_email = self._parse_email(email_message, email_id.decode()) if parsed_email: new_emails.append(parsed_email) except Exception as e: logger.error(f"メール解析エラー (ID: {email_id}): {str(e)}") continue return new_emails except Exception as e: logger.error(f"新しいメールの取得に失敗しました: {str(e)}") return [] def _parse_email(self, email_message, uid: str) -> Optional[EmailMessage]: """メールメッセージを解析""" try: # ヘッダー情報を取得 subject = self._decode_header(email_message.get('Subject', '')) sender = self._decode_header(email_message.get('From', '')) date = email_message.get('Date', '') # メール本文を取得 body = self._get_email_body(email_message) return EmailMessage( subject=subject, sender=sender, date=date, body=body, uid=uid ) except Exception as e: logger.error(f"メール解析エラー: {str(e)}") return None def _decode_header(self, header: str) -> str: """メールヘッダーをデコード""" if not header: return '' try: decoded_header = email.header.decode_header(header) result = '' for text, encoding in decoded_header: if isinstance(text, bytes): if encoding: text = text.decode(encoding, errors='ignore') else: text = text.decode('utf-8', errors='ignore') result += text return result except Exception as e: logger.warning(f"ヘッダーデコードエラー: {str(e)}") return str(header) def _get_email_body(self, email_message) -> str: """メール本文を取得""" body = '' try: if email_message.is_multipart(): for part in email_message.walk(): content_type = part.get_content_type() content_disposition = str(part.get('Content-Disposition')) if content_type == 'text/plain' and 'attachment' not in content_disposition: charset = part.get_content_charset() or 'utf-8' body_bytes = part.get_payload(decode=True) if body_bytes: body = body_bytes.decode(charset, errors='ignore') break else: charset = email_message.get_content_charset() or 'utf-8' body_bytes = email_message.get_payload(decode=True) if body_bytes: body = body_bytes.decode(charset, errors='ignore') # 本文が長い場合は切り詰める（Discord制限対策） if len(body) > 1900: body = body[:1900] + '\n...(本文が切り詰められました)' return body except Exception as e: logger.warning(f"メール本文取得エラー: {str(e)}") return '本文の取得に失敗しました' def send_to_discord(self, email_msg: EmailMessage) -> bool: """DiscordにWebhookでメールを送信""" try: # Discord Embed形式でメッセージを構築 embed = { "title": "📧 新しいメール", "color": 0x3498db, "fields": [ { "name": "件名", "value": email_msg.subject or "件名なし", "inline": False }, { "name": "送信者", "value": email_msg.sender or "送信者不明", "inline": True }, { "name": "日時", "value": email_msg.date or "日時不明", "inline": True }, { "name": "本文", "value": email_msg.body[:1000] if email_msg.body else "本文なし", "inline": False } ], "timestamp": datetime.utcnow().isoformat() } webhook_data = { "username": "Email Monitor", "avatar_url": "https://cdn-icons-png.flaticon.com/512/732/732200.png", "embeds": [embed] } response = requests.post( self.discord_webhook_url, json=webhook_data, timeout=10 ) if response.status_code == 204: logger.info(f"Discordにメールを送信しました: {email_msg.subject}") return True else: logger.error(f"Discord送信エラー: {response.status_code} - {response.text}") return False except Exception as e: logger.error(f"Discord送信エラー: {str(e)}") return False def start_monitoring(self): """メール監視を開始""" logger.info("メール監視を開始します...") self.running = True while self.running: try: if not self.connect_to_email(): logger.warning(f"{self.check_interval}秒後に再試行します...") time.sleep(self.check_interval) continue if self.last_processed_uid is None: self.last_processed_uid = self.get_latest_email_uid() logger.info(f"初期UID: {self.last_processed_uid}") new_emails = self.get_new_emails_since_uid(self.last_processed_uid) if new_emails: logger.info(f"{len(new_emails)}件の新しいメールが見つかりました") for email_msg in new_emails: if self.send_to_discord(email_msg): logger.info(f"処理完了: {email_msg.subject}") self.last_processed_uid = email_msg.uid # 最後に処理したUIDを更新 else: logger.warning(f"Discord送信失敗: {email_msg.subject}") else: logger.debug("新しいメールはありません") self.disconnect_from_email() logger.debug(f"{self.check_interval}秒後に次のチェックを実行します") time.sleep(self.check_interval) except KeyboardInterrupt: logger.info("キーボード割り込みを受信しました") break except Exception as e: logger.error(f"監視ループでエラーが発生しました: {str(e)}") self.disconnect_from_email() time.sleep(self.check_interval) logger.info("メール監視を停止しました") def stop_monitoring(self): """メール監視を停止""" self.running = False self.disconnect_from_email() def signal_handler(signum, frame, monitor): """シグナルハンドラー""" logger.info(f"シグナル {signum} を受信しました。アプリケーションを終了します...") monitor.stop_monitoring() sys.exit(0) def main(): """メイン関数""" logger.info("Email to Discord Webhook Forwarder を開始します") logger.info(f"チェック間隔: {os.getenv('CHECK_INTERVAL', '60')}秒") logger.info(f"メールボックス: {os.getenv('MAILBOX', 'INBOX')}") logger.info(f"IMAPサーバー: {os.getenv('IMAP_SERVER', 'imap.gmail.com')}") # メール監視インスタンスを作成 monitor = EmailMonitor() # シグナルハンドラーを設定 signal.signal(signal.SIGINT, lambda s, f: signal_handler(s, f, monitor)) signal.signal(signal.SIGTERM, lambda s, f: signal_handler(s, f, monitor)) try: # 監視開始 monitor.start_monitoring() except Exception as e: logger.error(f"アプリケーションエラー: {str(e)}") sys.exit(1) if __name__ == "__main__": main()