37b032771f
Monitor an IMAP mailbox via IDLE for iMIP emails (invitations, RSVPs, cancellations) and ICS attachments, then update a CalDAV calendar. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
135 lines
5.0 KiB
Python
135 lines
5.0 KiB
Python
"""CalDAV operations: create, update, and cancel events."""
|
|
|
|
import logging
|
|
|
|
import caldav
|
|
import icalendar
|
|
|
|
from .config import CaldavConfig
|
|
from .email_processor import CalendarEvent
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class CalDAVHandler:
|
|
def __init__(self, config: CaldavConfig):
|
|
self._config = config
|
|
self._client: caldav.DAVClient | None = None
|
|
self._calendar: caldav.Calendar | None = None
|
|
|
|
def connect(self) -> None:
|
|
"""Connect to the CalDAV server and locate the calendar."""
|
|
self._client = caldav.DAVClient(
|
|
url=self._config.url,
|
|
username=self._config.username,
|
|
password=self._config.password,
|
|
)
|
|
self._calendar = caldav.Calendar(client=self._client, url=self._config.url)
|
|
log.info("Connected to CalDAV server: %s", self._config.url)
|
|
|
|
def handle_event(self, event: CalendarEvent) -> None:
|
|
"""Dispatch a calendar event based on its METHOD."""
|
|
try:
|
|
if event.method == "REQUEST" or event.method == "PUBLISH":
|
|
self._handle_request(event)
|
|
elif event.method == "REPLY":
|
|
self._handle_reply(event)
|
|
elif event.method == "CANCEL":
|
|
self._handle_cancel(event)
|
|
else:
|
|
log.warning("Unknown METHOD %s for UID %s, treating as REQUEST", event.method, event.uid)
|
|
self._handle_request(event)
|
|
except Exception:
|
|
log.error("CalDAV error processing UID %s (method=%s)", event.uid, event.method, exc_info=True)
|
|
|
|
def _find_event(self, uid: str) -> caldav.CalendarObjectResource | None:
|
|
"""Search the calendar for an event by UID."""
|
|
try:
|
|
return self._calendar.event_by_uid(uid)
|
|
except caldav.error.NotFoundError:
|
|
return None
|
|
except Exception:
|
|
log.debug("Error searching for UID %s", uid, exc_info=True)
|
|
return None
|
|
|
|
def _handle_request(self, event: CalendarEvent) -> None:
|
|
"""Handle REQUEST/PUBLISH: create or update an event."""
|
|
existing = self._find_event(event.uid)
|
|
|
|
cal = icalendar.Calendar()
|
|
cal.add("PRODID", "-//imap-idle-caldav//EN")
|
|
cal.add("VERSION", "2.0")
|
|
cal.add_component(event.vevent)
|
|
ical_data = cal.to_ical().decode("utf-8")
|
|
|
|
if existing:
|
|
existing.data = ical_data
|
|
existing.save()
|
|
log.info("Updated event UID %s", event.uid)
|
|
else:
|
|
self._calendar.save_event(ical_data)
|
|
log.info("Created event UID %s", event.uid)
|
|
|
|
def _handle_reply(self, event: CalendarEvent) -> None:
|
|
"""Handle REPLY: update attendee PARTSTAT on an existing event."""
|
|
existing = self._find_event(event.uid)
|
|
if not existing:
|
|
log.warning("REPLY for unknown UID %s, skipping", event.uid)
|
|
return
|
|
|
|
existing_cal = icalendar.Calendar.from_ical(existing.data)
|
|
reply_attendees = event.vevent.get("ATTENDEE")
|
|
if reply_attendees is None:
|
|
log.warning("REPLY with no ATTENDEE for UID %s", event.uid)
|
|
return
|
|
|
|
# Normalize to list
|
|
if not isinstance(reply_attendees, list):
|
|
reply_attendees = [reply_attendees]
|
|
|
|
# Build a map of reply attendee addresses to their PARTSTAT
|
|
reply_map: dict[str, str] = {}
|
|
for att in reply_attendees:
|
|
addr = str(att).lower()
|
|
partstat = att.params.get("PARTSTAT", "NEEDS-ACTION")
|
|
reply_map[addr] = str(partstat)
|
|
|
|
updated = False
|
|
for component in existing_cal.walk():
|
|
if component.name != "VEVENT":
|
|
continue
|
|
existing_attendees = component.get("ATTENDEE")
|
|
if existing_attendees is None:
|
|
continue
|
|
if not isinstance(existing_attendees, list):
|
|
existing_attendees = [existing_attendees]
|
|
|
|
for att in existing_attendees:
|
|
addr = str(att).lower()
|
|
if addr in reply_map:
|
|
att.params["PARTSTAT"] = reply_map[addr]
|
|
updated = True
|
|
log.info("Updated PARTSTAT for %s to %s on UID %s", addr, reply_map[addr], event.uid)
|
|
|
|
if updated:
|
|
existing.data = existing_cal.to_ical().decode("utf-8")
|
|
existing.save()
|
|
else:
|
|
log.warning("REPLY attendee not found in existing event UID %s", event.uid)
|
|
|
|
def _handle_cancel(self, event: CalendarEvent) -> None:
|
|
"""Handle CANCEL: mark event as CANCELLED."""
|
|
existing = self._find_event(event.uid)
|
|
if not existing:
|
|
log.warning("CANCEL for unknown UID %s, skipping", event.uid)
|
|
return
|
|
|
|
existing_cal = icalendar.Calendar.from_ical(existing.data)
|
|
for component in existing_cal.walk():
|
|
if component.name == "VEVENT":
|
|
component["STATUS"] = "CANCELLED"
|
|
|
|
existing.data = existing_cal.to_ical().decode("utf-8")
|
|
existing.save()
|
|
log.info("Cancelled event UID %s", event.uid)
|