"""MIME parsing via the stdlib ``email`` package with ``policy.default`` (SPEC §6.4). Primary parsing is stdlib per SPEC §3; ``mail-parser`` is available for helpers only. Header RFC2047 decoding is automatic under ``policy.default``. """ from __future__ import annotations import email import os from dataclasses import dataclass, field from datetime import datetime from email.headerregistry import AddressHeader from email.message import EmailMessage from email.policy import default as default_policy import nh3 from app.threading_.normalize import normalize_message_id, normalize_reference_list _SNIPPET_LEN = 160 @dataclass(frozen=True) class ParsedAttachment: filename: str mime_type: str content_id: str | None payload: bytes @dataclass class ParsedEmail: message_id: str | None in_reply_to: str | None references: list[str] # References + In-Reply-To, normalized, in order (§6.4) subject: str from_addr: list[dict[str, str]] to_addr: list[dict[str, str]] cc_addr: list[dict[str, str]] bcc_addr: list[dict[str, str]] date: datetime | None text_body: str html_body: str | None attachments: list[ParsedAttachment] = field(default_factory=list) @property def has_attachments(self) -> bool: # Inline resources (cid: images) carry a content_id and render in-body, not # as attachment chips — only real attachments flip the paperclip. return any(a.content_id is None for a in self.attachments) @property def snippet(self) -> str: return " ".join(self.text_body.split())[:_SNIPPET_LEN] def _addresses(msg: EmailMessage, header: str) -> list[dict[str, str]]: value = msg[header] if value is None: return [] out: list[dict[str, str]] = [] if isinstance(value, AddressHeader): for addr in value.addresses: if addr.addr_spec: out.append({"name": str(addr.display_name or ""), "email": addr.addr_spec}) return out def _html_to_text(html: str) -> str: """Derive plain text from HTML by stripping all tags (SPEC §6.4 nh3 strip).""" return nh3.clean(html, tags=set(), attributes={}, strip_comments=True) def _get_body_content(msg: EmailMessage, subtype: str) -> str | None: part = msg.get_body(preferencelist=(subtype,)) if part is None: return None try: content = part.get_content() except (LookupError, ValueError, KeyError): payload = part.get_payload(decode=True) if not isinstance(payload, bytes): return None content = payload.decode("utf-8", "replace") return content if isinstance(content, str) else None def _extract_attachments(msg: EmailMessage) -> list[ParsedAttachment]: # Walk the whole tree, not iter_attachments(): the latter is shallow and treats # multipart/related as a body, so an inline image nested in mixed>related>alt is # never yielded (its cid: reference then renders broken). A part is an attachment # or inline resource if it has a filename, a Content-ID, or attachment disposition; # the body text/plain and text/html parts have none of these and are skipped. out: list[ParsedAttachment] = [] for part in msg.walk(): if part.is_multipart() or not isinstance(part, EmailMessage): continue cid = part.get("Content-ID") filename = part.get_filename() disposition = part.get_content_disposition() if not (filename or cid or disposition == "attachment"): continue payload = part.get_payload(decode=True) if not isinstance(payload, bytes): continue # A part is an INLINE resource (cid: image rendered in-body, hidden from the # attachment list) only when it carries a Content-ID AND is not explicitly an # attachment. A part marked `Content-Disposition: attachment` is a real # attachment even when it also has a Content-ID (e.g. Fastmail's invite.ics), # so we drop the cid for it and it shows as a downloadable chip. inline = cid is not None and disposition != "attachment" out.append( ParsedAttachment( filename=filename or "attachment", mime_type=part.get_content_type(), content_id=cid.strip("<>") if (inline and cid is not None) else None, payload=payload, ) ) return out def parse_message(raw: bytes) -> ParsedEmail: """Parse a raw RFC822 message (in memory) into a :class:`ParsedEmail`.""" msg = email.message_from_bytes(raw, policy=default_policy) assert isinstance(msg, EmailMessage) return _parse_email(msg) def parse_message_file(path: str | os.PathLike[str]) -> ParsedEmail: """Parse a message straight from a blob file handle (large-message path, §7).""" with open(path, "rb") as handle: msg = email.message_from_binary_file(handle, policy=default_policy) assert isinstance(msg, EmailMessage) return _parse_email(msg) def _parse_email(msg: EmailMessage) -> ParsedEmail: message_id = normalize_message_id(msg.get("Message-ID")) in_reply_to_ids = normalize_reference_list(msg.get("In-Reply-To")) in_reply_to = in_reply_to_ids[0] if in_reply_to_ids else None # references_hdr = References then In-Reply-To message-ids, in order, deduped (§6.4). references: list[str] = [] for ref in normalize_reference_list(msg.get("References")) + in_reply_to_ids: if ref not in references: references.append(ref) subject = str(msg.get("Subject", "") or "") date: datetime | None = None date_header = msg["Date"] if date_header is not None: try: date = date_header.datetime except (AttributeError, ValueError): date = None html_body = _get_body_content(msg, "html") text_body = _get_body_content(msg, "plain") if text_body is None and html_body is not None: text_body = _html_to_text(html_body) return ParsedEmail( message_id=message_id, in_reply_to=in_reply_to, references=references, subject=subject, from_addr=_addresses(msg, "From"), to_addr=_addresses(msg, "To"), cc_addr=_addresses(msg, "Cc"), bcc_addr=_addresses(msg, "Bcc"), date=date, text_body=text_body or "", html_body=html_body, attachments=_extract_attachments(msg), )