# -*- coding: utf-8 -*- """ Trixy Email-Plugin. Prueft E-Mail-Konten via IMAP, benachrichtigt bei neuen Mails und beantwortet Fragen zu E-Mails per Sprachsteuerung. Konfiguration analog zum Kalender-Plugin: - Accounts koennen per Satellite oder global zugewiesen werden. - Periodische Pruefung auf neue E-Mails mit Benachrichtigung. """ from __future__ import annotations import asyncio from datetime import datetime from trixy_core.plugins.trixy_plugin import TrixyPlugin from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn from trixy_core.utils.logging.file_router import log_as from plugins.email.imap_client import ImapClient, EmailMessage from plugins.email.mapping import SatelliteEmailMapper # Intent-Dekoratoren from trixy_core.nlp.decorators import intent, pattern, example from trixy_core.nlp.handler import IntentResult, IntentReceivedData class EmailPlugin(TrixyPlugin): """E-Mail-Plugin: Zugriff auf IMAP-Konten per Sprache.""" NAME = "email" VERSION = "1.0.0" DESCRIPTION = "E-Mail-Benachrichtigungen und Abfragen per Sprache" AUTHOR = "Trixy" def __init__(self, **kwargs) -> None: super().__init__(**kwargs) self._mapper: SatelliteEmailMapper | None = None self._clients: dict[str, ImapClient] = {} self._check_task: asyncio.Task | None = None self._notification_pcm: bytes = b"" # Ungelesen-Zaehler pro Account (fuer Benachrichtigungen) self._last_unseen_counts: dict[str, int] = {} # Letzte abgerufene Mails pro Satellite (fuer "lies vor") self._last_fetched: dict[str, list[EmailMessage]] = {} async def on_load(self) -> None: """Initialisiert Mapper, IMAP-Clients und Benachrichtigungs-Sound.""" self._mapper = SatelliteEmailMapper(self.config, self.application) # IMAP-Clients erstellen for key in self._mapper.get_all_account_keys(): account_cfg = self._mapper.get_account_config(key) if not account_cfg.get("host"): pwarn(f"Email-Account '{key}' hat keinen Host — uebersprungen") continue self._clients[key] = ImapClient(account_cfg) if not self._clients: pwarn("Keine gueltigen E-Mail-Accounts konfiguriert") # Benachrichtigungs-Sound laden self._load_notification_sound() # Initiale Zaehler setzen (ohne Benachrichtigung) await self._init_unseen_counts() # Periodische Pruefung starten interval = self.get_config_value("check_interval_minutes", 5) * 60 self._check_task = asyncio.create_task(self._check_loop(interval)) pinfo(f"Email-Plugin geladen: {len(self._clients)} Account(s)") async def on_unload(self) -> None: """Stoppt die periodische Pruefung.""" if self._check_task and not self._check_task.done(): self._check_task.cancel() try: await self._check_task except asyncio.CancelledError: pass self._check_task = None self._clients.clear() pinfo("Email-Plugin entladen") def _load_notification_sound(self) -> None: """Laedt den Benachrichtigungs-Sound aus den Assets.""" sound_name = self.get_config_value("notification_sound", "notification.wav") try: asset_service = self.application.services.get_service("AssetService") if not asset_service: pdebug("AssetService nicht verfuegbar") return audio_path = asset_service.get_audio(sound_name) if audio_path and audio_path.exists(): raw = audio_path.read_bytes() self._notification_pcm = raw[44:] if len(raw) > 44 else raw pdebug(f"Notification-Sound geladen: {sound_name}") else: pdebug(f"Notification-Sound nicht gefunden: {sound_name}") except Exception as e: pdebug(f"Notification-Sound Fehler: {e}") async def _init_unseen_counts(self) -> None: """Setzt die initialen Ungelesen-Zaehler (ohne Benachrichtigung).""" for key, client in self._clients.items(): try: count = await client.count_unseen() self._last_unseen_counts[key] = count pdebug(f"Email '{key}': {count} ungelesen (initial)") except Exception as e: perror(f"Email '{key}' Initialpruefung fehlgeschlagen: {e}") self._last_unseen_counts[key] = 0 # ========================================================================= # Periodische Pruefung # ========================================================================= async def _check_loop(self, interval: float) -> None: """Prueft periodisch auf neue E-Mails.""" while True: try: await asyncio.sleep(interval) with log_as("plugin.email"): await self._check_for_new_emails() except asyncio.CancelledError: break except Exception as e: perror(f"Email-Pruefung fehlgeschlagen: {e}") async def _check_for_new_emails(self) -> None: """Prueft alle Accounts auf neue E-Mails und benachrichtigt.""" for key, client in self._clients.items(): try: current = await client.count_unseen() previous = self._last_unseen_counts.get(key, 0) if current > previous: new_count = current - previous pinfo(f"Email '{key}': {new_count} neue E-Mail(s)") await self._notify_new_emails(key, new_count) self._last_unseen_counts[key] = current except Exception as e: perror(f"Email '{key}' Pruefung fehlgeschlagen: {e}") async def _notify_new_emails(self, account_key: str, new_count: int) -> None: """Benachrichtigt zugehoerige Satellites ueber neue E-Mails.""" if not self._mapper: return satellite_ids = self._mapper.get_satellites_for_account(account_key) if not satellite_ids: # Standalone-Modus: Nur TTS if new_count == 1: text = "Du hast eine neue E-Mail." else: text = f"Du hast {new_count} neue E-Mails." await self.application.events.emit("tts_request", { "request_id": f"email-notify-{account_key}", "satellite_id": "", "text": text, }) return satellites = getattr(self.application, "satellites", None) if not satellites: return for sat_id in satellite_ids: satellite = satellites.get(sat_id) if not satellite or not satellite.is_connected: continue # Notification-Sound abspielen if self._notification_pcm: try: await satellite.say(self._notification_pcm) except Exception: pass # Text vorlesen if new_count == 1: text = "Du hast eine neue E-Mail." else: text = f"Du hast {new_count} neue E-Mails." try: await satellite.speak(text) except Exception as e: perror(f"Email-Benachrichtigung an '{sat_id}' fehlgeschlagen: {e}") # ========================================================================= # Hilfsmethoden # ========================================================================= def _get_accounts_for_request(self, data: IntentReceivedData) -> list[str]: """Ermittelt die Account-Keys fuer eine Anfrage.""" if not self._mapper: return list(self._clients.keys()) satellite_id = data.satellite_id or "" accounts = self._mapper.get_accounts_for_satellite(satellite_id) return accounts if accounts else list(self._clients.keys()) async def _count_unseen_for_request(self, data: IntentReceivedData) -> int: """Zaehlt ungelesene E-Mails fuer alle Accounts eines Satellites.""" total = 0 for key in self._get_accounts_for_request(data): client = self._clients.get(key) if client: try: total += await client.count_unseen() except Exception as e: perror(f"Email '{key}' Zaehlung fehlgeschlagen: {e}") return total async def _fetch_recent_for_request( self, data: IntentReceivedData, count: int = 5 ) -> list[EmailMessage]: """Holt die neuesten E-Mails fuer alle Accounts eines Satellites.""" max_body = self.get_config_value("max_body_length", 500) all_messages: list[EmailMessage] = [] for key in self._get_accounts_for_request(data): client = self._clients.get(key) if client: try: messages = await client.fetch_recent(count=count, max_body_length=max_body) all_messages.extend(messages) except Exception as e: perror(f"Email '{key}' Abruf fehlgeschlagen: {e}") # Nach Datum sortieren (neuste zuerst) all_messages.sort(key=lambda m: m.date or datetime.min, reverse=True) # Cachen fuer "vorlesen" sat_id = data.satellite_id or "standalone" self._last_fetched[sat_id] = all_messages return all_messages[:count] # ========================================================================= # Intent-Handler # ========================================================================= @intent("check_new_emails", description="Prueft ob neue E-Mails vorhanden sind") async def handle_check_new(self, data: IntentReceivedData) -> IntentResult: """Habe ich neue E-Mails?""" try: count = await self._count_unseen_for_request(data) except Exception as e: return IntentResult.failure(f"E-Mail-Pruefung fehlgeschlagen: {e}") if count == 0: return IntentResult.success_with_response("Du hast keine neuen E-Mails.") elif count == 1: return IntentResult.success_with_response("Du hast eine ungelesene E-Mail.") else: return IntentResult.success_with_response( f"Du hast {count} ungelesene E-Mails." ) @intent("count_unread_emails", description="Zaehlt ungelesene E-Mails") async def handle_count_unread(self, data: IntentReceivedData) -> IntentResult: """Wie viele ungelesene E-Mails habe ich?""" try: count = await self._count_unseen_for_request(data) except Exception as e: return IntentResult.failure(f"E-Mail-Zaehlung fehlgeschlagen: {e}") if count == 0: return IntentResult.success_with_response( "Alle E-Mails sind gelesen. Du hast keine ungelesenen Nachrichten." ) elif count == 1: return IntentResult.success_with_response( "Du hast eine ungelesene E-Mail." ) else: return IntentResult.success_with_response( f"Du hast {count} ungelesene E-Mails." ) @intent("read_newest_email", description="Liest Absender und Betreff der neusten E-Mail vor") async def handle_read_newest(self, data: IntentReceivedData) -> IntentResult: """Was ist meine neuste E-Mail?""" try: messages = await self._fetch_recent_for_request(data, count=1) except Exception as e: return IntentResult.failure(f"E-Mail-Abruf fehlgeschlagen: {e}") if not messages: return IntentResult.success_with_response("Du hast keine E-Mails.") msg = messages[0] response = f"Deine neuste E-Mail ist von {msg.sender_name}. " response += f"Betreff: {msg.subject}." if msg.date_str: response += f" Empfangen am {msg.date_str}." return IntentResult.success_with_response(response) @intent("read_email_body", description="Liest den Inhalt der letzten E-Mail vor") async def handle_read_body(self, data: IntentReceivedData) -> IntentResult: """Lies die E-Mail vor / Was steht drin?""" sat_id = data.satellite_id or "standalone" cached = self._last_fetched.get(sat_id, []) if not cached: # Noch keine E-Mails abgerufen — hole die neueste try: cached = await self._fetch_recent_for_request(data, count=1) except Exception as e: return IntentResult.failure(f"E-Mail-Abruf fehlgeschlagen: {e}") if not cached: return IntentResult.success_with_response("Keine E-Mail zum Vorlesen vorhanden.") msg = cached[0] if not msg.body_text: return IntentResult.success_with_response( f"Die E-Mail von {msg.sender_name} mit Betreff '{msg.subject}' hat keinen lesbaren Inhalt." ) response = f"E-Mail von {msg.sender_name}. Betreff: {msg.subject}. " response += msg.body_text return IntentResult.success_with_response(response) @intent("list_recent_emails", description="Listet die letzten E-Mails auf") async def handle_list_recent(self, data: IntentReceivedData) -> IntentResult: """Zeig mir meine E-Mails.""" try: messages = await self._fetch_recent_for_request(data, count=5) except Exception as e: return IntentResult.failure(f"E-Mail-Abruf fehlgeschlagen: {e}") if not messages: return IntentResult.success_with_response("Du hast keine E-Mails.") if len(messages) == 1: response = "Du hast eine E-Mail. " else: response = f"Hier sind deine letzten {len(messages)} E-Mails. " for i, msg in enumerate(messages, 1): unread = " neu," if msg.is_unread else "" response += f"{i}:{unread} von {msg.sender_name}, Betreff: {msg.subject}. " return IntentResult.success_with_response(response) # ========================================================================= # Config-Tool Tabs # ========================================================================= def get_config_tool_tabs(self) -> list[dict]: """Bietet keine zusaetzlichen Config-Tool-Tabs.""" return []