Metadata-Version: 2.4
Name: mailnorm
Version: 0.1.0
Summary: Normalize heterogeneous email archives (MSG, MBOX, EML, ICS) into a canonical AI/RAG-ready schema
Author: ocastrup
License: MIT
Keywords: email,normalization,mbox,msg,eml,ical,RAG,LLM
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Communications :: Email
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.12
Description-Content-Type: text/markdown
Requires-Dist: pydantic>=2.0
Requires-Dist: extract-msg>=0.48
Requires-Dist: beautifulsoup4
Requires-Dist: lxml
Requires-Dist: compressed-rtf
Requires-Dist: html2text
Requires-Dist: readability-lxml
Requires-Dist: pyyaml>=6.0.3
Requires-Dist: spacy>=3.7
Requires-Dist: markitdown[docx,pptx,xlsx]>=0.1.5
Requires-Dist: icalendar>=7.1.1
Requires-Dist: loguru>=0.7.3
Requires-Dist: fastmcp>=2.0
Provides-Extra: pst
Requires-Dist: libpff-python; extra == "pst"
Provides-Extra: quotes
Requires-Dist: talon; extra == "quotes"

# 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

```bash
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

```bash
# 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

```python
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

```python
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)

```python
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):

```python
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:

```python
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

```bash
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:

```powershell
# 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`](plan.md) for the full specification and implementation status.
