90 lines
3.1 KiB
Python
90 lines
3.1 KiB
Python
import re
|
|
from datetime import timezone
|
|
from email import policy
|
|
from email.parser import BytesParser
|
|
from email.utils import getaddresses, parsedate_to_datetime
|
|
from html import unescape
|
|
|
|
|
|
def _addresses(values: list[str]) -> list[dict[str, str]]:
|
|
return [
|
|
{
|
|
"name": name.strip() or address.split("@", 1)[0],
|
|
"address": address.lower(),
|
|
}
|
|
for name, address in getaddresses(values)
|
|
if address
|
|
]
|
|
|
|
|
|
def _text(part) -> str:
|
|
try:
|
|
value = part.get_content()
|
|
except (LookupError, UnicodeError):
|
|
payload = part.get_payload(decode=True) or b""
|
|
value = payload.decode("utf-8", errors="replace")
|
|
return value if isinstance(value, str) else ""
|
|
|
|
|
|
def _snippet(text: str, html: str | None, limit: int = 240) -> str:
|
|
source = text
|
|
if not source and html:
|
|
source = re.sub(r"<[^>]+>", " ", html)
|
|
source = unescape(source)
|
|
return re.sub(r"\s+", " ", source).strip()[:limit]
|
|
|
|
|
|
def parse_mime(raw: bytes) -> dict:
|
|
message = BytesParser(policy=policy.default).parsebytes(raw)
|
|
sender = _addresses(message.get_all("from", []))
|
|
from_value = sender[0] if sender else {"name": "Unknown sender", "address": ""}
|
|
text_parts: list[str] = []
|
|
html_parts: list[str] = []
|
|
attachments: list[dict] = []
|
|
|
|
for part in message.walk():
|
|
if part.is_multipart():
|
|
continue
|
|
disposition = part.get_content_disposition()
|
|
filename = part.get_filename()
|
|
content_type = part.get_content_type()
|
|
payload = part.get_payload(decode=True) or b""
|
|
if disposition == "attachment" or filename:
|
|
attachments.append(
|
|
{
|
|
"filename": filename or "attachment",
|
|
"content_type": content_type,
|
|
"size": len(payload),
|
|
"content_id": (part.get("Content-ID") or "").strip("<>") or None,
|
|
"disposition": disposition or "attachment",
|
|
}
|
|
)
|
|
elif content_type == "text/plain":
|
|
text_parts.append(_text(part))
|
|
elif content_type == "text/html":
|
|
html_parts.append(_text(part))
|
|
|
|
date = None
|
|
if message.get("Date"):
|
|
try:
|
|
date = parsedate_to_datetime(message["Date"])
|
|
if date.tzinfo is None:
|
|
date = date.replace(tzinfo=timezone.utc)
|
|
date = date.astimezone(timezone.utc)
|
|
except (TypeError, ValueError, OverflowError):
|
|
pass
|
|
body_text = "\n".join(text_parts).strip()
|
|
body_html = "\n".join(html_parts).strip() or None
|
|
return {
|
|
"message_id": (message.get("Message-ID") or "").strip() or None,
|
|
"sender": from_value,
|
|
"to": _addresses(message.get_all("to", [])),
|
|
"cc": _addresses(message.get_all("cc", [])),
|
|
"reply_to": _addresses(message.get_all("reply-to", [])),
|
|
"subject": str(message.get("Subject") or ""),
|
|
"sent_at": date,
|
|
"body_text": body_text,
|
|
"body_html": body_html,
|
|
"snippet": _snippet(body_text, body_html),
|
|
"attachments": attachments,
|
|
}
|