84 lines
2.8 KiB
Python
84 lines
2.8 KiB
Python
"""IMAP mailbox handling and SMTP sending for chancen@destengs.com."""
|
|
from __future__ import annotations
|
|
|
|
import email
|
|
import email.utils
|
|
import imaplib
|
|
import smtplib
|
|
import ssl
|
|
from email.message import EmailMessage
|
|
from email.policy import default as default_policy
|
|
|
|
IMAP_HOST, IMAP_PORT = "mail.destengs.com", 993
|
|
SMTP_HOST, SMTP_PORT = "mail.destengs.com", 587
|
|
KEYWORD = "$ProjektChecked"
|
|
|
|
|
|
class MailBox:
|
|
def __init__(self, user, password, conn=None):
|
|
self.conn = conn or imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT)
|
|
if conn is None:
|
|
self.conn.login(user, password)
|
|
self.conn.select("INBOX")
|
|
self._trash = None
|
|
|
|
def unchecked_uids(self):
|
|
typ, data = self.conn.uid("SEARCH", None, f"UNKEYWORD {KEYWORD}")
|
|
if typ != "OK" or not data or not data[0]:
|
|
return []
|
|
return [u.decode() for u in data[0].split()]
|
|
|
|
def fetch(self, uid):
|
|
typ, data = self.conn.uid("FETCH", uid, "(BODY.PEEK[])")
|
|
if typ != "OK" or not data or not data[0] or not isinstance(data[0], tuple):
|
|
raise RuntimeError(f"FETCH {uid} fehlgeschlagen: {typ}")
|
|
return email.message_from_bytes(data[0][1], policy=default_policy)
|
|
|
|
def flag_checked(self, uid):
|
|
self.conn.uid("STORE", uid, "+FLAGS", f"({KEYWORD})")
|
|
|
|
def trash_folder(self):
|
|
if self._trash:
|
|
return self._trash
|
|
typ, data = self.conn.list()
|
|
candidates = []
|
|
for raw in data or []:
|
|
line = raw.decode() if isinstance(raw, bytes) else str(raw)
|
|
name = line.rsplit(" ", 1)[-1].strip('"')
|
|
if "\\Trash" in line.split(")")[0]:
|
|
self._trash = name
|
|
return name
|
|
candidates.append(name)
|
|
for cand in ("Trash", "INBOX.Trash"):
|
|
if cand in candidates:
|
|
self._trash = cand
|
|
return cand
|
|
self._trash = "Trash"
|
|
return self._trash
|
|
|
|
def move_to_trash(self, uid):
|
|
folder = self.trash_folder()
|
|
typ, _ = self.conn.uid("MOVE", uid, folder)
|
|
if typ != "OK":
|
|
self.conn.uid("COPY", uid, folder)
|
|
self.conn.uid("STORE", uid, "+FLAGS", "(\\Deleted)")
|
|
self.conn.expunge()
|
|
|
|
def close(self):
|
|
try:
|
|
self.conn.logout()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def send_mail(user, password, to, subject, body):
|
|
msg = EmailMessage()
|
|
msg["From"], msg["To"], msg["Subject"] = user, to, subject
|
|
msg["Date"] = email.utils.formatdate(localtime=True)
|
|
msg["Message-ID"] = email.utils.make_msgid(domain="destengs.com")
|
|
msg.set_content(body)
|
|
with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=30) as server:
|
|
server.starttls(context=ssl.create_default_context())
|
|
server.login(user, password)
|
|
server.send_message(msg)
|