IMAP Email Fetcher and Parser

IMAP Email Fetcher and Parser

Problem Statement

Automated systems often need to programmatically read email — think of an alerting pipeline that watches a shared inbox for vendor notifications, a reporting job that ingests daily CSV attachments, or a compliance tool that archives messages from a specific sender. Doing this reliably across providers (Gmail, Outlook, Fastmail, self-hosted Dovecot) means dealing with IMAP quirks, SSL/TLS negotiation, MIME parsing, and attachment decoding — all of which are tedious to reimplement.

This script solves that by providing a single, reusable CLI that connects to any IMAP server, searches messages by date/sender/subject, parses the full MIME structure (including nested multipart bodies and attachments), and emits clean JSON that downstream tools can consume. It handles UTF-8, quoted-printable, and base64 encodings transparently, so you get structured data instead of raw RFC 822 text.

How It Works

The script follows a straightforward connect → search → fetch → parse → emit pipeline. It uses imapclient.IMAPClient for the network layer because it wraps Python's imaplib with a saner API (UID-based operations, automatic folder selection, and TLS by default). For parsing, mailparser.parse_from_bytes() decodes the raw RFC 822 message into a MailParser object exposing subject, from_, to, date, text_plain, text_html, and attachments — each attachment being a dict with filename, payload, and mail_content_type.

Key design decisions:

• UID-based fetching — UIDs are stable across sessions, unlike sequence numbers, so re-runs don't accidentally fetch the wrong message.
• Search-before-fetch — the SEARCH command runs server-side, so we only download the messages we actually need (important on large mailboxes).
• Streaming JSON output — messages are accumulated into a list and dumped once with json.dump(), keeping memory bounded for typical inbox sizes (thousands of messages).
• Fail-fast on auth, fail-soft on individual messages — a bad password aborts immediately, but a single malformed message is logged and skipped so one broken email doesn't kill the whole batch.

Full Script

#!/usr/bin/env python3
"""
IMAP Email Fetcher and Parser

Connects to an IMAP server, searches for messages matching the given
criteria, parses each message's MIME structure, and writes the results
as JSON. Attachments are base64-encoded inline so the output is
self-contained.

Usage:
    python3 imap_fetch.py --host imap.gmail.com --folder INBOX --since 7d --output emails.json
"""

from __future__ import annotations

import argparse
import base64
import json
import logging
import os
import re
import ssl
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any

from imapclient import IMAPClient
from mailparser import parse_from_bytes

# ---------------------------------------------------------------------------
# Configuration (environment variables with sensible defaults)
# ---------------------------------------------------------------------------
IMAP_HOST = os.environ.get("IMAP_HOST", "imap.gmail.com")
IMAP_PORT = int(os.environ.get("IMAP_PORT", "993"))
IMAP_USER = os.environ.get("IMAP_USER", "")
IMAP_PASSWORD = os.environ.get("IMAP_PASSWORD", "")
IMAP_USE_SSL = os.environ.get("IMAP_USE_SSL", "true").lower() in ("1", "true", "yes")
IMAP_TIMEOUT = int(os.environ.get("IMAP_TIMEOUT", "30"))
DEFAULT_FOLDER = os.environ.get("IMAP_FOLDER", "INBOX")
MAX_MESSAGES = int(os.environ.get("IMAP_MAX_MESSAGES", "500"))

logging.basicConfig(
    level=os.environ.get("LOG_LEVEL", "INFO").upper(),
    format="%(asctime)s [%(levelname)s] %(message)s",
)
log = logging.getLogger("imap_fetch")


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def parse_since(value: str) -> datetime:
    """Parse a '--since' argument like '7d', '24h', '30m', or an ISO date.

    Returns a timezone-aware UTC datetime representing the lower bound.
    """
    if not value:
        return datetime.now(timezone.utc) - timedelta(days=7)

    # Relative form: <number><unit> where unit is m/h/d/w
    match = re.fullmatch(r"(\d+)([mhdw])", value.strip().lower())
    if match:
        amount, unit = int(match.group(1)), match.group(2)
        delta = {
            "m": timedelta(minutes=amount),
            "h": timedelta(hours=amount),
            "d": timedelta(days=amount),
            "w": timedelta(weeks=amount),
        }[unit]
        return datetime.now(timezone.utc) - delta

    # Absolute form: ISO 8601 date or datetime
    try:
        dt = datetime.fromisoformat(value)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return dt
    except ValueError as exc:
        raise argparse.ArgumentTypeError(
            f"Invalid --since value {value!r}; use e.g. '7d', '24h', or '2024-01-01'"
        ) from exc


def build_search_criteria(since: datetime, sender: str | None, subject: str | None) -> list[str]:
    """Build an IMAP SEARCH criteria list.

    IMAP date search uses day granularity only, so we pass the date portion.
    """
    criteria: list[str] = ["SINCE", since.strftime("%d-%b-%Y")]
    if sender:
        criteria += ["FROM", sender]
    if subject:
        criteria += ["SUBJECT", subject]
    return criteria


def serialize_attachment(att: dict[str, Any]) -> dict[str, Any]:
    """Convert a mailparser attachment dict into JSON-safe output."""
    payload = att.get("payload") or b""
    if isinstance(payload, str):
        payload = payload.encode("utf-8", errors="replace")
    return {
        "filename": att.get("filename"),
        "content_type": att.get("mail_content_type"),
        "size_bytes": len(payload),
        "content_base64": base64.b64encode(payload).decode("ascii"),
    }


def parse_message(uid: int, raw: bytes) -> dict[str, Any]:
    """Parse a raw RFC 822 message into a flat dict suitable for JSON."""
    mail = parse_from_bytes(raw)

    # mailparser exposes headers as attributes; fall back to empty lists.
    def _as_list(value: Any) -> list[str]:
        if value is None:
            return []
        if isinstance(value, list):
            return [str(v) for v in value]
        return [str(value)]

    return {
        "uid": uid,
        "message_id": (mail.message_id or "").strip("<>"),
        "subject": mail.subject or "",
        "from": _as_list(mail.from_),
        "to": _as_list(mail.to),
        "cc": _as_list(mail.cc),
        "date": mail.date.isoformat() if mail.date else None,
        "text_plain": mail.text_plain[0] if mail.text_plain else "",
        "text_html": mail.text_html[0] if mail.text_html else "",
        "attachments": [serialize_attachment(a) for a in mail.attachments],
    }


# ---------------------------------------------------------------------------
# Core fetch logic
# ---------------------------------------------------------------------------
def fetch_emails(
    host: str,
    port: int,
    user: str,
    password: str,
    folder: str,
    since: datetime,
    sender: str | None,
    subject: str | None,
    use_ssl: bool,
    max_messages: int,
) -> list[dict[str, Any]]:
    """Connect, search, fetch, and parse messages. Returns a list of dicts."""
    if not user or not password:
        raise RuntimeError(
            "IMAP credentials missing. Set IMAP_USER and IMAP_PASSWORD env vars, "
            "or pass them via CLI flags."
        )

    ssl_context = ssl.create_default_context() if use_ssl else None
    results: list[dict[str, Any]] = []

    log.info("Connecting to %s:%d (ssl=%s)", host, port, use_ssl)
    # IMAPClient handles the SSL handshake and LOGIN command internally.
    with IMAPClient(
        host,
        port=port,
        ssl=use_ssl,
        ssl_context=ssl_context,
        timeout=IMAP_TIMEOUT,
    ) as client:
        client.login(user, password)
        log.info("Authenticated as %s", user)

        # Select the folder read-only so we never accidentally mark messages
        # as \Seen just by fetching them.
        client.select_folder(folder, readonly=True)
        log.info("Selected folder %s", folder)

        criteria = build_search_criteria(since, sender, subject)
        log.info("Search criteria: %s", criteria)
        uids = client.search(criteria)
        log.info("Found %d matching messages", len(uids))

        if not uids:
            return results

        # Cap the number of messages to avoid pulling a giant mailbox.
        uids = uids[-max_messages:]

        # Fetch the raw RFC822 body in one round-trip per batch.
        # BODY.PEEK[] avoids setting the \Seen flag (belt-and-braces with readonly).
        response = client.fetch(uids, ["BODY.PEEK[]", "RFC822.SIZE"])
        for uid, data in response.items():
            raw = data.get(b"BODY[]") or data.get(b"RFC822")
            if raw is None:
                log.warning("UID %s returned no body; skipping", uid)
                continue
            try:
                results.append(parse_message(uid, raw))
            except Exception as exc:  # noqa: BLE001 — one bad message shouldn't abort the run
                log.error("Failed to parse UID %s: %s", uid, exc)

    return results


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def build_arg_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="Fetch and parse emails from an IMAP server into JSON.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    p.add_argument("--host", default=IMAP_HOST, help="IMAP server hostname")
    p.add_argument("--port", type=int, default=IMAP_PORT, help="IMAP server port")
    p.add_argument("--user", default=IMAP_USER, help="IMAP username (env: IMAP_USER)")
    p.add_argument(
        "--password",
        default=IMAP_PASSWORD,
        help="IMAP password (env: IMAP_PASSWORD; prefer env over CLI for security)",
    )
    p.add_argument("--folder", default=DEFAULT_FOLDER, help="Mailbox folder to select")
    p.add_argument(
        "--since",
        default="7d",
        help="Lower bound: relative ('7d', '24h', '30m', '2w') or ISO date",
    )
    p.add_argument("--from", dest="sender", default=None, help="Filter by sender address")
    p.add_argument("--subject", default=None, help="Filter by subject substring")
    p.add_argument(
        "--no-ssl",
        action="store_true",
        help="Disable SSL (only use for local testing against a plain IMAP server)",
    )
    p.add_argument(
        "--max",
        type=int,
        default=MAX_MESSAGES,
        help="Maximum number of messages to fetch (most recent first)",
    )
    p.add_argument(
        "--output",
        type=Path,
        default=Path("emails.json"),
        help="Output JSON file path (use '-' for stdout)",
    )
    return p


def main(argv: list[str] | None = None) -> int:
    args = build_arg_parser().parse_args(argv)

    try:
        since = parse_since(args.since)
    except argparse.ArgumentTypeError as exc:
        log.error("%s", exc)
        return 2

    try:
        emails = fetch_emails(
            host=args.host,
            port=args.port,
            user=args.user,
            password=args.password,
            folder=args.folder,
            since=since,
            sender=args.sender,
            subject=args.subject,
            use_ssl=not args.no_ssl,
            max_messages=args.max,
        )
    except Exception as exc:  # noqa: BLE001 — surface a clean error to the shell
        log.error("Fetch failed: %s", exc)
        return 1

    payload = {
        "fetched_at": datetime.now(timezone.utc).isoformat(),
        "host": args.host,
        "folder": args.folder,
        "since": since.isoformat(),
        "count": len(emails),
        "messages": emails,
    }

    if str(args.output) == "-":
        json.dump(payload, sys.stdout, indent=2, ensure_ascii=False)
        sys.stdout.write("\n")
    else:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        with args.output.open("w", encoding="utf-8") as fh:
            json.dump(payload, fh, indent=2, ensure_ascii=False)
        log.info("Wrote %d messages to %s", len(emails), args.output)

    return 0


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

Configuration

The script reads configuration from environment variables first, then lets CLI flags override them. This makes it easy to drop into a Docker container or CI job without changing the invocation.

Security note: For Gmail and most providers, you must use an app password or OAuth token — regular account passwords are rejected when 2FA is enabled. Prefer setting IMAP_PASSWORD via env var or a secrets manager rather than passing --password on the command line, since CLI args are visible in ps output.

Usage Examples

1. Fetch the last 7 days from Gmail INBOX into a JSON file:

export IMAP_USER="alerts@example.com"
export IMAP_PASSWORD="xxxx xxxx xxxx xxxx"
python3 imap_fetch.py --host imap.gmail.com --folder INBOX --since 7d --output emails.json

Expected output:

2024-06-01 10:15:02,113 [INFO] Connecting to imap.gmail.com:993 (ssl=True)
2024-06-01 10:15:03,401 [INFO] Authenticated as alerts@example.com
2024-06-01 10:15:03,612 [INFO] Selected folder INBOX
2024-06-01 10:15:03,612 [INFO] Search criteria: ['SINCE', '25-May-2024']
2024-06-01 10:15:04,022 [INFO] Found 42 matching messages
2024-06-01 10:15:05,887 [INFO] Wrote 42 messages to emails.json

2. Stream to stdout for piping into jq:

python3 imap_fetch.py --since 24h --from "noreply@github.com" --output - \
  | jq '.messages[] | {subject, date}'

3. Fetch a month of messages from a specific subject into a named folder:

python3 imap_fetch.py \
  --host imap.fastmail.com \
  --folder "Archive/Receipts" \
  --since 30d \
  --subject "Invoice" \
  --max 200 \
  --output receipts.json

Error Handling

The script is deliberately fail-fast on connection/auth but fail-soft on individual messages — a single corrupted email should never block an entire ingestion run.

Extending Ideas

• Idempotent state tracking: Persist the highest UID seen per folder (e.g., to a SQLite file or a .lastuid sidecar) and pass UID <lastuid+1:* as the search criterion so subsequent runs only fetch new mail.
• Attachment extraction to disk: Instead of base64-inlining attachments, write them to a directory using filename and message_id for collision-free naming, then reference paths in the JSON.
• OAuth2 for Gmail/Outlook: Replace the LOGIN command with XOAUTH2 (client.oauth2_login(user, access_token)) and plug in google-auth or msal to obtain the token — required for Workspace accounts that disable app passwords.
• Filtering rules engine: Add a --rules rules.yaml flag that matches on sender domain, subject regex, or attachment MIME type, and routes each message to a different output file or webhook.
• Parallel parsing: For very large batches, offload parse_message() to a concurrent.futures.ProcessPoolExecutor — MIME parsing is CPU-bound and benefits from multiple cores.