Python自動化マスター第6編：メールと通知の自動化
Python Automation Master Part 6: Email and Notification Automation
はじめに：通知自動化の必要性
自動化システムの完成度は通知機能で決まります。どんなに優れた自動化スクリプトを作っても、その結果をタイムリーに確認できなければ意味が半減します。サーバー障害、バッチ処理完了、日次レポート送信など、様々な状況で適切な通知を受け取ることは、業務効率を大幅に向上させます。
今回の第6編では、Pythonを活用して様々なチャネルに通知を送る方法を学びます。伝統的なメールからSlack、Telegram、Discordまで、実務で最も多く使用される通知方法をすべて取り上げます。
1. smtplibでメールを送信する
Pythonの組み込みライブラリであるsmtplibを使用すると、追加インストールなしでメールを送信できます。
1.1 基本的なメール送信
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
def send_simple_email(sender, password, recipient, subject, body):
"""シンプルなテキストメールの送信"""
# メッセージオブジェクトの作成
msg = MIMEMultipart()
msg['From'] = sender
msg['To'] = recipient
msg['Subject'] = subject
# 本文の追加
msg.attach(MIMEText(body, 'plain', 'utf-8'))
try:
# Gmail SMTPサーバーに接続
server = smtplib.SMTP('smtp.gmail.com', 587)
server.starttls() # TLS暗号化
server.login(sender, password)
# メール送信
server.send_message(msg)
server.quit()
print("メール送信成功！")
return True
except Exception as e:
print(f"メール送信失敗: {e}")
return False
# 使用例
send_simple_email(
sender='your_email@gmail.com',
password='アプリパスワード', # 2段階認証後のアプリパスワードを使用
recipient='recipient@example.com',
subject='[自動化] 日次レポート',
body='本日の作業が正常に完了しました。'
)
2. Gmail/Yahoo SMTP設定
2.1 Gmail SMTP設定
Gmailを使用するには、まずアプリパスワードを生成する必要があります。
- Googleアカウント設定にアクセス
- セキュリティタブで2段階認証を有効化
- アプリパスワードを生成（その他のアプリを選択）
- 生成された16桁のパスワードをコードで使用
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
class GmailSender:
"""Gmail SMTPを使用したメール送信クラス"""
def __init__(self, email, app_password):
self.email = email
self.password = app_password
self.smtp_server = 'smtp.gmail.com'
self.smtp_port = 587
def send(self, to, subject, body, html=False):
"""メール送信"""
msg = MIMEMultipart('alternative')
msg['From'] = self.email
msg['To'] = to if isinstance(to, str) else ', '.join(to)
msg['Subject'] = subject
# テキストまたはHTML本文
content_type = 'html' if html else 'plain'
msg.attach(MIMEText(body, content_type, 'utf-8'))
with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
server.starttls()
server.login(self.email, self.password)
# 複数の受信者に送信
recipients = [to] if isinstance(to, str) else to
server.send_message(msg)
return True
# 使用例
gmail = GmailSender('your_email@gmail.com', 'アプリパスワード')
gmail.send(
to='recipient@example.com',
subject='テストメール',
body='Pythonから送信したメールです。'
)
2.2 Yahoo SMTP設定
class YahooSender:
"""Yahoo SMTPを使用したメール送信クラス"""
def __init__(self, email, password):
self.email = email
self.password = password
self.smtp_server = 'smtp.mail.yahoo.co.jp'
self.smtp_port = 587
def send(self, to, subject, body, html=False):
"""メール送信"""
msg = MIMEMultipart('alternative')
msg['From'] = self.email
msg['To'] = to if isinstance(to, str) else ', '.join(to)
msg['Subject'] = subject
content_type = 'html' if html else 'plain'
msg.attach(MIMEText(body, content_type, 'utf-8'))
with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
server.starttls()
server.login(self.email, self.password)
server.send_message(msg)
return True
yahoo = YahooSender('your_id@yahoo.co.jp', 'Yahoo_パスワード')
yahoo.send(
to='recipient@example.com',
subject='Yahooメールテスト',
body='Yahoo SMTPから送信したメールです。'
)
3. メール本文の作成（テキスト、HTML）
3.1 HTML形式のメール
def create_html_email():
"""HTML形式のメールを作成"""
html_content = """
<!DOCTYPE html>
<html>
<head>
<style>
body { font-family: 'メイリオ', sans-serif; }
.header { background-color: #2c3e50; color: white; padding: 20px; }
.content { padding: 20px; }
.footer { background-color: #ecf0f1; padding: 10px; font-size: 12px; }
table { border-collapse: collapse; width: 100%; }
th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
th { background-color: #3498db; color: white; }
</style>
</head>
<body>
<div class="header">
<h1>日次業務レポート</h1>
</div>
<div class="content">
<p>お疲れ様です。</p>
<p>本日の業務処理結果をお知らせします。</p>
<h2>処理結果</h2>
<table>
<tr>
<th>項目</th>
<th>件数</th>
<th>状態</th>
</tr>
<tr>
<td>データ処理</td>
<td>1,234件</td>
<td>完了</td>
</tr>
<tr>
<td>レポート生成</td>
<td>15件</td>
<td>完了</td>
</tr>
</table>
</div>
<div class="footer">
<p>このメールは自動送信されました。</p>
</div>
</body>
</html>
"""
return html_content
# HTML メール送信
gmail = GmailSender('your_email@gmail.com', 'アプリパスワード')
gmail.send(
to='recipient@example.com',
subject='[日次レポート] 2026-01-22',
body=create_html_email(),
html=True
)
4. 添付ファイル付きメール
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders
import os
def send_email_with_attachment(sender, password, recipient, subject, body, files):
"""添付ファイル付きメールを送信"""
msg = MIMEMultipart()
msg['From'] = sender
msg['To'] = recipient
msg['Subject'] = subject
# 本文追加
msg.attach(MIMEText(body, 'plain', 'utf-8'))
# 添付ファイル追加
for file_path in files:
if os.path.exists(file_path):
with open(file_path, 'rb') as f:
part = MIMEBase('application', 'octet-stream')
part.set_payload(f.read())
encoders.encode_base64(part)
filename = os.path.basename(file_path)
part.add_header(
'Content-Disposition',
f'attachment; filename="{filename}"'
)
msg.attach(part)
with smtplib.SMTP('smtp.gmail.com', 587) as server:
server.starttls()
server.login(sender, password)
server.send_message(msg)
print("添付ファイル付きメール送信完了")
# 使用例
send_email_with_attachment(
sender='your_email@gmail.com',
password='アプリパスワード',
recipient='recipient@example.com',
subject='月次レポート添付',
body='月次レポートを添付します。ご確認ください。',
files=['report.xlsx', 'summary.pdf']
)
5. Slack Webhook通知
import requests
import json
class SlackNotifier:
"""Slack Webhook通知クラス"""
def __init__(self, webhook_url):
self.webhook_url = webhook_url
def send_message(self, text, channel=None, username=None, icon_emoji=None):
"""シンプルなメッセージを送信"""
payload = {"text": text}
if channel:
payload["channel"] = channel
if username:
payload["username"] = username
if icon_emoji:
payload["icon_emoji"] = icon_emoji
response = requests.post(
self.webhook_url,
json=payload,
headers={'Content-Type': 'application/json'}
)
return response.status_code == 200
def send_rich_message(self, title, text, color="good", fields=None):
"""リッチフォーマットメッセージを送信"""
attachment = {
"fallback": title,
"color": color, # good(緑), warning(黄), danger(赤), または hex color
"title": title,
"text": text,
"fields": fields or [],
"footer": "自動通知システム",
"ts": int(time.time())
}
payload = {"attachments": [attachment]}
response = requests.post(
self.webhook_url,
json=payload
)
return response.status_code == 200
# 使用例
slack = SlackNotifier("https://hooks.slack.com/services/YOUR/WEBHOOK/URL")
# シンプルメッセージ
slack.send_message("サーバー監視: すべて正常に動作しています ✅")
# リッチメッセージ
slack.send_rich_message(
title="日次バッチ処理完了",
text="すべての処理が正常に完了しました。",
color="good",
fields=[
{"title": "処理件数", "value": "1,234件", "short": True},
{"title": "所要時間", "value": "15分32秒", "short": True}
]
)
6. Telegramボット通知
import requests
class TelegramNotifier:
"""Telegramボット通知クラス"""
def __init__(self, bot_token, chat_id):
self.bot_token = bot_token
self.chat_id = chat_id
self.base_url = f"https://api.telegram.org/bot{bot_token}"
def send_message(self, text, parse_mode='HTML'):
"""テキストメッセージを送信"""
url = f"{self.base_url}/sendMessage"
payload = {
"chat_id": self.chat_id,
"text": text,
"parse_mode": parse_mode
}
response = requests.post(url, json=payload)
return response.json()
def send_document(self, file_path, caption=None):
"""ファイルを送信"""
url = f"{self.base_url}/sendDocument"
with open(file_path, 'rb') as f:
files = {"document": f}
data = {
"chat_id": self.chat_id,
"caption": caption or ""
}
response = requests.post(url, data=data, files=files)
return response.json()
def send_photo(self, image_path, caption=None):
"""画像を送信"""
url = f"{self.base_url}/sendPhoto"
with open(image_path, 'rb') as f:
files = {"photo": f}
data = {
"chat_id": self.chat_id,
"caption": caption or ""
}
response = requests.post(url, data=data, files=files)
return response.json()
# 使用例
telegram = TelegramNotifier(
bot_token="YOUR_BOT_TOKEN",
chat_id="YOUR_CHAT_ID"
)
# HTMLフォーマットメッセージ
telegram.send_message("""
サーバーアラート
状態: 正常
CPU: 45%
メモリ: 62%
ディスク: 38%
詳細はダッシュボードをご覧ください。
""")
7. Discord Webhook通知
import requests
from datetime import datetime
class DiscordNotifier:
"""Discord Webhook通知クラス"""
def __init__(self, webhook_url):
self.webhook_url = webhook_url
def send_message(self, content, username=None, avatar_url=None):
"""シンプルメッセージを送信"""
payload = {"content": content}
if username:
payload["username"] = username
if avatar_url:
payload["avatar_url"] = avatar_url
response = requests.post(self.webhook_url, json=payload)
return response.status_code == 204
def send_embed(self, title, description, color=0x3498db, fields=None,
footer=None, thumbnail_url=None):
"""Embedメッセージを送信"""
embed = {
"title": title,
"description": description,
"color": color,
"timestamp": datetime.utcnow().isoformat(),
"fields": fields or []
}
if footer:
embed["footer"] = {"text": footer}
if thumbnail_url:
embed["thumbnail"] = {"url": thumbnail_url}
payload = {"embeds": [embed]}
response = requests.post(self.webhook_url, json=payload)
return response.status_code == 204
# 使用例
discord = DiscordNotifier("https://discord.com/api/webhooks/YOUR/WEBHOOK")
# シンプルメッセージ
discord.send_message("バッチ処理が完了しました！ 🎉")
# Embedメッセージ
discord.send_embed(
title="システム状態レポート",
description="すべてのサービスが正常に動作しています。",
color=0x2ecc71, # 緑色
fields=[
{"name": "API Server", "value": "稼働中 ✅", "inline": True},
{"name": "Database", "value": "稼働中 ✅", "inline": True},
{"name": "最終確認", "value": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "inline": False}
],
footer="自動監視システム"
)
8. 統合通知システムの構築
from abc import ABC, abstractmethod
import logging
class NotificationChannel(ABC):
"""通知チャネルの抽象基底クラス"""
@abstractmethod
def send(self, message: str, **kwargs) -> bool:
pass
class NotificationManager:
"""統合通知管理クラス"""
def __init__(self):
self.channels = {}
self.logger = logging.getLogger(__name__)
def add_channel(self, name: str, channel: NotificationChannel):
"""通知チャネルを追加"""
self.channels[name] = channel
def notify(self, message: str, channels: list = None, **kwargs):
"""指定されたチャネルに通知を送信"""
target_channels = channels or list(self.channels.keys())
results = {}
for name in target_channels:
if name in self.channels:
try:
success = self.channels[name].send(message, **kwargs)
results[name] = success
self.logger.info(f"{name}通知: {'成功' if success else '失敗'}")
except Exception as e:
results[name] = False
self.logger.error(f"{name}通知エラー: {e}")
return results
def notify_all(self, message: str, **kwargs):
"""すべてのチャネルに通知を送信"""
return self.notify(message, channels=None, **kwargs)
# 使用例
manager = NotificationManager()
manager.add_channel('email', EmailChannel(...))
manager.add_channel('slack', SlackChannel(...))
manager.add_channel('telegram', TelegramChannel(...))
# すべてのチャネルに送信
manager.notify_all("システムアラート: サーバーを再起動しました")
# 特定のチャネルのみ
manager.notify("緊急アラート", channels=['slack', 'telegram'])
まとめ
今回の編では、Pythonを活用したメールと通知の自動化について学びました。
- smtplib：基本的なメール送信に使用
- Slack Webhook：チーム協業ツールへの通知
- Telegram Bot：個人/グループへの即座の通知
- Discord Webhook：コミュニティ通知に最適
実務では状況に応じて適切なチャネルを選択し、緊急度に応じて複数のチャネルを組み合わせて使用することをお勧めします。次回の第7編では、API活用とデータ収集について学びます。