Pythonでsmtplibを使ってメールを送信する方法

このシリーズでは、色々なプログラムでメールを送る方法を紹介していきます。今回は第3回として、Pythonでメールを送る方法を紹介します。

環境情報

本記事では、メールサーバとは別のサーバでプログラムを実行します。
メールサーバに接続可能でプログラムが実行できる環境であれば、クライアントPCやメールサーバと同一環境でも問題ありません。

プログラムを実行可能なサーバやクライアントPC

実行環境のOSは「AlmaLinux 8.10」を使用しています。

# cat /etc/redhat-release
AlmaLinux release 8.10 (Cerulean Leopard)

メールサーバ

OSは「AlmaLinux 8.9」、SMTPサーバとして「postfix 3.5.8」を使用しています。

# cat /etc/redhat-release
AlmaLinux release 8.9 (Midnight Oncilla)

# postconf | grep mail_version
mail_version = 3.5.8
milter_macro_v = $mail_name $mail_version

プログラムの実行環境とメールサーバ間は587ポートで接続し、SMTP AUTHで認証します。

シンプルですが今回の構成を図にするとこんな感じです。

Pythonでメールを送信する場合の選択肢

Pythonはシンプルで読みやすい構文を持つ、汎用性の高いプログラミング言語です。ライブラリやフレームワークも豊富で、Webアプリ開発、データ分析、AI・機械学習など、用途に合わせた機能の拡充が可能です。
また、オープンソースでコミュニティも活発であり、プログラム初心者から専門家まで幅広い層に人気の高いプログラミング言語ですので、まだ触れたことのない方はこれを機に触れてみてはいかがでしょうか。

Pythonを使ってメールを送信する方法はいくつかあります。以下に主要な選択肢と、それぞれのメリット・デメリットを示します。

smtplib

Pythonの標準ライブラリであるsmptlibを使うことで、SMTPサーバに接続してメールを送信することができます。

メリット

標準ライブラリを使うため、外部パッケージのインストールが不要
基本的なSMTP接続の知識があれば簡単に使える

デメリット

認証やセキュリティに関する細かい設定が手動で必要
手動でヘッダーや本文を構築する必要があるため、複雑なメールの作成には手間がかかる
大量のメール送信には向いていない

Yagmail

Yagmailは、Gmailを使ったメール送信を簡単にするためのPythonライブラリです。

メリット

GmailのAPIを使うことで、より簡単にメールを送信できる
Gmail用の設定が容易で、セキュリティを高めつつパスワード管理を簡略化できる

デメリット

Gmailに最適化されているので、他のメールサーバーを使う場合は設定がやや複雑になる
大量のメール送信には向いていない

DjangoやFlaskのメール送信機能を使用

Webフレームワークに付属するメール送信機能を利用することもできます。

メリット

Webアプリケーション開発と統合でき、フォームや認証システムと連携しやすい
設定が簡単
フレームワークの設定を使えるため学習コストが低い

デメリット

Webアプリケーションと無関係のプロジェクトには適さない

Python・smtplibでメールを送ってみる

「smtplib」は必要最低限の機能に絞られてはいるものの、SMTPサーバとの通信という意味では必要な機能は十分です。複雑な仕組みを必要としない点や、コードをシンプルにしたいという点から、今回はsmtplibを使用します。

本記事で使用するバージョンは3.11.9になります。 Python3系では問題ないと思いますが、Python2系では動かない可能性があります。

# python3 --version
Python 3.11.9

smtplibは標準ライブラリに含まれていますので、別途インストールする必要はありません。

SMTP設定ファイルの作成

まずはSMTP接続情報やヘッダFrom、エンベロープFromを記載するための設定ファイルを作成します。

ファイル名は「smtp_config.py」としています。

# vi smtp_config.py

※ここから
-------
config = {
    'host': 'SERVER_HOST_NAME',        // SMTPサーバのホスト名
    'username': 'SMTP_USERNAME,       //SMTPユーザ名
    'password': 'SMTP_PASSWORD',       // SMTPパスワード
    'port': 587,                           // SMTPポート番号
    'use_tls': False,                         //TLSの使用有無。使用する場合はTrue
    'from_email': 'HEADER_FROM_ADD',   // ヘッダFromメールアドレス
    'from_name': 'YOUR_NAME',           // ヘッダFrom表示名
    'envelope_from': 'ENVELOPE_FROM_ADD', // エンベロープFromアドレス
}
-------
※ここまで

Pythonのプログラムの作成

メールを送るプログラム本体を作っていきます。

本記事では「send_mail.py」という名前で作成しますが、他の名前でも問題ありません。

# vi send_mail.py

※ここから
-------
import smtplib
from email.message import EmailMessage
from smtp_config import config
import uuid

def send_email(to_address, subject, message):

    msg = EmailMessage()
    msg.set_content(message)
    msg['Subject'] = subject
    msg['From'] = f"{config['from_name']} <{config['from_email']}>"
    msg['To'] = to_address
    msg['Message-ID'] = f"<{uuid.uuid4()}@{config['host']}>"

    # SMTPサーバへの接続とメールの送信
    try:
        with smtplib.SMTP(config['host'], config['port']) as server:
            if config['use_tls']:
                server.starttls()  # TLSの開始
            server.login(config['username'], config['password'])
            server.send_message(msg, from_addr=config['envelope_from'])
        print(f"Message has been sent to {to_address}")
    except Exception as e:
        print(f"Message could not be sent. Error: {e}")

if __name__ == "__main__":
    import sys
    if len(sys.argv) != 4:
        print("Usage: python send_mail.py   ")
        sys.exit(1)

    to_address = sys.argv[1]
    subject = sys.argv[2]
    message = sys.argv[3]

    send_email(to_address, subject, message)

-------
※ここまで

実行

保存したら、早速実行してみましょう！

# python3 send_mail.py "xxxxxxx@xxx.xxx" "Test Mail from Python" "Python"

実行コマンドが「python3」となっていますが、環境によっては「python」となっていたり、「python39」や「python3.11」となっている場合があります。

実行が出来ない場合は、pythonコマンド名が間違っている可能性があるため、利用中の環境で確認して下さい。

本記事で使用している環境では、下記のコマンドで調べることが出来ます。

# ll /usr/bin/python*
lrwxrwxrwx. 1 root root   25  7月  1 10:42 /usr/bin/python3 -> /etc/alternatives/python3
-rwxr-xr-x. 1 root root 7752  6月 24 19:23 /usr/bin/python3.11

正常に実行されると、下記のメッセージが出力されます。

# python3 send_mail.py "xxxxxxx@xxx.xxx" "Test Mail from Python" "Python"
Message has been sent to xxxxxxx@xxx.xxx

正常に送れたらメールボックスを確認します。

こちらもちゃんと届いていますね。

注意事項

今回作成したプログラムではSMTP接続に失敗した場合でもリトライされません。
また、エラー処理も最低限となっておりますので、実際に業務やサービスとして使用される場合は、SMTP接続に失敗した場合のリトライ処理、プログラムの再実行処理、エラー処理やログの生成などのカスタマイズが必要になります。

具体的な利用例

最後に、作成したスクリプトの活用例として、エラーログを検出した際にメールを送信するサンプルコードを紹介します。

エラーログが出力された場合、エラーが発生したことを何かしらの方法で通知することは、いち早く障害発生に気付くために大切な仕組みです。

利便性と汎用性の観点から、「エラーログを検出するプログラム」と「メールを送信するプログラム」を別にするケースが多いと思います。以下に用意したサンプルコードでは、「メールを送信するプログラム」は本記事で作成したプログラムと定義ファイルをそのまま利用し、「エラーログを検出するプログラム」から呼び出すことで自動的にメールを送る仕組みになっています。

1. システムエラー通知のメール送信

本記事では「エラーログを検出するプログラム」を「err_chk.py」という名前で作成していますが、任意の名前で問題ありません。

# vi err_chk.py

※ここから
-------
import os

LOG_FILE = "/var/log/test_log"                      # 監視するログファイル
KEYWORDS = ["TEST", "ERROR", "FATAL", "CRITICAL"]   # 検知したいキーワード
STATE_FILE = "/tmp/log_monitor_state.txt"           # 最後に読んだ行の位置を記録するファイル
MAIL_SCRIPT = "/home/link/mail/send_mail.py"        # メール送信スクリプトの保存場所
SYSTEM_NAME = "System Monitoring"                   # 通知時に表示するシステム名
PROCESS_NAME = "MySystem"                           # 監視しているシステム名
TO_ADDRESS = "admin@example.com"                    # 通知を送るメールアドレス


def get_last_position(state_file):
    """
    前回の実行時に読んだログファイルの行数を取得。

    Args:
        state_file (str): 状態を記録するファイルのパス。

    Returns:
        int: 前回読んだ行数。存在しない場合は0。
    """
    if os.path.exists(state_file):
        with open(state_file, "r") as f:
            return int(f.read().strip())
    return 0


def save_last_position(state_file, position):
    """
    ログファイルの最後の行の位置を記録。

    Args:
        state_file (str): 状態を記録するファイルのパス。
        position (int): 現在のログファイルの行番号。
    """
    with open(state_file, "w") as f:
        f.write(str(position))


def detect_errors(log_file, keywords, last_position):
    """
    ログファイルを読み取り、エラーを検知。

    Args:
        log_file (str): 監視するログファイルのパス。
        keywords (list): 検知するキーワードのリスト。
        last_position (int): 前回の読み取り行数。

    Returns:
        list: 検知したエラー行のリスト。
        int: 最後に読んだファイルの行数。
    """
    errors = []
    try:
        with open(log_file, "r") as f:
            # 最初にlast_positionまで読み飛ばす
            for _ in range(last_position):
                f.readline()

            # 残りの行を検査
            line_number = last_position
            for line in f:
                line_number += 1
                if any(keyword in line for keyword in keywords):
                    errors.append(line.strip())
            current_position = line_number
        return errors, current_position
    except Exception as e:
        print(f"ログファイルの読み取り中にエラーが発生しました: {e}")
        return [], last_position


def send_email_notification(errors):
    """
    検知したエラーをメール通知プログラムに渡す。

    Args:
        errors (list): 検知したエラー行のリスト。
    """
    try:
        for error_message in errors:
            subject = f"【エラーログ通知】{PROCESS_NAME}で障害が発生しました。"
            body = (
                f"以下のシステムでエラーが発生しました：\n"
                f"    システム名: {PROCESS_NAME}\n\n"
                f"    詳細メッセージ:\n"
                f"    {error_message}\n"
            )
            # send_mail.py に引数を渡して実行
            os.system(
                f"python {MAIL_SCRIPT} '{TO_ADDRESS}' '{subject}' '{body}'"
            )
            print(f"メール通知を送信しました: {TO_ADDRESS},{subject},{error_message}")
    except Exception as e:
        print(f"メール送信中にエラーが発生しました: {e}")


def handle_log_rotation():
    """
    ローテーションされていないか確認し、されていれば位置をリセット。
    """
    if os.path.exists(LOG_FILE):
        current_line_count = sum(1 for _ in open(LOG_FILE, 'r'))
        last_position = get_last_position(STATE_FILE)
        if last_position > current_line_count:
            print(f"{LOG_FILE} の行数が減少しました。ログローテーションが行われた可能性があります。")
            if os.path.exists(STATE_FILE):
                os.remove(STATE_FILE) 
        else:
            print(f"{LOG_FILE} の行数は変動していません。位置情報は保持します。")


if __name__ == "__main__":
    # ログローテーションのチェック
    handle_log_rotation()
    last_position = get_last_position(STATE_FILE)
    errors, current_position = detect_errors(LOG_FILE, KEYWORDS, last_position)
    if errors:
        print(f"エラーを検知しました: {errors}")
        send_email_notification(errors)
    save_last_position(STATE_FILE, current_position)

-------
※ここまで

実行すると、最後に読み込んだ行数と、ファイルの行数を比較し、新しい行が追加されていた場合、検知バターンの確認を行い、マッチした場合は下記のようなメールが TO_ADDRESS で指定したメールアドレス宛に届きます。

メールタイトル：

【エラーログ通知】MySystemで障害が発生しました

メッセージ：以下のシステムでエラーが発生しました：
システム名: MySystem

詳細メッセージ:
FATAL