Trixy7/plugins/calendar/providers/caldav_provider.py

# -*- 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