"""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 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]] date: datetime | None text_body: str html_body: str | None attachments: list[ParsedAttachment] = field(default_factory=list) @property def has_attachments(self) -> bool: return len(self.attachments) > 0 @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]: out: list[ParsedAttachment] = [] for part in msg.iter_attachments(): if not isinstance(part, EmailMessage): continue payload = part.get_payload(decode=True) if not isinstance(payload, bytes): continue cid = part.get("Content-ID") out.append( ParsedAttachment( filename=part.get_filename() or "attachment", mime_type=part.get_content_type(), content_id=cid.strip("<>") if cid else None, payload=payload, ) ) return out def parse_message(raw: bytes) -> ParsedEmail: """Parse a raw RFC822 message into a :class:`ParsedEmail`.""" msg = email.message_from_bytes(raw, policy=default_policy) assert isinstance(msg, EmailMessage) 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"), date=date, text_body=text_body or "", html_body=html_body, attachments=_extract_attachments(msg), )