# -*- coding: utf-8 -*- """ Calendar Cloud Plugin fuer Trixy. Ermoeglicht Kalender-Anbindung an Microsoft Outlook, Google Calendar und CalDAV/Nextcloud. Mitarbeiter koennen per Sprache Termine abfragen, suchen und erstellen. Bei Erinnerungen und Terminbeginn werden Sounds auf den zugehoerigen Satellites abgespielt. """ from datetime import date, datetime, timedelta from pathlib import Path from trixy_core.plugins import TrixyPlugin from trixy_core.nlp.intent_registry import IntentRegistry from trixy_core.nlp.decorators import intent, pattern, example from trixy_core.nlp.handler import IntentReceivedData, IntentResult from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn from plugins.calendar.models import CalendarEvent from plugins.calendar.providers.base import CalendarProvider from plugins.calendar.providers.microsoft import MicrosoftProvider, HTTPX_AVAILABLE from plugins.calendar.providers.google import GoogleProvider, GOOGLE_AUTH_AVAILABLE, GOOGLE_API_AVAILABLE from plugins.calendar.providers.caldav_provider import CalDAVProvider, CALDAV_AVAILABLE, ICALENDAR_AVAILABLE from plugins.calendar.mapping import SatelliteCalendarMapper from plugins.calendar.sync import CalendarSyncManager from plugins.calendar.notifications import NotificationManager from plugins.calendar import utils # Provider-Typ-Mapping PROVIDER_TYPES: dict[str, type] = { "microsoft": MicrosoftProvider, "google": GoogleProvider, "caldav": CalDAVProvider, } class CalendarPlugin(TrixyPlugin): """ Calendar Cloud Plugin. Orchestriert Provider, Sync, Notifications und Intent-Handler. """ NAME = "calendar" VERSION = "1.0.0" DESCRIPTION = "Kalender-Anbindung an Cloud-Dienste (Outlook, Google, CalDAV)" AUTHOR = "Trixy" def __init__(self, application, plugin_path, config=None): super().__init__(application, plugin_path, config) self._registry = IntentRegistry.get_instance() self._providers: dict[str, CalendarProvider] = {} self._mapper: SatelliteCalendarMapper | None = None self._sync_manager: CalendarSyncManager | None = None self._notification_manager: NotificationManager | None = None self._scheduler_job_id: str | None = None # ========================================================================= # Lifecycle # ========================================================================= async def on_load(self) -> None: """Initialisiert das Plugin.""" pinfo(f"CalendarPlugin: Lade Plugin v{self.VERSION}") # 1. Provider initialisieren await self._init_providers() if not self._providers: pwarn("CalendarPlugin: Keine Provider verfuegbar — Plugin geladen ohne Cloud-Anbindung") # 2. Mapper erstellen self._mapper = SatelliteCalendarMapper( config=self._config, application=self.application, ) # 3. Sync-Manager erstellen sync_config = self.get_config_value("sync", {}) calendars_config = self.get_config_value("calendars", {}) self._sync_manager = CalendarSyncManager( providers=self._providers, calendars_config=calendars_config, lookahead_hours=sync_config.get("lookahead_hours", 48), max_retries=sync_config.get("max_retries", 3), ) # 4. Initialer Sync if self._providers: try: results = await self._sync_manager.sync_all() pdebug(f"CalendarPlugin: Initialer Sync: {results}") except Exception as e: perror(f"CalendarPlugin: Initialer Sync fehlgeschlagen: {e}") # 5. Notification-Manager erstellen und Sounds laden notify_config = self.get_config_value("notifications", {}) self._notification_manager = NotificationManager( application=self.application, sync_manager=self._sync_manager, mapper=self._mapper, config=notify_config, ) self._load_notification_sounds(notify_config) # 6. Scheduler-Job registrieren self._register_scheduler_job(sync_config) # 7. Notification-Check-Task starten await self._notification_manager.start_checking() # 8. Intents werden automatisch vom PluginManager registriert # (via @intent/@pattern/@example Dekoratoren auf den Handler-Methoden) pinfo(f"CalendarPlugin: Geladen ({len(self._providers)} Provider aktiv)") async def on_unload(self) -> None: """Entlaedt das Plugin.""" # 1. Notification-Task stoppen if self._notification_manager: await self._notification_manager.stop_checking() # 2. Scheduler-Job entfernen self._remove_scheduler_job() # 3. Intents deregistrieren self._registry.unregister_plugin(self.NAME) # 4. Provider disconnect for name, provider in self._providers.items(): try: await provider.disconnect() except Exception as e: perror(f"CalendarPlugin: Fehler beim Trennen von '{name}': {e}") self._providers.clear() pdebug("CalendarPlugin: Entladen") async def on_enable(self) -> None: """Aktiviert das Plugin.""" pinfo("CalendarPlugin: Aktiviert") async def on_disable(self) -> None: """Deaktiviert das Plugin.""" pinfo("CalendarPlugin: Deaktiviert") # ========================================================================= # Provider-Initialisierung # ========================================================================= async def _init_providers(self) -> None: """Initialisiert alle konfigurierten Provider.""" providers_config = self.get_config_value("providers", {}) for name, prov_config in providers_config.items(): prov_type = prov_config.get("type", "") if prov_type not in PROVIDER_TYPES: pwarn(f"CalendarPlugin: Unbekannter Provider-Typ '{prov_type}' fuer '{name}'") continue # Verfuegbarkeit pruefen if prov_type == "microsoft" and not HTTPX_AVAILABLE: pwarn(f"CalendarPlugin: httpx nicht installiert, '{name}' uebersprungen") continue if prov_type == "google" and (not GOOGLE_AUTH_AVAILABLE or not GOOGLE_API_AVAILABLE): pwarn(f"CalendarPlugin: google-auth/google-api-python-client nicht installiert, '{name}' uebersprungen") continue if prov_type == "caldav" and (not CALDAV_AVAILABLE or not ICALENDAR_AVAILABLE): pwarn(f"CalendarPlugin: caldav/icalendar nicht installiert, '{name}' uebersprungen") continue # Credentials pruefen if not self._validate_provider_config(name, prov_type, prov_config): continue provider_class = PROVIDER_TYPES[prov_type] provider = provider_class(name=name, config=prov_config) try: connected = await provider.connect() if connected: self._providers[name] = provider pinfo(f"CalendarPlugin: Provider '{name}' ({prov_type}) verbunden") else: pwarn(f"CalendarPlugin: Provider '{name}' konnte nicht verbunden werden") except Exception as e: perror(f"CalendarPlugin: Fehler bei Provider '{name}': {e}") def _validate_provider_config(self, name: str, prov_type: str, config: dict) -> bool: """Validiert die Provider-Konfiguration.""" if prov_type == "microsoft": if not config.get("tenant_id") or not config.get("client_id"): pwarn(f"CalendarPlugin: '{name}' — tenant_id und client_id erforderlich") return False auth_flow = config.get("auth_flow", "client_credentials") if auth_flow == "client_credentials" and not config.get("client_secret"): pwarn(f"CalendarPlugin: '{name}' — client_secret erforderlich fuer client_credentials") return False elif prov_type == "google": auth_flow = config.get("auth_flow", "service_account") if auth_flow == "service_account" and not config.get("service_account_file"): pwarn(f"CalendarPlugin: '{name}' — service_account_file erforderlich") return False elif prov_type == "caldav": if not config.get("url"): pwarn(f"CalendarPlugin: '{name}' — URL erforderlich") return False return True # ========================================================================= # Sound-Laden # ========================================================================= def _load_notification_sounds(self, notify_config: dict) -> None: """Laedt die Benachrichtigungs-Sounds.""" asset_service = self.application.services.get_service("AssetService") if not asset_service: pwarn("CalendarPlugin: AssetService nicht verfuegbar") return reminder_pcm = self._load_sound(asset_service, notify_config.get("reminder_sound", "calender_remember.wav")) start_pcm = self._load_sound(asset_service, notify_config.get("start_sound", "calender.wav")) if self._notification_manager: self._notification_manager.set_sounds(reminder_pcm, start_pcm) def _load_sound(self, asset_service, sound_name: str) -> bytes | None: """Laedt einen Sound und gibt die rohen PCM-Daten zurueck.""" audio_path: Path | None = asset_service.get_audio(sound_name) if not audio_path or not audio_path.exists(): pwarn(f"CalendarPlugin: Sound nicht gefunden: {sound_name}") return None try: raw = audio_path.read_bytes() pcm = raw[44:] # WAV-Header (44 Bytes) abschneiden pdebug(f"CalendarPlugin: Sound geladen: {sound_name} ({len(pcm)} bytes PCM)") return pcm except Exception as e: perror(f"CalendarPlugin: Fehler beim Laden von {sound_name}: {e}") return None # ========================================================================= # Scheduler # ========================================================================= def _register_scheduler_job(self, sync_config: dict) -> None: """Registriert den periodischen Sync-Job.""" scheduler = self.application.services.get_service("SchedulerService") if not scheduler: pwarn("CalendarPlugin: SchedulerService nicht verfuegbar, kein automatischer Sync") return from trixy_core.scheduler.trigger.interval import IntervalTrigger from trixy_core.scheduler.action.callback import CallbackAction interval_minutes = sync_config.get("interval_minutes", 30) try: job = scheduler.create_job( trigger=IntervalTrigger(minutes=interval_minutes), actions=CallbackAction(self._sync_manager.sync_all), job_id="calendar_sync", save=False, ) self._scheduler_job_id = job.id if hasattr(job, "id") else "calendar_sync" pinfo(f"CalendarPlugin: Sync-Job registriert (alle {interval_minutes} Minuten)") except Exception as e: perror(f"CalendarPlugin: Scheduler-Job konnte nicht erstellt werden: {e}") def _remove_scheduler_job(self) -> None: """Entfernt den Sync-Job.""" if not self._scheduler_job_id: return scheduler = self.application.services.get_service("SchedulerService") if scheduler: try: scheduler.remove_job(self._scheduler_job_id, delete_file=False) except Exception: pass self._scheduler_job_id = None # ========================================================================= # Intent-Handler # ========================================================================= @intent("next_appointment", description="Naechsten bevorstehenden Termin abfragen") @pattern("(wann|was) ist [mein] naechster termin") @pattern("naechster termin") @pattern("habe ich [bald] [einen] termin") @example("Wann ist mein naechster Termin?", "Naechster Termin") @example("Habe ich bald einen Termin?") async def handle_next_appointment(self, data: IntentReceivedData) -> IntentResult: """Gibt den naechsten Termin zurueck.""" calendar_keys = self._get_calendar_keys(data.satellite_id) if not calendar_keys: return IntentResult.failure( "Keine Kalender konfiguriert", response_text="Fuer diesen Raum sind keine Kalender konfiguriert.", ) event = self._sync_manager.get_next_event(calendar_keys) if not event: return IntentResult.success_with_response( "Du hast keine bevorstehenden Termine." ) response = f"Dein naechster Termin ist '{event.title}' {utils.format_event_date(event)} {utils.format_event_time(event)}." return IntentResult.success_with_response(response) @intent("search_appointment", description="Termin anhand eines Suchbegriffs suchen") @pattern("wann ist [der] {query} [termin]") @pattern("um wieviel uhr ist {query}") @pattern("wann (habe ich|findet) [den|das] {query} [statt]") @example("Wann ist der Zahnarzt Termin?", "Um wieviel Uhr ist das Stand-Up?") @example("Wann findet das Meeting statt?") async def handle_search_appointment(self, data: IntentReceivedData, query: str = "") -> IntentResult: """Sucht einen Termin anhand eines Suchbegriffs.""" query = data.get_slot("query", "") if not query: return IntentResult.failure( "Kein Suchbegriff", response_text="Was fuer einen Termin suchst du?", ) calendar_keys = self._get_calendar_keys(data.satellite_id) if not calendar_keys: return IntentResult.failure( "Keine Kalender konfiguriert", response_text="Fuer diesen Raum sind keine Kalender konfiguriert.", ) results = self._sync_manager.search_events(calendar_keys, query) if not results: return IntentResult.success_with_response( f"Ich habe keinen Termin zu '{query}' gefunden." ) event = results[0] response = f"Der Termin '{event.title}' ist {utils.format_event_date(event)} {utils.format_event_time(event)}." if event.location: response += f" Ort: {event.location}." return IntentResult.success_with_response(response) @intent("appointment_attendees", description="Teilnehmer eines Termins abfragen") @pattern("wer ist [alles] (in|bei) [dem|dem] {query}") @pattern("wer nimmt [am|an] {query} teil") @pattern("welche teilnehmer hat {query}") @pattern("wer kommt [zum|zur] {query}") @example("Wer ist alles in dem Stand-Up?", "Wer nimmt am Meeting teil?") async def handle_appointment_attendees(self, data: IntentReceivedData, query: str = "") -> IntentResult: """Gibt die Teilnehmer eines Termins zurueck.""" query = data.get_slot("query", "") if not query: return IntentResult.failure( "Kein Terminname", response_text="Fuer welchen Termin moechtest du die Teilnehmer wissen?", ) calendar_keys = self._get_calendar_keys(data.satellite_id) if not calendar_keys: return IntentResult.failure( "Keine Kalender konfiguriert", response_text="Fuer diesen Raum sind keine Kalender konfiguriert.", ) results = self._sync_manager.search_events(calendar_keys, query) if not results: return IntentResult.success_with_response( f"Ich habe keinen Termin zu '{query}' gefunden." ) event = results[0] if not event.attendees: return IntentResult.success_with_response( f"Fuer '{event.title}' sind keine Teilnehmer eingetragen." ) attendees_str = utils.format_attendees_for_tts(event.attendees) response = f"Im {event.title} sind: {attendees_str}." return IntentResult.success_with_response(response) @intent("check_person_schedule", description="Terminplan einer bestimmten Person abfragen") @pattern("wann hat {person} [seinen|ihren] [naechsten] termin") @pattern("wann hat {person} zeit") @pattern("hat {person} [heute] [noch] termine") @example("Wann hat André seinen naechsten Termin?", "Hat Max heute noch Termine?") async def handle_check_person_schedule(self, data: IntentReceivedData, person: str = "") -> IntentResult: """Gibt den Terminplan einer Person zurueck.""" person = data.get_slot("person", "") if not person: return IntentResult.failure( "Kein Personenname", response_text="Fuer wen moechtest du die Termine abfragen?", ) calendar_keys = self._mapper.find_person_calendars(person) if not calendar_keys: return IntentResult.failure( f"Keine Kalender fuer '{person}'", response_text=f"Ich habe keine Kalender fuer {person} gefunden.", ) event = self._sync_manager.get_next_event(calendar_keys) if not event: return IntentResult.success_with_response( f"{person} hat keine bevorstehenden Termine." ) response = f"{person} hat den naechsten Termin '{event.title}' {utils.format_event_date(event)} {utils.format_event_time(event)}." return IntentResult.success_with_response(response) @intent("create_appointment", description="Neuen Termin im Kalender erstellen") @pattern("(richte|trage|erstelle) [einen] [termin] {title} [um|fuer] {time} [ein]") @pattern("neuer termin {title} [am] {date} [um] {time}") @example("Richte einen Termin Zahnarzt um 15 Uhr ein") @example("Erstelle einen Termin Meeting um 14 Uhr fuer 30 Minuten") async def handle_create_appointment( self, data: IntentReceivedData, title: str = "", time: str = "", date: str = "", duration: int = 60, ) -> IntentResult: """Erstellt einen neuen Termin.""" title = data.get_slot("title", "") time_str = data.get_slot("time", "") if not title: return IntentResult.failure( "Kein Titel", response_text="Wie soll der Termin heissen?", ) if not time_str: return IntentResult.failure( "Keine Uhrzeit", response_text="Um wie viel Uhr soll der Termin sein?", ) # Schreib-Kalender ermitteln cal_info = self._mapper.get_write_calendar(data.satellite_id) if not cal_info: return IntentResult.failure( "Kein beschreibbarer Kalender", response_text="Fuer diesen Raum ist kein beschreibbarer Kalender konfiguriert.", ) # Provider holen provider = self._providers.get(cal_info.provider_name) if not provider or not provider.is_connected: return IntentResult.failure( "Provider nicht verfuegbar", response_text="Der Kalender-Dienst ist gerade nicht erreichbar.", ) # Zeit parsen parsed_time = utils.parse_time_string(time_str) if not parsed_time: return IntentResult.failure( f"Uhrzeit nicht erkannt: {time_str}", response_text=f"Ich habe die Uhrzeit '{time_str}' nicht verstanden.", ) # Datum parsen date_str = data.get_slot("date", "heute") parsed_date = utils.parse_date_string(date_str or "heute") if not parsed_date: parsed_date = date.today() # Dauer duration_minutes = data.get_slot("duration", 60) try: duration_minutes = int(duration_minutes) except (TypeError, ValueError): duration_minutes = 60 # Datetime zusammenbauen hour, minute = parsed_time start_dt = datetime(parsed_date.year, parsed_date.month, parsed_date.day, hour, minute) end_dt = start_dt + timedelta(minutes=duration_minutes) # User-Email aus Kalender-Config aufloeser cal_key = self._get_write_calendar_key(data.satellite_id) cal_config = self._mapper.get_calendar_config(cal_key) if cal_key else {} user = cal_config.get("user", "") user_email = self._resolve_user_email(cal_info.provider_name, user) # Event erstellen event = await provider.create_event( calendar_id=cal_info.calendar_id, title=title, start=start_dt, end=end_dt, user=user_email, ) if event: date_str_tts = utils.format_event_date(event) time_str_tts = utils.format_event_time(event) response = f"Termin '{title}' {date_str_tts} {time_str_tts} erstellt." return IntentResult.success_with_response(response) return IntentResult.failure( "Termin konnte nicht erstellt werden", response_text="Der Termin konnte leider nicht erstellt werden.", ) @intent("list_appointments", description="Alle Termine fuer einen bestimmten Tag auflisten") @pattern("was steht [heute|morgen] an") @pattern("welche termine habe ich {date}") @pattern("termine [fuer] {date}") @pattern("was habe ich {date} [auf dem plan]") @pattern("zeig [meine] termine [fuer] {date}") @example("Was steht heute an?", "Welche Termine habe ich morgen?") @example("Termine fuer Freitag", "Zeig mir meine Termine fuer morgen") async def handle_list_appointments(self, data: IntentReceivedData, date: str = "") -> IntentResult: """Listet alle Termine fuer einen Tag auf.""" date_str = data.get_slot("date", "heute") target_date = utils.parse_date_string(date_str or "heute") if not target_date: target_date = date.today() calendar_keys = self._get_calendar_keys(data.satellite_id) if not calendar_keys: return IntentResult.failure( "Keine Kalender konfiguriert", response_text="Fuer diesen Raum sind keine Kalender konfiguriert.", ) day_start, day_end = utils.get_day_range(target_date) events = self._sync_manager.get_events(calendar_keys, start=day_start, end=day_end) if not events: date_label = utils.format_event_date(CalendarEvent( event_id="", calendar_id="", provider_name="", title="", start=datetime.combine(target_date, datetime.min.time()), end=datetime.combine(target_date, datetime.min.time()), )) return IntentResult.success_with_response( f"Du hast {date_label} keine Termine." ) # Datum-Label date_label = utils.format_event_date(events[0]) if len(events) == 1: response = f"Du hast {date_label} einen Termin. {utils.format_event_for_tts(events[0])}." else: response = f"Du hast {date_label} {len(events)} Termine. " parts = [utils.format_event_for_tts(e) for e in events] response += ", ".join(parts[:-1]) + " und " + parts[-1] + "." return IntentResult.success_with_response(response) # ========================================================================= # Hilfsmethoden # ========================================================================= def _get_calendar_keys(self, satellite_id: str) -> list[str]: """Gibt die Kalender-Keys fuer einen Satellite zurueck.""" if not self._mapper: return [] infos = self._mapper.get_calendars_for_satellite(satellite_id) return [ key for key, cfg in self.get_config_value("calendars", {}).items() if any( i.calendar_id == cfg.get("calendar_id", "") and i.provider_name == cfg.get("provider", "") for i in infos ) ] def _get_write_calendar_key(self, satellite_id: str) -> str: """Gibt den Kalender-Key des Schreib-Kalenders zurueck.""" if not self._mapper: return "" # Aus dem Mapping-Eintrag holen entry = self._mapper._find_entry_for_satellite(satellite_id) if entry: return entry.get("default_write_calendar", "") return "" def _resolve_user_email(self, provider_name: str, user: str) -> str: """Loest einen User-Key in eine Email-Adresse auf.""" if not user: return "" provider = self._providers.get(provider_name) if not provider: return user user_mapping = provider.config.get("user_mapping", {}) return user_mapping.get(user, user)