#!/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())