You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
201 lines
5.8 KiB
201 lines
5.8 KiB
from enum import IntEnum
|
|
import datetime
|
|
import re
|
|
|
|
from dateutil import rrule
|
|
|
|
from .calendar_api import calendar_api
|
|
from .utils import to_dateTime, from_dateTime
|
|
|
|
|
|
class weekdays(IntEnum):
|
|
SUN = 6
|
|
MON = 0
|
|
TUE = 1
|
|
WED = 2
|
|
THU = 3
|
|
FRI = 4
|
|
SAT = 5
|
|
|
|
|
|
class Event:
|
|
"""Model for a calendar event that can be uploaded"""
|
|
|
|
def __init__(
|
|
self,
|
|
start,
|
|
end,
|
|
name,
|
|
description=None,
|
|
recurrences=None,
|
|
reminders=None,
|
|
attendees=None,
|
|
location=None,
|
|
id=None,
|
|
):
|
|
self.start = start
|
|
self.attendees = attendees
|
|
self.end = end
|
|
self.name = name
|
|
self.description = description
|
|
self.location = location
|
|
|
|
if recurrences is None:
|
|
self.recurrences = []
|
|
else:
|
|
self.recurrences = recurrences
|
|
self.id = id
|
|
if reminders is None:
|
|
self.reminders = {"useDefault": True}
|
|
else:
|
|
self.reminders = reminders
|
|
|
|
def add_reminder(self, until, method="popup"):
|
|
"""Add a reminder minutes before an event.
|
|
Use either a notification (popup) or email"""
|
|
assert method in ("email", "popup")
|
|
self.reminders["useDefault"] = False
|
|
if isinstance(until, datetime.timedelta):
|
|
minutes_until = until.days * 24 * 60
|
|
minutes_until += until.seconds // 60
|
|
else:
|
|
minutes_until = until
|
|
self.reminders.setdefault("overrides", []).append(
|
|
{"method": method, "minutes": minutes_until}
|
|
)
|
|
|
|
def add_weekly_recurrence(self, until: datetime.datetime, *days):
|
|
if not until.tzinfo:
|
|
until = until.astimezone()
|
|
self.recurrences.append(
|
|
rrule.rrule(
|
|
freq=rrule.WEEKLY,
|
|
dtstart=self.start,
|
|
wkst=weekdays.SUN,
|
|
until=until,
|
|
byweekday=days,
|
|
)
|
|
)
|
|
|
|
def serialize_recurrences(self):
|
|
for _rrule in self.recurrences:
|
|
ret_str = str(_rrule).split("\n")[-1]
|
|
ret_str = re.sub(r"(UNTIL=[^;]+)", r"\1Z", ret_str)
|
|
yield ret_str
|
|
|
|
def to_json(self):
|
|
keys = ("attendees", "description", "location", "reminders")
|
|
ret = {
|
|
"summary": self.name,
|
|
"start": to_dateTime(self.start),
|
|
"end": to_dateTime(self.end),
|
|
}
|
|
for _rrule in self.serialize_recurrences():
|
|
ret.setdefault("recurrence", []).append(_rrule)
|
|
for key in keys:
|
|
try:
|
|
value = self.__getattribute__(key)
|
|
if value:
|
|
ret[key] = value
|
|
except AttributeError:
|
|
pass
|
|
return ret
|
|
|
|
def add_attendees(self):
|
|
pass
|
|
|
|
@classmethod
|
|
def from_json(cls, body):
|
|
args = {}
|
|
args["name"] = body.get("summary", "unnamed")
|
|
args["start"] = from_dateTime(body["start"])
|
|
args["end"] = from_dateTime(body["end"])
|
|
keys = (
|
|
"attendees",
|
|
"description",
|
|
"location",
|
|
"recurrences",
|
|
"reminders",
|
|
"id",
|
|
)
|
|
for key in keys:
|
|
try:
|
|
args[key] = body[key]
|
|
except KeyError:
|
|
pass
|
|
instance = cls(**args)
|
|
return instance
|
|
|
|
@classmethod
|
|
def from_id(cls, api: calendar_api, calendar_id, event_id):
|
|
"""creates an event model from specified calendar and event_id"""
|
|
event = api.get_event_by_id(calendar_id, event_id)
|
|
return cls.from_json(event)
|
|
|
|
def upload(self, api: calendar_api, calendar_id):
|
|
"""Upload an event to calendar.
|
|
Either modifies an event in place or creates a new event."""
|
|
if self.id is not None:
|
|
event_id = api.update_event(calendar_id, self.id, self.to_json())
|
|
else:
|
|
event_id = api.create_event(calendar_id, self.to_json())
|
|
self.id = event_id
|
|
return event_id
|
|
|
|
|
|
class Calendar:
|
|
"""Model for representing a Google calendar"""
|
|
|
|
def __init__(self, api: calendar_api, calendar_id):
|
|
self.api = api
|
|
|
|
if calendar_id in api.ids.keys():
|
|
self.name = calendar_id
|
|
self.id = self.api.ids[self.name]
|
|
elif calendar_id in api.ids.values():
|
|
self.name = self.api.calendars[calendar_id]["summary"]
|
|
self.id = calendar_id
|
|
else:
|
|
raise ValueError("Non-existent calendar specified")
|
|
self.calendar_id = calendar_id
|
|
|
|
def update_or_add_event(self, event: Event):
|
|
event.upload(self.api, self.calendar_id)
|
|
|
|
def get_events(
|
|
self, start: datetime.datetime = None, end: datetime.datetime = None
|
|
):
|
|
return (
|
|
Event(event_representation)
|
|
for event_representation in self.api.get_events_in_range(
|
|
self.id, start, end
|
|
)
|
|
)
|
|
|
|
def delete_event(self, event: Event):
|
|
if event.id is not None:
|
|
return self.api.delete_event(self.calendar_id, event.id)
|
|
|
|
def search_events(
|
|
self, event_name_regex=None, event_description_regex=None, start=None, end=None
|
|
):
|
|
for event in self.get_events(start, end):
|
|
will_yield = False
|
|
if event_name_regex is not None:
|
|
will_yield = re.search(event_name_regex, Event.name) is not None
|
|
|
|
if event_description_regex is not None:
|
|
will_yield = (
|
|
re.search(event_description_regex, Event.description) is not None
|
|
or will_yield # noqa
|
|
)
|
|
|
|
if will_yield:
|
|
yield event
|
|
|
|
|
|
def calendar_dict(api: calendar_api):
|
|
return dict(
|
|
(calendar_name, Calendar(api, calendar_name))
|
|
for calendar_name in api.ids.keys()
|
|
)
|