add proxy for the first part and try to develop the second part

This commit is contained in:
Pierre Geeraert
2025-12-03 12:08:56 +00:00
parent 6c34d510e5
commit 460f18d85a
1095 changed files with 358799 additions and 5 deletions
@@ -0,0 +1,7 @@
from .__meta__ import (__author__, __copyright__, __license__, __title__,
__version__)
from .alarm import AudioAlarm, DisplayAlarm
from .attendee import Attendee, Organizer
from .event import Event, Geo
from .icalendar import Calendar
from .todo import Todo
@@ -0,0 +1,5 @@
__title__ = "ics"
__version__ = "0.7.2"
__author__ = "Nikita Marchant"
__license__ = "Apache License, Version 2.0"
__copyright__ = "Copyright 2013-2020 Nikita Marchant and individual contributors"
@@ -0,0 +1,3 @@
from ics.alarm.audio import AudioAlarm
from ics.alarm.display import DisplayAlarm
from ics.alarm.email import EmailAlarm
@@ -0,0 +1,44 @@
from typing import Optional
from ics.alarm.base import BaseAlarm
from ics.grammar.parse import ContentLine
from typing import Union
from datetime import datetime, timedelta
from ics.serializers.alarm_serializer import AudioAlarmSerializer
from ics.parsers.alarm_parser import AudioAlarmParser
class AudioAlarm(BaseAlarm):
"""
A calendar event VALARM with AUDIO option.
"""
class Meta:
name = "VALARM"
parser = AudioAlarmParser
serializer = AudioAlarmSerializer
def __init__(
self,
trigger: Union[timedelta, datetime] = None,
repeat: int = None,
duration: timedelta = None,
):
super().__init__(trigger, repeat, duration)
self._sound: Optional[ContentLine] = None
@property
def action(self):
return "AUDIO"
@property
def sound(self):
return self._sound
@sound.setter
def sound(self, sound):
assert isinstance(sound, ContentLine)
self._sound = sound
@@ -0,0 +1,154 @@
import copy
from datetime import datetime, timedelta
from typing import Optional, Union, Any
from abc import ABCMeta, abstractmethod
from ics.component import Component
from ics.utils import (
get_arrow,
get_lines
)
from ics.grammar.parse import Container
from ics.serializers.alarm_serializer import BaseAlarmSerializer
from ics.parsers.alarm_parser import BaseAlarmParser
class BaseAlarm(Component, metaclass=ABCMeta):
"""
A calendar event VALARM base class
"""
class Meta:
name = "VALARM"
parser = BaseAlarmParser
serializer = BaseAlarmSerializer
def __init__(
self,
trigger: Union[timedelta, datetime] = None,
repeat: int = None,
duration: timedelta = None,
) -> None:
"""
Instantiates a new :class:`ics.alarm.BaseAlarm`.
Adheres to RFC5545 VALARM standard: http://icalendar.org/iCalendar-RFC-5545/3-6-6-alarm-component.html
Args:
trigger (datetime.timedelta OR datetime.datetime) : Timespan to alert before parent action, or absolute time to alert
repeat (integer) : How many times to repeat the alarm
duration (datetime.timedelta) : Duration between repeats
Raises:
ValueError: If trigger, repeat, or duration do not match the RFC spec.
"""
# Set initial values
self._trigger: Optional[Union[timedelta, datetime]] = None
self._repeat: Optional[int] = None
self._duration: Optional[timedelta] = None
# Validate and parse
self.trigger = trigger
# XOR repeat and duration
if (repeat is not None) and (duration is None):
raise ValueError(
"A definition of an alarm with a repeating trigger MUST include both the DURATION and REPEAT properties."
)
if repeat:
self.repeat = repeat
if duration:
self.duration = duration
self.extra = Container(name="VALARM")
@classmethod
def _from_container(cls, container: Container, *args: Any, **kwargs: Any):
ret = super()._from_container(container, *args, **kwargs)
get_lines(ret.extra, "ACTION", keep=False) # Just drop the ACTION line
return ret
@property
def trigger(self) -> Optional[Union[timedelta, datetime]]:
"""The trigger condition for the alarm
| Returns either a timedelta or datetime object
"""
return self._trigger
@trigger.setter
def trigger(self, value: Optional[Union[timedelta, datetime]]) -> None:
if isinstance(value, datetime):
value = get_arrow(value)
self._trigger = value
@property
def repeat(self) -> Optional[int]:
"""Number of times to repeat alarm
| Returns an integer for number of alarm repeats
| Value must be >= 0
"""
return self._repeat
@repeat.setter
def repeat(self, value: Optional[int]) -> None:
if value is not None and value < 0:
raise ValueError("Repeat must be great than or equal to 0.")
self._repeat = value
@property
def duration(self) -> Optional[timedelta]:
"""Duration between alarm repeats
| Returns a timedelta object
| Timespan must return positive total_seconds() value
"""
return self._duration
@duration.setter
def duration(self, value: Optional[timedelta]) -> None:
if value is not None and value.total_seconds() < 0:
raise ValueError("Alarm duration timespan must be positive.")
self._duration = value
@property
@abstractmethod
def action(self):
""" VALARM action to be implemented by concrete classes
"""
raise NotImplementedError("Base class cannot be instantiated directly")
def __repr__(self):
value = "{0} trigger:{1}".format(type(self).__name__, self.trigger)
if self.repeat:
value += " repeat:{0} duration:{1}".format(self.repeat, self.duration)
return "<{0}>".format(value)
def __ne__(self, other) -> bool:
return not self.__eq__(other)
def __eq__(self, other) -> bool:
"""Two alarms are considered equal if they have the same type and base values."""
return (
type(self) is type(other)
and self.trigger == other.trigger
and self.repeat == other.repeat
and self.duration == other.duration
)
def clone(self):
"""
Returns:
Alarm: an exact copy of self"""
clone = copy.copy(self)
clone.extra = clone.extra.clone()
return clone
@@ -0,0 +1,33 @@
from ics.alarm.base import BaseAlarm
import copy
from typing import Union
from datetime import datetime, timedelta
from ics.serializers.alarm_serializer import CustomAlarmSerializer
from ics.parsers.alarm_parser import CustomAlarmParser
class CustomAlarm(BaseAlarm):
"""
A calendar event VALARM with custom ACTION.
"""
class Meta:
name = "VALARM"
parser = CustomAlarmParser
serializer = CustomAlarmSerializer
def __init__(
self,
trigger: Union[timedelta, datetime] = None,
repeat: int = None,
duration: timedelta = None,
action: str = None,
) -> None:
super().__init__(trigger, repeat, duration)
self._action = action
@property
def action(self):
return self._action
@@ -0,0 +1,34 @@
from ics.alarm.base import BaseAlarm
from typing import Union
from datetime import datetime, timedelta
from ics.serializers.alarm_serializer import DisplayAlarmSerializer
from ics.parsers.alarm_parser import DisplayAlarmParser
class DisplayAlarm(BaseAlarm):
"""
A calendar event VALARM with DISPLAY option.
"""
class Meta:
name = "VALARM"
parser = DisplayAlarmParser
serializer = DisplayAlarmSerializer
def __init__(
self,
trigger: Union[timedelta, datetime] = None,
repeat: int = None,
duration: timedelta = None,
display_text: str = None,
):
super().__init__(trigger, repeat, duration)
self.display_text = display_text
@property
def action(self):
return "DISPLAY"
@@ -0,0 +1,38 @@
from ics.attendee import Attendee
from ics.alarm.base import BaseAlarm
from typing import Union, List
from datetime import datetime, timedelta
from ics.serializers.alarm_serializer import EmailAlarmSerializer
from ics.parsers.alarm_parser import EmailAlarmParser
class EmailAlarm(BaseAlarm):
"""
A calendar event VALARM with Email option.
"""
class Meta:
name = "VALARM"
parser = EmailAlarmParser
serializer = EmailAlarmSerializer
def __init__(
self,
trigger: Union[timedelta, datetime] = None,
repeat: int = None,
duration: timedelta = None,
subject: str = None,
body: str = None,
recipients: List[Attendee] = None,
):
super().__init__(trigger, repeat, duration)
self.subject = subject
self.body = body
self.recipients = recipients if recipients else []
@property
def action(self):
return "EMAIL"
@@ -0,0 +1,19 @@
from ics.alarm.base import BaseAlarm
from ics.serializers.alarm_serializer import NoneAlarmSerializer
from ics.parsers.alarm_parser import NoneAlarmParser
class NoneAlarm(BaseAlarm):
"""
A calendar event VALARM with NONE option.
"""
class Meta:
name = "VALARM"
parser = NoneAlarmParser
serializer = NoneAlarmSerializer
@property
def action(self):
return "NONE"
@@ -0,0 +1,28 @@
from ics.alarm.audio import AudioAlarm
from ics.alarm.none import NoneAlarm
from ics.alarm.custom import CustomAlarm
from ics.alarm.display import DisplayAlarm
from ics.alarm.email import EmailAlarm
from ics.utils import get_lines
def get_type_from_action(action_type):
if action_type == "DISPLAY":
return DisplayAlarm
elif action_type == "AUDIO":
return AudioAlarm
elif action_type == "NONE":
return NoneAlarm
elif action_type == 'EMAIL':
return EmailAlarm
else:
return CustomAlarm
def get_type_from_container(container):
action_type_lines = get_lines(container, "ACTION", keep=True)
if len(action_type_lines) > 1:
raise ValueError("Too many ACTION parameters in VALARM")
action_type = action_type_lines[0]
return get_type_from_action(action_type.value)
@@ -0,0 +1,92 @@
import warnings
from typing import Dict, List
from ics.grammar.parse import ContentLine
from ics.parsers.attendee_parser import AttendeeParser, PersonParser
from ics.serializers.attendee_serializer import AttendeeSerializer, PersonSerializer
from ics.utils import escape_string, unescape_string
class Person(object):
class Meta:
name = "ABSTRACT-PERSON"
parser = PersonParser
serializer = PersonSerializer
def __init__(self, email: str, common_name: str = None, dir: str = None, sent_by: str = None) -> None:
self.email = email
self.common_name = common_name or email
self.dir = dir
self.sent_by = sent_by
self.extra: Dict[str, List[str]] = {}
@classmethod
def parse(cls, line: ContentLine) -> "Person":
email = unescape_string(line.value)
if email.lower().startswith("mailto:"):
email = email[len("mailto:"):]
val = cls(email)
val.populate(line)
return val
def populate(self, line: ContentLine) -> None:
if line.name != self.Meta.name:
raise ValueError("line isn't an {}".format(self.Meta.name))
params = dict(line.params)
for param_name, (parser, options) in self.Meta.parser.get_parsers().items():
values = params.pop(param_name, [])
if not values and options.required:
if options.default:
values = options.default
default_str = "\\n".join(map(str, options.default))
message = ("The %s property was not found and is required by the RFC." +
" A default value of \"%s\" has been used instead") % (param_name, default_str)
warnings.warn(message)
else:
raise ValueError('A {} must have at least one {}'.format(line.name, param_name))
if not options.multiple and len(values) > 1:
raise ValueError('A {} must have at most one {}'.format(line.name, param_name))
if options.multiple:
parser(self, values) # Send a list or empty list
else:
if len(values) == 1:
parser(self, values[0]) # Send the element
else:
parser(self, None) # Send None
self.extra = params # Store unused lines
def serialize(self) -> ContentLine:
line = ContentLine(self.Meta.name, params=self.extra, value=escape_string('mailto:%s' % self.email))
for output in self.Meta.serializer.get_serializers():
output(self, line)
return line
def __str__(self) -> str:
"""Returns the attendee in an ContentLine format."""
return str(self.serialize())
class Organizer(Person):
class Meta:
name = 'ORGANIZER'
parser = PersonParser
serializer = PersonSerializer
class Attendee(Person):
def __init__(self, email: str, common_name: str = None, dir: str = None, sent_by: str = None,
rsvp: bool = None, role: str = None, partstat: str = None, cutype: str = None) -> None:
super().__init__(email, common_name, dir, sent_by)
self.rsvp = rsvp
self.role = role
self.partstat = partstat
self.cutype = cutype
class Meta:
name = 'ATTENDEE'
parser = AttendeeParser
serializer = AttendeeSerializer
@@ -0,0 +1,89 @@
import warnings
from collections import namedtuple
from typing import Any, Dict, Tuple, Iterable
from ics.grammar.parse import Container
from .utils import get_lines
from .serializers.serializer import Serializer
from .parsers.parser import Parser
Extractor = namedtuple(
'Extractor',
['function', 'type', 'required', 'multiple', 'default']
)
class Component(object):
class Meta:
name = "ABSTRACT"
parser = Parser
serializer = Serializer
_classmethod_args: Tuple
_classmethod_kwargs: Dict
@classmethod
def _from_container(cls, container: Container, *args: Any, **kwargs: Any):
k = cls()
k._classmethod_args = args
k._classmethod_kwargs = kwargs
k._populate(container)
return k
def _populate(self, container: Container) -> None:
if container.name != self.Meta.name:
raise ValueError("container isn't an {}".format(self.Meta.name))
for line_name, (parser, options) in self.Meta.parser.get_parsers().items():
lines = get_lines(container, line_name)
if not lines and options.required:
if options.default:
lines = options.default
default_str = "\\n".join(map(str, options.default))
message = ("The %s property was not found and is required by the RFC." +
" A default value of \"%s\" has been used instead") % (line_name, default_str)
warnings.warn(message)
else:
raise ValueError(
'A {} must have at least one {}'
.format(container.name, line_name))
if not options.multiple and len(lines) > 1:
raise ValueError(
'A {} must have at most one {}'
.format(container.name, line_name))
if options.multiple:
parser(self, lines) # Send a list or empty list
else:
if len(lines) == 1:
parser(self, lines[0]) # Send the element
else:
parser(self, None) # Send None
self.extra = container # Store unused lines
def serialize(self) -> str:
"""Returns the component in an iCalendar format."""
container = self.extra.clone()
for output in self.Meta.serializer.get_serializers():
output(self, container)
return str(container)
def serialize_iter(self) -> Iterable[str]:
"""Returns the component in an iCalendar format.
This returns an Iterable of multiple string chunks which should be concatenated to form the actual ics representation.
Note that individual items of the returned Iterable not necessarily correspond to individual lines,
linebreaks are contained at the right places within the items."""
return self.serialize().splitlines(keepends=True)
def __str__(self) -> str:
"""Starting from version 0.9, returns a short description of the Component."""
warnings.warn(
"Behaviour of str(Component) will change in version 0.9 to only return a short description, NOT the ics representation. "
"Use the explicit Component.serialize() to get the ics representation.", FutureWarning
)
return self.serialize()
@@ -0,0 +1,489 @@
import copy
from datetime import datetime, timedelta
from typing import (Dict, Iterable, List, NamedTuple, Optional, Set,
Tuple, Union)
from arrow import Arrow
from .alarm.base import BaseAlarm
from .attendee import Attendee, Organizer
from .component import Component
from ics.grammar.parse import Container
from .types import ArrowLike
from .utils import (get_arrow, uid_gen)
from ics.parsers.event_parser import EventParser
from ics.serializers.event_serializer import EventSerializer
class Geo(NamedTuple):
latitude: float
longitude: float
class Event(Component):
"""A calendar event.
Can be full-day or between two instants.
Can be defined by a beginning instant and
a duration *or* end instant.
Unsupported event attributes can be found in `event.extra`,
a :class:`ics.parse.Container`. You may add some by appending a
:class:`ics.parse.ContentLine` to `.extra`
"""
class Meta:
name = "VEVENT"
parser = EventParser
serializer = EventSerializer
def __init__(self,
name: str = None,
begin: ArrowLike = None,
end: ArrowLike = None,
duration: timedelta = None,
uid: str = None,
description: str = None,
created: ArrowLike = None,
last_modified: ArrowLike = None,
location: str = None,
url: str = None,
transparent: bool = None,
alarms: Iterable[BaseAlarm] = None,
attendees: Iterable[Attendee] = None,
categories: Iterable[str] = None,
status: str = None,
organizer: Organizer = None,
geo=None,
classification: str = None,
) -> None:
"""Instantiates a new :class:`ics.event.Event`.
Args:
name: rfc5545 SUMMARY property
begin (Arrow-compatible)
end (Arrow-compatible)
duration
uid: must be unique
description
created (Arrow-compatible)
last_modified (Arrow-compatible)
location
url
transparent
alarms
attendees
categories
status
organizer
classification
Raises:
ValueError: if `end` and `duration` are specified at the same time
"""
self._duration: Optional[timedelta] = None
self._end_time: Optional[ArrowLike] = None
self._begin: Optional[ArrowLike] = None
self._begin_precision = None
self._status: Optional[str] = None
self._classification: Optional[str] = None
self.organizer: Optional[Organizer] = organizer
self.uid: str = uid_gen() if not uid else uid
self.description: Optional[str] = description
self.created: Optional[ArrowLike] = get_arrow(created)
self.last_modified: Optional[ArrowLike] = get_arrow(last_modified)
self.location: Optional[str] = location
self.url: Optional[str] = url
self.transparent: Optional[bool] = transparent
self.alarms: List[BaseAlarm] = list()
self.attendees: Set[Attendee] = set()
self.categories: Set[str] = set()
self.geo = geo
self.extra = Container(name='VEVENT')
self.name = name
self.begin = begin
if duration and end:
raise ValueError(
'Event() may not specify an end and a duration \
at the same time')
elif end: # End was specified
self.end = end
elif duration: # Duration was specified
self.duration = duration
if alarms is not None:
self.alarms = list(alarms)
self.status = status
self.classification = classification
if categories is not None:
self.categories.update(set(categories))
if attendees is not None:
self.attendees.update(set(attendees))
def has_end(self) -> bool:
"""
Return:
bool: self has an end
"""
return bool(self._end_time or self._duration)
def add_attendee(self, attendee: Attendee):
""" Add an attendee to the attendees set
"""
self.attendees.add(attendee)
@property
def begin(self) -> Arrow:
"""Get or set the beginning of the event.
| Will return an :class:`Arrow` object.
| May be set to anything that :func:`Arrow.get` understands.
| If an end is defined (not a duration), .begin must not
be set to a superior value.
"""
return self._begin
@begin.setter
def begin(self, value: ArrowLike):
value = get_arrow(value)
if value and self._end_time and value > self._end_time:
raise ValueError('Begin must be before end')
self._begin = value
self._begin_precision = 'second'
@property
def end(self) -> Arrow:
"""Get or set the end of the event.
| Will return an :class:`Arrow` object.
| May be set to anything that :func:`Arrow.get` understands.
| If set to a non null value, removes any already
existing duration.
| Setting to None will have unexpected behavior if
begin is not None.
| Must not be set to an inferior value than self.begin.
"""
if self._duration: # if end is duration defined
# return the beginning + duration
return self.begin + self._duration
elif self._end_time: # if end is time defined
if self.all_day:
return self._end_time
else:
return self._end_time
elif self._begin: # if end is not defined
if self.all_day:
return self._begin + timedelta(days=1)
else:
# instant event
return self._begin
else:
return None
@end.setter
def end(self, value: ArrowLike):
value = get_arrow(value)
if value and self._begin and value < self._begin:
raise ValueError('End must be after begin')
self._end_time = value
if value:
self._duration = None
@property
def duration(self) -> Optional[timedelta]:
"""Get or set the duration of the event.
| Will return a timedelta object.
| May be set to anything that timedelta() understands.
| May be set with a dict ({"days":2, "hours":6}).
| If set to a non null value, removes any already
existing end time.
"""
if self._duration:
return self._duration
elif self.end:
# because of the clever getter for end, this also takes care of all_day events
return self.end - self.begin
else:
# event has neither start, nor end, nor duration
return None
@duration.setter
def duration(self, value: timedelta):
if isinstance(value, dict):
value = timedelta(**value)
elif isinstance(value, timedelta):
value = value
elif value is not None:
value = timedelta(value)
if value:
self._end_time = None
self._duration = value
@property
def geo(self) -> Optional[Geo]:
"""Get or set the geo position of the event.
| Will return a namedtuple object.
| May be set to any Geo, tuple or dict with latitude and longitude keys.
| If set to a non null value, removes any already
existing geo.
"""
return self._geo
@geo.setter
def geo(self, value: Union[Dict[str, float], Tuple[float, float], Geo, None]):
if isinstance(value, dict):
latitude, longitude = value['latitude'], value['longitude']
value = Geo(latitude, longitude)
elif value is not None:
latitude, longitude = value
value = Geo(latitude, longitude)
self._geo = value
@property
def all_day(self):
"""
Return:
bool: self is an all-day event
"""
# the event may have an end, also given in 'day' precision
return self._begin_precision == 'day'
def make_all_day(self) -> None:
"""Transforms self to an all-day event.
The event will span all the days from the begin to the end day.
"""
if self.all_day:
# Do nothing if we already are a all day event
return
begin_day = self.begin.floor('day')
end_day = self.end.floor('day')
self._begin = begin_day
# for a one day event, we don't need a _end_time
if begin_day == end_day:
self._end_time = None
else:
self._end_time = end_day + timedelta(days=1)
self._duration = None
self._begin_precision = 'day'
@property
def status(self) -> Optional[str]:
return self._status
@status.setter
def status(self, value: Optional[str]):
if isinstance(value, str):
value = value.upper()
statuses = (None, 'TENTATIVE', 'CONFIRMED', 'CANCELLED')
if value not in statuses:
raise ValueError('status must be one of %s' % ", ".join([repr(x) for x in statuses]))
self._status = value
@property
def classification(self):
return self._classification
@classification.setter
def classification(self, value):
if value is not None:
if not isinstance(value, str):
raise ValueError('classification must be a str')
self._classification = value
else:
self._classification = None
def __repr__(self) -> str:
name = "'{}' ".format(self.name) if self.name else ''
if self.all_day:
assert self._begin
if not self._end_time or self._begin == self._end_time:
return "<all-day Event {}{}>".format(name, self.begin.strftime('%Y-%m-%d'))
else:
return "<all-day Event {}begin:{} end:{}>".format(name, self._begin.strftime('%Y-%m-%d'), self._end_time.strftime('%Y-%m-%d'))
elif self.begin is None:
return "<Event '{}'>".format(self.name) if self.name else "<Event>"
else:
return "<Event {}begin:{} end:{}>".format(name, self.begin, self.end)
def starts_within(self, other) -> bool:
if not isinstance(other, Event):
raise NotImplementedError(
'Cannot compare Event and {}'.format(type(other)))
return self.begin >= other.begin and self.begin <= other.end
def ends_within(self, other) -> bool:
if not isinstance(other, Event):
raise NotImplementedError(
'Cannot compare Event and {}'.format(type(other)))
return self.end >= other.begin and self.end <= other.end
def intersects(self, other) -> bool:
if not isinstance(other, Event):
raise NotImplementedError(
'Cannot compare Event and {}'.format(type(other)))
return (self.starts_within(other)
or self.ends_within(other)
or other.starts_within(self)
or other.ends_within(self))
__xor__ = intersects
def includes(self, other) -> bool:
if isinstance(other, Event):
return other.starts_within(self) and other.ends_within(self)
if isinstance(other, datetime):
return self.begin <= other and self.end >= other
raise NotImplementedError(
'Cannot compare Event and {}'.format(type(other)))
def is_included_in(self, other) -> bool:
if isinstance(other, Event):
return other.includes(self)
raise NotImplementedError(
'Cannot compare Event and {}'.format(type(other)))
__in__ = is_included_in
def __lt__(self, other) -> bool:
if isinstance(other, Event):
if self.begin is None and other.begin is None:
if self.name is None and other.name is None:
return False
elif self.name is None:
return True
elif other.name is None:
return False
else:
return self.name < other.name
# if we arrive here, at least one of self.begin
# and other.begin is not None
# so if they are equal, they are both Arrow
elif self.begin == other.begin:
if self.end is None:
return True
elif other.end is None:
return False
else:
return self.end < other.end
else:
return self.begin < other.begin
if isinstance(other, datetime):
return self.begin < other
raise NotImplementedError(
'Cannot compare Event and {}'.format(type(other)))
def __le__(self, other) -> bool:
if isinstance(other, Event):
if self.begin is None and other.begin is None:
if self.name is None and other.name is None:
return True
elif self.name is None:
return True
elif other.name is None:
return False
else:
return self.name <= other.name
elif self.begin == other.begin:
if self.end is None:
return True
elif other.end is None:
return False
else:
return self.end <= other.end
else:
return self.begin <= other.begin
if isinstance(other, datetime):
return self.begin <= other
raise NotImplementedError(
'Cannot compare Event and {}'.format(type(other)))
def __gt__(self, other) -> bool:
return not self.__le__(other)
def __ge__(self, other) -> bool:
return not self.__lt__(other)
def __eq__(self, other: object) -> bool:
if isinstance(other, Event):
return (self.name == other.name
and self.begin == other.begin
and self.end == other.end
and self.duration == other.duration
and self.description == other.description
and self.created == other.created
and self.last_modified == other.last_modified
and self.location == other.location
and self.url == other.url
and self.transparent == other.transparent
and self.alarms == other.alarms
and self.attendees == other.attendees
and self.categories == other.categories
and self.status == other.status
and self.organizer == other.organizer)
raise NotImplementedError(
'Cannot compare Event and {}'.format(type(other)))
def time_equals(self, other) -> bool:
return (self.begin == other.begin) and (self.end == other.end)
def join(self, other, *args, **kwarg):
"""Create a new event which covers the time range of two intersecting events
All extra parameters are passed to the Event constructor.
Args:
other: the other event
Returns:
a new Event instance
"""
event = Event(*args, **kwarg)
if self.intersects(other):
if self.starts_within(other):
event.begin = other.begin
else:
event.begin = self.begin
if self.ends_within(other):
event.end = other.end
else:
event.end = self.end
return event
raise ValueError('Cannot join {} with {}: they don\'t intersect.'.format(self, other))
__and__ = join
def clone(self):
"""
Returns:
Event: an exact copy of self"""
clone = copy.copy(self)
clone.extra = clone.extra.clone()
clone.alarms = copy.copy(self.alarms)
clone.categories = copy.copy(self.categories)
return clone
def __hash__(self) -> int:
"""
Returns:
int: hash of self. Based on self.uid."""
return int(''.join(map(lambda x: '%.3d' % ord(x), self.uid)))
@@ -0,0 +1,18 @@
@@grammar :: contentline
@@whitespace :: None
@@left_recursion :: False
# generate with tatsu --generate-parser --outfile ics/grammar/contentline.py ics/grammar/contentline.ebnf
start = @:contentline $;
full = @+:contentline { /(\r?\n)+/ @+:contentline }+ /(\r?\n)*/ $;
contentline = name:ALPHADIGIT_MINUS_PLUS { ";" params+:param }* ":" value:VALUE_CHAR_STAR ;
param = name:ALPHADIGIT_MINUS_PLUS "=" values+:param_value { "," values+:param_value }* ;
param_value = DQUOTE ~ value:QSAFE_CHAR_STAR DQUOTE quoted:`true` | value:SAFE_CHAR_STAR quoted:`false` ;
ALPHADIGIT_MINUS_PLUS = /[a-zA-Z0-9-]+/;
QSAFE_CHAR_STAR = /[^\x00-\x08\x0A-\x1F\x22\x7F]*/;
SAFE_CHAR_STAR = /[^\x00-\x08\x0A-\x1F\x22\x2C\x3A\x3B\x7F]*/;
VALUE_CHAR_STAR = /[^\x00-\x08\x0A-\x1F\x7F]*/;
DQUOTE = '"';
@@ -0,0 +1,235 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# CAVEAT UTILITOR
#
# This file was automatically generated by TatSu.
#
# https://pypi.python.org/pypi/tatsu/
#
# Any changes you make to it will be overwritten the next time
# the file is generated.
from __future__ import print_function, division, absolute_import, unicode_literals
import sys
from tatsu.buffering import Buffer
from tatsu.parsing import Parser
from tatsu.parsing import tatsumasu, leftrec, nomemo
from tatsu.parsing import leftrec, nomemo # noqa
from tatsu.util import re, generic_main # noqa
KEYWORDS = {} # type: ignore
class contentlineBuffer(Buffer):
def __init__(
self,
text,
whitespace=None,
nameguard=None,
comments_re=None,
eol_comments_re=None,
ignorecase=None,
namechars='',
**kwargs
):
super(contentlineBuffer, self).__init__(
text,
whitespace=whitespace,
nameguard=nameguard,
comments_re=comments_re,
eol_comments_re=eol_comments_re,
ignorecase=ignorecase,
namechars=namechars,
**kwargs
)
class contentlineParser(Parser):
def __init__(
self,
whitespace=None,
nameguard=None,
comments_re=None,
eol_comments_re=None,
ignorecase=None,
left_recursion=False,
parseinfo=True,
keywords=None,
namechars='',
buffer_class=contentlineBuffer,
**kwargs
):
if keywords is None:
keywords = KEYWORDS
super(contentlineParser, self).__init__(
whitespace=whitespace,
nameguard=nameguard,
comments_re=comments_re,
eol_comments_re=eol_comments_re,
ignorecase=ignorecase,
left_recursion=left_recursion,
parseinfo=parseinfo,
keywords=keywords,
namechars=namechars,
buffer_class=buffer_class,
**kwargs
)
@tatsumasu()
def _start_(self): # noqa
self._contentline_()
self.name_last_node('@')
self._check_eof()
@tatsumasu()
def _full_(self): # noqa
self._contentline_()
self.add_last_node_to_name('@')
def block1():
self._pattern('(\\r?\\n)+')
self._contentline_()
self.add_last_node_to_name('@')
self._positive_closure(block1)
self._pattern('(\\r?\\n)*')
self._check_eof()
@tatsumasu()
def _contentline_(self): # noqa
self._ALPHADIGIT_MINUS_PLUS_()
self.name_last_node('name')
def block1():
self._token(';')
self._param_()
self.add_last_node_to_name('params')
self._closure(block1)
self._token(':')
self._VALUE_CHAR_STAR_()
self.name_last_node('value')
self.ast._define(
['name', 'value'],
['params']
)
@tatsumasu()
def _param_(self): # noqa
self._ALPHADIGIT_MINUS_PLUS_()
self.name_last_node('name')
self._token('=')
self._param_value_()
self.add_last_node_to_name('values')
def block2():
self._token(',')
self._param_value_()
self.add_last_node_to_name('values')
self._closure(block2)
self.ast._define(
['name'],
['values']
)
@tatsumasu()
def _param_value_(self): # noqa
with self._choice():
with self._option():
self._DQUOTE_()
self._cut()
self._QSAFE_CHAR_STAR_()
self.name_last_node('value')
self._DQUOTE_()
self._constant('true')
self.name_last_node('quoted')
with self._option():
self._SAFE_CHAR_STAR_()
self.name_last_node('value')
self._constant('false')
self.name_last_node('quoted')
self._error('no available options')
self.ast._define(
['quoted', 'value'],
[]
)
@tatsumasu()
def _ALPHADIGIT_MINUS_PLUS_(self): # noqa
self._pattern('[a-zA-Z0-9-]+')
@tatsumasu()
def _QSAFE_CHAR_STAR_(self): # noqa
self._pattern('[^\\x00-\\x08\\x0A-\\x1F\\x22\\x7F]*')
@tatsumasu()
def _SAFE_CHAR_STAR_(self): # noqa
self._pattern('[^\\x00-\\x08\\x0A-\\x1F\\x22\\x2C\\x3A\\x3B\\x7F]*')
@tatsumasu()
def _VALUE_CHAR_STAR_(self): # noqa
self._pattern('[^\\x00-\\x08\\x0A-\\x1F\\x7F]*')
@tatsumasu()
def _DQUOTE_(self): # noqa
self._token('"')
class contentlineSemantics(object):
def start(self, ast): # noqa
return ast
def full(self, ast): # noqa
return ast
def contentline(self, ast): # noqa
return ast
def param(self, ast): # noqa
return ast
def param_value(self, ast): # noqa
return ast
def ALPHADIGIT_MINUS_PLUS(self, ast): # noqa
return ast
def QSAFE_CHAR_STAR(self, ast): # noqa
return ast
def SAFE_CHAR_STAR(self, ast): # noqa
return ast
def VALUE_CHAR_STAR(self, ast): # noqa
return ast
def DQUOTE(self, ast): # noqa
return ast
def main(filename, start=None, **kwargs):
if start is None:
start = 'start'
if not filename or filename == '-':
text = sys.stdin.read()
else:
with open(filename) as f:
text = f.read()
parser = contentlineParser()
return parser.parse(text, rule_name=start, filename=filename, **kwargs)
if __name__ == '__main__':
import json
from tatsu.util import asjson
ast = generic_main(main, contentlineParser, name='contentline')
print('AST:')
print(ast)
print()
print('JSON:')
print(json.dumps(asjson(ast), indent=2))
print()
@@ -0,0 +1,227 @@
import collections
import re
from typing import Dict, List
import attr
import tatsu
from ics.grammar.contentline import contentlineParser
# grammar_path = Path(__file__).parent.joinpath('contentline.ebnf')
#
# with open(grammar_path) as fd:
# GRAMMAR = tatsu.compile(fd.read())
GRAMMAR = contentlineParser()
@attr.s(auto_exc=True)
class ParseError(Exception):
line: str = attr.ib(default="")
nr: int = attr.ib(default=-1)
class ContentLine:
"""
Represents one property line.
For example:
``FOO;BAR=1:YOLO`` is represented by
``ContentLine('FOO', {'BAR': ['1']}, 'YOLO'))``
Args:
name: The name of the property (uppercased for consistency and
easier comparison)
params: A map name:list of values
value: The value of the property
"""
def __eq__(self, other):
ret = (self.name == other.name and
self.params == other.params and
self.value == other.value)
return ret
def __ne__(self, other):
return not self.__eq__(other)
def __init__(self, name: str, params: Dict[str, List[str]] = {}, value: str = ''):
self.name = name.upper()
self.params = params
self.value = value
def __str__(self):
params_str = ''
for pname in self.params:
params_str += ';{}={}'.format(pname, ','.join(self.params[pname]))
return "{}{}:{}".format(self.name, params_str, self.value)
def __repr__(self):
return "<ContentLine '{}' with {} parameter{}. Value='{}'>".format(
self.name,
len(self.params),
"s" if len(self.params) > 1 else "",
self.value,
)
def __getitem__(self, item):
return self.params[item]
def __setitem__(self, item, *values):
self.params[item] = [val for val in values]
@classmethod
def parse(cls, line, nr=-1):
"""Parse a single iCalendar-formatted line into a ContentLine"""
if "\n" in line or "\r" in line:
raise ValueError("ContentLine can only contain escaped newlines")
try:
ast = GRAMMAR.parse(line)
except tatsu.exceptions.FailedToken:
raise ParseError(line, nr)
name = ''.join(ast['name'])
value = ''.join(ast['value'])
params = {}
for param_ast in ast.get('params', []):
param_name = ''.join(param_ast["name"])
param_values = [''.join(x["value"]) for x in param_ast["values"]]
params[param_name] = param_values
return cls(name, params, value)
def clone(self):
"""Makes a copy of itself"""
# dict(self.params) -> Make a copy of the dict
return self.__class__(self.name, dict(self.params), self.value)
class Container(list):
"""Represents an iCalendar object.
Contains a list of ContentLines or Containers.
Args:
name: the name of the object (VCALENDAR, VEVENT etc.)
items: Containers or ContentLines
"""
def __init__(self, name: str, *items: List):
super(Container, self).__init__(items)
self.name = name
def __str__(self):
name = self.name
ret = ['BEGIN:' + name]
for line in self:
ret.append(str(line))
ret.append('END:' + name)
return "\r\n".join(ret)
def __repr__(self):
return "<Container '{}' with {} element{}>" \
.format(self.name, len(self), "s" if len(self) > 1 else "")
@classmethod
def parse(cls, name, tokenized_lines):
items = []
for line in tokenized_lines:
if line.name == 'BEGIN':
items.append(Container.parse(line.value, tokenized_lines))
elif line.name == 'END':
if line.value != name:
raise ParseError(
"Expected END:{}, got END:{}".format(name, line.value))
break
else:
items.append(line)
return cls(name, *items)
def clone(self):
"""Makes a copy of itself"""
c = self.__class__(self.name)
for elem in self:
c.append(elem.clone())
return c
def unfold_lines(physical_lines, with_linenr=False):
if not isinstance(physical_lines, collections.abc.Iterable):
raise ParseError('Parameter `physical_lines` must be an iterable')
current_nr = -1
current_line = ''
for nr, line in enumerate(physical_lines):
if len(line.strip()) == 0:
continue
elif not current_line:
current_nr = nr
current_line = line.strip('\r')
elif line[0] in (' ', '\t'):
current_line += line[1:].strip('\r')
else:
if with_linenr:
yield current_nr, current_line
else:
yield current_line
current_nr = nr
current_line = line.strip('\r')
if current_line:
if with_linenr:
yield current_nr, current_line
else:
yield current_line
def tokenize_line(unfolded_lines):
for line in unfolded_lines:
if isinstance(line, tuple):
yield ContentLine.parse(line[1], line[0])
else:
yield ContentLine.parse(line)
def parse(tokenized_lines, block_name=None):
res = []
for line in tokenized_lines:
if line.name == 'BEGIN':
res.append(Container.parse(line.value, tokenized_lines))
else:
res.append(line)
return res
def lines_to_container(lines, linewise=True):
if linewise:
return parse(tokenize_line(unfold_lines(lines, with_linenr=True))) # linewise
else:
return string_to_container("\r\n".join(lines), linewise) # full-string
def string_to_container(txt, linewise=True):
if linewise:
return lines_to_container(txt.split("\n"), linewise) # linewise
else:
return parse(string_to_content_lines(txt)) # full-string
def string_to_content_lines(txt):
txt = re.sub("\r?\n[ \t]", "", txt)
ast = GRAMMAR.parse(txt, rule_name='full')
for line in ast:
line = line[0]
name = ''.join(line['name'])
value = ''.join(line['value'])
params = {}
for param_ast in line.get('params', []):
param_name = ''.join(param_ast["name"])
param_values = [''.join(x["value"]) for x in param_ast["values"]]
params[param_name] = param_values
yield ContentLine(name, params, value)
def calendar_string_to_containers(string):
if not isinstance(string, str):
raise TypeError("Expecting a string")
return string_to_container(string)
@@ -0,0 +1,148 @@
import copy
import warnings
from typing import Dict, Iterable, Optional, Set, Union
from six import text_type
from .component import Component
from .event import Event
from ics.grammar.parse import Container, calendar_string_to_containers
from .timeline import Timeline
from .todo import Todo
from ics.parsers.icalendar_parser import CalendarParser
from ics.serializers.icalendar_serializer import CalendarSerializer
class Calendar(Component):
"""
Represents an unique rfc5545 iCalendar.
Attributes:
events: a set of Event contained in the Calendar
todos: a set of Todo contained in the Calendar
timeline: a Timeline instance linked to this Calendar
"""
class Meta:
name = 'VCALENDAR'
parser = CalendarParser
serializer = CalendarSerializer
def __init__(
self,
imports: Union[str, Container] = None,
events: Iterable[Event] = None,
todos: Iterable[Todo] = None,
creator: str = None
):
"""Instantiates a new Calendar.
Args:
imports (**str**): data to be imported into the Calendar,
events (**Set[Event]**): Events to be added to the calendar
todos (Set[Todo]): Todos to be added to the calendar
creator (string): uid of the creator program.
If ``imports`` is specified, every other argument will be ignored.
"""
self._timezones: Dict = {} # FIXME mypy
self.events: Set[Event] = set()
self.todos: Set[Todo] = set()
self.extra = Container(name='VCALENDAR')
self.scale = None
self.method = None
self.timeline = Timeline(self)
if imports is not None:
if isinstance(imports, Container):
self._populate(imports)
else:
containers = calendar_string_to_containers(imports)
if len(containers) != 1:
raise NotImplementedError(
'Multiple calendars in one file are not supported by this method. Use ics.Calendar.parse_multiple()')
self._populate(containers[0]) # Use first calendar
else:
if events is not None:
self.events.update(set(events))
if todos is not None:
self.todos.update(set(todos))
self._creator = creator
@classmethod
def parse_multiple(cls, string):
""""
Parses an input string that may contain mutiple calendars
and retruns a list of :class:`ics.event.Calendar`
"""
containers = calendar_string_to_containers(string)
return [cls(imports=c) for c in containers]
def __repr__(self) -> str:
return "<Calendar with {} event{} and {} todo{}>" \
.format(len(self.events),
"s" if len(self.events) > 1 else "",
len(self.todos),
"s" if len(self.todos) > 1 else "")
def __iter__(self) -> Iterable[str]:
"""Returns:
iterable: an iterable version of seralize(), line per line
(with line-endings).
Example:
Can be used to write calendar to a file:
>>> c = Calendar(); c.events.add(Event(name="My cool event"))
>>> open('my.ics', 'w').writelines(c)
"""
warnings.warn(
"Using Calendar as Iterable is deprecated and will be removed in version 0.8. "
"Use the explicit calendar.serialize_iter() instead.", DeprecationWarning)
yield from self.serialize_iter()
def __eq__(self, other: object) -> bool:
if not isinstance(other, Calendar):
raise NotImplementedError
for attr in ('extra', 'scale', 'method', 'creator'):
if self.__getattribute__(attr) != other.__getattribute__(attr):
return False
return (self.events == other.events) and (self.todos == other.todos)
def __ne__(self, other: object) -> bool:
return not self.__eq__(other)
@property
def creator(self) -> Optional[str]:
"""Get or set the calendar's creator.
| Will return a string.
| May be set to a string.
| Creator is the PRODID iCalendar property.
| It uniquely identifies the program that created the calendar.
"""
return self._creator
@creator.setter
def creator(self, value: Optional[str]) -> None:
if not isinstance(value, text_type):
raise ValueError('Event.creator must be unicode data not {}'.format(type(value)))
self._creator = value
def clone(self):
"""
Returns:
Calendar: an exact deep copy of self
"""
clone = copy.copy(self)
clone.extra = clone.extra.clone()
clone.events = copy.copy(self.events)
clone.todos = copy.copy(self.todos)
clone._timezones = copy.copy(self._timezones)
return clone
@@ -0,0 +1,65 @@
import warnings
from ics.attendee import Attendee
from ics.parsers.parser import Parser, option
from ics.utils import iso_to_arrow, parse_duration, unescape_string
class BaseAlarmParser(Parser):
@option(required=True)
def parse_trigger(alarm, line):
if line.params.get("VALUE", [""])[0] == "DATE-TIME":
alarm.trigger = iso_to_arrow(line)
elif line.params.get("VALUE", ["DURATION"])[0] == "DURATION":
alarm.trigger = parse_duration(line.value)
else:
warnings.warn(
"ics.py encountered a TRIGGER of unknown type '%s'. It has been ignored."
% line.params["VALUE"][0]
)
def parse_duration(alarm, line):
if line:
alarm._duration = parse_duration(line.value)
def parse_repeat(alarm, line):
if line:
alarm._repeat = int(line.value)
class CustomAlarmParser(BaseAlarmParser):
def parse_action(alarm, line, required=True):
print(line)
if line:
alarm._action = line.value
class AudioAlarmParser(BaseAlarmParser):
def parse_attach(alarm, line):
if line:
alarm._sound = line
class DisplayAlarmParser(BaseAlarmParser):
@option(required=True)
def parse_description(alarm, line):
alarm.display_text = unescape_string(line.value) if line else None
class EmailAlarmParser(BaseAlarmParser):
@option(required=True)
def parse_description(alarm, line):
alarm.body = unescape_string(line.value) if line else None
@option(required=True)
def parse_summary(alarm, line):
alarm.subject = unescape_string(line.value) if line else None
@option(required=True, multiple=True)
def parse_attendee(alarm, lines):
for line in lines:
alarm.recipients.append(Attendee.parse(line))
class NoneAlarmParser(BaseAlarmParser):
pass
@@ -0,0 +1,34 @@
from ics.parsers.parser import Parser
from ics.utils import unescape_string
class PersonParser(Parser):
def parse_cn(person: "Person", value):
if value:
person.common_name = unescape_string(value)
def parse_dir(person: "Person", value):
if value:
person.dir = unescape_string(value)
def parse_sent_by(person: "Person", value):
if value:
person.sent_by = unescape_string(value)
class AttendeeParser(PersonParser):
def parse_rsvp(attendee: "Attendee", value):
if value:
attendee.rsvp = bool(value)
def parse_role(attendee: "Attendee", value):
if value:
attendee.role = unescape_string(value)
def parse_partstat(attendee: "Attendee", value):
if value:
attendee.partstat = unescape_string(value)
def parse_cutype(attendee: "Attendee", value):
if value:
attendee.cutype = unescape_string(value)
@@ -0,0 +1,98 @@
import re
from ics import Organizer, Attendee
from ics.alarm.utils import get_type_from_container
from ics.parsers.parser import Parser, option
from ics.utils import (iso_precision, iso_to_arrow, parse_duration,
unescape_string)
class EventParser(Parser):
def parse_dtstamp(event, line):
if line:
# get the dict of vtimezones passed to the classmethod
tz_dict = event._classmethod_kwargs["tz"]
event.created = iso_to_arrow(line, tz_dict)
def parse_last_modified(event, line):
if line:
tz_dict = event._classmethod_kwargs["tz"]
event.last_modified = iso_to_arrow(line, tz_dict)
def parse_dtstart(event, line):
if line:
# get the dict of vtimezones passed to the classmethod
tz_dict = event._classmethod_kwargs["tz"]
event.begin = iso_to_arrow(line, tz_dict)
event._begin_precision = iso_precision(line.value)
def parse_duration(event, line):
if line:
# TODO: DRY [1]
if event._end_time: # pragma: no cover
raise ValueError("An event can't have both DTEND and DURATION")
event._duration = parse_duration(line.value)
def parse_dtend(event, line):
if line:
# TODO: DRY [1]
if event._duration:
raise ValueError("An event can't have both DTEND and DURATION")
# get the dict of vtimezones passed to the classmethod
tz_dict = event._classmethod_kwargs["tz"]
event._end_time = iso_to_arrow(line, tz_dict)
# one could also save the end_precision to check that if begin_precision is day, end_precision also is
def parse_summary(event, line):
event.name = unescape_string(line.value) if line else None
def parse_organizer(event, line):
event.organizer = Organizer.parse(line) if line else None
@option(multiple=True)
def parse_attendee(event, lines):
for line in lines:
event.attendees.add(Attendee.parse(line))
def parse_description(event, line):
event.description = unescape_string(line.value) if line else None
def parse_location(event, line):
event.location = unescape_string(line.value) if line else None
def parse_geo(event, line):
if line:
latitude, _, longitude = unescape_string(line.value).partition(";")
event.geo = float(latitude), float(longitude)
def parse_url(event, line):
event.url = unescape_string(line.value) if line else None
def parse_transp(event, line):
if line and line.value in ["TRANSPARENT", "OPAQUE"]:
event.transparent = line.value == "TRANSPARENT"
# TODO : make uid required ?
# TODO : add option somewhere to ignore some errors
def parse_uid(event, line):
if line:
event.uid = line.value
@option(multiple=True)
def parse_valarm(event, lines):
event.alarms = [get_type_from_container(x)._from_container(x) for x in lines]
def parse_status(event, line):
if line:
event.status = line.value
def parse_class(event, line):
if line:
event.classification = line.value
def parse_categories(event, line):
event.categories = set()
if line:
# In the regular expression: Only match unquoted commas.
for cat in re.split("(?<!\\\\),", line.value):
event.categories.update({unescape_string(cat)})
@@ -0,0 +1,80 @@
from dateutil.tz import tzical
from six import StringIO
from ics.event import Event
from ics.grammar.parse import ContentLine
from ics.parsers.parser import Parser, option
from ics.todo import Todo
from ics.utils import remove_sequence, remove_x
class CalendarParser(Parser):
@option(required=True)
def parse_prodid(calendar, prodid):
calendar._creator = prodid.value
_version_default = [ContentLine(name="VERSION", value="2.0")]
@option(required=True, default=_version_default)
def parse_version(calendar, line):
version = line
# TODO : should take care of minver/maxver
if ";" in version.value:
_, calendar.version = version.value.split(";")
else:
calendar.version = version.value
def parse_calscale(calendar, line):
calscale = line
if calscale:
calendar.scale = calscale.value.lower()
calendar.scale_params = calscale.params
else:
calendar.scale = "georgian"
calendar.scale_params = {}
def parse_method(calendar, line):
method = line
if method:
calendar.method = method.value
calendar.method_params = method.params
else:
calendar.method = None
calendar.method_params = {}
@option(multiple=True)
def parse_vtimezone(calendar, vtimezones):
"""Receives a list of VTIMEZONE blocks.
Parses them and adds them to calendar._timezones.
"""
for vtimezone in vtimezones:
remove_x(vtimezone) # Remove non standard lines from the block
remove_sequence(
vtimezone
) # Remove SEQUENCE lines because tzical does not understand them
fake_file = StringIO()
fake_file.write(str(vtimezone)) # Represent the block as a string
fake_file.seek(0)
timezones = tzical(fake_file) # tzical does not like strings
# timezones is a tzical object and could contain multiple timezones
for key in timezones.keys():
calendar._timezones[key] = timezones.get(key)
@option(multiple=True)
def parse_vevent(calendar, lines):
# tz=calendar._timezones gives access to the event factory to the
# timezones list
def event_factory(x):
return Event._from_container(x, tz=calendar._timezones)
calendar.events = set(map(event_factory, lines))
@option(multiple=True)
def parse_vtodo(calendar, lines):
# tz=calendar._timezones gives access to the event factory to the
# timezones list
def todo_factory(x):
return Todo._from_container(x, tz=calendar._timezones)
calendar.todos = set(map(todo_factory, lines))
@@ -0,0 +1,45 @@
from typing import Callable, Dict, List, NamedTuple, Optional, Tuple
from ics.grammar.parse import ContentLine
class ParserOption(NamedTuple):
required: bool = False
multiple: bool = False
default: Optional[List[ContentLine]] = None
class Parser:
@classmethod
def get_parsers(cls) -> Dict[str, Tuple[Callable, ParserOption]]:
methods = [
(method_name, getattr(cls, method_name))
for method_name in dir(cls)
if callable(getattr(cls, method_name))
]
parsers = [
(method_name, method_callable)
for (method_name, method_callable) in methods
if method_name.startswith("parse_")
]
return {
method_name.split("_", 1)[1]
.upper()
.replace("_", "-"): (
method_callable,
getattr(method_callable, "options", ParserOption()),
)
for (method_name, method_callable) in parsers
}
def option(
required: bool = False,
multiple: bool = False,
default: Optional[List[ContentLine]] = None,
) -> Callable:
def decorator(fn):
fn.options = ParserOption(required, multiple, default)
return fn
return decorator
@@ -0,0 +1,83 @@
from typing import TYPE_CHECKING, List
from ics.alarm.utils import get_type_from_container
from ics.grammar.parse import ContentLine
from ics.parsers.parser import Parser, option
from ics.utils import iso_to_arrow, parse_duration, unescape_string
if TYPE_CHECKING:
from .todo_parser import Todo
class TodoParser(Parser):
@option(required=True)
def parse_dtstamp(todo: "Todo", line: ContentLine):
if line:
# get the dict of vtimezones passed to the classmethod
tz_dict = todo._classmethod_kwargs["tz"]
todo.dtstamp = iso_to_arrow(line, tz_dict)
@option(required=True)
def parse_uid(todo: "Todo", line: ContentLine):
if line:
todo.uid = line.value
def parse_completed(todo: "Todo", line: ContentLine):
if line:
# get the dict of vtimezones passed to the classmethod
tz_dict = todo._classmethod_kwargs["tz"]
todo.completed = iso_to_arrow(line, tz_dict)
def parse_created(todo: "Todo", line: ContentLine):
if line:
# get the dict of vtimezones passed to the classmethod
tz_dict = todo._classmethod_kwargs["tz"]
todo.created = iso_to_arrow(line, tz_dict)
def parse_description(todo: "Todo", line: ContentLine):
todo.description = unescape_string(line.value) if line else None
def parse_dtstart(todo: "Todo", line: ContentLine):
if line:
# get the dict of vtimezones passed to the classmethod
tz_dict = todo._classmethod_kwargs["tz"]
todo.begin = iso_to_arrow(line, tz_dict)
def parse_location(todo: "Todo", line: ContentLine):
todo.location = unescape_string(line.value) if line else None
def parse_percent_complete(todo: "Todo", line: ContentLine):
todo.percent = int(line.value) if line else None
def parse_priority(todo: "Todo", line: ContentLine):
todo.priority = int(line.value) if line else None
def parse_summary(todo: "Todo", line: ContentLine):
todo.name = unescape_string(line.value) if line else None
def parse_url(todo: "Todo", line: ContentLine):
todo.url = unescape_string(line.value) if line else None
def parse_due(todo: "Todo", line: ContentLine):
if line:
# TODO: DRY [1]
if todo._duration:
raise ValueError("A todo can't have both DUE and DURATION")
# get the dict of vtimezones passed to the classmethod
tz_dict = todo._classmethod_kwargs["tz"]
todo._due_time = iso_to_arrow(line, tz_dict)
def parse_duration(todo: "Todo", line: ContentLine):
if line:
# TODO: DRY [1]
if todo._due_time: # pragma: no cover
raise ValueError("An todo can't have both DUE and DURATION")
todo._duration = parse_duration(line.value)
@option(multiple=True)
def parse_valarm(todo: "Todo", lines: List[ContentLine]):
todo.alarms = [get_type_from_container(x)._from_container(x) for x in lines]
def parse_status(todo: "Todo", line: ContentLine):
if line:
todo.status = line.value
@@ -0,0 +1,72 @@
from datetime import timedelta
from ics.grammar.parse import ContentLine
from ics.serializers.serializer import Serializer
from ics.utils import arrow_to_iso, escape_string, timedelta_to_duration
class BaseAlarmSerializer(Serializer):
def serialize_trigger(alarm, container):
if alarm.trigger is None:
raise ValueError("Alarm must have a trigger")
if isinstance(alarm.trigger, timedelta):
representation = timedelta_to_duration(alarm.trigger)
container.append(ContentLine("TRIGGER", value=representation))
else:
container.append(
ContentLine(
"TRIGGER",
params={"VALUE": ["DATE-TIME"]},
value=arrow_to_iso(alarm.trigger),
)
)
def serialize_duration(alarm, container):
if alarm.duration:
representation = timedelta_to_duration(alarm.duration)
container.append(ContentLine("DURATION", value=representation))
def serialize_repeat(alarm, container):
if alarm.repeat:
container.append(ContentLine("REPEAT", value=alarm.repeat))
def serialize_action(alarm, container):
container.append(ContentLine("ACTION", value=alarm.action))
class CustomAlarmSerializer(BaseAlarmSerializer):
pass
class AudioAlarmSerializer(BaseAlarmSerializer):
def serialize_attach(alarm, container):
if alarm._sound:
container.append(str(alarm._sound))
class DisplayAlarmSerializer(BaseAlarmSerializer):
def serialize_description(alarm, container):
container.append(
ContentLine("DESCRIPTION", value=escape_string(alarm.display_text or ""))
)
class EmailAlarmSerializer(BaseAlarmSerializer):
def serialize_body(alarm, container):
container.append(
ContentLine("DESCRIPTION", value=escape_string(alarm.body or ""))
)
def serialize_subject(alarm, container):
container.append(
ContentLine("SUMMARY", value=escape_string(alarm.subject or ""))
)
def serialize_recipients(alarm, container):
for attendee in alarm.recipients:
container.append(attendee.serialize())
class NoneAlarmSerializer(BaseAlarmSerializer):
pass
@@ -0,0 +1,35 @@
from ics.grammar.parse import ContentLine
from ics.serializers.serializer import Serializer
from ics.utils import escape_string
class PersonSerializer(Serializer):
def serialize_cn(person: "Person", line: ContentLine):
if person.common_name:
line.params["CN"] = [escape_string(person.common_name)]
def serialize_dir(person: "Person", line: ContentLine):
if person.dir:
line.params["DIR"] = [escape_string(person.dir)]
def serialize_sent_by(person: "Person", line: ContentLine):
if person.sent_by:
line.params["SENT-BY"] = [escape_string(person.sent_by)]
class AttendeeSerializer(PersonSerializer):
def serialize_rsvp(attendee: "Attendee", line: ContentLine):
if attendee.rsvp is not None:
line.params["RSVP"] = [attendee.rsvp]
def serialize_role(attendee: "Attendee", line: ContentLine):
if attendee.role:
line.params["ROLE"] = [escape_string(attendee.role)]
def serialize_partstat(attendee: "Attendee", line: ContentLine):
if attendee.partstat:
line.params["PARTSTAT"] = [escape_string(attendee.partstat)]
def serialize_cutype(attendee: "Attendee", line: ContentLine):
if attendee.cutype:
line.params["CUTYPE"] = [escape_string(attendee.cutype)]
@@ -0,0 +1,121 @@
from ics.attendee import Attendee, Organizer
from ics.grammar.parse import ContentLine
from ics.serializers.serializer import Serializer
from ics.utils import (arrow_date_to_iso, arrow_to_iso, escape_string,
timedelta_to_duration, uid_gen)
class EventSerializer(Serializer):
def serialize_created(event, container):
if event.created:
container.append(ContentLine("DTSTAMP", value=arrow_to_iso(event.created)))
def serialize_last_modified(event, container):
if event.last_modified:
instant = event.last_modified
container.append(ContentLine("LAST-MODIFIED", value=arrow_to_iso(instant)))
def serialize_start(event, container):
if event.begin and not event.all_day:
container.append(ContentLine("DTSTART", value=arrow_to_iso(event.begin)))
def serialize_all_day(event, container):
if event.begin and event.all_day:
container.append(
ContentLine(
"DTSTART",
params={"VALUE": ["DATE"]},
value=arrow_date_to_iso(event.begin),
)
)
if event._end_time:
container.append(
ContentLine(
"DTEND",
params={"VALUE": ["DATE"]},
value=arrow_date_to_iso(event.end),
)
)
def serialize_duration(event, container):
if event._duration and event.begin:
representation = timedelta_to_duration(event._duration)
container.append(ContentLine("DURATION", value=representation))
def serialize_end(event, container):
if event.begin and event._end_time and not event.all_day:
container.append(ContentLine("DTEND", value=arrow_to_iso(event.end)))
def serialize_summary(event, container):
if event.name:
container.append(ContentLine("SUMMARY", value=escape_string(event.name)))
def serialize_organizer(event, container):
if event.organizer:
organizer = event.organizer
if isinstance(organizer, str):
organizer = Organizer(organizer)
container.append(organizer.serialize())
def serialize_attendee(event, container):
for attendee in event.attendees:
if isinstance(attendee, str):
attendee = Attendee(attendee)
container.append(attendee.serialize())
def serialize_description(event, container):
if event.description:
container.append(
ContentLine("DESCRIPTION", value=escape_string(event.description))
)
def serialize_location(event, container):
if event.location:
container.append(
ContentLine("LOCATION", value=escape_string(event.location))
)
def serialize_geo(event, container):
if event.geo:
container.append(ContentLine("GEO", value="%f;%f" % event.geo))
def serialize_url(event, container):
if event.url:
container.append(ContentLine("URL", value=escape_string(event.url)))
def serialize_transparent(event, container):
if event.transparent is None:
return
if event.transparent:
container.append(ContentLine("TRANSP", value=escape_string("TRANSPARENT")))
else:
container.append(ContentLine("TRANSP", value=escape_string("OPAQUE")))
def serialize_uid(event, container):
if event.uid:
uid = event.uid
else:
uid = uid_gen()
container.append(ContentLine("UID", value=uid))
def serialize_alarm(event, container):
for alarm in event.alarms:
container.append(str(alarm))
def serialize_status(event, container):
if event.status:
container.append(ContentLine("STATUS", value=event.status))
def serialize_class(event, container):
if event.classification:
container.append(ContentLine("CLASS", value=event.classification))
def serialize_categories(event, container):
if event.categories:
container.append(
ContentLine(
"CATEGORIES",
value=",".join([escape_string(s) for s in event.categories]),
)
)
@@ -0,0 +1,31 @@
from ics.grammar.parse import ContentLine
from ics.serializers.serializer import Serializer
class CalendarSerializer(Serializer):
def serialize_0version(calendar, container): # 0version will be sorted first
container.append(ContentLine("VERSION", value="2.0"))
def serialize_1prodid(calendar, container): # 1prodid will be sorted second
if calendar.creator:
creator = calendar.creator
else:
creator = "ics.py - http://git.io/lLljaA"
container.append(ContentLine("PRODID", value=creator))
def serialize_calscale(calendar, container):
if calendar.scale:
container.append(ContentLine("CALSCALE", value=calendar.scale.upper()))
def serialize_method(calendar, container):
if calendar.method:
container.append(ContentLine("METHOD", value=calendar.method.upper()))
def serialize_event(calendar, container):
for event in calendar.events:
container.append(event.serialize())
def serialize_todo(calendar, container):
for todo in calendar.todos:
container.append(todo.serialize())
@@ -0,0 +1,16 @@
from typing import Callable
class Serializer:
@classmethod
def get_serializers(cls) -> Callable:
methods = [
(method_name, getattr(cls, method_name))
for method_name in dir(cls)
if callable(getattr(cls, method_name))
]
return sorted([
method_callable
for (method_name, method_callable) in methods
if method_name.startswith("serialize_")
], key=lambda x: x.__name__)
@@ -0,0 +1,88 @@
from typing import TYPE_CHECKING
import arrow
from ics.grammar.parse import Container, ContentLine
from ics.serializers.serializer import Serializer
from ics.utils import (arrow_to_iso, escape_string, timedelta_to_duration,
uid_gen)
if TYPE_CHECKING:
from ics.todo import Todo
class TodoSerializer(Serializer):
def serialize_dtstamp(todo: "Todo", container: Container):
if todo.dtstamp:
instant = todo.dtstamp
else:
instant = arrow.now()
container.append(ContentLine("DTSTAMP", value=arrow_to_iso(instant)))
def serialize_uid(todo: "Todo", container: Container):
if todo.uid:
uid = todo.uid
else:
uid = uid_gen()
container.append(ContentLine("UID", value=uid))
def serialize_completed(todo: "Todo", container: Container):
if todo.completed:
container.append(
ContentLine("COMPLETED", value=arrow_to_iso(todo.completed))
)
def serialize_created(todo: "Todo", container: Container):
if todo.created:
container.append(ContentLine("CREATED", value=arrow_to_iso(todo.created)))
def serialize_description(todo: "Todo", container: Container):
if todo.description:
container.append(
ContentLine("DESCRIPTION", value=escape_string(todo.description))
)
def serialize_start(todo: "Todo", container: Container):
if todo.begin:
container.append(ContentLine("DTSTART", value=arrow_to_iso(todo.begin)))
def serialize_location(todo: "Todo", container: Container):
if todo.location:
container.append(
ContentLine("LOCATION", value=escape_string(todo.location))
)
def serialize_percent(todo: "Todo", container: Container):
if todo.percent is not None:
container.append(ContentLine("PERCENT-COMPLETE", value=str(todo.percent)))
def serialize_priority(todo: "Todo", container: Container):
if todo.priority is not None:
container.append(ContentLine("PRIORITY", value=str(todo.priority)))
def serialize_summary(todo: "Todo", container: Container):
if todo.name:
container.append(ContentLine("SUMMARY", value=escape_string(todo.name)))
def serialize_url(todo: "Todo", container: Container):
if todo.url:
container.append(ContentLine("URL", value=escape_string(todo.url)))
def serialize_due(todo: "Todo", container: Container):
if todo._due_time:
container.append(ContentLine("DUE", value=arrow_to_iso(todo._due_time)))
def serialize_duration(todo: "Todo", container: Container):
if todo._duration:
representation = timedelta_to_duration(todo._duration)
container.append(ContentLine("DURATION", value=representation))
def serialize_alarm(todo: "Todo", container: Container):
for alarm in todo.alarms:
container.append(alarm.serialize())
def serialize_status(todo: "Todo", container: Container):
if todo.status:
container.append(ContentLine("STATUS", value=todo.status))
@@ -0,0 +1,110 @@
import heapq
from typing import Iterator
import arrow
from arrow.arrow import Arrow
from .event import Event
class Timeline(object):
def __init__(self, calendar) -> None:
"""Instanciates a new Timeline.
(You should not have to instanciate a new timeline by yourself)
Args:
calendar : :class:`ics.icalendar.Calendar`
"""
self._calendar = calendar
def __iter__(self) -> Iterator[Event]:
"""Iterates on every event from the :class:`ics.icalendar.Calendar` in chronological order
Note :
- chronological order is defined by the comparaison operators in :class:`ics.event.Event`
- Events with no `begin` will not appear here. (To list all events in a `Calendar` use `Calendar.events`)
"""
# Using a heap is faster than sorting if the number of events (n) is
# much bigger than the number of events we extract from the iterator (k).
# Complexity: O(n + k log n).
heap = [x for x in self._calendar.events if x.begin is not None]
heapq.heapify(heap)
while heap:
yield heapq.heappop(heap)
def included(self, start: Arrow, stop: Arrow) -> Iterator[Event]:
"""Iterates (in chronological order) over every event that is included
in the timespan between `start` and `stop`
Args:
start : (Arrow object)
stop : (Arrow object)
"""
for event in self:
if (start <= event.begin <= stop # if start is between the bonds
and start <= event.end <= stop): # and stop is between the bonds
yield event
def overlapping(self, start: Arrow, stop: Arrow) -> Iterator[Event]:
"""Iterates (in chronological order) over every event that has an intersection
with the timespan between `start` and `stop`
Args:
start : (Arrow object)
stop : (Arrow object)
"""
for event in self:
if ((start <= event.begin <= stop # if start is between the bonds
or start <= event.end <= stop) # or stop is between the bonds
or event.begin <= start and event.end >= stop): # or event is a superset of [start,stop]
yield event
def start_after(self, instant: Arrow) -> Iterator[Event]:
"""Iterates (in chronological order) on every event from the :class:`ics.icalendar.Calendar` in chronological order.
The first event of the iteration has a starting date greater (later) than `instant`
Args:
instant : (Arrow object) starting point of the iteration
"""
for event in self:
if event.begin > instant:
yield event
def at(self, instant: Arrow) -> Iterator[Event]:
"""Iterates (in chronological order) over all events that are occuring during `instant`.
Args:
instant (Arrow object)
"""
for event in self:
if event.begin <= instant <= event.end:
yield event
def on(self, day: Arrow, strict: bool = False) -> Iterator[Event]:
"""Iterates (in chronological order) over all events that occurs on `day`
Args:
day (Arrow object)
strict (bool): if True events will be returned only if they are\
strictly *included* in `day`.
"""
day_start, day_stop = day.floor('day').span('day')
if strict:
return self.included(day_start, day_stop)
else:
return self.overlapping(day_start, day_stop)
def today(self, strict: bool = False) -> Iterator[Event]:
"""Iterates (in chronological order) over all events that occurs today
Args:
strict (bool): if True events will be returned only if they are\
strictly *included* in `day`.
"""
return self.on(arrow.now(), strict=strict)
def now(self) -> Iterator[Event]:
"""Iterates (in chronological order) over all events that occurs now
"""
return self.at(arrow.now())
@@ -0,0 +1,336 @@
import copy
from datetime import datetime, timedelta
from typing import Iterable, List, Optional
import arrow
from six.moves import map
from .alarm.base import BaseAlarm
from .component import Component
from ics.grammar.parse import Container
from .utils import get_arrow, uid_gen
from ics.parsers.todo_parser import TodoParser
from ics.serializers.todo_serializer import TodoSerializer
class Todo(Component):
"""A todo list entry.
Can have a start time and duration, or start and due time,
or only start/due time.
"""
class Meta:
name = "VTODO"
parser = TodoParser
serializer = TodoSerializer
def __init__(self,
dtstamp=None,
uid: str = None,
completed=None,
created=None,
description: str = None,
begin=None,
location: str = None,
percent: int = None,
priority: int = None,
name: str = None,
url: str = None,
due=None,
duration: timedelta = None,
alarms: Iterable[BaseAlarm] = None,
status: str = None):
"""Instantiates a new :class:`ics.todo.Todo`.
Args:
uid (string): must be unique
dtstamp (Arrow-compatible)
completed (Arrow-compatible)
created (Arrow-compatible)
description (string)
begin (Arrow-compatible)
location (string)
percent (int): 0-100
priority (int): 0-9
name (string) : rfc5545 SUMMARY property
url (string)
due (Arrow-compatible)
duration (datetime.timedelta)
alarms (:class:`ics.alarm.Alarm`)
status (string)
Raises:
ValueError: if `duration` and `due` are specified at the same time
"""
self._percent: Optional[int] = None
self._priority: Optional[int] = None
self._begin = None
self._due_time = None
self._duration = None
self.uid = uid_gen() if not uid else uid
self.dtstamp = arrow.now() if not dtstamp else get_arrow(dtstamp)
self.completed = get_arrow(completed)
self.created = get_arrow(created)
self.description = description
self.begin = begin
self.location = location
self.percent = percent
self.priority = priority
self.name = name
self.url = url
self.alarms: List[BaseAlarm] = list()
self.extra = Container(name='VTODO')
if duration and due:
raise ValueError(
'Todo() may not specify a duration and due date\
at the same time')
elif duration:
if not begin:
raise ValueError(
'Todo() must specify a begin if a duration\
is specified')
self.duration = duration
elif due:
self.due = due
if alarms is not None:
self.alarms = list(alarms)
self.status = status
@property
def percent(self) -> Optional[int]:
return self._percent
@percent.setter
def percent(self, value: Optional[int]):
if value:
value = int(value)
if value < 0 or value > 100:
raise ValueError('percent must be [0, 100]')
self._percent = value
@property
def priority(self) -> Optional[int]:
return self._priority
@priority.setter
def priority(self, value: Optional[int]):
if value:
value = int(value)
if value < 0 or value > 9:
raise ValueError('priority must be [0, 9]')
self._priority = value
@property
def begin(self):
"""Get or set the beginning of the todo.
| Will return an :class:`Arrow` object.
| May be set to anything that :func:`Arrow.get` understands.
| If a due time is defined (not a duration), .begin must not
be set to a superior value.
"""
return self._begin
@begin.setter
def begin(self, value):
value = get_arrow(value)
if value and self._due_time and value > self._due_time:
raise ValueError('Begin must be before due time')
self._begin = value
@property
def due(self):
"""Get or set the end of the todo.
| Will return an :class:`Arrow` object.
| May be set to anything that :func:`Arrow.get` understands.
| If set to a non null value, removes any already
existing duration.
| Setting to None will have unexpected behavior if
begin is not None.
| Must not be set to an inferior value than self.begin.
"""
if self._duration:
# if due is duration defined return the beginning + duration
return self.begin + self._duration
elif self._due_time:
# if due is time defined
return self._due_time
else:
return None
@due.setter
def due(self, value):
value = get_arrow(value)
if value and self._begin and value < self._begin:
raise ValueError('Due must be after begin')
self._due_time = value
if value:
self._duration = None
@property
def duration(self):
"""Get or set the duration of the todo.
| Will return a timedelta object.
| May be set to anything that timedelta() understands.
| May be set with a dict ({"days":2, "hours":6}).
| If set to a non null value, removes any already
existing end time.
"""
if self._duration:
return self._duration
elif self.due:
return self.due - self.begin
else:
# todo has neither due, nor start and duration
return None
@duration.setter
def duration(self, value):
if isinstance(value, dict):
value = timedelta(**value)
elif isinstance(value, timedelta):
value = value
elif value is not None:
value = timedelta(value)
self._duration = value
if value:
self._due_time = None
@property
def status(self) -> Optional[str]:
return self._status
@status.setter
def status(self, value: Optional[str]):
if isinstance(value, str):
value = value.upper()
statuses = (None, 'NEEDS-ACTION', 'COMPLETED', 'IN-PROCESS', 'CANCELLED')
if value not in statuses:
raise ValueError('status must be one of %s' % ", ".join([repr(x) for x in statuses]))
self._status = value
def __repr__(self) -> str:
if self.name is None:
return "<Todo>"
if self.begin is None and self.due is None:
return "<Todo '{}'>".format(self.name)
if self.due is None:
return "<Todo '{}' begin:{}>".format(self.name, self.begin)
if self.begin is None:
return "<Todo '{}' due:{}>".format(self.name, self.due)
return "<Todo '{}' begin:{} due:{}>".format(self.name, self.begin, self.due)
def __lt__(self, other) -> bool:
if isinstance(other, Todo):
if self.due is None and other.due is None:
if self.name is None and other.name is None:
return False
elif self.name is None:
return True
elif other.name is None:
return False
else:
return self.name < other.name
return self.due < other.due
if isinstance(other, datetime):
if self.due:
return self.due < other
raise NotImplementedError(
'Cannot compare Todo and {}'.format(type(other)))
def __le__(self, other) -> bool:
if isinstance(other, Todo):
if self.due is None and other.due is None:
if self.name is None and other.name is None:
return True
elif self.name is None:
return True
elif other.name is None:
return False
else:
return self.name <= other.name
return self.due <= other.due
if isinstance(other, datetime):
if self.due:
return self.due <= other
raise NotImplementedError(
'Cannot compare Todo and {}'.format(type(other)))
def __gt__(self, other) -> bool:
if isinstance(other, Todo):
if self.due is None and other.due is None:
if self.name is None and other.name is None:
return False
elif self.name is None:
return False
elif other.name is None:
return True
else:
return self.name > other.name
return self.due > other.due
if isinstance(other, datetime):
if self.due:
return self.due > other
raise NotImplementedError(
'Cannot compare Todo and {}'.format(type(other)))
def __ge__(self, other) -> bool:
if isinstance(other, Todo):
if self.due is None and other.due is None:
if self.name is None and other.name is None:
return True
elif self.name is None:
return True
elif other.name is None:
return False
else:
return self.name >= other.name
return self.due >= other.due
if isinstance(other, datetime):
if self.due:
return self.due >= other
raise NotImplementedError(
'Cannot compare Todo and {}'.format(type(other)))
def __eq__(self, other) -> bool:
"""Two todos are considered equal if they have the same uid."""
if isinstance(other, Todo):
return self.uid == other.uid
raise NotImplementedError(
'Cannot compare Todo and {}'.format(type(other)))
def __ne__(self, other) -> bool:
"""Two todos are considered not equal if they do not have the same uid."""
if isinstance(other, Todo):
return self.uid != other.uid
raise NotImplementedError(
'Cannot compare Todo and {}'.format(type(other)))
def clone(self):
"""
Returns:
Todo: an exact copy of self"""
clone = copy.copy(self)
clone.extra = clone.extra.clone()
clone.alarms = copy.copy(self.alarms)
return clone
def __hash__(self):
"""
Returns:
int: hash of self. Based on self.uid."""
return int(''.join(map(lambda x: '%.3d' % ord(x), self.uid)))
@@ -0,0 +1,26 @@
import re
def striphtml(data):
p = re.compile(r'<.*?>')
return p.sub('', data)
def validate(string):
import requests
payload = {'snip': string}
ret = requests.post('http://severinghaus.org/projects/icv/', data=payload)
if 'Sorry, your calendar could not be parsed.' in ret.text:
i = ret.text.index('<div class="parse-error">')
j = ret.text[i:].index('</div>')
usefull = ret.text[i:i + j]
usefull_clean = striphtml(usefull)
lines = usefull_clean.split('\n')
lines_clean = map(lambda x: x.strip(), lines)
lines_no_empty = filter(lambda x: x != '', lines_clean)
return '\n'.join(lines_no_empty)
elif 'Congratulations; your calendar validated!' in ret.text:
return True
else:
return None
@@ -0,0 +1,6 @@
from datetime import datetime
from typing import Union
from arrow import Arrow
ArrowLike = Union[datetime, Arrow]
@@ -0,0 +1,235 @@
import re
from datetime import timedelta
from typing import Dict, List, Optional, Tuple, Union
from uuid import uuid4
import arrow
from arrow.arrow import Arrow
from dateutil.tz import gettz
from ics.grammar import parse
from ics.grammar.parse import Container, ContentLine
tzutc = arrow.utcnow().tzinfo
def remove_x(container: Container) -> None:
for i in reversed(range(len(container))):
item = container[i]
if item.name.startswith('X-'):
del container[i]
def remove_sequence(container: Container) -> None:
for i in reversed(range(len(container))):
item = container[i]
if item.name == 'SEQUENCE':
del container[i]
DATE_FORMATS: Dict[int, str] = dict((len(k), k) for k in (
'YYYYMM',
'YYYYMMDD',
'YYYYMMDDTHH',
'YYYYMMDDTHHmm',
'YYYYMMDDTHHmmss'))
def arrow_get(string: str) -> Arrow:
'''this function exists because ICS uses ISO 8601 without dashes or
colons, i.e. not ISO 8601 at all.'''
# replace slashes with dashes
if '/' in string:
string = string.replace('/', '-')
# if string contains dashes, assume it to be proper ISO 8601
if '-' in string:
return arrow.get(string)
string = string.rstrip('Z')
return arrow.get(string, DATE_FORMATS[len(string)])
def iso_to_arrow(time_container: Optional[ContentLine], available_tz={}) -> Arrow:
if time_container is None:
return None
# TODO : raise if not iso date
tz_list = time_container.params.get('TZID')
# TODO : raise if len(tz_list) > 1 or if tz is not a valid tz
# TODO : see if timezone is registered as a VTIMEZONE
tz: Optional[str]
if tz_list and len(tz_list) > 0:
tz = tz_list[0]
else:
tz = None
if ('T' not in time_container.value) and \
'DATE' in time_container.params.get('VALUE', []):
val = time_container.value + 'T0000'
else:
val = time_container.value
if tz and not (val[-1].upper() == 'Z'):
naive = arrow_get(val).naive
selected_tz = gettz(tz)
if not selected_tz:
selected_tz = available_tz.get(tz, 'UTC')
return arrow.get(naive, selected_tz)
else:
return arrow_get(val)
# TODO : support floating (ie not bound to any time zone) times (cf
# http://www.kanzaki.com/docs/ical/dateTime.html)
def iso_precision(string: str) -> str:
has_time = 'T' in string
if has_time:
date_string, time_string = string.split('T', 1)
time_parts = re.split('[+-]', time_string, 1)
has_seconds = time_parts[0].count(':') > 1
has_seconds = not has_seconds and len(time_parts[0]) == 6
if has_seconds:
return 'second'
else:
return 'minute'
else:
return 'day'
def get_lines(container: Container, name: str, keep: bool = False) -> List[ContentLine]:
lines = []
for i in reversed(range(len(container))):
item = container[i]
if item.name == name:
lines.append(item)
if not keep:
del container[i]
return lines
def parse_duration(line: str) -> timedelta:
"""
Return a timedelta object from a string in the DURATION property format
"""
DAYS = {'D': 1, 'W': 7}
SECS = {'S': 1, 'M': 60, 'H': 3600}
sign, i = 1, 0
if line[i] in '-+':
if line[i] == '-':
sign = -1
i += 1
if line[i] != 'P':
raise parse.ParseError("Error while parsing %s" % line)
i += 1
days, secs = 0, 0
while i < len(line):
if line[i] == 'T':
i += 1
if i == len(line):
break
j = i
while line[j].isdigit():
j += 1
if i == j:
raise parse.ParseError("Error while parsing %s" % line)
val = int(line[i:j])
if line[j] in DAYS:
days += val * DAYS[line[j]]
DAYS.pop(line[j])
elif line[j] in SECS:
secs += val * SECS[line[j]]
SECS.pop(line[j])
else:
raise parse.ParseError("Error while parsing %s" % line)
i = j + 1
return timedelta(sign * days, sign * secs)
def timedelta_to_duration(dt: timedelta) -> str:
"""
Return a string according to the DURATION property format
from a timedelta object
"""
ONE_DAY_IN_SECS = 3600 * 24
total = abs(int(dt.total_seconds()))
days = total // ONE_DAY_IN_SECS
seconds = total % ONE_DAY_IN_SECS
res = 'P'
if days // 7:
res += str(days // 7) + 'W'
days %= 7
if days:
res += str(days) + 'D'
if seconds:
res += 'T'
if seconds // 3600:
res += str(seconds // 3600) + 'H'
seconds %= 3600
if seconds // 60:
res += str(seconds // 60) + 'M'
seconds %= 60
if seconds:
res += str(seconds) + 'S'
if dt.total_seconds() >= 0:
return res
else:
return "-%s" % res
def get_arrow(value: Union[None, Arrow, Tuple, Dict]) -> Arrow:
if value is None:
return None
elif isinstance(value, Arrow):
return value
elif isinstance(value, tuple):
return arrow.get(*value)
elif isinstance(value, dict):
return arrow.get(**value)
else:
return arrow.get(value)
def arrow_to_iso(instant: Arrow) -> str:
# set to utc, make iso, remove timezone
instant = arrow.get(instant.astimezone(tzutc)).format('YYYYMMDDTHHmmss')
return instant + 'Z'
def arrow_date_to_iso(instant: Arrow) -> str:
# date-only for all day events
# set to utc, make iso, remove timezone
instant = arrow.get(instant.astimezone(tzutc)).format('YYYYMMDD')
return instant # no TZ for all days
def uid_gen() -> str:
uid = str(uuid4())
return "{}@{}.org".format(uid, uid[:4])
def escape_string(string: str) -> str:
string = string.replace("\\", "\\\\")
string = string.replace(";", "\\;")
string = string.replace(",", "\\,")
string = string.replace("\n", "\\n")
string = string.replace("\r", "\\r")
return string
def unescape_string(string: str) -> str:
string = string.replace("\\;", ";")
string = string.replace("\\,", ",")
string = string.replace("\\n", "\n")
string = string.replace("\\N", "\n")
string = string.replace("\\r", "\r")
string = string.replace("\\R", "\r")
string = string.replace("\\\\", "\\")
return string