📄 calendar.py 4,971 bytes Apr 24, 2026 📋 Raw

"""Radicale CalDAV client for family calendar."""

import os
import logging
from datetime import datetime
from typing import Dict, Any, Optional
import httpx

logger = logging.getLogger(name)

RADICALE_HOST = os.getenv("RADICALE_HOST", "127.0.0.1")
RADICALE_PORT = os.getenv("RADICALE_PORT", "5232")
RADICALE_USER = os.getenv("RADICALE_USER", "assistant")
RADICALE_PASS = os.getenv("RADICALE_PASS", "family-assistant-2026")

Radicale uses UUIDs for calendar URLs - discovered via PROPFIND

FAMILY_CALENDAR_UUID = os.getenv("FAMILY_CALENDAR_UUID", "c8b37772-1e70-41d7-b325-44588049bd4d")

class CalendarClient:
"""Minimal CalDAV client for Radicale family calendar."""

def __init__(self):
    self.base_url = f"http://{RADICALE_HOST}:{RADICALE_PORT}"
    self.calendar_url = f"{self.base_url}/{RADICALE_USER}/{FAMILY_CALENDAR_UUID}"
    self.client = httpx.AsyncClient(
        auth=(RADICALE_USER, RADICALE_PASS),
        timeout=30.0
    )

async def health(self) -> Dict[str, Any]:
    """Check Radicale connectivity."""
    try:
        response = await self.client.request("PROPFIND", self.calendar_url)
        return {
            "connected": response.status_code == 207,
            "url": self.calendar_url
        }
    except Exception as e:
        logger.warning(f"Calendar health check failed: {e}")
        return {"connected": False, "url": self.calendar_url}

async def create_event(
    self,
    summary: str,
    start_datetime: str,
    end_datetime: str,
    description: str = "",
    location: str = "",
    uid: Optional[str] = None
) -> Dict[str, Any]:
    """Create a calendar event via CalDAV.

    Args:
        summary: Event title
        start_datetime: ISO 8601 start time
        end_datetime: ISO 8601 end time
        description: Event details
        location: Event location
        uid: Optional unique ID (auto-generated if not provided)

    Returns:
        Dict with success status and event details
    """
    from uuid import uuid4
    from datetime import datetime

    uid = uid or str(uuid4())
    timestamp = datetime.now().strftime("%Y%m%dT%H%M%SZ")

    # Convert ISO datetimes to iCalendar format
    def to_ical(dt_str: str) -> str:
        dt = datetime.fromisoformat(dt_str.replace('Z', '+00:00'))
        return dt.strftime("%Y%m%dT%H%M%S")

    start_ical = to_ical(start_datetime)
    end_ical = to_ical(end_datetime)

    ical = f"""BEGIN:VCALENDAR

VERSION:2.0
PRODID:-//HoffDesk Family Automation//EN
BEGIN:VEVENT
UID:{uid}@hoffdesk.family
DTSTAMP:{timestamp}
DTSTART:{start_ical}
DTEND:{end_ical}
SUMMARY:{summary}
DESCRIPTION:{description.replace(chr(10), '\n')}
LOCATION:{location}
END:VEVENT
END:VCALENDAR"""

    event_url = f"{self.calendar_url}/{uid}.ics"

    try:
        response = await self.client.put(
            event_url,
            content=ical,
            headers={"Content-Type": "text/calendar"}
        )

        success = response.status_code in (201, 204)

        if success:
            logger.info(f"Created calendar event: {summary}")
        else:
            logger.error(f"Calendar create failed: {response.status_code}")

        return {
            "created": success,
            "uid": uid,
            "url": event_url if success else None,
            "status_code": response.status_code
        }

    except Exception as e:
        logger.exception("Calendar create failed")
        return {"created": False, "error": str(e)}

async def close(self):
    await self.client.aclose()

async def delete_event(self, uid: str) -> Dict[str, Any]:
    """Delete a calendar event by UID.

    Args:
        uid: Event UID to delete

    Returns:
        Dict with deleted status
    """
    event_url = f"{self.calendar_url}/{uid}.ics"

    try:
        response = await self.client.delete(event_url)

        # 204 = deleted, 404 = already gone (still success)
        success = response.status_code in (204, 404)

        if success:
            logger.info(f"Deleted calendar event: {uid}")
        else:
            logger.error(f"Calendar delete failed: {response.status_code}")

        return {
            "deleted": success,
            "uid": uid,
            "status_code": response.status_code
        }

    except Exception as e:
        logger.exception(f"Calendar delete failed for {uid}")
        return {"deleted": False, "uid": uid, "error": str(e)}