189 lines
6.8 KiB
Python
189 lines
6.8 KiB
Python
"""The `Message` object is the primary interface object in `python-oxmsg`.
|
|
|
|
This object is returned by the `Message.load()` classmethod:
|
|
|
|
```python
|
|
>>> from oxmsg import Message
|
|
|
|
>>> msg = Message.load("message.msg")
|
|
>>> msg.subject
|
|
'Agenda for the Acme meeting'
|
|
```
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import email.message
|
|
import email.parser
|
|
import os
|
|
import struct
|
|
from typing import IO, Iterator
|
|
|
|
from olefile import OleFileIO, isOleFile
|
|
|
|
from oxmsg.attachment import Attachment
|
|
from oxmsg.domain import constants as c
|
|
from oxmsg.domain import model as m
|
|
from oxmsg.properties import Properties
|
|
from oxmsg.recipient import Recipient
|
|
from oxmsg.storage import Storage
|
|
from oxmsg.util import lazyproperty
|
|
|
|
|
|
class Message:
|
|
"""An Outlook email message loaded from an OXMSG format (.msg) file."""
|
|
|
|
def __init__(self, storage: m.StorageT):
|
|
self._storage = storage
|
|
|
|
@classmethod
|
|
def load(cls, msg_file: str | IO[bytes] | bytes) -> Message:
|
|
"""Load an instance from `msg_file`.
|
|
|
|
`msg_file` can be a file-path, a file-like object, or the bytes of the MSG file.
|
|
"""
|
|
# -- ensure magic bytes on file are as expected --
|
|
is_olefile = (
|
|
isOleFile(data=msg_file) if isinstance(msg_file, bytes) else isOleFile(msg_file)
|
|
)
|
|
if not is_olefile:
|
|
filename = os.path.basename(msg_file) if isinstance(msg_file, str) else "msg_file"
|
|
raise ValueError(f"{filename} is not an Outlook MSG file")
|
|
|
|
with OleFileIO(msg_file) as ole:
|
|
root_storage = Storage.from_ole(ole)
|
|
|
|
self = cls(root_storage)
|
|
self._validate()
|
|
return self
|
|
|
|
@property
|
|
def attachment_count(self) -> int:
|
|
"""Number of attachments on this message."""
|
|
return self._header_prop_values[3]
|
|
|
|
@property
|
|
def attachments(self) -> tuple[Attachment, ...]:
|
|
"""Attachment object for each attachment in this message."""
|
|
return tuple(self._iter_attachments())
|
|
|
|
@property
|
|
def body(self) -> str | None:
|
|
"""The plain-text body of this message; `None` if not present."""
|
|
return self.properties.str_prop_value(c.PID_BODY)
|
|
|
|
@property
|
|
def html_body(self) -> str | None:
|
|
"""The HTML body of this message if present, `None` otherwise."""
|
|
# -- body HTML most commonly (these days) is bytes encoded with internet-codepage --
|
|
if html_bytes := self.properties.binary_prop_value(c.PID_HTML):
|
|
return html_bytes.decode(self.properties.body_encoding)
|
|
# -- however older emails may encode the HTML as Unicode or an 8-bit string --
|
|
if html_str := self.properties.str_prop_value(c.PID_BODY_HTML):
|
|
return html_str
|
|
return None
|
|
|
|
@property
|
|
def message_headers(self) -> dict[str, str]:
|
|
"""From, To, Content-Type, etc. headers for this message as {name: value} mapping."""
|
|
return dict(self._message_headers)
|
|
|
|
@property
|
|
def message_class(self) -> str:
|
|
"""Outlook Message Class identifier like "IPM.Note" for this message."""
|
|
return self.properties.str_prop_value(c.PID_MESSAGE_CLASS) or ""
|
|
|
|
@property
|
|
def properties(self) -> Properties:
|
|
"""Provides access to the properties of this OXMSG object."""
|
|
return self._properties
|
|
|
|
@property
|
|
def recipients(self) -> tuple[Recipient, ...]:
|
|
"""`Recipient` object for each recipient of this message."""
|
|
return tuple(self._iter_recipients())
|
|
|
|
@property
|
|
def sender(self) -> str | None:
|
|
"""Name and email address of the message sender.
|
|
|
|
Like '"John Doe" <john@bigisp.com>'.
|
|
|
|
None if it is not recorded in the message. This may occur when the message is a draft or
|
|
system-generated or perhaps in other cases.
|
|
|
|
Note that the value of the From: message header is returned when present (usually I expect)
|
|
and may contain multiple addresses separated by commas.
|
|
"""
|
|
# -- start by looking in the message "From:" header --
|
|
if from_header_value := self._message_headers["from"]:
|
|
return from_header_value
|
|
# -- assemble from parts --
|
|
props = self.properties
|
|
raw_name = (props.str_prop_value(c.PID_SENDER_NAME) or "").strip()
|
|
name = f'"{raw_name}" ' if raw_name else ""
|
|
email = props.str_prop_value(c.PID_SENDER_EMAIL_ADDRESS) or props.str_prop_value(
|
|
c.PID_SENDER_SMTP_ADDRESS
|
|
)
|
|
return f"{name}<{email}>" if email else props.str_prop_value(c.PID_SENT_REPRESENTING_NAME)
|
|
|
|
@property
|
|
def sent_date(self) -> dt.datetime | None:
|
|
"""When this message was submitted by the sender.
|
|
|
|
This value will be `None` if the message has not been sent and possibly in other cases.
|
|
"""
|
|
return self.properties.date_prop_value(c.PID_CLIENT_SUBMIT_TIME)
|
|
|
|
@property
|
|
def subject(self) -> str:
|
|
"""Subject line of this message."""
|
|
return self.properties.str_prop_value(c.PID_SUBJECT) or ""
|
|
|
|
@lazyproperty
|
|
def _header_prop_values(self) -> tuple[int, int, int, int]:
|
|
"""The property values in the MSG-root properties header.
|
|
|
|
It is a tuple of the four int values:
|
|
- next_recipient_id
|
|
- next_attachment_id
|
|
- recipient_count
|
|
- attachment_count
|
|
"""
|
|
return struct.unpack("<8x4I", self._storage.properties_stream_bytes[:24])
|
|
|
|
def _iter_attachments(self) -> Iterator[Attachment]:
|
|
"""Generate `Attachment` object for each attachment in this message.
|
|
|
|
Should need to be called at most once.
|
|
"""
|
|
return (Attachment(storage) for storage in self._storage.iter_attachment_storages())
|
|
|
|
def _iter_recipients(self) -> Iterator[Recipient]:
|
|
"""Generate `Recipient` object for each recipient in this message.
|
|
|
|
Should need to be called at most once.
|
|
"""
|
|
return (Recipient(storage) for storage in self._storage.iter_recipient_storages())
|
|
|
|
@lazyproperty
|
|
def _message_headers(self) -> email.message.Message:
|
|
"""From, To, Content-Type, etc. headers for this message as Message object.
|
|
|
|
This provides case-insensitive access along with some other convenient behaviors not
|
|
available from a `dict`.
|
|
"""
|
|
headers = self.properties.str_prop_value(c.PID_TRANSPORT_MESSAGE_HEADERS) or ""
|
|
return email.parser.HeaderParser().parsestr(headers)
|
|
|
|
@lazyproperty
|
|
def _properties(self) -> Properties:
|
|
return Properties(self._storage, properties_header_offset=m.MSG_HDR_OFFSET)
|
|
|
|
def _validate(self) -> None:
|
|
"""Raise if this message is invalid for one of a variety of possible reasons."""
|
|
# -- for now, we only process email messages --
|
|
# if not self.message_class.startswith("IPM.Note"):
|
|
# raise ValueError(f"{self.message_class} files not supported")
|