# Copyright Kevin Deldycke <kevin@deldycke.com> and contributors.
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
"""Utilities to read and write mail boxes in various formats.
Based on [Python's standard library mailbox module](https://docs.python.org/3.11/library/mailbox.html).
"""
from __future__ import annotations
import logging
import mailbox
import os
from email.generator import BytesGenerator
from enum import Enum, auto
from functools import partial
from mailbox import MH, MMDF, Babyl, ExternalClashError, Mailbox, Maildir, mbox
from pathlib import Path
from typing import cast
from uuid import uuid4
from click_extra import ClickException, get_current_theme
from .mail import DedupMailMixin
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Callable, Iterator
[docs]
class MailSourceError(ClickException):
"""A mail source that cannot be used as given.
Raised for a problem the user can fix: a folder with no structure mdedup
recognizes, a box another program locked, a source given twice. Click prints it
as a one-line `Error:`, with no traceback and no invitation to report a bug.
"""
[docs]
def maildir_subpath(box: Mailbox, key: str) -> str:
"""Path of a `maildir` mail below its box, read from the table of contents.
A maildir key drops the `:2,<flags>` suffix the file name carries, so the name
cannot be rebuilt from the key alone. The box refreshed its table of contents to
hand out the mail in the first place, so it is only re-validated on a miss.
"""
try:
return cast("str", box._toc[key]) # type: ignore[attr-defined]
except KeyError:
return cast("str", box._lookup(key)) # type: ignore[attr-defined]
[docs]
def maildir_mail_path(box: Mailbox, key: str) -> str:
"""Location of a `maildir` mail, read from the box's table of contents."""
return os.path.join(box._path, maildir_subpath(box, key))
[docs]
def keyed_mail_path(box: Mailbox, key: str) -> str:
"""Location of an `MH` or `eml` mail, whose file is named after its key."""
return os.path.join(box._path, str(key))
[docs]
def box_file_path(box: Mailbox, key: str) -> str:
"""Location of a mail from a file-based box: the box's own single file.
Every mail of an `mbox`, `babyl` or `mmdf` box is packed into it, so they all
share this path and are told apart by their mail ID.
"""
return box._path
[docs]
def iter_mail_ids(box: Mailbox) -> Iterator[str]:
"""Yields the key of every mail held by a box.
`Maildir.iterkeys()` confirms that each key still resolves to a file, one `stat`
per mail, on top of the directory listing it has just built. Every caller here
goes on to stat or open that same file anyway, and copes with its disappearance,
so the check is paid for twice and needed once: reading the refreshed table of
contents directly skips it.
The other formats list their mails without that extra round, and are left to
their own iterator.
"""
if isinstance(box, Maildir):
box._refresh() # type: ignore[attr-defined]
# Iterate a copy: any later refresh rebuilds the table in place.
yield from list(box._toc) # type: ignore[attr-defined]
else:
yield from box.iterkeys()
[docs]
def resolve_mail_path(box: Mailbox, key: str) -> str:
"""Location of a mail in its box, without instantiating the mail.
Lets a caller reach a mail's file before deciding to read it, which is how the
hash cache checks whether a mail changed without paying for its parsing.
"""
factory = box._factory
assert factory is not None, "Box opened without a DedupMail factory."
return cast("str", factory.resolve_path(box, key)) # type: ignore[attr-defined]
[docs]
def make_dedup_mail(
name: str,
base: type,
path_resolver: Callable[[Mailbox, str], str],
) -> type:
"""Create a DedupMail class for a mailbox message type.
Deriving a mail's own location from its box is format-specific, so the resolver
is baked into the class here instead of being branched on at runtime.
"""
return type(
name,
(DedupMailMixin, base),
{"resolve_path": staticmethod(path_resolver)},
)
MaildirDedupMail = make_dedup_mail(
"MaildirDedupMail", mailbox.MaildirMessage, maildir_mail_path
)
mboxDedupMail = make_dedup_mail("mboxDedupMail", mailbox.mboxMessage, box_file_path)
MHDedupMail = make_dedup_mail("MHDedupMail", mailbox.MHMessage, keyed_mail_path)
BabylDedupMail = make_dedup_mail("BabylDedupMail", mailbox.BabylMessage, box_file_path)
MMDFDedupMail = make_dedup_mail("MMDFDedupMail", mailbox.MMDFMessage, box_file_path)
EMLDedupMail = make_dedup_mail("EMLDedupMail", mailbox.Message, keyed_mail_path)
[docs]
class EML(Mailbox):
"""A folder of loose `.eml` files, walked recursively.
Supports mail archives exported as individual RFC 5322 files, one mail per
file, as produced by Outlook PST/OST conversion tools for instance. See:
https://github.com/kdeldycke/mail-deduplicate/issues/760
Keys are the paths of the mail files, relative to the folder's root. Files
without the `.eml` extension (case-insensitive) are ignored, as well as
hidden files and directories.
Follows the interface of Python's
[mailbox.Mailbox](https://docs.python.org/3/library/mailbox.html#mailbox.Mailbox).
Like `maildir`, the one-file-per-mail storage needs no locking.
"""
def __init__(self, dirname, factory=None, create=True) -> None:
super().__init__(dirname, factory, create)
if not os.path.exists(self._path):
if create:
os.mkdir(self._path, 0o700)
else:
raise mailbox.NoSuchMailboxError(self._path)
def _full_path(self, key: str) -> str:
return os.path.join(self._path, key)
[docs]
def iterkeys(self):
for dirpath, dirnames, filenames in os.walk(self._path):
dirnames[:] = sorted(d for d in dirnames if not d.startswith("."))
for filename in sorted(filenames):
if not filename.startswith(".") and filename.lower().endswith(".eml"):
yield os.path.relpath(os.path.join(dirpath, filename), self._path)
def __contains__(self, key) -> bool:
return key.lower().endswith(".eml") and os.path.isfile(self._full_path(key))
def __len__(self) -> int:
return sum(1 for _ in self.iterkeys())
[docs]
def get_file(self, key):
try:
file = open(self._full_path(key), "rb") # noqa: SIM115
except FileNotFoundError:
raise KeyError(key) from None
# A BufferedReader covers _ProxyFile's runtime needs; typeshed's stricter
# _GetFileReturn protocol (read1/readlines typed with int | None) rejects it.
return mailbox._ProxyFile(file) # type: ignore[arg-type]
[docs]
def get_bytes(self, key) -> bytes:
try:
with open(self._full_path(key), "rb") as file:
return file.read()
except FileNotFoundError:
raise KeyError(key) from None
[docs]
def get_message(self, key):
return mailbox.Message(self.get_bytes(key))
[docs]
def add(self, message) -> str:
key = f"{uuid4().hex}.eml"
# Bytes are written as they are, with LF line endings. The stdlib
# `_dump_message` rewrites `\n` to `os.linesep`, which emits CRLF on Windows
# and makes the same mail serialize (and hash) differently across platforms.
with open(self._full_path(key), "wb") as file:
if isinstance(message, (bytes, bytearray)):
file.write(message)
else:
# A message is written as `_dump_message` writes one: `maxheaderlen=0`
# keeps each header folded as it was parsed, where `as_bytes()` would
# refold them at the 78 columns of the default policy.
BytesGenerator(file, mangle_from_=False, maxheaderlen=0).flatten(
message
)
return key
[docs]
def remove(self, key) -> None:
try:
os.remove(self._full_path(key))
except FileNotFoundError:
raise KeyError(key) from None
def __setitem__(self, key, message) -> None:
"""Replacing a mail in place is not supported.
`EML` keys are content paths and new mails get a fresh UUID filename from
`add()`, so there is no meaningful in-place replacement by key. Overriding
the abstract `mailbox.Mailbox.__setitem__` (which already raises) keeps the
class concrete and instantiable, matching `get_folder` below.
"""
raise NotImplementedError("EML mails are added and removed, not replaced.")
[docs]
def list_folders(self) -> list[str]:
"""No subfolder objects: the recursive walk covers nested directories."""
return []
[docs]
def get_folder(self, folder):
raise NotImplementedError("EML folders are walked recursively instead.")
[docs]
def flush(self) -> None:
"""Mails are written straight to the filesystem: nothing to flush."""
[docs]
def lock(self) -> None:
"""One-file-per-mail storage needs no locking."""
[docs]
def unlock(self) -> None:
"""One-file-per-mail storage needs no locking."""
[docs]
def close(self) -> None:
"""No resource is kept open between operations."""
[docs]
class BoxStructure(Enum):
"""Box structures can be file-based or folder-based."""
# We use auto() as we don't care about the actual values here.
FOLDER = auto()
FILE = auto()
FOLDER_FORMATS = tuple(box for box in BoxFormat if box.structure == BoxStructure.FOLDER)
"""Box formats implementing a folder-based structure.
Is a tuple to keep natural order defined by `BoxFormat`.
"""
FILE_FORMATS = tuple(box for box in BoxFormat if box.structure == BoxStructure.FILE)
"""Box formats implementing a file-based structure.
Is a tuple to keep natural order defined by `BoxFormat`.
"""
FOLDER_FORMAT_CLASSES = tuple(box.base_class for box in FOLDER_FORMATS)
"""Base classes of folder-based box formats, as a tuple ready for `isinstance`."""
MAILDIR_SUBDIRS = frozenset(("cur", "new", "tmp"))
"""List of required sub-folders defining a properly structured maildir."""
MBOX_FAMILY = (mbox, MMDF)
"""Formats keeping an envelope `From ` line before each mail.
The standard library reads and writes both with the same code, so the line travels
from one to the other.
"""
[docs]
def is_maildir(path: Path) -> bool:
"""Returns `True` when the path holds all the sub-directories of a maildir."""
return all(path.joinpath(subdir).is_dir() for subdir in MAILDIR_SUBDIRS)
[docs]
def contains_maildir(path: Path) -> bool:
"""Returns `True` when the path is a maildir or holds one at any depth.
Allows the discovery of nested maildir folders stored as plain directories, as
produced by
[isync/mbsync's Verbatim naming style](https://isync.sourceforge.io/mbsync.html).
See: https://github.com/kdeldycke/mail-deduplicate/issues/973
Dot-prefixed directories are ignored, as they are covered by the `Maildir++`
folder convention. The mail-holding sub-directories of maildirs are not
explored either.
"""
if is_maildir(path):
return True
return any(
contains_maildir(sub)
for sub in path.iterdir()
if sub.is_dir()
and not sub.name.startswith(".")
and sub.name not in MAILDIR_SUBDIRS
)
[docs]
def contains_eml(path: Path) -> bool:
"""Returns `True` when the path holds at least one `.eml` file, at any depth.
Hidden files and directories are ignored, and the extension is matched
case-insensitively, mirroring the walk of `EML` boxes.
"""
for dirpath, dirnames, filenames in os.walk(path):
dirnames[:] = [d for d in dirnames if not d.startswith(".")]
if any(not f.startswith(".") and f.lower().endswith(".eml") for f in filenames):
return True
return False
[docs]
def autodetect_box_type(path: Path) -> BoxFormat:
"""Auto-detect the format of the mailbox located at the provided path.
If the path is a file, then it is considered as an `mbox`. Else, if the
provided path is a folder and features the `MAILDIR_SUBDIRS` sub-directories,
or holds nested maildir folders at any depth, it is parsed as a `maildir`.
A folder holding loose `.eml` files instead is parsed as an `eml` source.
```{todo}
Future finer autodetection heuristics should be implemented here. Some ideas:
- single mail from a `maildir`
- plain text mail content
- other mailbox formats supported in Python's standard library:
- `MH`
- `Babyl`
- `MMDF`
```
"""
box_format = None
# Validates folder as a maildir, either by its own structure or by the nested
# maildir folders it contains. Falls back to a folder of loose .eml files.
if path.is_dir():
if contains_maildir(path):
box_format = BoxFormat.MAILDIR
elif contains_eml(path):
box_format = BoxFormat.EML
else:
raise MailSourceError(
f"Unrecognized folder: no {'/'.join(sorted(MAILDIR_SUBDIRS))} "
"maildir structure, no nested maildir folders, and no .eml files "
f"found in {path}. Force a format with --input-format."
)
# A single file is read as an mbox.
elif path.is_file():
box_format = BoxFormat.MBOX
if not box_format:
raise MailSourceError(f"Unrecognized mail source type: {path}")
logging.info(f"{get_current_theme().choice(str(box_format))} detected.")
return box_format
[docs]
def open_box(
path: Path,
box_format: BoxFormat | None = None,
force_unlock: bool = False,
) -> list[Mailbox]:
"""Open a mail box.
Returns a list of boxes, one per sub-folder. All are locked, ready for operations.
If `box_format` is provided, forces the opening of the box in the specified format.
Else, defaults to autodetection.
"""
logging.info(f"Opening {get_current_theme().choice(str(path))} ...")
if not box_format:
box_format = autodetect_box_type(path)
else:
logging.warning(f"Forcing {box_format} format.")
# Do not allow the constructor to create a new mailbox if not found.
box = box_format.constructor(path, create=False)
return open_subfolders(box, force_unlock)
[docs]
def lock_box(box: Mailbox, force_unlock: bool) -> Mailbox:
"""Lock an opened box and allows for forced unlocking.
Returns the locked box. Raises `MailSourceError` if another lock holds the box
and `force_unlock` is not set.
"""
try:
logging.debug("Locking box...")
box.lock()
except ExternalClashError as expt:
if not force_unlock:
# Release the file handle before aborting. On Windows a lingering open
# handle keeps the box file locked, so a later `--force-unlock` run in the
# same process cannot rewrite it (WinError 32).
release_box(box)
raise MailSourceError(
f"{box._path} is already locked. If no other program uses it, run "
"again with --force-unlock."
) from expt
logging.warning("Box already locked! Forcing removal of lock...")
box._locked = True # type: ignore[attr-defined]
box.unlock()
box.lock()
logging.debug("Box opened.")
return box
[docs]
def release_box(box: Mailbox) -> None:
"""Unlock a box and close its file, without writing its pending changes.
`Mailbox.close()` first writes the pending changes of a single-file box, which
rewrites the whole file when a mail was removed. A run cut short by an error or
an interrupt calls this instead: its single-file boxes stay as the run found
them, and an interrupt does not start a rewrite.
Unlike `MH.close()`, this also closes the `.mh_sequences` file that a refused
`MH.lock()` leaves open.
"""
box.unlock()
# Single-file boxes hold their file open between operations.
file = getattr(box, "_file", None)
if file is not None:
file.close()
[docs]
def open_subfolders(box: Mailbox, force_unlock: bool) -> list[Mailbox]:
"""Browse recursively the subfolder tree of a box.
Returns a list of opened and locked boxes, each for one subfolder. If a subfolder
cannot be locked, because another program holds it for instance, the boxes locked
before it are unlocked again and the error is raised: the next run then needs no
`--force-unlock` for them.
Skips box types not supporting subfolders. For `maildir`, both the
`Maildir++` convention (dot-prefixed folders) and Verbatim-style layouts
(nested plain directories, each a maildir of its own) are browsed. A directory
without the maildir structure only acts as a container of nested folders and
carries no mail of its own.
"""
folder_list = []
try:
if isinstance(box, Maildir) and not is_maildir(Path(box._path)):
logging.info("No mails at this level: only browse nested folders.")
else:
folder_list.append(lock_box(box, force_unlock))
if isinstance(box, FOLDER_FORMAT_CLASSES):
# Asserts to please the type checker.
assert hasattr(box, "list_folders")
assert hasattr(box, "get_folder")
for folder_id in box.list_folders():
logging.info(f"Opening subfolder {folder_id} ...")
folder_list += open_subfolders(box.get_folder(folder_id), force_unlock)
# Python's mailbox module only lists dot-prefixed Maildir++ folders, so
# browse the filesystem for Verbatim-style nested maildir folders.
if isinstance(box, Maildir):
for sub_path in sorted(Path(box._path).iterdir()):
if (
sub_path.is_dir()
and not sub_path.name.startswith(".")
and sub_path.name not in MAILDIR_SUBDIRS
and contains_maildir(sub_path)
):
logging.info(f"Opening subfolder {sub_path.name} ...")
sub_box = Maildir(sub_path, factory=box._factory, create=False)
folder_list += open_subfolders(sub_box, force_unlock)
except BaseException:
# A nested call that failed unlocked its own boxes already: unlock the ones
# this call locked before the error.
for locked in folder_list:
release_box(locked)
raise
return folder_list
[docs]
def carry_mh_sequences(source: MH, key: str, target: MH, new_key: str) -> None:
"""Enroll a copied `MH` mail in the sequences of its original.
MH keeps the state of a mail (`unseen`, `replied`, `flagged`) in the
`.mh_sequences` file of its folder, by the number of the mail. `MH.add()` takes
them from an `MHMessage` alone, so they are written here for a mail added as
bytes, the way `MH.add()` writes them.
"""
names = [name for name, keys in source.get_sequences().items() if int(key) in keys]
if not names:
return
sequences = target.get_sequences()
for name in names:
sequences.setdefault(name, []).append(int(new_key))
target.set_sequences(sequences)
[docs]
def add_mail(box: Mailbox, mail: DedupMailMixin) -> None:
r"""Add a mail to a box, as the bytes its source box holds.
Every box format writes a mail handed over as a message with `BytesGenerator`,
from the message parsed out of it. The generator does not write the mail back:
it rebuilds it. It redraws each multipart boundary, closes one left open, turns
a bare `\r` into a line break, and cannot write at all a multipart part parsed
with no subpart, which
[python/cpython#136686](https://github.com/python/cpython/issues/136686) tracks.
Deduplicating a mail is no reason to repair it, so every mail goes in as bytes,
which each format writes as they are. See:
https://github.com/kdeldycke/mail-deduplicate/issues/1090
Line endings are the one conversion left. They are reduced to LF before the
write, as every box format but `eml` writes `\n` as the line separator of the
platform, and would double the CR of a CRLF ending on Windows. An mbox or MMDF
box also quotes the body lines starting with `From `, as its format requires.
The bytes do not include what a format keeps beside each mail: the envelope
`From ` line of an mbox or MMDF mail, the folder, flags and modification time of
a maildir mail, the sequences of an MH mail, the labels of a Babyl mail. The
first three travel with the mail into a box of the same format, and the
envelope line also travels between mbox and MMDF, which keep it alike. A box of
another format has no place for them, and nothing is translated: a maildir mail
copied into an mbox gets no `Status` header. Babyl labels are not carried, as
the standard library takes them from a `BabylMessage` alone.
"""
source = mail.box
key = mail.mail_id
# Asserts to please the type checker: every mail reaching an action was read
# from its box.
assert source is not None
assert key is not None
if isinstance(source, MBOX_FAMILY) and isinstance(box, MBOX_FAMILY):
raw = source.get_bytes(key, from_=True)
else:
raw = source.get_bytes(key)
new_key = box.add(raw.replace(b"\r\n", b"\n"))
if isinstance(source, Maildir) and isinstance(box, Maildir):
carry_maildir_metadata(source, key, box, new_key)
elif isinstance(source, MH) and isinstance(box, MH):
carry_mh_sequences(source, key, box, new_key)
[docs]
def create_box(
path: Path,
box_format: BoxFormat,
export_append: bool = False,
) -> Mailbox:
"""Creates a brand new box from scratch."""
theme = get_current_theme()
logging.info(
f"Creating new {theme.choice(str(box_format))} box "
f"at {theme.choice(str(path))} ..."
)
if path.exists() and export_append is not True:
raise FileExistsError(path)
# Allow the constructor to create a new mail box as we already double-checked
# beforehand it does not exist.
box: Mailbox = box_format.constructor(path, create=True)
logging.debug("Locking box...")
box.lock()
return box