# retoor <retoor@molodetz.nl>
from __future__ import annotations
import email
import imaplib
import re
import smtplib
import ssl
from contextlib import contextmanager
from email.message import EmailMessage
from email.policy import default as default_policy
from email.utils import formataddr, getaddresses, parsedate_to_datetime
from typing import Any, Iterator
from devplacepy.net_guard import BlockedAddressError, guard_public_host_sync
from .config import EmailAccount
from .errors import EmailError
DEFAULT_TIMEOUT_SECONDS = 30.0
MAX_BODY_CHARS = 80_000
HEADER_FIELDS = "(FROM TO CC SUBJECT DATE MESSAGE-ID)"
SUMMARY_ITEMS = f"(UID FLAGS BODY.PEEK[HEADER.FIELDS {HEADER_FIELDS}])"
_UID_PATTERN = re.compile(rb"UID (\d+)")
def imap_quote(value: str) -> str:
return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'
def _header_addresses(value: str) -> list[str]:
if not value:
return []
return [formataddr(pair) for pair in getaddresses([value]) if pair[1]]
def _decode_date(raw: str) -> str | None:
if not raw:
return None
try:
return parsedate_to_datetime(raw).isoformat()
except (TypeError, ValueError):
return raw
class EmailClient:
def __init__(
self, account: EmailAccount, timeout: float = DEFAULT_TIMEOUT_SECONDS
) -> None:
self.account = account
self.timeout = timeout
@contextmanager
def _imap(self) -> Iterator[imaplib.IMAP4]:
account = self.account
if not account.imap_host:
raise EmailError("This account has no IMAP host configured.", kind="config")
try:
guard_public_host_sync(account.imap_host)
except BlockedAddressError as exc:
raise EmailError(str(exc), kind="blocked") from exc
context = ssl.create_default_context()
try:
if account.imap_ssl:
conn: imaplib.IMAP4 = imaplib.IMAP4_SSL(
account.imap_host,
account.imap_port,
ssl_context=context,
timeout=self.timeout,
)
else:
conn = imaplib.IMAP4(
account.imap_host, account.imap_port, timeout=self.timeout
)
if account.imap_starttls:
conn.starttls(context)
conn.login(account.username, account.password)
except (OSError, imaplib.IMAP4.error) as exc:
raise EmailError(f"IMAP connection failed: {exc}", kind="connect") from exc
try:
yield conn
finally:
try:
conn.logout()
except (OSError, imaplib.IMAP4.error):
pass
def _select(self, conn: imaplib.IMAP4, folder: str, readonly: bool) -> None:
typ, data = conn.select(imap_quote(folder), readonly=readonly)
if typ != "OK":
raise EmailError(f"Cannot open folder '{folder}': {_first(data)}")
def _uid_search(self, conn: imaplib.IMAP4, criteria: list[str]) -> list[str]:
if not criteria:
typ, data = conn.uid("SEARCH", None, "ALL")
else:
args: list[Any] = [b"CHARSET", b"UTF-8"]
args.extend(token.encode("utf-8") for token in criteria)
typ, data = conn.uid("SEARCH", *args)
if typ != "OK":
raise EmailError(f"Search failed: {_first(data)}")
if not data or not data[0]:
return []
return [chunk.decode() for chunk in data[0].split()]
def list_folders(self) -> list[dict[str, Any]]:
with self._imap() as conn:
typ, data = conn.list()
if typ != "OK":
raise EmailError(f"Cannot list folders: {_first(data)}")
folders: list[dict[str, Any]] = []
for entry in data or []:
if entry is None:
continue
line = entry.decode("utf-8", "replace") if isinstance(entry, bytes) else str(entry)
match = re.match(r"\((?P<flags>[^)]*)\) \"?(?P<delim>[^\"]*)\"? (?P<name>.+)", line)
if not match:
continue
name = match.group("name").strip().strip('"')
folders.append({"name": name, "flags": match.group("flags").split()})
return folders
def list_messages(
self,
folder: str,
criteria: list[str],
limit: int,
offset: int,
) -> dict[str, Any]:
with self._imap() as conn:
self._select(conn, folder, readonly=True)
uids = self._uid_search(conn, criteria)
uids.reverse()
total = len(uids)
window = uids[offset : offset + limit]
messages = self._fetch_summaries(conn, window)
return {
"folder": folder,
"total": total,
"offset": offset,
"limit": limit,
"count": len(messages),
"messages": messages,
}
def _fetch_summaries(
self, conn: imaplib.IMAP4, uids: list[str]
) -> list[dict[str, Any]]:
if not uids:
return []
typ, data = conn.uid("FETCH", ",".join(uids), SUMMARY_ITEMS)
if typ != "OK":
raise EmailError(f"Fetch failed: {_first(data)}")
order = {uid: index for index, uid in enumerate(uids)}
summaries: list[dict[str, Any]] = []
for part in data or []:
if not isinstance(part, tuple) or len(part) < 2:
continue
descriptor, header_bytes = part[0], part[1]
uid_match = _UID_PATTERN.search(descriptor or b"")
if not uid_match:
continue
uid = uid_match.group(1).decode()
flags = [flag.decode() if isinstance(flag, bytes) else str(flag) for flag in imaplib.ParseFlags(descriptor)]
headers = email.message_from_bytes(header_bytes or b"", policy=default_policy)
summaries.append(
{
"uid": uid,
"flags": flags,
"seen": "\\Seen" in flags,
"from": str(headers.get("From", "")),
"to": _header_addresses(str(headers.get("To", ""))),
"subject": str(headers.get("Subject", "")),
"date": _decode_date(str(headers.get("Date", ""))),
"message_id": str(headers.get("Message-ID", "")),
}
)
summaries.sort(key=lambda item: order.get(item["uid"], 0))
return summaries
def read_message(self, folder: str, uid: str) -> dict[str, Any]:
with self._imap() as conn:
self._select(conn, folder, readonly=True)
typ, data = conn.uid("FETCH", uid, "(FLAGS BODY.PEEK[])")
if typ != "OK":
raise EmailError(f"Fetch failed: {_first(data)}")
raw = _first_payload(data)
if raw is None:
raise EmailError(f"Message {uid} not found in '{folder}'.", kind="not_found")
flags = [flag.decode() if isinstance(flag, bytes) else str(flag) for flag in imaplib.ParseFlags(_first_descriptor(data) or b"")]
message = email.message_from_bytes(raw, policy=default_policy)
return self._render_message(uid, folder, flags, message)
def _render_message(
self, uid: str, folder: str, flags: list[str], message: Any
) -> dict[str, Any]:
text_part = message.get_body(preferencelist=("plain",))
html_part = message.get_body(preferencelist=("html",))
attachments: list[dict[str, Any]] = []
for part in message.iter_attachments():
payload = part.get_content()
size = len(payload) if isinstance(payload, (bytes, str)) else 0
attachments.append(
{
"filename": part.get_filename() or "",
"content_type": part.get_content_type(),
"size": size,
}
)
body = text_part.get_content().strip() if text_part else ""
html = html_part.get_content().strip() if html_part else ""
return {
"uid": uid,
"folder": folder,
"flags": flags,
"from": str(message.get("From", "")),
"to": _header_addresses(str(message.get("To", ""))),
"cc": _header_addresses(str(message.get("Cc", ""))),
"subject": str(message.get("Subject", "")),
"date": _decode_date(str(message.get("Date", ""))),
"message_id": str(message.get("Message-ID", "")),
"body": body[:MAX_BODY_CHARS],
"body_truncated": len(body) > MAX_BODY_CHARS,
"html": html[:MAX_BODY_CHARS],
"attachments": attachments,
}
def search(self, folder: str, criteria: list[str], limit: int) -> dict[str, Any]:
return self.list_messages(folder, criteria, limit, 0)
def set_flags(
self, folder: str, uid: str, flags: list[str], add: bool
) -> dict[str, Any]:
with self._imap() as conn:
self._select(conn, folder, readonly=False)
command = "+FLAGS" if add else "-FLAGS"
typ, data = conn.uid("STORE", uid, command, "(%s)" % " ".join(flags))
if typ != "OK":
raise EmailError(f"Flag update failed: {_first(data)}")
return {"uid": uid, "folder": folder, "flags": flags, "added": add}
def move_message(self, folder: str, uid: str, destination: str) -> dict[str, Any]:
with self._imap() as conn:
self._select(conn, folder, readonly=False)
typ, data = conn.uid("COPY", uid, imap_quote(destination))
if typ != "OK":
raise EmailError(f"Copy to '{destination}' failed: {_first(data)}")
conn.uid("STORE", uid, "+FLAGS", "(\\Deleted)")
conn.expunge()
return {"uid": uid, "from_folder": folder, "to_folder": destination}
def delete_message(
self, folder: str, uid: str, trash: str | None
) -> dict[str, Any]:
if trash and trash != folder:
result = self.move_message(folder, uid, trash)
result["deleted"] = True
return result
with self._imap() as conn:
self._select(conn, folder, readonly=False)
typ, data = conn.uid("STORE", uid, "+FLAGS", "(\\Deleted)")
if typ != "OK":
raise EmailError(f"Delete failed: {_first(data)}")
conn.expunge()
return {"uid": uid, "folder": folder, "deleted": True}
def send_message(
self,
to: list[str],
subject: str,
body: str,
cc: list[str] | None = None,
bcc: list[str] | None = None,
html: str | None = None,
in_reply_to: str | None = None,
) -> dict[str, Any]:
account = self.account
if not account.smtp_host:
raise EmailError("This account has no SMTP host configured.", kind="config")
message = EmailMessage()
message["From"] = (
formataddr((account.from_name, account.from_address))
if account.from_name
else account.from_address
)
message["To"] = ", ".join(to)
if cc:
message["Cc"] = ", ".join(cc)
message["Subject"] = subject
if in_reply_to:
message["In-Reply-To"] = in_reply_to
message["References"] = in_reply_to
message.set_content(body)
if html:
message.add_alternative(html, subtype="html")
recipients = list(to) + list(cc or []) + list(bcc or [])
try:
guard_public_host_sync(account.smtp_host)
except BlockedAddressError as exc:
raise EmailError(str(exc), kind="blocked") from exc
context = ssl.create_default_context()
try:
if account.smtp_ssl:
server: smtplib.SMTP = smtplib.SMTP_SSL(
account.smtp_host,
account.smtp_port,
timeout=self.timeout,
context=context,
)
else:
server = smtplib.SMTP(
account.smtp_host, account.smtp_port, timeout=self.timeout
)
if account.smtp_starttls:
server.starttls(context=context)
with server:
if account.username and account.password:
server.login(account.username, account.password)
server.send_message(
message, from_addr=account.from_address, to_addrs=recipients
)
except (OSError, smtplib.SMTPException) as exc:
raise EmailError(f"Sending failed: {exc}", kind="send") from exc
return {
"sent": True,
"from": account.from_address,
"to": to,
"cc": cc or [],
"bcc": bcc or [],
"subject": subject,
}
def _first(data: Any) -> str:
if not data:
return ""
head = data[0]
if isinstance(head, bytes):
return head.decode("utf-8", "replace")
return str(head)
def _first_descriptor(data: Any) -> bytes | None:
for part in data or []:
if isinstance(part, tuple) and part:
return part[0]
return None
def _first_payload(data: Any) -> bytes | None:
for part in data or []:
if isinstance(part, tuple) and len(part) >= 2:
return part[1]
return None
View Git Blame Copy Permalink