"""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)}