feat(projekt-matching): Cfg dataclass, IMAP mailbox with keyword/trash handling, SMTP send

projekt-matching/projektmatch/mailer.py

@@ -0,0 +1,78 @@
+"""IMAP mailbox handling and SMTP sending for chancen@destengs.com."""
+from __future__ import annotations
+
+import email
+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[])")
+        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.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)