100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
from typing import Optional
|
|
from caldav import DAVClient, Calendar
|
|
from caldav.objects import SynchronizableCalendarObjectCollection
|
|
# from caldav.lib.error import ResponseError
|
|
from pydantic import BaseModel, Field
|
|
from datetime import datetime
|
|
from icalendar import Event as iEvent
|
|
from uuid import uuid1
|
|
|
|
from masto_bridges.config import Config
|
|
|
|
class Event(BaseModel):
|
|
|
|
summary: str # the title of the event, use for username
|
|
dtstart: datetime
|
|
dtend: datetime
|
|
uid: str = Field(default_factory=uuid1)
|
|
url: str = ''
|
|
location:str = ''
|
|
description:str = '' # post body
|
|
|
|
|
|
|
|
|
|
|
|
def to_ical(self) -> str:
|
|
evt = iEvent()
|
|
evt.add('dtstart', self.dtstart)
|
|
evt.add('dtend', self.dtend)
|
|
evt.add('uid', self.uid)
|
|
evt.add('location', self.location)
|
|
evt.add('summary', self.summary)
|
|
evt.add('description', self.description)
|
|
evt.add('url', self.url)
|
|
return evt.to_ical()
|
|
|
|
|
|
class CalDAV:
|
|
def __init__(self, config:Config):
|
|
self.config = config
|
|
for value in ('CALDAV_URL', 'CALDAV_USER', 'CALDAV_PASSWORD', 'CALDAV_CALENDAR_NAME'):
|
|
if getattr(self.config, value) is None:
|
|
raise ValueError('Need all CALDAV fields to be present in config! Check your .env file!')
|
|
self._sync_token = None
|
|
self._sync = None # type: Optional[SynchronizableCalendarObjectCollection]
|
|
|
|
|
|
def get_calendar(self, name:str) -> Optional[Calendar]:
|
|
"""
|
|
Get a calendar by its name, creating one if none exists.
|
|
"""
|
|
with self.client as client:
|
|
cals = client.principal().calendars()
|
|
# try and find
|
|
cal = [cal for cal in cals if cal.name == name]
|
|
if len(cal) == 0:
|
|
cal = client.principal().make_calendar(name=name)
|
|
else:
|
|
cal = cal[0]
|
|
|
|
return cal
|
|
|
|
def post(self, event:Event) -> bool:
|
|
with self.client as client:
|
|
calendar = self.get_calendar(self.config.CALDAV_CALENDAR_NAME)
|
|
calendar.save_event(
|
|
**event.dict()
|
|
)
|
|
|
|
return True
|
|
|
|
def poll(self) -> SynchronizableCalendarObjectCollection:
|
|
cal = self.get_calendar(self.config.CALDAV_CALENDAR_NAME)
|
|
if self._sync_token is None:
|
|
# we haven't polled yet. get and ignore previous entries.
|
|
# we don't want to post a bunch of junk every time lol
|
|
self._sync = cal.objects(load_objects=False)
|
|
|
|
self._sync_token = self._sync.sync_token
|
|
|
|
try:
|
|
self._sync = cal.objects(sync_token=self._sync_token, load_objects=True)
|
|
self._sync_token = self._sync.sync_token
|
|
except:
|
|
# deleting calendar entries makes us choke!
|
|
self._sync = cal.objects(sync_token=self._sync_token, load_objects=False)
|
|
self._sync_token = self._sync.sync_token
|
|
|
|
return self._sync
|
|
|
|
|
|
|
|
@property
|
|
def client(self) -> DAVClient:
|
|
return DAVClient(
|
|
url=self.config.CALDAV_URL,
|
|
username=self.config.CALDAV_USER,
|
|
password=self.config.CALDAV_PASSWORD
|
|
)
|