scriptory/import-outlook-mail.py

#!/usr/bin/env python3
"""Bulk-import Outlook .eml files into the Parallels mail archive."""

import argparse
import csv
import fcntl
import hashlib
import os
import re
import shutil
import unicodedata
from datetime import timezone
from email import policy
from email.parser import BytesHeaderParser
from email.utils import parsedate_to_datetime
from pathlib import Path

DEFAULT_SOURCE = Path("~/Seafile/buffer/mail").expanduser()
DEFAULT_ARCHIVE = Path("~/Seafile/doc/job/Parallels/mail").expanduser()
INDEX_FIELDS = ("sha256", "message_id", "date", "subject", "archive_path")


def sha256(path):
    digest = hashlib.sha256()
    with path.open("rb") as file:
        for chunk in iter(lambda: file.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def clean(value):
    return " ".join(str(value or "").splitlines()).strip()


def slug(value):
    value = unicodedata.normalize("NFKD", value)
    value = value.encode("ascii", "ignore").decode().lower()
    value = re.sub(r"[^a-z0-9]+", "-", value).strip("-")
    return value[:80].rstrip("-") or "message"


def read_headers(path):
    with path.open("rb") as file:
        message = BytesHeaderParser(policy=policy.default).parse(file)

    date_header = clean(message.get("Date"))
    subject = clean(message.get("Subject")) or "(no subject)"
    message_id = clean(message.get("Message-ID")).strip("<>").lower()
    prefix = "unknown-date"
    year = "unknown-date"
    month = None

    try:
        date = parsedate_to_datetime(date_header)
        if date is not None and date.tzinfo is not None:
            date = date.astimezone(timezone.utc)
            prefix = date.strftime("%Y%m%d-%H%M%SZ")
            year = date.strftime("%Y")
            month = date.strftime("%m")
        elif date is not None:
            prefix = date.strftime("%Y%m%d-%H%M%S-local")
            year = date.strftime("%Y")
            month = date.strftime("%m")
    except (TypeError, ValueError, OverflowError):
        pass

    return {
        "message_id": message_id,
        "date": date_header,
        "subject": subject,
        "prefix": prefix,
        "year": year,
        "month": month,
    }


def scan_archive(archive):
    rows = []
    for path in sorted(archive.rglob("*.eml")) if archive.exists() else []:
        if not path.is_file() or path.is_symlink():
            continue
        digest = sha256(path)
        headers = read_headers(path)
        rows.append(
            {
                "sha256": digest,
                "message_id": headers["message_id"],
                "date": headers["date"],
                "subject": headers["subject"],
                "archive_path": str(path.relative_to(archive)),
            }
        )
    return rows


def write_index(archive, rows):
    temporary = archive / ".index.csv.tmp"
    with temporary.open("w", encoding="utf-8", newline="") as file:
        writer = csv.DictWriter(file, fieldnames=INDEX_FIELDS)
        writer.writeheader()
        writer.writerows(sorted(rows, key=lambda row: row["archive_path"]))
        file.flush()
        os.fsync(file.fileno())
    os.replace(temporary, archive / "index.csv")


def destination(archive, headers, digest):
    directory = archive / headers["year"]
    if headers["month"]:
        directory /= headers["month"]
    stem = f"{headers['prefix']}-{slug(headers['subject'])}"
    candidates = (
        directory / f"{stem}.eml",
        directory / f"{stem}-{digest[:8]}.eml",
        directory / f"{stem}-{digest}.eml",
    )
    for path in candidates:
        if not path.exists() or sha256(path) == digest:
            return path
    raise FileExistsError(f"all destination names are occupied for {stem}")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--source", type=Path, default=DEFAULT_SOURCE)
    parser.add_argument("--archive", type=Path, default=DEFAULT_ARCHIVE)
    args = parser.parse_args()

    source = args.source.expanduser().resolve()
    archive = args.archive.expanduser().resolve()
    if not source.is_dir():
        parser.error(f"source directory does not exist: {source}")
    if source == archive or source in archive.parents or archive in source.parents:
        parser.error("source and archive directories must not overlap")

    archive.mkdir(parents=True, exist_ok=True)
    lock = (archive / ".import.lock").open("w")
    fcntl.flock(lock, fcntl.LOCK_EX)

    rows = scan_archive(archive)
    hashes = {row["sha256"] for row in rows}
    message_ids = {row["message_id"] for row in rows if row["message_id"]}
    try:
        files = [
            path
            for path in source.rglob("*")
            if path.is_file() and not path.is_symlink()
        ]
    except OSError as error:
        parser.error(f"could not scan source directory: {error}")
    messages = [path for path in files if path.suffix.lower() == ".eml"]
    unsupported = [path for path in files if path.suffix.lower() != ".eml"]
    imported = duplicates = variants = failures = 0

    for path in sorted(messages):
        try:
            digest = sha256(path)
            if digest in hashes:
                print(f"DUPLICATE {path.relative_to(source)}")
                duplicates += 1
                continue

            headers = read_headers(path)
            target = destination(archive, headers, digest)
            is_variant = bool(
                headers["message_id"] and headers["message_id"] in message_ids
            )
            label = "VARIANT" if is_variant else "IMPORT"
            print(f"{label} {path.relative_to(source)} -> {target.relative_to(archive)}")

            target.parent.mkdir(parents=True, exist_ok=True)
            temporary = target.with_name(f".{target.name}.{os.getpid()}.tmp")
            try:
                shutil.copy2(path, temporary)
                if sha256(temporary) != digest:
                    raise OSError("copied file failed SHA-256 verification")
                os.link(temporary, target)
            finally:
                temporary.unlink(missing_ok=True)

            rows.append(
                {
                    "sha256": digest,
                    "message_id": headers["message_id"],
                    "date": headers["date"],
                    "subject": headers["subject"],
                    "archive_path": str(target.relative_to(archive)),
                }
            )

            hashes.add(digest)
            if headers["message_id"]:
                message_ids.add(headers["message_id"])
            if is_variant:
                variants += 1
            else:
                imported += 1
        except (OSError, ValueError) as error:
            print(f"FAILED {path.relative_to(source)}: {error}")
            failures += 1

    for path in sorted(unsupported):
        print(f"UNSUPPORTED {path.relative_to(source)}")

    write_index(archive, rows)

    print(
        f"{imported} imported, {variants} variants, {duplicates} duplicates, "
        f"{len(unsupported)} unsupported, {failures} failures"
    )
    return 1 if failures or unsupported else 0


if __name__ == "__main__":
    raise SystemExit(main())