gatekeeper/app/services/caldav.py

import logging
import threading
from datetime import UTC, date, datetime, timedelta

from caldav import get_calendars

from app.services.settings import settings

logger = logging.getLogger(__name__)


class CaldavEventCache:
    def __init__(self, url, username, password, calendarNames):
        self.url = url
        self.username = username
        self.password = password
        self.calendarNames = calendarNames
        self.events = []
        self.is_running = False
        self.thread: threading.Thread | None = None
        self._stopped = threading.Event()

    def start(self):
        if self.is_running:
            logger.error("CaldavEventCache already running")
            return
        self.is_running = True
        self._stopped.clear()
        self.thread = threading.Thread(target=self._refresh_loop, daemon=True)
        self.thread.start()
        logger.info("CaldavEventCache started")

    def stop(self):
        self.is_running = False
        self._stopped.set()
        if self.thread:
            self.thread.join()
        logger.info("CaldavEventCache stopped")

    def _refresh_loop(self):
        while self.is_running:
            try:
                self.refresh()
            except Exception as e:
                logger.error(f"Error updating CaldavEventCache: {e}", exc_info=True)
            self._stopped.wait(settings.caldav_refresh_interval)

    def refresh(self, daysBack=1, daysAhead=7):
        calendars = get_calendars(
            url=self.url,
            username=self.username,
            password=self.password,
            calendar_name=self.calendarNames,
        )

        cached = []
        now = datetime.now(UTC)
        for cal in calendars:
            my_events = cal.search(
                event=True,
                start=now - timedelta(days=daysBack),
                end=now + timedelta(days=daysAhead),
                expand=True,
            )
            for event in my_events:
                component = event.component
                start, end = component.start, component.end
                all_day = isinstance(start, date) and not isinstance(start, datetime)

                if not all_day:
                    # Treat timezone-naive event times as system-local time.
                    if start.tzinfo is None:
                        start = start.astimezone()
                    if end.tzinfo is None:
                        end = end.astimezone()

                cached.append(
                    {
                        "uid": str(component.get("uid", "")),
                        "summary": str(component.get("summary", "")),
                        "groups": str(component.get("description", "")).splitlines(),
                        "start": start,
                        "end": end,
                        "all_day": all_day,
                    }
                )

        # dont cache empty array.
        # it most probably means there was error during caldav connection
        # TODO: handle error properly
        if len(cached) == 0:
            return

        self.events = cached

    def get_active_events(self, active_time=None):
        active_time = active_time or datetime.now().astimezone()
        if active_time.tzinfo is None:
            active_time = active_time.astimezone()

        today = active_time.date()
        active = []

        for event in self.events:
            start, end = event["start"], event["end"]

            if event["all_day"]:
                # All-day event end dates are exclusive.
                is_active = start <= today < end
            else:
                is_active = start <= active_time < end

            if is_active:
                active.append(event)

        return active

    def get_future_events(self, active_time=None):
        active_time = active_time or datetime.now().astimezone()
        if active_time.tzinfo is None:
            active_time = active_time.astimezone()

        today = active_time.date()
        future = []

        for event in self.events:
            start = event["start"]

            if event["all_day"]:  # noqa: SIM108
                # All-day event end dates are exclusive.
                is_active = start <= today
            else:
                is_active = start <= active_time

            if is_active:
                future.append(event)

        return future

    def get_active_groups(self, active_time=None):
        active_events = self.get_active_events(active_time)

        # combine all groups into a list of lists
        active_groups = []
        for active_event in active_events:
            active_groups.append(active_event["groups"])

        # flatten the lists, make every group unique
        return set(group for groups in active_groups for group in groups)

    def is_group_active(self, group_name, active_time=None):
        return group_name in self.get_active_groups(active_time)


_caldav_handler: CaldavEventCache | None = None


def init_caldav_handler() -> CaldavEventCache | None:
    """Build and start the event cache from settings.
    Returns None when caldav handling is disabled
    """
    global _caldav_handler
    if settings.disable_caldav:
        logger.info("Caldav handling disabled")
        return None

    handler = CaldavEventCache(
        url=settings.caldav_url,
        username=settings.caldav_user,
        password=settings.caldav_password,
        calendarNames=settings.caldav_calendars,
    )
    handler.start()
    _caldav_handler = handler
    return handler


def get_caldav_handler() -> CaldavEventCache | None:
    return _caldav_handler


def stop_caldav_handler() -> None:
    global _caldav_handler
    if _caldav_handler is not None:
        _caldav_handler.stop()
        _caldav_handler = None