Email Attachment Downloader via IMAP

Email Attachment Downloader via IMAP

Problem Statement

Teams and automated systems frequently receive important files as email attachments: nightly report exports, invoice PDFs, log bundles from monitoring tools, CSV data feeds from partners, and so on. Manually opening a mailbox, finding the right messages, and saving each attachment is tedious and error-prone, especially when the same process must run daily. A script that connects to an IMAP server, scans a specific folder, and downloads every attachment to a local directory solves this cleanly and can be scheduled with cron or a CI job.

This tutorial builds a robust, production-ready downloader using the imapclient library. It supports authentication via password or environment variable, filters by folder, skips messages already processed (via a state file), handles duplicate filenames safely, and works with any standards-compliant IMAP server (Gmail, Outlook/Exchange, Dovecot, etc.) that allows IMAP access.

How It Works

The script follows a linear pipeline with clear separation of concerns:

1. Connect & authenticate — imapclient.IMAPClient handles the low-level IMAP protocol, TLS negotiation, and login. We use ssl=True by default and fall back to STARTTLS if configured.
2. Select the target folder — the --folder argument (default INBOX) is selected read-only so the script never marks messages as seen or modifies the mailbox.
3. Fetch message metadata — we search for messages matching an optional --since date filter and fetch only the RFC822 body plus UID. Fetching metadata first keeps memory usage bounded.
4. Parse & extract attachments — Python's email module parses each raw message. We walk the MIME tree, identify parts with a Content-Disposition: attachment header (or a filename), decode the payload, and write it to disk.
5. Deduplicate & persist state — a JSON state file tracks processed UIDs so re-runs are idempotent. Filename collisions are resolved with a numeric suffix.

Design decisions: read-only mailbox access (no side effects), a persistent state file instead of relying on \Seen flags (works with read-only folders), and streaming writes to avoid loading large attachments into memory twice.

Full Script

#!/usr/bin/env python3
"""
Email Attachment Downloader via IMAP.

Connects to an IMAP server, scans a folder for messages, and downloads
every attachment to a local output directory. Tracks processed message
UIDs in a JSON state file so repeated runs are idempotent.

Usage:
    python3 mail_attachments.py --host mail.example.com --user admin \
        --folder Reports --out ./downloads/
"""

from __future__ import annotations

import argparse
import email
import json
import logging
import os
import re
import ssl
import sys
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from email.message import Message
from email.utils import parsedate_to_datetime
from pathlib import Path
from typing import Iterable

from imapclient import IMAPClient
from imapclient.exceptions import IMAPClientError

# ---------------------------------------------------------------------------
# Configuration (environment variables with sensible defaults)
# ---------------------------------------------------------------------------
DEFAULT_HOST = os.environ.get("IMAP_HOST", "")
DEFAULT_PORT = int(os.environ.get("IMAP_PORT", "993"))
DEFAULT_USER = os.environ.get("IMAP_USER", "")
DEFAULT_PASSWORD = os.environ.get("IMAP_PASSWORD", "")
DEFAULT_FOLDER = os.environ.get("IMAP_FOLDER", "INBOX")
DEFAULT_OUT = os.environ.get("IMAP_OUT", "./downloads")
DEFAULT_STATE = os.environ.get("IMAP_STATE_FILE", ".mail_attachments_state.json")
DEFAULT_USE_SSL = os.environ.get("IMAP_USE_SSL", "true").lower() in ("1", "true", "yes")

# Characters that are illegal in filenames on Windows/macOS/Linux.
_UNSAFE_FILENAME = re.compile(r'[<>:"/\\|?*\x00-\x1f]')

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
log = logging.getLogger("mail_attachments")


@dataclass
class DownloadStats:
    """Simple counter for the end-of-run summary."""

    messages_scanned: int = 0
    attachments_saved: int = 0
    attachments_skipped: int = 0
    errors: int = 0
    saved_paths: list[Path] = field(default_factory=list)


def sanitize_filename(name: str) -> str:
    """Return a filesystem-safe version of an attachment filename."""
    name = _UNSAFE_FILENAME.sub("_", name).strip().strip(".")
    return name or "attachment.bin"


def unique_path(directory: Path, filename: str) -> Path:
    """Return a non-colliding path inside *directory* for *filename*."""
    candidate = directory / filename
    if not candidate.exists():
        return candidate
    stem, suffix = candidate.stem, candidate.suffix
    counter = 1
    while True:
        candidate = directory / f"{stem}_{counter}{suffix}"
        if not candidate.exists():
            return candidate
        counter += 1


def load_state(state_file: Path) -> dict:
    """Load the JSON state file, tolerating a missing or corrupt file."""
    if not state_file.exists():
        return {"processed_uids": []}
    try:
        with state_file.open("r", encoding="utf-8") as fh:
            data = json.load(fh)
        if not isinstance(data.get("processed_uids"), list):
            raise ValueError("malformed state")
        return data
    except (json.JSONDecodeError, ValueError, OSError) as exc:
        log.warning("Could not read state file %s (%s); starting fresh", state_file, exc)
        return {"processed_uids": []}


def save_state(state_file: Path, state: dict) -> None:
    """Persist the state file atomically to avoid partial writes."""
    tmp = state_file.with_suffix(state_file.suffix + ".tmp")
    tmp.write_text(json.dumps(state, indent=2), encoding="utf-8")
    tmp.replace(state_file)


def iter_attachments(message: Message) -> Iterable[tuple[str, bytes]]:
    """Yield (filename, payload) for every attachment part in *message*."""
    for part in message.walk():
        if part.get_content_maintype() == "multipart":
            continue
        disposition = (part.get_content_disposition() or "").lower()
        filename = part.get_filename()
        if disposition != "attachment" and not filename:
            continue
        if filename:
            # RFC 2231 / RFC 2047 encoded filenames are decoded by get_filename().
            filename = email.utils.collapse_rfc2231_value(filename) if "=" in filename else filename
        else:
            filename = "attachment.bin"
        payload = part.get_payload(decode=True)
        if payload is None:
            continue
        yield sanitize_filename(filename), payload


def download_attachments(
    host: str,
    user: str,
    password: str,
    folder: str,
    out_dir: Path,
    state_file: Path,
    port: int = 993,
    use_ssl: bool = True,
    since_days: int | None = None,
) -> DownloadStats:
    """Connect to IMAP, scan *folder*, and save all attachments to *out_dir*."""
    stats = DownloadStats()
    out_dir.mkdir(parents=True, exist_ok=True)
    state = load_state(state_file)
    processed: set[int] = set(state.get("processed_uids", []))

    ssl_context = ssl.create_default_context()
    log.info("Connecting to %s:%s (ssl=%s)", host, port, use_ssl)

    try:
        with IMAPClient(host, port=port, ssl=use_ssl, ssl_context=ssl_context) as client:
            client.login(user, password)
            log.info("Authenticated as %s", user)

            # Read-only selection: never mutate the mailbox or mark as seen.
            client.select_folder(folder, readonly=True)
            log.info("Selected folder %r", folder)

            # Build the IMAP search criteria.
            criteria: list = ["ALL"]
            if since_days is not None:
                since_date = (datetime.now() - timedelta(days=since_days)).strftime("%d-%b-%Y")
                criteria = ["SINCE", since_date]
                log.info("Filtering messages since %s", since_date)

            uids = client.search(criteria)
            log.info("Found %d message(s)", len(uids))

            for uid in uids:
                stats.messages_scanned += 1
                if uid in processed:
                    log.debug("Skipping UID %s (already processed)", uid)
                    continue

                try:
                    response = client.fetch([uid], ["RFC822"])
                    raw = response.get(uid, {}).get(b"RFC822")
                    if not raw:
                        log.warning("UID %s returned no body; skipping", uid)
                        stats.errors += 1
                        continue

                    msg = email.message_from_bytes(raw)
                    subject = msg.get("Subject", "(no subject)")
                    msg_date = msg.get("Date", "")
                    try:
                        date_str = parsedate_to_datetime(msg_date).isoformat()
                    except (TypeError, ValueError):
                        date_str = msg_date

                    for filename, payload in iter_attachments(msg):
                        target = unique_path(out_dir, filename)
                        target.write_bytes(payload)
                        stats.attachments_saved += 1
                        stats.saved_paths.append(target)
                        log.info(
                            "Saved %s (%d bytes) from UID %s [%s, %s]",
                            target.name,
                            len(payload),
                            uid,
                            subject,
                            date_str,
                        )
                    processed.add(uid)
                except (IMAPClientError, OSError) as exc:
                    stats.errors += 1
                    log.error("Failed to process UID %s: %s", uid, exc)

            # Persist state only after a successful pass.
            state["processed_uids"] = sorted(processed)
            save_state(state_file, state)
            log.info("State saved to %s", state_file)

    except IMAPClientError as exc:
        log.error("IMAP error: %s", exc)
        raise
    except ssl.SSLError as exc:
        log.error("TLS error: %s", exc)
        raise

    return stats


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Download email attachments from an IMAP folder.",
    )
    parser.add_argument("--host", default=DEFAULT_HOST, help="IMAP server hostname")
    parser.add_argument("--port", type=int, default=DEFAULT_PORT, help="IMAP port (default 993)")
    parser.add_argument("--user", default=DEFAULT_USER, help="IMAP username")
    parser.add_argument(
        "--password",
        default=DEFAULT_PASSWORD,
        help="IMAP password (prefer IMAP_PASSWORD env var over CLI)",
    )
    parser.add_argument("--folder", default=DEFAULT_FOLDER, help="Mailbox folder to scan")
    parser.add_argument("--out", default=DEFAULT_OUT, help="Output directory for attachments")
    parser.add_argument("--state-file", default=DEFAULT_STATE, help="Path to JSON state file")
    parser.add_argument(
        "--since-days",
        type=int,
        default=None,
        help="Only scan messages from the last N days",
    )
    parser.add_argument(
        "--no-ssl",
        action="store_true",
        help="Disable implicit SSL (use STARTTLS-style plain connect)",
    )
    parser.add_argument("--verbose", "-v", action="store_true", help="Enable debug logging")
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    if args.verbose:
        log.setLevel(logging.DEBUG)

    missing = [name for name, value in (("host", args.host), ("user", args.user)) if not value]
    if missing:
        log.error("Missing required argument(s): %s", ", ".join(missing))
        return 2
    if not args.password:
        log.error("No password provided (set IMAP_PASSWORD or use --password)")
        return 2

    try:
        stats = download_attachments(
            host=args.host,
            user=args.user,
            password=args.password,
            folder=args.folder,
            out_dir=Path(args.out),
            state_file=Path(args.state_file),
            port=args.port,
            use_ssl=not args.no_ssl,
            since_days=args.since_days,
        )
    except (IMAPClientError, ssl.SSLError, OSError) as exc:
        log.error("Fatal error: %s", exc)
        return 1

    log.info(
        "Done. Scanned=%d Saved=%d Errors=%d",
        stats.messages_scanned,
        stats.attachments_saved,
        stats.errors,
    )
    return 1 if stats.errors and not stats.attachments_saved else 0


if __name__ == "__main__":
    sys.exit(main())

Configuration

The script reads configuration from environment variables first, then allows command-line overrides. This makes it easy to run under cron, systemd, or Docker without embedding secrets in scripts.

Security note: never pass --password on the command line in shared environments — process listings (ps aux) expose it. Use IMAP_PASSWORD or a secrets manager. For Gmail, generate an App Password; for Exchange, ensure IMAP is enabled for the account.

Usage Examples

Example 1 — Download report attachments from a shared mailbox:

export IMAP_HOST=mail.example.com
export IMAP_USER=admin
export IMAP_PASSWORD='s3cr3t'
python3 mail_attachments.py --folder Reports --out ./downloads/

Expected output:

2024-06-01 09:14:02 [INFO] Connecting to mail.example.com:993 (ssl=True)
2024-06-01 09:14:03 [INFO] Authenticated as admin
2024-06-01 09:14:03 [INFO] Selected folder 'Reports'
2024-06-01 09:14:03 [INFO] Found 12 message(s)
2024-06-01 09:14:04 [INFO] Saved Q2_revenue.csv (18432 bytes) from UID 4211 [Q2 report, 2024-06-01T08:00:00+00:00]
2024-06-01 09:14:04 [INFO] Done. Scanned=12 Saved=8 Errors=0

Example 2 — Only scan the last 7 days, verbose logging:

python3 mail_attachments.py \
  --host imap.gmail.com --user me@example.com \
  --folder "Invoices" --since-days 7 --out /var/data/invoices -v

Example 3 — Cron-safe invocation using only env vars:

# /etc/cron.d/mail-attachments
0 * * * * deploy IMAP_HOST=mail.example.com IMAP_USER=admin \
  IMAP_PASSWORD_FILE=/run/secrets/imap \
  /usr/bin/python3 /opt/scripts/mail_attachments.py --folder Reports --out /srv/reports

Error Handling

• Authentication failures — IMAPClientError is caught in main() and returns exit code 1; the password is never logged.
• TLS errors — ssl.SSLError is caught and reported distinctly, since misconfigured certificates are a common ops issue.
• Network/connection drops — the with IMAPClient(...) context manager guarantees logout() is called even on exceptions.
• Malformed MIME messages — each UID is processed in its own try/except, so one bad message does not abort the run.
• Missing or corrupt state file — load_state() falls back to an empty state and logs a warning rather than crashing.
• Filename collisions — unique_path() appends _1, _2, etc., so no attachment is ever silently overwritten.
• Unsafe filenames — sanitize_filename() strips path separators and control characters to prevent directory traversal (e.g., ../../etc/passwd).
• Partial state writes — save_state() writes to a .tmp file and atomically renames it, so an interrupted run won't leave a half-written state file.

Extending Ideas

• Server-side filtering by sender/subject — extend the IMAP search() criteria with FROM, SUBJECT, or HEADER to process only messages from a specific partner.
• Attachment type allow-listing — add a --extensions .csv,.pdf flag and skip parts whose filename doesn't match, useful for noisy mailboxes.
• Move processed messages — after saving, call client.copy(uid, "Processed") and client.add_flags(uid, [b"\\Deleted"]) to keep the source folder clean (requires a non-read-only selection).
• Notification hooks — after the run, POST a summary to Slack/Teams or emit Prometheus metrics (attachments_saved_total) for observability.
• Cloud storage sink — replace target.write_bytes(payload) with an s3.put_object() or Azure Blob upload to archive attachments off-host.