# -*- coding: utf-8 -*- """ IMAP-Client fuer das Email-Plugin. Stellt eine einfache Schnittstelle fuer IMAP-Operationen bereit: - Ungelesene E-Mails zaehlen - Neueste E-Mails abrufen (Header + Body) """ from __future__ import annotations import asyncio import email import email.header import email.utils import imaplib import re from dataclasses import dataclass, field from datetime import datetime from html.parser import HTMLParser from trixy_core.utils.debug import pdebug, perror @dataclass class EmailMessage: """Einzelne E-Mail-Nachricht.""" uid: str = "" subject: str = "" sender_name: str = "" sender_address: str = "" date: datetime | None = None date_str: str = "" body_text: str = "" is_unread: bool = False class _HTMLStripper(HTMLParser): """Entfernt HTML-Tags und gibt reinen Text zurueck.""" def __init__(self) -> None: super().__init__() self._parts: list[str] = [] self._skip = False def handle_starttag(self, tag: str, attrs) -> None: if tag in ("script", "style"): self._skip = True def handle_endtag(self, tag: str) -> None: if tag in ("script", "style"): self._skip = False if tag in ("br", "p", "div", "li", "tr"): self._parts.append("\n") def handle_data(self, data: str) -> None: if not self._skip: self._parts.append(data) def get_text(self) -> str: return "".join(self._parts).strip() def _strip_html(html_text: str) -> str: """Konvertiert HTML in Klartext.""" stripper = _HTMLStripper() try: stripper.feed(html_text) return stripper.get_text() except Exception: # Fallback: Regex return re.sub(r"<[^>]+>", "", html_text).strip() def _decode_header(raw: str) -> str: """Dekodiert einen MIME-Header (Subject, From, etc.).""" if not raw: return "" parts = email.header.decode_header(raw) decoded: list[str] = [] for data, charset in parts: if isinstance(data, bytes): decoded.append(data.decode(charset or "utf-8", errors="replace")) else: decoded.append(data) return " ".join(decoded) def _extract_body(msg: email.message.Message, max_length: int) -> str: """Extrahiert den Nachrichtentext (Klartext bevorzugt, HTML als Fallback).""" text_part = "" html_part = "" if msg.is_multipart(): for part in msg.walk(): content_type = part.get_content_type() disposition = str(part.get("Content-Disposition", "")) if "attachment" in disposition: continue if content_type == "text/plain" and not text_part: payload = part.get_payload(decode=True) if payload: charset = part.get_content_charset() or "utf-8" text_part = payload.decode(charset, errors="replace") elif content_type == "text/html" and not html_part: payload = part.get_payload(decode=True) if payload: charset = part.get_content_charset() or "utf-8" html_part = payload.decode(charset, errors="replace") else: payload = msg.get_payload(decode=True) if payload: charset = msg.get_content_charset() or "utf-8" content_type = msg.get_content_type() raw = payload.decode(charset, errors="replace") if content_type == "text/html": html_part = raw else: text_part = raw body = text_part if text_part else _strip_html(html_part) # Mehrfache Leerzeilen reduzieren body = re.sub(r"\n{3,}", "\n\n", body).strip() if len(body) > max_length: body = body[:max_length] + "..." return body class ImapClient: """ IMAP-Client fuer ein einzelnes E-Mail-Konto. Alle Operationen laufen in einem Thread-Executor, damit der asyncio Event-Loop nicht blockiert wird. """ def __init__(self, account_config: dict) -> None: self._host = account_config.get("host", "") self._port = account_config.get("port", 993) self._ssl = account_config.get("ssl", True) self._username = account_config.get("username", "") self._password = account_config.get("password", "") self._folder = account_config.get("folder", "INBOX") def _connect(self) -> imaplib.IMAP4 | imaplib.IMAP4_SSL: """Stellt eine IMAP-Verbindung her (synchron).""" if self._ssl: conn = imaplib.IMAP4_SSL(self._host, self._port) else: conn = imaplib.IMAP4(self._host, self._port) conn.login(self._username, self._password) conn.select(self._folder, readonly=True) return conn def _count_unseen_sync(self) -> int: """Zaehlt ungelesene E-Mails (synchron).""" conn = self._connect() try: _, data = conn.search(None, "UNSEEN") uids = data[0].split() return len(uids) finally: try: conn.close() conn.logout() except Exception: pass def _fetch_recent_sync(self, count: int, max_body_length: int) -> list[EmailMessage]: """Holt die neuesten E-Mails (synchron).""" conn = self._connect() try: # Alle UIDs holen, die letzten N nehmen _, data = conn.search(None, "ALL") all_uids = data[0].split() if not all_uids: return [] # Ungelesene UIDs ermitteln _, unseen_data = conn.search(None, "UNSEEN") unseen_uids = set(unseen_data[0].split()) recent_uids = all_uids[-count:] recent_uids.reverse() # Neuste zuerst messages: list[EmailMessage] = [] for uid in recent_uids: _, msg_data = conn.fetch(uid, "(RFC822)") if not msg_data or not msg_data[0]: continue raw_email = msg_data[0][1] if not isinstance(raw_email, bytes): continue msg = email.message_from_bytes(raw_email) # Absender parsen from_raw = _decode_header(msg.get("From", "")) sender_name, sender_addr = email.utils.parseaddr(from_raw) if not sender_name: sender_name = sender_addr.split("@")[0] if sender_addr else "Unbekannt" # Datum parsen date_str = msg.get("Date", "") date_obj = None if date_str: try: date_tuple = email.utils.parsedate_to_datetime(date_str) date_obj = date_tuple date_str = date_obj.strftime("%d.%m.%Y %H:%M") except Exception: pass # Body extrahieren body = _extract_body(msg, max_body_length) messages.append(EmailMessage( uid=uid.decode() if isinstance(uid, bytes) else str(uid), subject=_decode_header(msg.get("Subject", "")), sender_name=_decode_header(sender_name), sender_address=sender_addr, date=date_obj, date_str=date_str, body_text=body, is_unread=uid in unseen_uids, )) return messages finally: try: conn.close() conn.logout() except Exception: pass async def count_unseen(self) -> int: """Zaehlt ungelesene E-Mails (async).""" loop = asyncio.get_running_loop() return await loop.run_in_executor(None, self._count_unseen_sync) async def fetch_recent( self, count: int = 5, max_body_length: int = 500 ) -> list[EmailMessage]: """Holt die neuesten E-Mails (async).""" loop = asyncio.get_running_loop() return await loop.run_in_executor( None, self._fetch_recent_sync, count, max_body_length )