|
import hashlib |
|
import imaplib |
|
import argparse |
|
import email |
|
import os |
|
import getpass |
|
import socket |
|
from contextlib import contextmanager |
|
from pathlib import Path |
|
from tqdm import tqdm |
|
|
|
|
|
def get_password(): |
|
password = os.getenv("IMAP_PASSWORD") |
|
if password is None: |
|
password = getpass.getpass() |
|
return password |
|
|
|
|
|
def message_id_hash(msg): |
|
mid = (msg["Message-Id"] or "").strip().lower().strip("<>") |
|
h = hashlib.sha256(mid.encode()).hexdigest() |
|
return h |
|
|
|
|
|
@contextmanager |
|
def imap_connection(host, username, password): |
|
# Set a reasonable timeout for the connection |
|
socket.setdefaulttimeout(30) |
|
|
|
try: |
|
mailbox = imaplib.IMAP4_SSL(host) |
|
try: |
|
mailbox.login(username, password) |
|
yield mailbox |
|
except imaplib.IMAP4.error as e: |
|
if "AUTHENTICATIONFAILED" in str(e): |
|
print( |
|
"Authentication failed. Please check your credentials and ensure you're using the correct password." |
|
) |
|
print( |
|
"If you're using a VPN, try disconnecting it or using a different VPN server." |
|
) |
|
raise |
|
else: |
|
print(f"IMAP error: {e}") |
|
raise |
|
except socket.timeout: |
|
print( |
|
"Connection timed out. This might be due to VPN issues or network problems." |
|
) |
|
raise |
|
except Exception as e: |
|
print(f"Unexpected error: {e}") |
|
raise |
|
finally: |
|
try: |
|
mailbox.close() |
|
except: # noqa: E722 |
|
pass |
|
try: |
|
mailbox.logout() |
|
except: # noqa: E722 |
|
pass |
|
|
|
|
|
def parse_args(): |
|
""" |
|
Parse command line arguments. |
|
|
|
Returns: |
|
argparse.Namespace: Parsed command line arguments |
|
""" |
|
argparser = argparse.ArgumentParser( |
|
description="Dump a IMAP folder into .eml files" |
|
) |
|
argparser.add_argument( |
|
"-s", |
|
dest="host", |
|
help="IMAP host, like imap.gmail.com", |
|
default="imap.gmail.com", |
|
) |
|
argparser.add_argument("-u", dest="username", help="IMAP username", required=True) |
|
argparser.add_argument( |
|
"-p", dest="password", help="IMAP password", default=get_password() |
|
) |
|
argparser.add_argument( |
|
"-r", dest="remote_folder", help="Remote folder to download", default="INBOX" |
|
) |
|
argparser.add_argument( |
|
"-l", |
|
dest="local_folder", |
|
help="Local folder where to save .eml files", |
|
default="./emails", |
|
) |
|
argparser.add_argument( |
|
"-t", |
|
dest="to_address", |
|
help="Only download messages sent to this email address", |
|
) |
|
return argparser.parse_args() |
|
|
|
|
|
def download_emails( |
|
host, |
|
username, |
|
password, |
|
remote_folder="INBOX", |
|
local_folder="./emails", |
|
to_address=None, |
|
): |
|
""" |
|
Download emails from an IMAP server. |
|
|
|
Args: |
|
host (str): IMAP server hostname |
|
username (str): IMAP username |
|
password (str): IMAP password |
|
remote_folder (str): Remote folder to download from |
|
local_folder (str): Local folder to save emails to |
|
to_address (str, optional): Only download messages sent to this email address |
|
""" |
|
with imap_connection(host, username, password) as mailbox: |
|
mailbox.select(remote_folder, readonly=True) |
|
# Use TO filter if to_address is provided, otherwise use ALL |
|
search_criteria = f'(TO "{to_address}")' if to_address else "ALL" |
|
status, data = mailbox.search(None, search_criteria) |
|
email_ids = data[0].split() |
|
|
|
tqdm.write(f"{len(email_ids)} emails found") |
|
local_folder = Path(local_folder) / username |
|
local_folder.mkdir(parents=True, exist_ok=True) |
|
|
|
for email_id in tqdm(email_ids, desc="Downloading emails"): |
|
status, data = mailbox.fetch(email_id, "(RFC822)") |
|
raw_email = data[0][1] |
|
msg = email.message_from_bytes(raw_email) |
|
filename = message_id_hash(msg) |
|
path = local_folder / ("%s.eml" % filename) |
|
if os.path.exists(path): |
|
tqdm.write("Skipping %s" % path) |
|
continue |
|
with open(path, "wb") as f: |
|
tqdm.write("Writing %s" % path) |
|
f.write(raw_email) |
|
|
|
|
|
if __name__ == "__main__": |
|
args = parse_args() |
|
download_emails( |
|
host=args.host, |
|
username=args.username, |
|
password=args.password, |
|
remote_folder=args.remote_folder, |
|
local_folder=args.local_folder, |
|
to_address=args.to_address, |
|
) |
