はじめに
特定のメールアドレス + 件名で届いたときに、自動で返信する仕組みを探していました。
fetchmail + Pythonで実装可能であったため、メモを残しておきます。
環境
・Ubuntu 24.04
・Python 3.12.3
・fetchmail 6.4.38
・Gmail (IMAP 有効化済み、アプリパスワード使用)
1. fetchmailの設定
sudo apt install fetchmail
sudo vi ~/.fetchmailrc
poll imap.gmail.com with proto IMAP
user 'your_account@gmail.com' there with password 'your_app_password' is 'your_user' here
folder 'INBOX'
ssl
sudo chmod 600 ~/.fetchmailrc
受信確認
Step1. メールを送ってみる
(文面が短いとGmailのスパム判定の影響で受信されないことがある)
Step2 Gmailを開いて受信されていることを確認
Step3 fetchmail -vk
・「fetchmail: バックグラウンドで fetchmail が動作中のためオプションが受け付けられません。」と出力された場合はfetchmail --quitを入力。
・受信メールは /var/mail に保存されている。
2. fetchmailのservice化
ログアウト中も動いてほしいので、
cd /etc/systemd/system
sudo vi fetchmail@<your_user>.service
[Unit]
Description=Fetchmail for %I
After=network-online.target
[Service]
User=%i
ExecStart=/usr/bin/fetchmail -f /home/%i/.fetchmailrc -d 60 --syslog
Restart=always
[Install]
WantedBy=multi-user.target
・After=network-online.target
→ ネットワークが準備完了したら実行してね
・-d 60
→ 60sごとにメールを取りに行く
・--syslog
→ ログをjournalctlに出力
sudo systemctl daemon-reload
sudo systemctl enable fetchmail@<your_user>.service
sudo journalctl -u fetchmail@<your_user>.service -f
・-u fetchmail@<your_user>.service
→ 特定のサービスだけを表示
・-f
→ 新しいログが出たらリアルタイムに表示する
・journalctl内で「...宛に届いた...サーバからメッセージを削除しました。」と表示される場合
→ Gmailの場合は、「受信トレイ」から「すべてのメール」に移動しただけ。
3. Pythonで自動返信する
仮想環境を作った後に、任意の場所に配置する。
方針として、
・mail.envに登録されたアドレス以外には返信しないことでスパムへの誤爆を防止
・本文がないものは返信しない
・件名に指定された文字がない場合は返信しない
・メールのidを.txtに記録して、多重送信防止
・相手も自動返信だとループが発生するかもしれないため「Re:」を付ける。
main.py
#!<your_pass>/venv/bin/python3
import os, mailbox, hashlib, subprocess, codecs, time
from typing import Iterable, Set, Optional
from email.utils import parseaddr
from dotenv import load_dotenv
from pathlib import Path
from datetime import datetime
class AutoReplier:
def __init__(self):
self.keyword = "test"
number = <your_name>
name = Path(__file__).parent
self.mbox_path = f"/var/mail/{number}"
self.env_path = f"{name}/templates/mail.env"
self.reply_body = "ありがとうございます。\nこれは自動返信です。"
self.subject_prefix = "Re: "
self.allow_senders = self._load_env()
# 返信済みログファイル
self.replied_log_path = f"{name}/auto_reply/log/replied_log.txt"
self.replied_ids = set()
if os.path.exists(self.replied_log_path):
with open(self.replied_log_path, "r") as f:
self.replied_ids = {line.strip() for line in f if line.strip()}
else:
open(self.replied_log_path, "w").close()
def run(self):
"""mbox を走査し、条件を満たすメッセージへ返信して記録。"""
while True:
time.sleep(60)
mbox = mailbox.mbox(self.mbox_path)
for message in mbox:
#print(message)
msg_id = message.get('Message-ID')
if msg_id in self.replied_ids:
continue
from_addr_full = message.get('From')
addr = self._norm_addr(self._extract_email(from_addr_full))
subject = message.get('Subject', '') or ''
text = self._extract_text(message)
if not text.strip() or self.keyword not in subject:
continue
#env内の人物は少ないのでひとつずつ探索
for i in self.allow_senders:
if not i == addr:
continue
else:
# 返信実行
now_str = datetime.now().strftime('%Y/%m/%d %H:%M:%S')
print(f'Started_reply | time:{now_str}, ID:{msg_id}')
self._reply(addr, subject)
# 記録
with open(self.replied_log_path, "a") as f:
f.write(msg_id + "\n")
self.replied_ids.add(msg_id)
def _load_env(self):
load_dotenv(self.env_path)
allow = os.getenv("EMAIL_TO", "")
allow_list = [addr.strip() for addr in os.getenv('EMAIL_TO', '').split(',') if addr.strip()]
return allow_list
def _extract_text(self, msg) -> str:
# text/plain を優先して抽出
if msg.is_multipart():
chunks = []
for part in msg.walk():
if part.get_content_type() == 'text/plain':
try:
chunks.append((part.get_payload(decode=True) or b'').decode(part.get_content_charset() or 'utf-8',errors='ignore'))
except Exception:
continue
return "\n".join(chunks)
else:
try:
return (msg.get_payload(decode=True) or b'').decode(msg.get_content_charset() or 'utf-8',errors='ignore')
except Exception:
return ""
def _reply(self, to_addr: str, original_subject: str) -> None:
subject = f"{self.subject_prefix}{original_subject}"
subprocess.run(["mail", "-s", subject, to_addr],input=self.reply_body,text=True,check=False)
def _extract_email(self, from_header):
if not from_header:
return None
_, email_addr = parseaddr(from_header)
return email_addr or None
def _norm_addr(self, addr):
if not addr:
return None
return addr.strip().lower()
if __name__ == "__main__":
replier = AutoReplier()
replier.run()
4. Pythonのサービス化
cd /etc/systemd/system
sudo vi fetchmail_auto_reply.service
[Unit]
Description=Auto_Reply_Mail_Service
After=network-online.target
Wants=network-online.target
[Service]
User=<your_name>
WorkingDirectory=<your_pass>
# ── 起動前にファイルを強制削除 ──
# 「-」を先頭につけると、削除失敗（ファイル未存在）を無視します
ExecStartPre=-/usr/bin/rm -f <your_pass>/log/general.log
ExecStartPre=-/usr/bin/rm -f <your_pass>/log/general_error.log
# 仮想環境の python を直接呼び出し、スクリプトも絶対パスで指定
ExecStart=<your_pass>/main.py
# 標準出力を一般ログへ追記
StandardOutput=append:<your_pass>/log/general.log
StandardError=append:<your_pass>/log/general_error.log
Restart=on-failure
#ログをリアルタイムに表示
Environment=PYTHONUNBUFFERED=1
[Install]
WantedBy=multi-user.target
sudo systemctl daemon-reload
sudo systemctl enable fetchmail_auto_reply.service
最後に
件名はtestを入力して、メールを送ってみてください。
現在は fetchmail と Python で別々にサービスを動かしていますが、将来的には「受信後すぐに処理」する形に統合できるかもしれません。