# 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.
"""A mail wrapped with the properties deduplication needs.
They are the canonical hash, the normalized headers, the timestamp, the size, and the
dehydration that keeps memory flat.
"""
from __future__ import annotations
import copy
import email.utils
import hashlib
import logging
import os
import re
from contextlib import contextmanager, suppress
from datetime import UTC, datetime
from email.header import Header
from enum import StrEnum
from functools import cached_property
from mailbox import Message
from typing import cast
from click_extra import get_current_context, render_table
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Callable, Generator, Iterator
from mailbox import Mailbox, _ProxyFile
from .cli import Config
[docs]
class TimeSource(StrEnum):
"""Enumeration of all supported mail timestamp sources."""
DATE_HEADER = "date-header"
"""Timestamp sourced from the message's `Date` header."""
CTIME = "ctime"
"""Timestamp is from the email's file on the filesystem.
```{attention}
Only meaningful for sources storing one mail per file, like `maildir`
and `eml`.
```
"""
ADDRESS_HEADERS = frozenset((
"from",
"to",
"cc",
"bcc",
"reply-to",
"sender",
"return-path",
"resent-from",
"resent-to",
"resent-cc",
"resent-bcc",
"resent-reply-to",
"resent-sender",
"delivered-to",
"x-original-to",
"envelope-to",
"x-envelope-from",
"x-envelope-to",
"disposition-notification-to",
"original-recipient",
))
"""Headers that contain email addresses.
```{hint}
Headers from which quotes should be discarded, so `"Bob" <bob@example.com>` hashes
to the same thing as `Bob <bob@example.com>`.
```
```{attention}
These IDs should be kept lower-case, because they are compared to the IDs provided
to the `-h`/`--hash-header` option, carried by the `hash_headers` entry of the
configuration.
```
"""
HEADER_LINE = re.compile(rb"\A(From |[!-9;-~]*:|[\t ])")
"""A line of the header block, as the parser reads one: an envelope line, a header
or a folded continuation.
The parser ends the headers on the first blank line, or on the first line that is
none of these when the blank line is missing. `split_raw()` follows the same rule.
"""
LINE = re.compile(rb"[^\r\n]*[\r\n]|[^\r\n]+\Z")
"""One line of a mail with LF line endings, with its line break when it has one.
A bare CR also ends a line, as it does for the parser.
"""
[docs]
class DedupMailMixin(Message):
"""Message with deduplication-specific properties and utilities.
Extends [standard library's mailbox.Message](https://github.com/python/cpython/blob/061965c/Lib/mailbox.py#L1564-L1598),
and shouldn't be used directly, but composed with `mailbox.Message` sub-classes.
"""
CONTENT_CACHES: tuple[str, ...] = (
"body_lines",
"canonical_headers",
"hash_decoded_body",
"hash_normalized_body",
"hash_raw_body",
)
"""Memoized properties derived from the message content.
They are only needed while computing hashes, and are dropped by `dehydrate()`
alongside the parsed message itself.
"""
resolve_path: Callable[[Mailbox, str], str]
"""Derives the real filesystem location of a mail from its box.
Set per box format by `make_dedup_mail()`, as folder-based boxes give each mail
its own file while file-based ones pack them all into the box's single file.
"""
defects: list = () # type: ignore[assignment]
"""Fallback for the parsing defects dropped by `dehydrate()`.
An empty tuple rather than a list, so it can be shared by every dehydrated mail
instead of costing one throw-away empty list each: reads still work, and the
appends only the parser performs fail loudly on a mail that has no message.
"""
PARSED_MESSAGE_ATTRS: tuple[str, ...] = (
"_headers",
"_payload",
"_unixfrom",
"preamble",
"epilogue",
"defects",
)
"""Attributes carrying the parsed message, as populated by `email.parser`.
They hold the bulk of a mail's memory footprint, and are the ones dropped by
`dehydrate()` and restored by `hydrate()`.
"""
def __init__(self, message: _ProxyFile | None = None) -> None:
super().__init__(message)
self.box: Mailbox | None = None
"""The box this message was read from, kept to re-fetch its content on
demand after dehydration."""
self.source_path: str | None = None
"""Normalized path to the mailbox this message originates from."""
self.mail_id: str | None = None
"""Mail ID used to uniquely refers to it in the context of its source."""
self.conf: Config
"""Global configuration"""
_path_override: str | None = None
"""Location of a mail read straight from its file, outside of any box.
A class attribute, so it costs nothing on the mails that do not need it. Only the
parallel hashing workers set it: they open a mail by path, and so have no box to
derive that path back from.
"""
@property
def path(self) -> str:
"""Real filesystem location of the mail.
Returns the individual mail's file for folder-based box types (`maildir` &
co.), but returns the whole box path for file-based boxes (`mbox` & co.).
Used by regexp-based selection strategies and to render the mail's repr.
Derived on access from the box rather than stored, so a retained mail does
not carry a copy of its own absolute path for the whole run. Raises
`AttributeError` before the box metadata is attached, so `getattr(mail,
"path", None)` still reads as absent.
"""
if self._path_override is not None:
return self._path_override
if self.box is None or self.mail_id is None:
raise AttributeError("No box metadata attached to this mail yet.")
return self.resolve_path(self.box, self.mail_id)
def __deepcopy__(self, memo: dict) -> DedupMailMixin:
"""Deep-copy the mail while sharing its box and configuration references.
`email.generator` deep-copies a message to flatten 8-bit payloads without
mutating the original, which `str(mail)` triggers. The parent box can
carry open file handles that cannot be copied, and both the box and the
configuration are shared references by design, so they are passed through
as-is.
"""
clone = copy.copy(self)
memo[id(self)] = clone
for name, value in vars(self).items():
if name not in ("box", "conf"):
setattr(clone, name, copy.deepcopy(value, memo))
return clone
@property
def is_hydrated(self) -> bool:
"""Whether the mail still carries its parsed message."""
return hasattr(self, "_payload")
[docs]
def dehydrate(self) -> None:
"""Reduce the mail to the lightweight metadata needed by the next steps.
Memoizes the scalar properties consumed by selection strategies (timestamp,
and size when the decoded body is at hand), then drops the parsed message and
every cached copy of its content. This cuts the resident footprint of a
retained mail from the full size of its message to a few hundred bytes,
whatever the number of mails processed. See:
https://github.com/kdeldycke/mail-deduplicate/issues/761
No-op if the mail is already dehydrated. The dropped content is re-read from
the source box by `hydrate()` when a later step needs it again.
"""
if not self.is_hydrated:
return
# Memoize the cheap scalars while the parsed message is still available.
_ = self.timestamp
if "body_lines" in self.__dict__:
_ = self.size
for name in self.CONTENT_CACHES:
self.__dict__.pop(name, None)
# Drop the parsed message. The two core attributes are deleted rather than
# emptied, so any unforeseen direct access fails loudly instead of silently
# misreading the mail as empty.
del self._headers # type: ignore[attr-defined]
del self._payload # type: ignore[attr-defined]
self._unixfrom = None
self.preamble = None
self.epilogue = None
# Fall back to the shared empty tuple on the class instead of holding a
# per-mail empty list.
del self.defects
[docs]
def hydrate(self) -> None:
"""Restore the full parsed message dropped by `dehydrate()`.
Re-reads the mail from its source box, through the same parsing path that
produced it in the first place, so the restored content is identical.
No-op if the mail still carries its parsed message.
"""
if self.is_hydrated:
return
# Asserts to please the type checker.
assert self.box is not None
assert self.mail_id is not None
fresh = self.box[self.mail_id]
for name in self.PARSED_MESSAGE_ATTRS:
setattr(self, name, getattr(fresh, name))
[docs]
def raw_bytes(self) -> bytes:
"""Read the bytes of the mail again, from its own file or from its box.
The parser consumed them without keeping them. A mail a worker opened by
path is read from that file. A mail of the main process is read from its
box, which hands it out without the envelope line and the state its format
keeps beside each mail.
"""
if self._path_override is not None:
with open(self._path_override, "rb") as file:
return file.read()
if self.box is None or self.mail_id is None:
raise LookupError("The mail has no box to be read from.")
return self.box.get_bytes(self.mail_id)
[docs]
@contextmanager
def hydrated(self) -> Generator[DedupMailMixin, None, None]:
"""Borrow the full parsed message for the duration of the block.
Restores the message on the way in and releases it on the way out, so a step
needing the whole mail after the hashing one keeps its content resident only
while it uses it, and memory stays flat across a loop of mails.
"""
self.hydrate()
try:
yield self
finally:
self.dehydrate()
def __repr__(self) -> str:
"""Renders the fully-qualified path of the mail's own file.
The path can be copy-pasted as-is for direct inspection.
Mails from file-based boxes share the box's path, so the mail ID is appended
to tell them apart. See:
https://github.com/kdeldycke/mail-deduplicate/issues/157
A mail with no box attached, or one that went away from its box, has no path
to render: the box path and mail ID name it instead, as warnings and bug
reports print them.
"""
try:
path = self.path
except (AttributeError, KeyError, OSError):
path = None
if path and path != self.source_path:
return f"<{self.__class__.__name__} {path}>"
return f"<{self.__class__.__name__} {self.source_path}:{self.mail_id}>"
[docs]
@cached_property
def parsed_date(self) -> float | None:
"""Parse the mail's date header into float timestamp.
Returns `None` if the mail has no valid date header, or one whose year no
timestamp can hold.
Self-hydrating: re-reads the message from its box if it was dehydrated.
"""
self.hydrate()
value = next(self.header_values("Date"), None)
parsed = email.utils.parsedate_tz(value)
if parsed:
# `parsedate_tz()` accepts a year like 99999, which `mktime_tz()` then
# refuses: `datetime` stops at 9999.
with suppress(OverflowError, ValueError):
return float(email.utils.mktime_tz(parsed))
logging.debug(f"Mail {self!r} has no valid Date header: {value!r}")
return None
[docs]
@cached_property
def timestamp(self) -> float | None:
"""Compute the normalized canonical timestamp of the mail.
Sourced from the message's `Date` header by default. In the case of
`maildir`, can be sourced from the email's file from the filesystem.
```{warning}
`ctime` does not refer to creation time on POSIX systems, but
rather [the time of the last metadata change](https://docs.python.org/3/library/os.path.html#os.path.getctime).
```
```{todo}
Investigate what [mailbox.MaildirMessage.get_date()](https://docs.python.org/3.11/library/mailbox.html#mailbox.MaildirMessage.get_date)
does and if we can use it.
```
"""
if self.conf["time_source"] == TimeSource.CTIME:
return os.path.getctime(self.path)
return self.parsed_date
[docs]
@cached_property
def size(self) -> int:
"""Returns the canonical size of the mail, in bytes.
The size is the byte length, in UTF-8, of the decoded body: the payload of
the mail stripped of its headers, its transfer encodings undone and its text
decoded, never the size of the mail on disk. Copies of a mail that differ in
their transfer encoding alone measure the same, and `--size-threshold`
compares them in the unit it names.
```{todo}
Allow customization of the way the size is computed, by getting the file
size instead, for example with `os.path.getsize(mail_file)`.
```
"""
return len("".join(self.body_lines).encode("utf-8"))
[docs]
@cached_property
def body_lines(self) -> list[str]:
"""Return the lines of the decoded body of the message.
Each text part has its transfer encoding undone and its text decoded with
its charset, so copies of a mail that differ in their encoding alone read the
same. A part that is not text is kept as the mail carries it. The preamble
and epilogue of a multipart are read as UTF-8.
Self-hydrating: re-reads the message from its box if it was dehydrated.
"""
self.hydrate()
body: list[str] = []
if self.preamble is not None:
body.extend(self.undo_surrogateescape(self.preamble).splitlines())
for part in self.walk():
if part.is_multipart():
continue
body.extend(self.decode_part(part).splitlines())
if self.epilogue is not None:
body.extend(self.undo_surrogateescape(self.epilogue).splitlines())
return body
[docs]
def decode_part(self, part: Message) -> str:
"""Decode a single message part to string."""
ctype = part.get_content_type()
cte = part.get_params(header="Content-Transfer-Encoding")
# Binary or 8bit content: return as-is
if (ctype and not ctype.startswith("text")) or (
cte and cte[0][0].lower() == "8bit"
):
return cast("str", part.get_payload(decode=False))
# Try to decode with charset(s)
charset = part.get_content_charset()
charsets = [charset] if charset else ["ascii", "utf-8"]
part_body_bytes = part.get_payload(decode=True)
if isinstance(part_body_bytes, bytes):
for enc in charsets:
try:
return part_body_bytes.decode(enc)
except (UnicodeDecodeError, LookupError):
continue
return cast("str", part.get_payload(decode=False))
[docs]
def undo_surrogateescape(self, text: str, errors: str = "replace") -> str:
"""Decode as UTF-8 the non-ASCII bytes that parsing left as lone surrogates.
A mail parsed from bytes keeps each non-ASCII byte as a lone surrogate, and
encoding such a string to UTF-8 raises `UnicodeEncodeError`. `get_payload()`
undoes this for each part, with the charset of the part. The preamble and the
epilogue have no charset, so they go through here instead. Bytes that are not
valid UTF-8 become `U+FFFD`, as `get_payload()` does with its charset, unless
`errors` names another way out, like `backslashreplace`.
Encoding with `utf-8` instead of `ascii` keeps the non-ASCII characters of a
mail parsed from a string, which `ascii` rejects.
"""
return text.encode("utf-8", "surrogateescape").decode("utf-8", errors)
[docs]
def hash_key(self) -> str:
"""Returns the canonical hash of a mail.
Under `--raw-headers`, the hash is that of the header block, as
`raw_headers()` reads it: every header takes part, so no minimal number of
them applies.
```{caution}
This method hasn't been made explicitly into a cached property in order to
reduce the overall memory footprint.
```
"""
if self.conf["raw_headers"]:
hash_value = hashlib.sha224(self.raw_headers()).hexdigest()
logging.debug(f"Header block hash: {hash_value}")
return hash_value
serialized_headers = self.serialized_headers()
logging.debug(f"Serialized headers: {serialized_headers!r}")
hash_value = hashlib.sha224(serialized_headers).hexdigest()
logging.debug(f"Hash: {hash_value}")
return hash_value
[docs]
def split_raw(self) -> tuple[bytes, bytes]:
"""The bytes of the mail as the box holds them, cut into headers and body.
Line endings are reduced to LF, as `add_mail()` writes them: every box format
rewrites them to the separator of the platform, and the mbox reader reduces
them on Windows, so a mail would otherwise never hash like its own export,
and a hash would depend on the platform. Nothing else is interpreted. A
header a server or a client added, an mbox `>From ` quote and a transfer
encoding all count.
The header block ends with the first blank line, which it keeps. When that
blank line is missing, it ends before the first line that is neither a
header nor a continuation, where the parser starts the body too. A mail of
headers alone has an empty body. The two halves always join back into the
whole mail, so `--raw-headers` with `--hash-body raw` hashes every byte.
"""
raw = self.raw_bytes().replace(b"\r\n", b"\n")
for match in LINE.finditer(raw):
line = match.group()
if line in (b"\n", b"\r"):
return raw[: match.end()], raw[match.end() :]
if not HEADER_LINE.match(line):
return raw[: match.start()], raw[match.start() :]
return raw, b""
[docs]
def raw_body(self) -> bytes:
"""The body bytes of the mail, as `split_raw()` cuts them."""
return self.split_raw()[1]
[docs]
@cached_property
def hash_raw_body(self) -> str:
"""Returns the hash of the body bytes, as the box holds them.
Nothing but the line endings is interpreted, see `raw_body()`: a transfer
encoding, a charset, a bare CR and an mbox `>From ` quote all count, where
`hash_decoded_body` reads through the first two.
"""
hash_value = hashlib.sha224(self.raw_body()).hexdigest()
logging.debug(f"Body raw hash: {hash_value}")
return hash_value
[docs]
@cached_property
def hash_decoded_body(self) -> str:
"""Returns the hash of the decoded body of a mail, line by line."""
serialized_decoded_body = "\n".join(self.body_lines).encode("utf-8")
hash_value = hashlib.sha224(serialized_decoded_body).hexdigest()
logging.debug(f"Body decoded hash: {hash_value}")
return hash_value
[docs]
@cached_property
def hash_normalized_body(self) -> str:
"""Returns the hash of the decoded body of a mail, stripped of whitespace."""
serialized_normalized_body = "".join(
[re.sub(r"\s", "", line) for line in self.body_lines],
).encode("utf-8")
hash_value = hashlib.sha224(serialized_normalized_body).hexdigest()
logging.debug(f"Body normalized hash: {hash_value}")
return hash_value
[docs]
def normalize_subject(self, subject: str) -> str:
"""Strip `Re:`/`Fwd:` and `[list-name]` prefixes from `Subject`.
This cleans up prefixes automatically added by mailing list software, since the
mail could have been `CC`'d to multiple lists, in which case it will receive a
different prefix for each.
"""
return re.sub(r"(?i)^(?:(?:re|fwd?): +|\[\w[\w_-]*\w?\] +)+", "", subject)
[docs]
def normalize_content_type(self, value: str) -> str:
"""Normalize `Content-Type` by stripping parameters.
Removes everything after the semicolon, keeping only the MIME type.
E.g., `text/plain; charset=utf-8` becomes `text/plain`.
Apparently list servers actually munge `Content-Type` e.g. by stripping the
quotes from `charset="us-ascii"`. Section 5.1 of RFC2045 says that either form
is valid (and they are equivalent).
Additionally, with multipart/mixed, boundary delimiters can vary by recipient.
We need to allow for duplicates coming from multiple recipients, since for
example you could be signed up to the same list twice with different
addresses. Or maybe someone bounces you a load of mail some of which is from a
mailing list you're both subscribed to - then it's still useful to be able to
eliminate duplicates.
"""
return re.sub(";.*", "", value)
[docs]
def normalize_date(self, value: str) -> str:
"""Normalize `Date` to `YYYY-MM-DD` format.
Date timestamps can differ by seconds or hours for various reasons, so only
the date is honoured, normalized to the UTC timezone.
```{todo}
Revisit the day-level granularity, and whether the time of the day should
take part in the hash.
```
"""
if self.parsed_date is not None:
utc_date = datetime.fromtimestamp(self.parsed_date, tz=UTC).date()
return utc_date.isoformat()
return value
[docs]
def normalize_message_id(self, value: str) -> str:
"""Normalize Message-ID header by stripping angle brackets.
E.g., `<unique-id@example.com>` becomes `unique-id@example.com`.
"""
return self.strip_angle_brackets(value)
[docs]
def strip_angle_brackets(self, value: str) -> str:
"""Strip angle brackets from a value if it's a single bracketed item.
Only strips if the value matches `<something>` with no commas.
```{note}
Sometimes `email.parser` strips the `<>` brackets from a `To:` header which has
a single address. I have seen this happen for only one mail in a duplicate pair.
I'm not sure why (presumably the parser uses `email.utils.unquote` somewhere in
its code path which was only triggered by that mail and not its sister mail),
but to be safe, we should always strip the `<>` brackets to avoid this
difference preventing duplicate detection.
```
"""
if re.match(r"^<[^<>,]+>$", value):
return str(email.utils.unquote(value))
return value