135 lines
4.9 KiB
Python
135 lines
4.9 KiB
Python
import logging
|
|
from typing import List, Tuple, Optional
|
|
|
|
import caldav
|
|
from caldav import Calendar, vcal
|
|
from django.db import models
|
|
from ics import Event
|
|
|
|
# Create your models here.
|
|
|
|
|
|
class CalDAVServer(models.Model):
|
|
name = models.CharField(max_length=191, verbose_name="Server name", unique=True)
|
|
url = models.URLField(verbose_name="CalDAV server URL")
|
|
username = models.CharField(max_length=191, verbose_name="CalDAV username")
|
|
password = models.CharField(max_length=191, verbose_name="CalDAV password")
|
|
ssl_verify = models.BooleanField(verbose_name="Verify SSL", default=True)
|
|
|
|
class Meta:
|
|
verbose_name = "CalDAV Server"
|
|
verbose_name_plural = "CalDAV Servers"
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
def get_client(self):
|
|
return caldav.DAVClient(url=self.url, username=self.username, password=self.password,
|
|
ssl_verify_cert=self.ssl_verify)
|
|
|
|
def test_connection(self) -> Tuple[bool, str]:
|
|
try:
|
|
self.get_client().principal().calendars()
|
|
return True, ""
|
|
except Exception as e:
|
|
return False, str(e)
|
|
|
|
def get_calendars(self) -> List[caldav.Calendar]:
|
|
return self.get_client().principal().calendars()
|
|
|
|
def get_calendar_name(self, url):
|
|
cals = self.get_client().principal().calendars()
|
|
for cal in cals:
|
|
if cal.url == url:
|
|
return cal.name
|
|
return None
|
|
|
|
|
|
class CalDAVCalendar(models.Model):
|
|
server = models.ForeignKey(to='CalDAVServer', on_delete=models.CASCADE, verbose_name="CalDAV server")
|
|
calendar_url = models.CharField(max_length=191, verbose_name="Calendar URL")
|
|
name = models.CharField(max_length=191, verbose_name="Calendar Name")
|
|
|
|
class Meta:
|
|
verbose_name = "CalDAV Calendar"
|
|
verbose_name_plural = "CalDAV Calendars"
|
|
unique_together = ['server', 'calendar_url']
|
|
|
|
def __str__(self):
|
|
return f"{self.name} on {self.server}"
|
|
|
|
def get_calendar(self) -> Optional[Calendar]:
|
|
cals = self.server.get_client().principal().calendars()
|
|
for cal in cals:
|
|
if cal.url == self.calendar_url:
|
|
return cal
|
|
return None
|
|
|
|
def upload_events(self, events: List[Event], purge: bool = True, ignore_invalid_recurring: bool = False) -> Tuple[int, int, int]:
|
|
"""
|
|
Upload events to this calendar. If purge is True, all old events will be removed.
|
|
:param events: The events to upload
|
|
:type events: List[Event]
|
|
:param purge: If old events in the calendar should be purged or not.
|
|
:type purge: bool
|
|
:param ignore_invalid_recurring: Whether to ignore errors when adding invalid recurring events (GCalendar does this)
|
|
:type ignore_invalid_recurring: bool
|
|
:return Amount of events added, updated, and deleted
|
|
:rtype Tuple[int, int, int]
|
|
"""
|
|
logger = logging.getLogger("davinci.icalendar.CalDAVCalendar.upload_events")
|
|
calendar = self.get_calendar()
|
|
if calendar is None:
|
|
logger.error(f"No calendar found on url '{self.calendar_url}'")
|
|
raise ValueError(f"No calendar found on url '{self.calendar_url}'")
|
|
|
|
# Keep track of which events were already in the calendar
|
|
purge_uids = [e.instance.vevent.uid.value for e in calendar.events()]
|
|
|
|
updated = 0
|
|
added = 0
|
|
|
|
# Import events
|
|
for event in events:
|
|
ev = f"BEGIN:VCALENDAR\n" \
|
|
f"VERSION:2.0\n" \
|
|
f"CALSCALE:GREGORIAN\n" \
|
|
f"PRODID:DAVinci.ical_utils\n" \
|
|
f"{event}\n" \
|
|
f"END:VCALENDAR"
|
|
|
|
if event.uid in purge_uids:
|
|
# Need to overwrite
|
|
try:
|
|
cevent = calendar.event_by_uid(event.uid)
|
|
cevent.data = vcal.fix(ev)
|
|
cevent.save()
|
|
except Exception as e:
|
|
if ignore_invalid_recurring and len([x for x in event.extra if x.name == "RECURRENCE-ID"]) > 0:
|
|
# Recurring event from GCal, ignore
|
|
pass
|
|
else:
|
|
logger.error(f"Error during saving of event {event.uid}: {e}")
|
|
logger.error(f"======EVENT======\n{ev}\n=================")
|
|
|
|
purge_uids.remove(event.uid)
|
|
updated += 1
|
|
|
|
else:
|
|
try:
|
|
calendar.save_event(ev)
|
|
except Exception as e:
|
|
logger.error(f"Error during saving of event {event.uid}: {e}")
|
|
logger.error(f"======EVENT======\n{ev}\n=================")
|
|
added += 1
|
|
|
|
# Purge old events
|
|
if purge:
|
|
deleted = len(purge_uids)
|
|
for uid in purge_uids:
|
|
calendar.event_by_uid(uid).delete()
|
|
else:
|
|
deleted = 0
|
|
|
|
return added, updated, deleted
|