"""Provides `partition_email()` function.
Suitable for use with `.eml` files, which can be exported from many email clients.
"""
from __future__ import annotations
import datetime as dt
import email
import email.policy
import email.utils
import io
import os
from email.message import EmailMessage, MIMEPart
from typing import IO, Any, Final, Iterator, cast
from unstructured.documents.elements import Element, ElementMetadata
from unstructured.file_utils.model import FileType
from unstructured.partition.common import UnsupportedFileFormatError
from unstructured.partition.common.metadata import get_last_modified_date
from unstructured.partition.html import partition_html
from unstructured.partition.text import partition_text
from unstructured.utils import lazyproperty
VALID_CONTENT_SOURCES: Final[tuple[str, ...]] = ("text/html", "text/plain")
def partition_email(
filename: str | None = None,
*,
file: IO[bytes] | None = None,
content_source: str = "text/html",
metadata_filename: str | None = None,
metadata_last_modified: str | None = None,
process_attachments: bool = True,
**kwargs: Any,
) -> list[Element]:
"""Partitions an .eml file into document elements.
Args:
filename: str path of the target file.
file: A file-like object open for reading bytes (not str) e.g. --> open(filename, "rb").
content_source: The preferred message body. Many emails contain both a plain-text and an
HTML version of the message body. By default, the HTML version will be used when
available. Specifying "text/plain" will cause the plain-text version to be preferred.
When the preferred version is not available, the other version will be used.
metadata_filename: The file-path to use for metadata purposes. Useful when the target file
is specified as a file-like object or when `filename` is a temporary file and the
original file-path is known or a more meaningful file-path is desired.
metadata_last_modified: The last-modified timestamp to be applied in metadata. Useful when
a file-like object (which can have no last-modified date) target is used. The
last-modified metadata is otherwise drawn from the filesystem when a path is provided.
process_attachments: When True, also partition any attachments in the message after
partitioning the message body. All document elements appear in the single returned
element list. The filename of the attachment, when available, is used as the
`filename` metadata value for elements arising from the attachment.
Note that all global keyword arguments such as `unique_element_ids`, `language` and
`chunking_strategy` can be used and will be passed along to the decorators that implement
those functions. Further, any keyword arguments applicable to HTML will be passed along to the
HTML partitioner when processing an HTML message body.
"""
ctx = EmailPartitioningContext.load(
file_path=filename,
file=file,
content_source=content_source,
metadata_file_path=metadata_filename,
metadata_last_modified=metadata_last_modified,
process_attachments=process_attachments,
kwargs=kwargs,
)
return list(_EmailPartitioner.iter_elements(ctx=ctx))
class EmailPartitioningContext:
"""Encapsulates partitioning option validation, computation, and application of defaults."""
def __init__(
self,
file_path: str | None = None,
file: IO[bytes] | None = None,
content_source: str = "text/html",
metadata_file_path: str | None = None,
metadata_last_modified: str | None = None,
process_attachments: bool = False,
kwargs: dict[str, Any] = {},
):
self._file_path = file_path
self._file = file
self._content_source = content_source
self._metadata_file_path = metadata_file_path
self._metadata_last_modified = metadata_last_modified
self._process_attachments = process_attachments
self._kwargs = kwargs
@classmethod
def load(
cls,
file_path: str | None,
file: IO[bytes] | None,
content_source: str,
metadata_file_path: str | None,
metadata_last_modified: str | None,
process_attachments: bool,
kwargs: dict[str, Any],
) -> EmailPartitioningContext:
"""Construct and validate an instance."""
return cls(
file_path=file_path,
file=file,
content_source=content_source,
metadata_file_path=metadata_file_path,
metadata_last_modified=metadata_last_modified,
process_attachments=process_attachments,
kwargs=kwargs,
)._validate()
@lazyproperty
def bcc_addresses(self) -> list[str] | None:
"""The "blind carbon-copy" Bcc: addresses of the message."""
bccs = self.msg.get_all("Bcc")
if not bccs:
return None
addrs = email.utils.getaddresses(bccs)
return [email.utils.formataddr(addr) for addr in addrs]
@lazyproperty
def body_part(self) -> MIMEPart | None:
"""The message part containing the actual textual email message.
This is as opposed to attachments or "related" parts like an image that appears in the
message etc.
"""
return self.msg.get_body(preferencelist=self.content_type_preference)
@lazyproperty
def cc_addresses(self) -> list[str] | None:
"""The "carbon-copy" Cc: addresses of the message."""
ccs = self.msg.get_all("Cc")
if not ccs:
return None
addrs = email.utils.getaddresses(ccs)
return [email.utils.formataddr(addr) for addr in addrs]
@lazyproperty
def content_type_preference(self) -> tuple[str, ...]:
"""Whether to prefer HTML or plain-text body when message-body has both.
The default order of preference is `("html", "plain")`. The order can be switched by
specifying `"text/plain"` as the `content_source` arg value.
"""
return ("plain", "html") if self._content_source == "text/plain" else ("html", "plain")
@lazyproperty
def email_metadata(self) -> ElementMetadata:
"""The email-specific metadata fields for this message.
Suitable for use with `.metadata.update()` on the base metadata applied to message body
elements by delegate partitioners for text and HTML.
"""
return ElementMetadata(
bcc_recipient=self.bcc_addresses,
cc_recipient=self.cc_addresses,
email_message_id=self.message_id,
sent_from=[self.from_address] if self.from_address else None,
sent_to=self.to_addresses,
subject=self.subject,
)
@lazyproperty
def from_address(self) -> str | None:
"""The address of the message sender."""
froms = self.msg.get_all("From")
if not froms:
# -- this should never occur because the From: header is mandatory per RFC 5322 --
return None
addrs = email.utils.getaddresses(froms)
formatted_addrs = [email.utils.formataddr(addr) for addr in addrs]
return formatted_addrs[0]
@lazyproperty
def message_id(self) -> str | None:
"""The value of the Message-ID: header, when present."""
raw_id = self.msg.get("Message-ID")
if not raw_id:
return None
return raw_id.strip().strip("<>")
@lazyproperty
def metadata_file_path(self) -> str | None:
"""The best available file-path information for this email message.
It's value is computed according to these rules, applied in order:
- The `metadata_filename` arg value when one was provided to `partition_email()`.
- The `file_path` value when one was provided.
- None otherwise.
This value is used as the `filename` metadata value for elements produced by partitioning
the email message (but not those from its attachments).
"""
return self._metadata_file_path or self._file_path or None
@lazyproperty
def metadata_last_modified(self) -> str | None:
"""The best available last-modified date for this message, as an ISO8601 string.
It's value is computed according to these rules, applied in order:
- The `metadata_last_modified` arg value when one was provided to `partition_email()`.
- The date-time in the `Sent:` header of the message, when present.
- The last-modified date recorded on the filesystem for `file_path` when it was provided.
- None otherwise.
This value is used as the `last_modified` metadata value for all elements produced by
partitioning this email message, including any attachments.
"""
return self._metadata_last_modified or self._sent_date or self._filesystem_last_modified
@lazyproperty
def msg(self) -> EmailMessage:
"""The Python stdlib `email.message.EmailMessage` object parsed from the EML file."""
if self._file_path is not None:
with open(self._file_path, "rb") as f:
return cast(
EmailMessage, email.message_from_binary_file(f, policy=email.policy.default)
)
assert self._file is not None
file_bytes = self._file.read()
return cast(EmailMessage, email.message_from_bytes(file_bytes, policy=email.policy.default))
@lazyproperty
def partitioning_kwargs(self) -> dict[str, Any]:
"""The "extra" keyword arguments received by `partition_email()`.
These are passed along to delegate partitioners which extract keyword args like
`chunking_strategy` etc. in their decorators to control metadata behaviors, etc.
"""
return self._kwargs
@lazyproperty
def process_attachments(self) -> bool:
"""When True, partition attachments in addition to the email message body.
Any attachment having file-format that cannot be partitioned by unstructured is silently
skipped.
"""
return self._process_attachments
@lazyproperty
def subject(self) -> str | None:
"""The value of the Subject: header, when present."""
subject = self.msg.get("Subject")
if not subject:
return None
return subject
@lazyproperty
def to_addresses(self) -> list[str] | None:
"""The To: addresses of the message."""
tos = self.msg.get_all("To")
if not tos:
return None
addrs = email.utils.getaddresses(tos)
return [email.utils.formataddr(addr) for addr in addrs]
@lazyproperty
def _filesystem_last_modified(self) -> str | None:
"""Last-modified retrieved from filesystem when a file-path was provided, None otherwise."""
return get_last_modified_date(self._file_path) if self._file_path else None
@lazyproperty
def _sent_date(self) -> str | None:
"""ISO-8601 str representation of message sent-date, if available."""
date_str = self.msg.get("Date")
if not date_str:
return None
sent_date = email.utils.parsedate_to_datetime(date_str)
return sent_date.astimezone(dt.timezone.utc).isoformat(timespec="seconds")
def _validate(self) -> EmailPartitioningContext:
"""Raise on first invalid option, return self otherwise."""
if not self._file_path and not self._file:
raise ValueError(
"no document specified; either a `filename` or `file` argument must be provided."
)
if self._file:
if not isinstance( # pyright: ignore[reportUnnecessaryIsInstance]
self._file.read(0), bytes
):
raise ValueError("file object must be opened in binary mode")
self._file.seek(0)
if self._content_source not in VALID_CONTENT_SOURCES:
raise ValueError(
f"{repr(self._content_source)} is not a valid value for content_source;"
f" must be one of: {VALID_CONTENT_SOURCES}",
)
return self
class _EmailPartitioner:
"""Encapsulates the partitioning logic for email documents."""
def __init__(self, ctx: EmailPartitioningContext):
self._ctx = ctx
@classmethod
def iter_elements(cls, ctx: EmailPartitioningContext) -> Iterator[Element]:
"""Generate the document elements for the email described by `ctx`."""
return cls(ctx=ctx)._iter_elements()
def _iter_elements(self) -> Iterator[Element]:
"""Generate the document elements for the email described in the partitioning context.
This optionally includes elements generated by partitioning any partitionable attachments
in the message as well.
"""
for e in self._iter_email_body_elements():
e.metadata.update(self._ctx.email_metadata)
yield e
if not self._ctx.process_attachments:
return
for attachment in self._ctx.msg.iter_attachments():
yield from _AttachmentPartitioner.iter_elements(attachment, self._ctx)
def _iter_email_body_elements(self) -> Iterator[Element]:
"""Generate document elements from the email body."""
body_part = self._ctx.body_part
# -- it's possible to have no body part; that translates to zero elements --
if body_part is None:
return
content_type = body_part.get_content_type()
content = body_part.get_content()
assert isinstance(content, str)
if content_type == "text/html":
yield from partition_html(
text=content,
metadata_filename=self._ctx.metadata_file_path,
metadata_file_type=FileType.EML,
metadata_last_modified=self._ctx.metadata_last_modified,
**self._ctx.partitioning_kwargs,
)
else:
yield from partition_text(
text=content,
metadata_filename=self._ctx.metadata_file_path,
metadata_file_type=FileType.EML,
metadata_last_modified=self._ctx.metadata_last_modified,
**self._ctx.partitioning_kwargs,
)
class _AttachmentPartitioner:
"""Partitions an attachment to a MSG file."""
def __init__(self, attachment: EmailMessage, ctx: EmailPartitioningContext):
self._attachment = attachment
self._ctx = ctx
@classmethod
def iter_elements(
cls, attachment: EmailMessage, ctx: EmailPartitioningContext
) -> Iterator[Element]:
"""Partition an attachment MIME-part from a MIME email message (.eml file)."""
return cls(attachment, ctx)._iter_elements()
def _iter_elements(self) -> Iterator[Element]:
"""Partition the byte-stream in the attachment MIME-part into elements.
Generates zero elements if the attachment is not partitionable.
"""
# -- `auto.partition()` imports this module, so we need to defer the import to here to
# -- avoid a circular import.
from unstructured.partition.auto import partition
file = io.BytesIO(self._file_bytes)
# -- partition the attachment --
try:
elements = partition(
file=file,
metadata_filename=self._attachment_file_name,
metadata_last_modified=self._ctx.metadata_last_modified,
**self._ctx.partitioning_kwargs,
)
except UnsupportedFileFormatError:
# -- indicates `auto.partition()` has no partitioner for this file-format;
# -- silently skip the attachment
return
for e in elements:
e.metadata.attached_to_filename = self._attached_to_filename
yield e
@lazyproperty
def _attached_to_filename(self) -> str | None:
"""The file-name (no path) of the message. `None` if not available."""
file_path = self._ctx.metadata_file_path
if file_path is None:
return None
return os.path.basename(file_path)
@lazyproperty
def _attachment_file_name(self) -> str | None:
"""The original name of the attached file, `None` if not present in the MIME part."""
return self._attachment.get_filename()
@lazyproperty
def _file_bytes(self) -> bytes:
"""The bytes of the attached file."""
content = self._attachment.get_content()
if isinstance(content, str):
return content.encode("utf-8")
assert isinstance(content, bytes)
return content