Zhengshou Lai
5a665e38bb
New `myclaude mail` group with: - config: interactive account setup - inbox: list recent messages - read: show message by UID - send: send email with attachments - reply/forward: reply to or forward messages
73 lines
2.2 KiB
Python
73 lines
2.2 KiB
Python
"""Email message parsing utilities."""
|
|
|
|
from email.message import Message
|
|
from typing import Any
|
|
|
|
from myclaude.mail.utils import decode_header_value, strip_html_tags
|
|
|
|
|
|
def _extract_payload(part: Message) -> str:
|
|
"""Extract text payload from a message part."""
|
|
charset = part.get_content_charset() or "utf-8"
|
|
payload = part.get_payload(decode=True)
|
|
if isinstance(payload, bytes):
|
|
return payload.decode(charset, errors="replace")
|
|
return str(payload or "")
|
|
|
|
|
|
def extract_text(msg: Message) -> tuple[str, str]:
|
|
"""Extract plain text and html body from a message.
|
|
|
|
Returns (plain_text, html_text). Either may be empty.
|
|
"""
|
|
plain_parts: list[str] = []
|
|
html_parts: list[str] = []
|
|
|
|
if msg.is_multipart():
|
|
for part in msg.walk():
|
|
ctype = part.get_content_type()
|
|
cdisp = str(part.get("Content-Disposition", ""))
|
|
if "attachment" in cdisp:
|
|
continue
|
|
if ctype == "text/plain":
|
|
plain_parts.append(_extract_payload(part))
|
|
elif ctype == "text/html":
|
|
html_parts.append(_extract_payload(part))
|
|
else:
|
|
ctype = msg.get_content_type()
|
|
payload = _extract_payload(msg)
|
|
if ctype == "text/html":
|
|
html_parts.append(payload)
|
|
else:
|
|
plain_parts.append(payload)
|
|
|
|
plain = "\n".join(plain_parts).strip()
|
|
html = "\n".join(html_parts).strip()
|
|
|
|
if not plain and html:
|
|
plain = strip_html_tags(html)
|
|
|
|
return plain, html
|
|
|
|
|
|
def extract_attachments(msg: Message) -> list[dict[str, Any]]:
|
|
"""Return list of attachment metadata dicts."""
|
|
attachments: list[dict[str, Any]] = []
|
|
if not msg.is_multipart():
|
|
return attachments
|
|
|
|
for part in msg.walk():
|
|
cdisp = str(part.get("Content-Disposition", ""))
|
|
if "attachment" not in cdisp:
|
|
continue
|
|
filename = decode_header_value(part.get_filename())
|
|
if filename:
|
|
attachments.append(
|
|
{
|
|
"filename": filename,
|
|
"content_type": part.get_content_type(),
|
|
"size": len(part.get_payload(decode=True) or b""),
|
|
}
|
|
)
|
|
return attachments
|