# -*- coding: utf-8 -*- """ CalDAV Calendar Provider. Unterstuetzt Nextcloud, iCloud und andere CalDAV-kompatible Server. Benoetigt: caldav, icalendar (optional) """ from datetime import datetime from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn from plugins.calendar.models import CalendarEvent, EventStatus from plugins.calendar.providers.base import CalendarProvider # Optionale Dependencies try: import caldav as caldav_lib CALDAV_AVAILABLE = True except ImportError: CALDAV_AVAILABLE = False try: from icalendar import Calendar as ICalCalendar ICALENDAR_AVAILABLE = True except ImportError: ICALENDAR_AVAILABLE = False class CalDAVProvider(CalendarProvider): """ CalDAV-basierter Calendar Provider. Authentifizierung: Username + Password (oder App-Passwort). """ def __init__(self, name: str, config: dict) -> None: super().__init__(name, config) self._client: object | None = None self._principal: object | None = None async def connect(self) -> bool: """Verbindet mit dem CalDAV-Server.""" if not CALDAV_AVAILABLE: pwarn(f"CalDAVProvider({self.name}): caldav nicht installiert") return False if not ICALENDAR_AVAILABLE: pwarn(f"CalDAVProvider({self.name}): icalendar nicht installiert") return False url = self.config.get("url", "") username = self.config.get("username", "") password = self.config.get("password", "") if not url: perror(f"CalDAVProvider({self.name}): Keine URL konfiguriert") return False try: self._client = caldav_lib.DAVClient( url=url, username=username, password=password, ) self._principal = self._client.principal() self._connected = True pinfo(f"CalDAVProvider({self.name}): Verbunden mit {url}") return True except Exception as e: perror(f"CalDAVProvider({self.name}): Verbindung fehlgeschlagen: {e}") self._connected = False return False async def disconnect(self) -> None: """Trennt die Verbindung.""" self._client = None self._principal = None self._connected = False pdebug(f"CalDAVProvider({self.name}): Getrennt") async def fetch_events( self, calendar_id: str, start: datetime, end: datetime, user: str = "", ) -> list[CalendarEvent]: """Ruft Events aus einem CalDAV-Kalender ab.""" if not self._connected or not self._principal: return [] try: # Kalender finden cal = self._find_calendar(calendar_id) if not cal: pwarn(f"CalDAVProvider({self.name}): Kalender '{calendar_id}' nicht gefunden") return [] # Events im Zeitfenster abrufen results = cal.search( start=start, end=end, event=True, expand=True, ) events: list[CalendarEvent] = [] for item in results: event = self._parse_vevent(item, calendar_id) if event: events.append(event) pdebug(f"CalDAVProvider({self.name}): {len(events)} Events aus '{calendar_id}' geladen") return events except Exception as e: perror(f"CalDAVProvider({self.name}): Fehler beim Abrufen von Events: {e}") return [] async def create_event( self, calendar_id: str, title: str, start: datetime, end: datetime, user: str = "", location: str = "", description: str = "", attendees: list[str] | None = None, ) -> CalendarEvent | None: """Erstellt einen neuen Termin im CalDAV-Kalender.""" if not self._connected or not self._principal: return None try: cal = self._find_calendar(calendar_id) if not cal: perror(f"CalDAVProvider({self.name}): Kalender '{calendar_id}' nicht gefunden") return None # iCalendar-Event erstellen import uuid event_uid = str(uuid.uuid4()) vcal = ( "BEGIN:VCALENDAR\r\n" "VERSION:2.0\r\n" "PRODID:-//Trixy//Calendar Plugin//DE\r\n" "BEGIN:VEVENT\r\n" f"UID:{event_uid}\r\n" f"DTSTART:{start.strftime('%Y%m%dT%H%M%S')}\r\n" f"DTEND:{end.strftime('%Y%m%dT%H%M%S')}\r\n" f"SUMMARY:{title}\r\n" ) if location: vcal += f"LOCATION:{location}\r\n" if description: vcal += f"DESCRIPTION:{description}\r\n" vcal += "END:VEVENT\r\nEND:VCALENDAR\r\n" cal.save_event(vcal) pinfo(f"CalDAVProvider({self.name}): Termin '{title}' erstellt") return CalendarEvent( event_id=event_uid, calendar_id=calendar_id, provider_name=self.name, title=title, start=start, end=end, location=location, description=description, attendees=attendees or [], ) except Exception as e: perror(f"CalDAVProvider({self.name}): Fehler beim Erstellen: {e}") return None async def health_check(self) -> bool: """Prueft die Verbindung zum CalDAV-Server.""" if not self._connected or not self._principal: return False try: self._principal.calendars() return True except Exception: return False # ========================================================================= # Hilfsmethoden # ========================================================================= def _find_calendar(self, calendar_id: str) -> object | None: """Findet einen Kalender anhand seiner ID.""" if not self._principal: return None try: for cal in self._principal.calendars(): # Vergleich per Name oder URL-Fragment cal_name = getattr(cal, "name", "") cal_url = str(getattr(cal, "url", "")) if calendar_id in (cal_name, cal_url) or cal_url.rstrip("/").endswith(calendar_id): return cal except Exception as e: perror(f"CalDAVProvider({self.name}): Fehler beim Suchen des Kalenders: {e}") return None def _parse_vevent(self, item: object, calendar_id: str) -> CalendarEvent | None: """Parst ein CalDAV-Event in ein CalendarEvent.""" try: ical_data = item.data if not ICALENDAR_AVAILABLE: return None cal = ICalCalendar.from_ical(ical_data) for component in cal.walk(): if component.name != "VEVENT": continue uid = str(component.get("UID", "")) summary = str(component.get("SUMMARY", "")) dtstart = component.get("DTSTART") dtend = component.get("DTEND") if not dtstart: continue start_dt = dtstart.dt end_dt = dtend.dt if dtend else start_dt # date zu datetime konvertieren (ganztaegige Termine) all_day = not isinstance(start_dt, datetime) if all_day: start_dt = datetime.combine(start_dt, datetime.min.time()) end_dt = datetime.combine(end_dt, datetime.min.time()) location = str(component.get("LOCATION", "")) description = str(component.get("DESCRIPTION", "")) organizer = str(component.get("ORGANIZER", "")) # Teilnehmer attendees: list[str] = [] attendee_list = component.get("ATTENDEE", []) if not isinstance(attendee_list, list): attendee_list = [attendee_list] for att in attendee_list: att_str = str(att).replace("mailto:", "") if att_str: attendees.append(att_str) # Status status_str = str(component.get("STATUS", "CONFIRMED")).upper() status_map = { "CONFIRMED": EventStatus.CONFIRMED, "TENTATIVE": EventStatus.TENTATIVE, "CANCELLED": EventStatus.CANCELLED, } status = status_map.get(status_str, EventStatus.CONFIRMED) return CalendarEvent( event_id=uid, calendar_id=calendar_id, provider_name=self.name, title=summary, start=start_dt, end=end_dt, all_day=all_day, location=location, description=description, attendees=attendees, organizer=organizer, status=status, raw_data={"ical": ical_data}, ) except Exception as e: perror(f"CalDAVProvider({self.name}): Fehler beim Parsen eines Events: {e}") return None