wp-cal-integration/adapters/abc.py

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()