85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
import abc
|
|
import typing
|
|
import datetime
|
|
import copy
|
|
|
|
import schema
|
|
import dateutil.rrule
|
|
|
|
class Adapter(abc.ABC):
|
|
schema: schema.Schema
|
|
|
|
def __init__(self, config, *args, **kwargs):
|
|
_ = args, kwargs
|
|
self.config = config
|
|
|
|
@classmethod
|
|
def new(cls, config: dict, *args, **kwargs):
|
|
return cls(cls.schema.validate(config) , *args, **kwargs)
|
|
|
|
@abc.abstractmethod
|
|
def login(self) -> bool:
|
|
raise NotImplementedError()
|
|
|
|
class Source(abc.ABC):
|
|
|
|
@abc.abstractmethod
|
|
def get_events(
|
|
self,
|
|
start: datetime.datetime | None=None,
|
|
until: datetime.timedelta | None=None,
|
|
limit=None,
|
|
) -> typing.Iterable[dict]:
|
|
raise NotImplementedError()
|
|
|
|
def get_events_resolved(
|
|
self,
|
|
start: datetime.datetime | None=None,
|
|
until: datetime.timedelta | None=None,
|
|
limit=None,
|
|
) -> typing.Iterable[dict]:
|
|
until = until if until else datetime.timedelta(days=365)
|
|
now = start if start else datetime.datetime.utcnow().astimezone()
|
|
now_365 = now + until
|
|
for e in self.get_events(
|
|
start=start,
|
|
until=until,
|
|
limit=limit,
|
|
):
|
|
if 'recurrence' not in e:
|
|
yield e
|
|
continue
|
|
r = e.pop('recurrence')
|
|
r = dateutil.rrule.rrulestr(
|
|
'\n'.join(r),
|
|
unfold=True,
|
|
ignoretz=True,
|
|
dtstart=datetime.datetime.fromisoformat(
|
|
e['start']['dateTime']
|
|
if 'dateTime' in e['start'] else
|
|
e['start']['date']
|
|
).replace(tzinfo=None)
|
|
)
|
|
for t_ in r.between(now.replace(tzinfo=None), now_365.replace(tzinfo=None)):
|
|
e_ = copy.deepcopy(e)
|
|
if 'dateTime' in e['start']:
|
|
e_['start']['dateTime'] = datetime.datetime.combine(t_.date(), datetime.datetime.fromisoformat(e['start']['dateTime']).time()).isoformat()
|
|
elif 'date' in e['start']:
|
|
e_['start']['date'] = datetime.datetime.combine(t_.date(), datetime.time())
|
|
if 'dateTime' in e['end']:
|
|
e_['end']['dateTime'] = datetime.datetime.combine(t_.date(), datetime.datetime.fromisoformat(e['end']['dateTime']).time()).isoformat()
|
|
elif 'date' in e['end']:
|
|
e_['end']['date'] = datetime.datetime.combine(t_.date(), datetime.time())
|
|
yield e_
|
|
yield e
|
|
|
|
class Sink(abc.ABC):
|
|
|
|
@abc.abstractmethod
|
|
def post_events(
|
|
self,
|
|
events,
|
|
start: datetime.datetime | None=None,
|
|
until: datetime.timedelta | None=None,
|
|
) -> bool:
|
|
raise NotImplementedError()
|