Help us understand the problem. What is going on with this article?

imaplib でメールボックス内の全メールを別サーバのメールボックスへ一括転送する

More than 1 year has passed since last update.

やったこと

Python の imaplib を使い、あるメールサーバのメールボックス内の全メールを別のメールサーバのメールボックスへ一括転送するツールを簡単に作ってみた。

経緯

メールサーバを移転するにあたり、旧メールサーバに保管していたメールを新メールサーバに引き継ぎたいので、何かいい方法があるのか調べてみたけど見つからなかった。

旧サーバ(共用レンタルサーバ)の cPanel でメールを一括エクスポートしようとするも、なぜかエラーログしかダウンロードできない。
メールクライアントでエクスポート&インポートもできなくはないけど、面倒そうだし、できれば既読やフラグ、受信日時なども引き継ぎたい。

一括転送ツールがあれば楽だと思っていたが、探しても見つからなかったので、いっそ自分で作ってしまおうと思い立った。

作るのは面倒かもしれないが、一度作ってしまえば次回引っ越し以降も使えるツールになるはず。今後引っ越しの機会があるかは分からないが。

ソースコード

transfer_mails.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import sys
import getpass
import re
import time
import imaplib
import email

# 転送元メールアカウント情報
src_mail_server = "mail.example.jp"
src_mail_user = "example"

# 転送先メールアカウント情報
dst_mail_server = "example.com"
dst_mail_user = "foo"

# 転送対象メールボックス
target_mailboxes = [
  ["INBOX.foo@example_com", "INBOX"],
  ["INBOX.Drafts", "Drafts"],
  ["INBOX.Sent", "Sent"],
  ["INBOX.Junk", "Junk"],
  ["INBOX.Trash", "Trash"],
  ["INBOX.Archive", "Archive"]
]

# 正規表現コンパイル
msg_id_re = re.compile(r"<([^>]+)>")
rcv_date_re = re.compile(r"^(?:.+)\s+for\s?<[^>]+>\s?;\s?([A-Za-z]{3}\s?,\s?[0-9]+\s+[A-Za-z]{3}\s+[0-9]{4}\s+[0-9]+:[0-9]+:[0-9]+\s+[\+\-]?[0-9]{4})", re.MULTILINE | re.DOTALL)
flags_re = re.compile(r"^[0-9]+ \(FLAGS \(([A-Za-z\\\$ ]*)\)\)$")

# 指定したメールボックス内の全メールのMessage-IDのsetを返す
def get_message_id_set(imap, mailbox):
  typ, msg = imap.select(mailbox)
  if typ != "OK":
    return None
  typ, [data] = imap.search(None, "ALL")
  ret = set()
  for num in data.split():
    typ, d = imap.fetch(num, '(RFC822)')
    msg = email.message_from_bytes(d[0][1])
    m = msg_id_re.match(msg['Message-ID'])
    if m:
      ret.add(m.group(1))
  imap.close()
  return ret

# 指定したメールボックス内のメールのうち、未転送のメールを転送する
def transfer_messages(src_imap, src_mailbox, dst_imap, dst_mailbox, excluded_message_id_set):
  print("=== START: transfer mails from '" + src_mailbox + "' to '" + dst_mailbox + "'. ===")
  src_imap.select(src_mailbox)
  typ, [data] = src_imap.search(None, "ALL")

  for num in data.split():
    typ, d = src_imap.fetch(num, '(RFC822)')
    msg = email.message_from_bytes(d[0][1])

    m = msg_id_re.match(msg['Message-ID'])
    if not m:
      print("WARNING: failed to get Message-ID for '" + num + "'.")
      continue
    msg_id = m.group(1)
    if excluded_message_id_set is not None and msg_id in excluded_message_id_set:
      print("Mail <" + msg_id + "> has already been transfered.")
      continue

    typ, d = src_imap.fetch(num, '(FLAGS)')
    if typ != "OK":
      print("WARNING: failed to get FLAGS for <" + msg_id + ">.")
      flags = ""
    else:
      m = flags_re.match(d[0].decode('ascii'))
      if not m:
        print("WARNING: failed to parse FLAGS for <" + msg_id + ">.")
        print(d[0].decode('ascii'))
        flags = ""
      else:
        flags = m.group(1)

    m = rcv_date_re.match(msg['Received'])
    if not m:
      print("WARNING: failed to get received datetime for <" + msg_id + ">.")
      dt = None
    else:
      dt = time.mktime( email.utils.parsedate(m.group(1)) )

    typ, [m] = dst_imap.append(dst_mailbox, flags, dt, bytes(msg))
    if typ != "OK":
      sys.stderr.write(m.decode('ascii') + "\n")
      continue

    print("Transfered <" + msg_id + "> (FLAGS: " + flags + ").")

  src_imap.close()
  print("=== END:   transfer mails from '" + src_mailbox + "' to '" + dst_mailbox + "'. ===")

# エントリーポイント
if __name__ == "__main__":
  try:
    src_imap = imaplib.IMAP4_SSL(src_mail_server)
    src_imap.login(src_mail_user, getpass.getpass("Password for " + src_mail_user + " (" + src_mail_server + "): "))
    dst_imap = imaplib.IMAP4_SSL(dst_mail_server)
    dst_imap.login(dst_mail_user, getpass.getpass("Password for " + dst_mail_user + " (" + dst_mail_server + "): "))

    for src_mailbox, dst_mailbox in target_mailboxes:
      excluded_message_id_set = get_message_id_set(dst_imap, dst_mailbox)
      if excluded_message_id_set is None:
        dst_imap.create(dst_mailbox)
        print("Created a new mailbox '" + dst_mailbox + "'.")
      elif len(excluded_message_id_set) == 0:
        excluded_message_id_set = None
      transfer_messages(src_imap, src_mailbox, dst_imap, dst_mailbox, excluded_message_id_set)

    dst_imap.logout()
    src_imap.logout()

  except imaplib.IMAP4_SSL.error as e:
    sys.stderr.write(str(e) + "\n")

メールアカウント情報とメールボックス情報は適宜置き換えて下さい。

実行例

console
$ ./transfer_mails.py
=== START: transfer mails from 'INBOX.foo@example_com' to 'INBOX'. ===
Transfered <***************************************************@mail.gmail.com> (FLAGS: \Seen).
Transfered <******************************$@example.jp> (FLAGS: \Seen).
Transfered <******************************$@@example.jp> (FLAGS: \Answered \Seen).
Transfered <******************************$@@example.jp> (FLAGS: \Seen $NotJunk).
Transfered <********************************@www.example.com> (FLAGS: \Flagged \Seen $NotJunk).
Transfered <*********.****.*************.JavaMail.******@***> (FLAGS: \Seen $NotJunk JunkRecorded).
=== END:   transfer mails from 'INBOX.foo@example_com' to 'INBOX'. ===
=== START: transfer mails from 'INBOX.Drafts' to 'Drafts'. ===
=== END:   transfer mails from 'INBOX.Drafts' to 'Drafts'. ===
=== START: transfer mails from 'INBOX.Sent' to 'Sent'. ===
=== END:   transfer mails from 'INBOX.Sent' to 'Sent'. ===
=== START: transfer mails from 'INBOX.Junk' to 'Junk'. ===
=== END:   transfer mails from 'INBOX.Junk' to 'Junk'. ===
=== START: transfer mails from 'INBOX.Trash' to 'Trash'. ===
=== END:   transfer mails from 'INBOX.Trash' to 'Trash'. ===
=== START: transfer mails from 'INBOX.Archive' to 'Archive'. ===
=== END:   transfer mails from 'INBOX.Archive' to 'Archive'. ===

受信メールは転送できたが、送信済みメールはツールでは検知できなかった。
あと、以前 Gmail から転送してきたメールがなぜが Message-ID チェックに引っかからず、実行のたびに転送しようとしてしまう(おそらくプログラムのバグ)。

送信済みメールについては、受信メールの返信内容を見れば内容を確認できるので、なくても問題ないだろう。
Gmail からの転送メールもまた然り。

参考ページ

renny1398
7月よりインターネットメディア会社で勤務中。最近職場ではCodeIgniter3やMySQL、jQuery、Bootstrapなどを触っています。プライベートではC++、Python、Ruby(主にon Rails)なども。GoやRustにも興味あり。
Why not register and get more from Qiita?
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
Why do not you register as a user and use Qiita more conveniently?
You need to log in to use this function. Qiita can be used more conveniently after logging in.
You seem to be reading articles frequently this month. Qiita can be used more conveniently after logging in.
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
ユーザーは見つかりませんでした