from __future__ import annotations import os import re import tempfile from typing import IO, Any, Iterator, Optional from oxmsg import Message from oxmsg.attachment import Attachment from unstructured.documents.elements import Element, ElementMetadata from unstructured.file_utils.model import FileType from unstructured.logger import logger 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 is_temp_file_path, lazyproperty def partition_msg( filename: Optional[str] = None, *, file: Optional[IO[bytes]] = None, metadata_filename: Optional[str] = None, metadata_last_modified: Optional[str] = None, process_attachments: bool = True, **kwargs: Any, ) -> list[Element]: """Partitions a MSFT Outlook .msg file Parameters ---------- filename A string defining the target filename path. file A file-like object using "rb" mode --> open(filename, "rb"). metadata_filename The filename to use for the metadata. metadata_last_modified The last modified date for the document. process_attachments If True, partition_email will process email attachments in addition to processing the content of the email itself. """ opts = MsgPartitionerOptions( file=file, file_path=filename, metadata_file_path=metadata_filename, metadata_last_modified=metadata_last_modified, partition_attachments=process_attachments, kwargs=kwargs, ) return list(_MsgPartitioner.iter_message_elements(opts)) class MsgPartitionerOptions: """Encapsulates partitioning option validation, computation, and application of defaults.""" def __init__( self, *, file: IO[bytes] | None, file_path: str | None, metadata_file_path: str | None, metadata_last_modified: str | None, partition_attachments: bool, kwargs: dict[str, Any], ): self._file = file self._file_path = file_path self._metadata_file_path = metadata_file_path self._metadata_last_modified = metadata_last_modified self._partition_attachments = partition_attachments self._kwargs = kwargs @lazyproperty def extra_msg_metadata(self) -> ElementMetadata: """ElementMetadata suitable for use on an element formed from message content. These are only the metadata fields specific to email messages. The remaining metadata fields produced by the delegate partitioner are used as produced. None of these metadata fields change based on the element, so we just compute it once. """ msg = self.msg sent_from = [s.strip() for s in sender.split(",")] if (sender := msg.sender) else None sent_to = [r.email_address for r in msg.recipients] or None bcc_recipient = ( [c.strip() for c in bcc.split(",")] if (bcc := msg.message_headers.get("Bcc")) else None ) cc_recipient = ( [c.strip() for c in cc.split(",")] if (cc := msg.message_headers.get("Cc")) else None ) if email_message_id := msg.message_headers.get("Message-Id"): email_message_id = re.sub(r"^<|>$", "", email_message_id) # Strip angle brackets element_metadata = ElementMetadata( bcc_recipient=bcc_recipient, cc_recipient=cc_recipient, email_message_id=email_message_id, sent_from=sent_from, sent_to=sent_to, subject=msg.subject or None, ) element_metadata.detection_origin = "msg" return element_metadata @lazyproperty def is_encrypted(self) -> bool: """True when message is encrypted.""" # NOTE(robinson) - Per RFC 2015, the content type for emails with PGP encrypted content # is multipart/encrypted (ref: https://www.ietf.org/rfc/rfc2015.txt) # NOTE(scanny) - pretty sure we're going to want to dig deeper to discover messages that are # encrypted with something other than PGP. # - might be able to distinguish based on PID_MESSAGE_CLASS = 'IPM.Note.Signed' # - Content-Type header might include "application/pkcs7-mime" for Microsoft S/MIME # encryption. return "encrypted" in self.msg.message_headers.get("Content-Type", "") @lazyproperty def metadata_file_path(self) -> str | None: """Best available path for MSG file. The value is the caller supplied `metadata_filename` if present, falling back to the source file-path if that was provided, otherwise `None`. """ return self._metadata_file_path or self._file_path @lazyproperty def metadata_last_modified(self) -> str | None: """Caller override for `.metadata.last_modified` to be applied to all elements.""" email_date = sent_date.isoformat() if (sent_date := self.msg.sent_date) else None return self._metadata_last_modified or email_date or self._last_modified @lazyproperty def msg(self) -> Message: """The `oxmsg.Message` object loaded from file or filename.""" return Message.load(self._msg_file) @lazyproperty def partition_attachments(self) -> bool: """True when message attachments should also be partitioned.""" return self._partition_attachments @lazyproperty def partitioning_kwargs(self) -> dict[str, Any]: """The "extra" keyword arguments received by `partition_msg()`. 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 _last_modified(self) -> str | None: """The best last-modified date available from source-file, None if not available.""" if not self._file_path or is_temp_file_path(self._file_path): return None return get_last_modified_date(self._file_path) @lazyproperty def _msg_file(self) -> str | IO[bytes]: """The source for the bytes of the message, either a file-path or a file-like object.""" if file_path := self._file_path: return file_path if file := self._file: return file raise ValueError("one of `file` or `filename` arguments must be provided") class _MsgPartitioner: """Partitions Outlook email message (MSG) files.""" def __init__(self, opts: MsgPartitionerOptions): self._opts = opts @classmethod def iter_message_elements(cls, opts: MsgPartitionerOptions) -> Iterator[Element]: """Partition MS Outlook email messages (.msg files) into elements.""" if opts.is_encrypted: logger.warning("Encrypted email detected. Partitioner will return an empty list.") return yield from cls(opts)._iter_message_elements() def _iter_message_elements(self) -> Iterator[Element]: """Partition MS Outlook email messages (.msg files) into elements.""" yield from self._iter_message_body_elements() if not self._opts.partition_attachments: return for attachment in self._attachments: yield from _AttachmentPartitioner.iter_elements(attachment, self._opts) @lazyproperty def _attachments(self) -> tuple[Attachment, ...]: """The `oxmsg.attachment.Attachment` objects for this message.""" return tuple(self._opts.msg.attachments) def _iter_message_body_elements(self) -> Iterator[Element]: """Partition the message body (but not the attachments).""" msg = self._opts.msg if html_body := msg.html_body: elements = partition_html( text=html_body, metadata_filename=self._opts.metadata_file_path, metadata_file_type=FileType.MSG, metadata_last_modified=self._opts.metadata_last_modified, **self._opts.partitioning_kwargs, ) elif msg.body: elements = partition_text( text=msg.body, metadata_filename=self._opts.metadata_file_path, metadata_file_type=FileType.MSG, metadata_last_modified=self._opts.metadata_last_modified, **self._opts.partitioning_kwargs, ) else: elements: list[Element] = [] # -- augment the element metadata with email-specific values -- email_specific_metadata = self._opts.extra_msg_metadata for e in elements: e.metadata.update(email_specific_metadata) yield e class _AttachmentPartitioner: """Partitions an attachment to a MSG file.""" def __init__(self, attachment: Attachment, opts: MsgPartitionerOptions): self._attachment = attachment self._opts = opts @classmethod def iter_elements( cls, attachment: Attachment, opts: MsgPartitionerOptions ) -> Iterator[Element]: """Partition an `oxmsg.attachment.Attachment` from an Outlook email message (.msg file).""" return cls(attachment, opts)._iter_elements() def _iter_elements(self) -> Iterator[Element]: """Partition the file in an `oxmsg.attachment.Attachment` into elements.""" from unstructured.partition.auto import partition with tempfile.TemporaryDirectory() as tmp_dir_path: # -- save attachment as file in this temporary directory -- detached_file_path = os.path.join(tmp_dir_path, self._attachment_file_name) with open(detached_file_path, "wb") as f: f.write(self._file_bytes) # -- partition the attachment -- try: elements = partition( detached_file_path, metadata_filename=self._attachment_file_name, metadata_last_modified=self._attachment_last_modified, **self._opts.partitioning_kwargs, ) except UnsupportedFileFormatError: return for e in elements: e.metadata.attached_to_filename = self._opts.metadata_file_path yield e @lazyproperty def _attachment_file_name(self) -> str: """The original name of the attached file, no path. This value is 'unknown' if it is not present in the MSG file (not expected). """ return self._attachment.file_name or "unknown" @lazyproperty def _attachment_last_modified(self) -> str | None: """ISO8601 string timestamp of attachment last-modified date. This value generally available on the attachment and will be the most reliable last-modifed time. There are fallbacks for when it is not present, ultimately `None` if we have no way of telling. """ if last_modified := self._attachment.last_modified: return last_modified.isoformat() return self._opts.metadata_last_modified @lazyproperty def _file_bytes(self) -> bytes: """The bytes of the attached file.""" return self._attachment.file_bytes or b""