mail-normalizer

A production-grade Python package for normalizing heterogeneous email archives into a unified, AI/RAG-ready canonical schema.

What it does

mail-normalizer ingests email from multiple formats (Outlook MSG/PST, Gmail MBOX, EML, iCal/ICS) and produces deterministic, semantically clean CanonicalMessage and CanonicalEvent objects — suitable for embedding, semantic retrieval, thread reconstruction, and agent memory systems.

The normalizer is stage 1 of a three-stage AI wiki pipeline:

source/   .msg / .ics / .mbox files
    ↓  mail-normalizer  (this package)
raw/      CanonicalMessage + CanonicalEvent JSON  ← machine API
    ↓  AI agent  (downstream)
vault/    Obsidian markdown knowledge base        ← human readable

The normalizer's sole responsibility is producing complete, deterministic JSON. The AI agent reads from raw/ and writes the Obsidian vault.

---

This package is not a mail client, IMAP wrapper, or LLM orchestration framework. It does not implement IMAP/Gmail sync, vector database integration, or mail sending. Those belong in downstream systems that consume CanonicalMessage.

---

Installation

uv add mail-normalizer

# Optional: PST archive support
uv add "mail-normalizer[pst]"

# Optional: quote stripping via Talon
uv add "mail-normalizer[quotes]"

Requires Python 3.12+.

---

Quick Start

# Ingest Outlook .msg files from a directory (.msg, .ics, .mbox, .eml all auto-detected)
mail-fetch ingest --folder /path/to/msg-files --archive-root /path/to/archive --account-id work

# Ingest Gmail Takeout .mbox archives (auto-detected — no flag needed)
mail-fetch ingest --folder /path/to/mbox --archive-root /path/to/archive --account-id gmail

# All formats + feature extraction + write to SQLite
mail-fetch ingest --folder /path/to/export --archive-root /path/to/archive --account-id work \
    --enable-features --db-path /path/to/mail.db

# Export from Outlook then ingest (Windows only)
mail-fetch outlook --out-dir C:/MailArchive/source --account-id work --raw-root C:/MailArchive/raw

# Export personal calendar only, last 2 years, up to today
mail-fetch outlook --out-dir C:/MailArchive/source --account-id work \
    --include-calendar --calendar-name "john@company.com" --calendar-since 2023-01-01

# Check last ingest state
mail-fetch status --out-dir /path/to/source

Programmatic usage

from pathlib import Path
from mail_normalizer.pipelines import MsgPipelineRunner, MsgPipelineConfig, StorageConfig

config = MsgPipelineConfig(
    archive_root=Path("path/to/source"),
    storage=StorageConfig(raw_root=Path("path/to/raw"), enable_features=False),
)
runner = MsgPipelineRunner(config)
result = runner.run_directory(Path("path/to/source"))
print(f"Processed {result.messages_ok} messages → {result.raw_records_written} JSON files")

SQLite store

from mail_normalizer.storage.sqlite_store import SQLiteStore
from mail_normalizer.storage.merger import StorageRecord

store = SQLiteStore("path/to/mail.db")

# Write (upsert — safe to re-run)
store.write(StorageRecord.merge(canonical_message))

# Write many in one transaction
store.write_batch([StorageRecord.merge(m) for m in messages])

# Query
record = store.get("<message-id@example.com>")
ids = store.list_message_ids(account_id="work", thread_id="thread-abc", limit=50)
print(store.count())           # total stored
store.delete("<old@example.com>")

Unified runner (programmatic)

from pathlib import Path
from mail_normalizer.pipelines import UnifiedPipelineRunner, StorageConfig
from mail_normalizer.pipelines.unified_runner import UnifiedIngestConfig

config = UnifiedIngestConfig(
    account_id="work",
    archive_root=Path("path/to/source"),
    recursive=True,
    ingest_msg=True,
    ingest_mbox=True,
    ingest_eml=True,
    ingest_ics=True,
    storage=StorageConfig(raw_root=Path("path/to/raw"), enable_features=False),
)
runner = UnifiedPipelineRunner(config)
run = runner.run_directory(Path("path/to/source"))
print(f"Processed {run.messages_ok} messages, {run.events_ok} events")

All ingest operations run the full pipeline: ingest → normalize → enrich → features (optional) → raw JSON output.

mail-fetch ingest

Ingest an already-exported folder of mail files.

mail-fetch ingest --folder PATH --archive-root PATH [options]

Required:
  --folder PATH         Folder containing mail files to ingest (source/)
  --archive-root PATH   Base path for computing relative provenance paths

Format detection (automatic — no flags needed):
  .msg                  Outlook MSG files — always processed
  .ics                  iCal calendar files — auto-detected
  .mbox                 Gmail Takeout archives — auto-detected
  .eml                  EML / RFC 822 files — auto-detected

Storage flags:
  --raw-root PATH       Output root for canonical JSON files
                        (defaults to <archive-root>/raw/)
  --db-path PATH        SQLite database file; each normalized message is
                        upserted after raw JSON writing
  --enable-features     Run spaCy NLP extraction, write feat-*.json files
  --no-event-store      Skip CanonicalEvent JSON writes for calendar events
  --rules-file PATH     YAML file with extra enrichment rules to merge with defaults

Other:
  --account-id ID       Account label stamped on every canonical record
  --recursive           Descend into subdirectories
  --no-threading        Skip thread resolution
  --no-cleanup          Skip clean_text generation

mail-fetch outlook

Export from Outlook via PowerShell, then ingest (Windows only).

mail-fetch outlook --out-dir PATH [options]

  --out-dir PATH          Output directory for .msg/.ics source files
  --mail-folder NAME      Outlook folder (default: Inbox)
  --mode incremental|all
  --include-calendar      Also export calendar → ingest .ics
  --calendar-folder NAME  Outlook calendar folder (default: Calendar)
  --calendar-name STR     Only export from the store whose name contains STR
                          (case-insensitive). Use to skip public-holiday or
                          shared calendars (e.g. --calendar-name "john@co.com")
  --calendar-since DATE   Only export events on or after DATE (e.g. 2023-01-01)
  --calendar-until DATE   Only export events on or before DATE (default: today)
  --max-items N           Cap on items exported per folder (default: 10000)
  + all storage flags from `ingest`

mail-fetch rules

Print the effective enrichment rules in human-readable form (default rules + any --rules-file additions).

mail-fetch rules [--rules-file PATH]

mail-fetch status

Show the last-ingest state recorded in <out-dir>/.ingest-state.json.

---

Format	Status
Outlook MSG	✅ Implemented
Outlook PST	✅ Implemented (optional: libpff-python)
iCal / ICS	✅ Implemented
Gmail MBOX	✅ Implemented
EML / RFC822	✅ Implemented

---

Canonical Schema

All formats normalize into a single CanonicalMessage (Pydantic v2):

class CanonicalMessage(BaseModel):
    id: str                   # internally generated stable ID
    source_type: str
    account_id: str
    message_id: str
    thread_id: str            # internally generated; never raw header value
    subject: str
    timestamp: datetime
    from_: Participant
    to: list[Participant]
    cc: list[Participant]
    bcc: list[Participant]
    body_html: str            # original HTML preserved
    body_text: str            # extracted plain text
    clean_text: str           # AI-clean semantic text
    attachments: list[Attachment]
    labels: list[str]
    references: list[str]
    headers: dict             # raw headers preserved
    metadata: dict
    hashes: MessageHashes     # SHA-256 deduplication fingerprints

Calendar events normalize into CanonicalEvent (separate schema).

Three text layers (body_html, body_text, clean_text) are always preserved — intermediate stages are never overwritten.

---

Pipeline Directory Layout

The normalizer reads from a source folder and writes canonical JSON to a raw output folder. An AI agent (separate package) reads raw/ and produces the Obsidian vault.

source/                               ← --folder / --out-dir  (.msg / .ics / .mbox)
    ↓  mail-fetch ingest
raw/                                  ← --raw-root  (defaults to <archive-root>/raw/)
├── mail_normalizer.log               ← structured JSON log (rotates at 10 MB, 10 files kept)
├── log.md                            ← human-readable run summary
├── msg-<sha256>.json                 ← one CanonicalMessage per email (full schema)
├── evt-<sha256>.json                 ← one CanonicalEvent per calendar entry
├── feat-<sha256>.json                ← NLP features (written when --enable-features)
└── attachments/                      ← extracted attachment content
    └── <content-hash>-<name-slug>/

    ↓  AI agent  (future)
vault/                                ← Obsidian markdown knowledge base
├── important/
├── unimportant/
├── threads/
├── people/
└── attachments/

The JSON filename IS the deduplication key — re-ingesting the same message overwrites the same file. The AI agent can track processed files by manifest or mtime and restart safely from raw/ without data loss.

The source and raw-root paths are independent and can point to different drives or network shares.

---

Architecture

mail_normalizer/
├── ingest/         # file discovery, source dispatch, archive traversal
├── parsers/        # format-specific parsers → intermediate representation
├── normalize/      # core engine: MIME, charset, date, metadata normalization
├── threading/      # conversation graph reconstruction, stable thread IDs
├── cleanup/        # HTML cleaning, quote/signature stripping, tracking removal
├── attachments/    # MIME detect → specialized parser → Tika fallback → OCR
├── identity/       # participant normalization and canonical identity resolution
├── hashing/        # SHA-256 content/attachment hashes for deduplication
├── enrich/         # rule-based + spaCy entity/topic/keyword extraction
├── features/       # feature extraction layer over CanonicalMessage / CanonicalEvent
├── schema/         # Pydantic v2 canonical schemas
├── storage/        # raw JSON, SQLite, filesystem/Obsidian vault, JSON store
├── pipelines/      # end-to-end runners: ingest → features → storage
├── cli/            # mail-fetch CLI entrypoint
└── utils/          # loguru-based structured logging

Orchestration runners

The pipelines/ module provides end-to-end runners that wire ingest → enrich → features → raw JSON → SQLite (optional) in a single call:

from pathlib import Path
from mail_normalizer.pipelines import MsgPipelineRunner, MsgPipelineConfig, StorageConfig

config = MsgPipelineConfig(
    archive_root=Path("path/to/source"),
    storage=StorageConfig(
        raw_root=Path("path/to/raw"),
        enable_features=False,
        db_path=Path("path/to/mail.db"),  # omit to skip SQLite
    ),
)
runner = MsgPipelineRunner(config)
result = runner.run_directory(Path("path/to/source"))
print(f"Processed {result.messages_ok} messages → {result.raw_records_written} JSON, {result.sqlite_records_written} SQLite rows")

Runners available: MsgPipelineRunner, MboxPipelineRunner, EmlPipelineRunner, IcsPipelineRunner.

Two-stage parsing contract: parsers produce an intermediate representation; the normalize/ module consumes it and produces CanonicalMessage. These two stages are never collapsed.

Attachment pipeline

attachment → MIME detection → specialized parser → Tika fallback → OCR → normalized attachment object

Supports nested attachments, embedded MSG files, and TNEF/winmail.dat.

---

Key Design Principles

• Deterministic — same input always produces same output; no nondeterministic LLM calls in core normalization.
• Fault-tolerant — malformed mail never crashes the pipeline; all failures are structured, logged, and recoverable.
• Streaming / low-memory — supports millions of emails without loading entire archives into RAM.
• Source-preserving — raw headers, original HTML, threading metadata, and attachment metadata are never discarded.
• Storage-agnostic — normalization engine does not depend on any specific database implementation.
• Security-conscious — all input is treated as untrusted; defends against decompression bombs, recursive attachment abuse, and HTML injection.

---

Storage Backends

Backend	Status
Raw JSON (raw/ flat files)	✅ Implemented — primary normalizer output
Filesystem / Obsidian vault	✅ Implemented — reserved for AI agent stage
SQLite	✅ Implemented — storage/sqlite_store/
PostgreSQL	🔲 Planned
Vector DB (Qdrant)	🔲 Planned

The normalizer writes canonical JSON to raw/ only. The storage/filesystem/ (Obsidian vault writer) is intentionally kept intact for the downstream AI agent that will produce the human-readable knowledge base.

After every run, log.md is written (overwritten) to the raw root with a summary of counts, duration, and any errors.

---

Development

uv run pytest                    # full test suite (~613 tests)
uv run pytest -m "not live"      # skip tests requiring Outlook/win32com
uv run pytest -m live            # live Outlook integration tests only
uv run ruff check .              # lint
uv run ruff format .             # format
uv run mypy .                    # type check

PowerShell Outlook export

The bundled script is shipped inside the wheel at mail_normalizer/scripts/Export-OutlookItems.ps1 and invoked automatically by mail-fetch outlook. You can also run it directly:

# Incremental export of Inbox
.\scripts\Export-OutlookItems.ps1 -OutDir C:\MailArchive\ingest

# Personal calendar only, since 2023, up to today
.\scripts\Export-OutlookItems.ps1 -OutDir C:\MailArchive\ingest `
    -IncludeCalendar `
    -CalendarName "john@company.com" `
    -CalendarSince "2023-01-01"

Key script parameters:

Parameter	Default	Description
-OutDir	(required)	Output folder for .msg / .ics files
-MailFolder	Inbox	Outlook mail folder (well-known name or backslash path)
-Mode	incremental	incremental (since last run) or all
-IncludeCalendar	off	Also export calendar appointments
-CalendarFolder	Calendar	Outlook calendar folder name
-CalendarName	(any)	Restrict to stores whose name contains this string — use to skip public-holiday or shared calendars
-CalendarSince	(none)	Only export events on or after this date (e.g. 2023-01-01)
-CalendarUntil	today	Only export events on or before this date
-MaxItems	10000	Per-folder export cap

---

Technology Stack

Concern	Library
MIME / email parsing	stdlib email, mailbox, policy, headerregistry
Outlook MSG	extract-msg
PST	libpff-python (optional)
iCalendar	icalendar
Quote stripping	Mailgun Talon (optional) / custom heuristics
HTML cleaning	beautifulsoup4, lxml, readability-lxml, html2text
Document extraction	markitdown (DOCX, PPTX, XLSX)
NLP / entity extraction	spaCy
Logging	loguru (JSON file + coloured console)
Schema / validation	pydantic v2

---

Roadmap

• PostgreSQL storage backend
• Async / parallel attachment extraction
• OCR via Tesseract
• Apache Tika fallback parser
• Live incremental mail sync

See project_plan.md for the full specification and implementation status.