Repository URL to install this package:
|
Version:
1.6.0 ▾
|
django-recurrence
/
base.py
|
|---|
"""
Wrapper around the ``dateutil.rrule`` module.
Provides more consistent behavior with the rfc2445 specification,
notably differing from `dateutil.rrule`` in the handling of the
`dtstart` parameter and the additional handling of a `dtend`
parameter. Also, the `byweekday` parameter in `dateutil.rrule` is
`byday` in this package to reflect the specification. See the `Rule`
and `Recurrence` class documentation for details on the differences.
"""
import re
import datetime
import calendar
import pytz
import dateutil.rrule
from django.utils import dateformat, timezone
from django.utils.encoding import python_2_unicode_compatible
from django.utils.translation import ugettext as _, pgettext as _p
from django.utils.six import string_types
from recurrence import exceptions
YEARLY, MONTHLY, WEEKLY, DAILY, HOURLY, MINUTELY, SECONDLY = range(7)
(JANUARY, FEBRUARY, MARCH, APRIL, MAY, JUNE, JULY, AUGUST,
SEPTEMBER, OCTOBER, NOVEMBER, DECEMBER) = range(1, 13)
def localtz():
return timezone.get_current_timezone()
class Rule(object):
"""
A recurrence rule.
`Rule` is a representation of a rfc2445 `RECUR` type, used in
the `RRULE` and `EXRULE` properties. More information about the
`RECUR` type specification can be found in the rfc at
http://www.ietf.org/rfc/rfc2445.txt.
An `Rrule` wraps the `dateutil.rrule.rrule` class while adhering
to the rfc2445 spec. Notably a `dtstart` parameter cannot be
specified with a `Rule` unlike `dateutil.rrule.rrule` as only one
`dtstart` can be used with a set of `RRULE` and `EXRULE` rfc2445
properties, therefore the `Recurrence` class (which is based on
`dateutil.rrule.rruleset`) accepts a `dtstart` parameter instead.
`Recurrence` also accepts a `dtend` parameter.
Documentation is largely sourced from the `dateutil.rrule.rrule`
documentation at http://labix.org/python-dateutil
:Variables:
`freq` : int
One of the enumerated constants `YEARLY`, `MONTHLY`,
`WEEKLY`, `DAILY`, `HOURLY`, `MINUTELY`, or `SECONDLY`,
specifying the base recurring frequency.
`interval` : int
The interval between each freq iteration. For example,
when using YEARLY, an interval of 2 means once every two
years, but with HOURLY, it means once every two hours. The
default interval is 1.
`wkst` : int
The week start day. Must be one of the `MO`, `TU`, `WE`,
`TH`, `FR`, `SA`, `SU` constants, or an integer,
specifying the first day of the week. This will affect
recurrences based on weekly periods. The default week
start is got from `calendar.firstweekday()`, and may be
modified by `calendar.setfirstweekday()`.
`count` : int
How many occurrences will be generated by this rule.
`until` : datetime.datetime
If given, this must be a `datetime.datetime` instance,
that will specify the limit of the recurrence. If a
recurrence instance happens to be the same as the
`datetime.datetime` instance given in the `until` keyword,
this will be the last occurrence.
`bysetpos` : int or sequence
If given, it must be either an integer, or a sequence of
integers, positive or negative. Each given integer will
specify an occurrence number, corresponding to the nth
occurrence of the rule inside the frequency period. For
example, a `bysetpos` of `-1` if combined with a `MONTHLY`
frequency, and a `byday` of `(MO, TU, WE, TH, FR)`, will
result in the last work day of every month.
`bymonth` : int or sequence
If given, it must be either an integer, or a sequence of
integers, meaning the months to apply the recurrence to.
`bymonthday` : int or sequence
If given, it must be either an integer, or a sequence of
integers, meaning the month days to apply the recurrence
to.
`byyearday` : int or sequence
If given, it must be either an integer, or a sequence of
integers, meaning the year days to apply the recurrence
to.
`byweekno` : int or sequence
If given, it must be either an integer, or a sequence of
integers, meaning the week numbers to apply the recurrence
to. Week numbers have the meaning described in ISO8601,
that is, the first week of the year is that containing at
least four days of the new year.
`byday` : int or sequence
If given, it must be either an integer `(0 == MO)`, a
sequence of integers, one of the weekday constants `(MO,
TU, ...)`, or a sequence of these constants. When given,
these variables will define the weekdays where the
recurrence will be applied. It's also possible to use an
argument n for the weekday instances, which will mean the
nth occurrence of this weekday in the period. For example,
with `MONTHLY`, or with `YEARLY` and `BYMONTH`, using
`FR(1)` in byweekday will specify the first friday of the
month where the recurrence happens.
`byhour` : int or sequence
If given, it must be either an integer, or a sequence of
integers, meaning the hours to apply the recurrence to.
`byminute` : int or sequence
If given, it must be either an integer, or a sequence of
integers, meaning the minutes to apply the recurrence to.
`bysecond` : int or sequence
If given, it must be either an integer, or a sequence of
integers, meaning the seconds to apply the recurrence to.
"""
byparams = (
'bysetpos', 'bymonth', 'bymonthday', 'byyearday',
'byweekno', 'byday', 'byhour', 'byminute', 'bysecond'
)
frequencies = (
'YEARLY', 'MONTHLY', 'WEEKLY', 'DAILY',
'HOURLY', 'MINUTELY', 'SECONDLY'
)
weekdays = (
'MO', 'TU', 'WE', 'TH', 'FR', 'SA', 'SU'
)
firstweekday = calendar.firstweekday()
def __init__(
self, freq,
interval=1, wkst=None, count=None, until=None, **kwargs
):
"""
Create a new rule.
See `Rule` class documentation for available `**kwargs` and
parameter usage.
"""
self.freq = freq
self.interval = interval
self.wkst = wkst
self.count = count
self.until = until
for param in self.byparams:
if param in kwargs:
value = kwargs[param]
if hasattr(value, '__iter__'):
value = list(value)
if not value:
value = []
elif value is not None:
value = [value]
else:
value = []
setattr(self, param, value)
else:
setattr(self, param, [])
def __hash__(self):
byparam_values = []
for param in self.byparams:
byparam_values.append(param)
byparam_values.extend(getattr(self, param, []) or [])
return hash((
self.freq, self.interval, self.wkst, self.count, self.until,
tuple(byparam_values)))
def __eq__(self, other):
if not isinstance(other, Rule):
raise TypeError('object to compare must be Rule object')
return hash(self) == hash(other)
def __ne__(self, other):
return not self.__eq__(other)
def to_text(self, short=False):
return rule_to_text(self, short)
def to_dateutil_rrule(self, dtstart=None, dtend=None, cache=False):
"""
Create a `dateutil.rrule.rrule` instance from this `Rule`.
:Parameters:
`dtstart` : datetime.datetime
The date/time the recurrence rule starts.
`dtend` : datetime.datetime
The rule should not yield occurrences past this
date. Replaces `until` if `until` is greater than
`dtend`. Note: `dtend` in this case does not count for
an occurrence itself.
`cache` : bool
If given, it must be a boolean value specifying to
enable or disable caching of results. If you will use
the same `dateutil.rrule.rrule` instance multiple
times, enabling caching will improve the performance
considerably.
:Returns:
A `dateutil.rrule.rrule` instance.
"""
kwargs = dict((p, getattr(self, p) or None) for p in self.byparams)
# dateutil.rrule renames the parameter 'byweekday' by we're using
# the parameter name originally specified by rfc2445.
kwargs['byweekday'] = kwargs.pop('byday')
until = self.until
if until:
until = normalize_offset_awareness(until, dtstart)
if dtend:
if until > dtend:
until = dtend
elif dtend:
until = dtend
return dateutil.rrule.rrule(
self.freq, dtstart, self.interval, self.wkst, self.count, until,
cache=cache, **kwargs)
@python_2_unicode_compatible
class Recurrence(object):
"""
A combination of `Rule` and `datetime.datetime` instances.
A `Recurrence` instance provides the combined behavior of the
rfc2445 `DTSTART`, `DTEND`, `RRULE`, `EXRULE`, `RDATE`, and
`EXDATE` properties in generating recurring date/times.
This is a wrapper around the `dateutil.rrule.rruleset` class while
adhering to the rfc2445 spec. Notably a `dtstart` parameter can be
given which cascades to all `dateutil.rrule.rrule` instances
generated by included `Rule` instances. A `dtend` parameter has
also been included to reflect the `DTEND` rfc2445 parameter.
:Variables:
`dtstart` : datetime.datetime
Optionally specify the first occurrence. This defaults to
`datetime.datetime.now()` when the occurrence set is
generated.
`dtend` : datetime.datetime
Optionally specify the last occurrence.
`rrules` : list
A list of `Rule` instances to include in the recurrence
set generation.
`exrules` : list
A list of `Rule` instances to include in the recurrence
set exclusion list. Dates which are part of the given
recurrence rules will not be generated, even if some
inclusive `Rule` or `datetime.datetime` instances matches
them.
`rdates` : list
A list of `datetime.datetime` instances to include in the
occurrence set generation.
`exdates` : list
A list of `datetime.datetime` instances to exclude in the
occurrence set generation. Dates included that way will
not be generated, even if some inclusive `Rule` or
`datetime.datetime` instances matches them.
`include_dtstart` : bool
Defines if `dtstart` is included in the recurrence set as
the first occurrence. With `include_dtstart == True` it is
both the starting point for recurrences and the first
recurrence in the set (according to the rfc2445 spec).
With `include_dtstart == False` `dtstart` is only the rule's
starting point like in python's `dateutil.rrule`.
"""
def __init__(
self, dtstart=None, dtend=None, rrules=(), exrules=(),
rdates=(), exdates=(), include_dtstart=True
):
"""
Create a new recurrence.
Parameters map directly to instance attributes, see
`Recurrence` class documentation for usage.
"""
self._cache = {}
self.dtstart = dtstart
self.dtend = dtend
self.rrules = list(rrules)
self.exrules = list(exrules)
self.rdates = list(rdates)
self.exdates = list(exdates)
self.include_dtstart = include_dtstart
def __iter__(self):
return self.occurrences()
def __str__(self):
return serialize(self)
def __hash__(self):
return hash((
self.dtstart, self.dtend,
tuple(self.rrules), tuple(self.exrules),
tuple(self.rdates), tuple(self.exdates)))
def __bool__(self):
if (self.dtstart or self.dtend or
tuple(self.rrules) or tuple(self.exrules) or
tuple(self.rdates) or tuple(self.exdates)):
return True
else:
return False
def __nonzero__(self):
# Required for Python 2 compatibility
return type(self).__bool__(self)
def __eq__(self, other):
if type(other) != type(self):
return False
if not isinstance(other, Recurrence):
raise TypeError('object to compare must be Recurrence object')
return hash(self) == hash(other)
def __ne__(self, other):
return not self.__eq__(other)
def occurrences(
self, dtstart=None, dtend=None, cache=False
):
"""
Get a generator yielding `datetime.datetime` instances in this
occurrence set.
:Parameters:
`dtstart` : datetime.datetime
Optionally specify the first occurrence of the
occurrence set. Defaults to `self.dtstart` if specified
or `datetime.datetime.now()` if not when the
occurrence set is generated.
`dtend` : datetime.datetime
Optionally specify the last occurrence of the
occurrence set. Defaults to `self.dtend` if specified.
`cache` : bool
Whether to cache the occurrence set generator.
:Returns:
A sequence of `datetime.datetime` instances.
"""
return self.to_dateutil_rruleset(dtstart, dtend, cache)
def count(self, dtstart=None, dtend=None, cache=False):
"""
Returns the number of occurrences in this occurrence set.
:Parameters:
`dtstart` : datetime.datetime
Optionally specify the first occurrence of the
occurrence set. Defaults to `self.dtstart` if specified
or `datetime.datetime.now()` if not when the
occurrence set is generated.
`dtend` : datetime.datetime
Optionally specify the last occurrence of the
occurrence set. Defaults to `self.dtend` if specified.
`cache` : bool
Whether to cache the occurrence set generator.
:Returns:
The number of occurrences in this occurrence set.
"""
return self.to_dateutil_rruleset(dtstart, dtend, cache).count()
def before(
self, dt, inc=False,
dtstart=None, dtend=None, cache=False
):
"""
Returns the last recurrence before the given
`datetime.datetime` instance.
:Parameters:
`dt` : datetime.datetime
The date to use as the threshold.
`inc` : bool
Defines what happens if `dt` is an occurrence. With
`inc == True`, if `dt` itself is an occurrence, it
will be returned.
`dtstart` : datetime.datetime
Optionally specify the first occurrence of the
occurrence set. Defaults to `self.dtstart` if specified
or `datetime.datetime.now()` if not when the
occurrence set is generated.
`dtend` : datetime.datetime
Optionally specify the last occurrence of the
occurrence set. Defaults to `self.dtend` if specified.
`cache` : bool
Whether to cache the occurrence set generator.
:Returns:
A `datetime.datetime` instance.
"""
return self.to_dateutil_rruleset(
dtstart, dtend, cache).before(dt, inc)
def after(
self, dt, inc=False,
dtstart=None, dtend=None, cache=False
):
"""
Returns the first recurrence after the given
`datetime.datetime` instance.
:Parameters:
`dt` : datetime.datetime
The date to use as the threshold.
`inc` : bool
Defines what happens if `dt` is an occurrence. With
`inc == True`, if `dt` itself is an occurrence, it
will be returned.
`dtstart` : datetime.datetime
Optionally specify the first occurrence of the
occurrence set. Defaults to `self.dtstart` if specified
or `datetime.datetime.now()` if not when the
occurrence set is generated.
`dtend` : datetime.datetime
Optionally specify the last occurrence of the
occurrence set. Defaults to `self.dtend` if specified.
`cache` : bool
Whether to cache the occurrence set generator.
:Returns:
A `datetime.datetime` instance.
"""
return self.to_dateutil_rruleset(dtstart, dtend, cache).after(dt, inc)
def between(
self, after, before,
inc=False, dtstart=None, dtend=None, cache=False
):
"""
Returns the first recurrence after the given
`datetime.datetime` instance.
:Parameters:
`after` : datetime.datetime
Return dates after this date.
`before` : datetime.datetime
Return dates before this date.
`inc` : bool
Defines what happens if `after` and/or `before` are
themselves occurrences. With `inc == True`, they will
be included in the list, if they are found in the
occurrence set.
`dtstart` : datetime.datetime
Optionally specify the first occurrence of the
occurrence set. Defaults to `self.dtstart` if specified
or `datetime.datetime.now()` if not when the
occurrence set is generated.
`dtend` : datetime.datetime
Optionally specify the last occurrence of the
occurrence set. Defaults to `self.dtend` if specified.
`cache` : bool
Whether to cache the occurrence set generator.
:Returns:
A sequence of `datetime.datetime` instances.
"""
return self.to_dateutil_rruleset(
dtstart, dtend, cache).between(after, before, inc)
def to_dateutil_rruleset(self, dtstart=None, dtend=None, cache=False):
"""
Create a `dateutil.rrule.rruleset` instance from this
`Recurrence`.
:Parameters:
`dtstart` : datetime.datetime
The date/time the recurrence rule starts. This value
overrides the `dtstart` property specified by the
`Recurrence` instance if its set.
`dtstart` : datetime.datetime
Optionally specify the first occurrence of the
occurrence set. Defaults to `self.dtstart` if specified
or `datetime.datetime.now()` if not when the
occurrence set is generated.
`cache` : bool
If given, it must be a boolean value specifying to
enable or disable caching of results. If you will use
the same `dateutil.rrule.rrule` instance multiple
times, enabling caching will improve the performance
considerably.
:Returns:
A `dateutil.rrule.rruleset` instance.
"""
# all datetimes used in dateutil.rrule objects will need to be
# normalized to either offset-aware or offset-naive datetimes
# to avoid exceptions. dateutil will use the tzinfo from the
# given dtstart, which will cascade to other datetime objects.
dtstart = dtstart or self.dtstart
dtend = dtend or self.dtend
include_dtstart = self.include_dtstart
if dtend:
dtend = normalize_offset_awareness(dtend or self.dtend, dtstart)
if cache:
# we need to cache an instance for each unique dtstart
# value because the occurrence values will differ.
cached = self._cache.get(dtstart)
if cached:
return cached
rruleset = dateutil.rrule.rruleset(cache=cache)
for rrule in self.rrules:
rruleset.rrule(rrule.to_dateutil_rrule(dtstart, dtend, cache))
for exrule in self.exrules:
rruleset.exrule(exrule.to_dateutil_rrule(dtstart, dtend, cache))
if include_dtstart and dtstart is not None:
rruleset.rdate(dtstart)
for rdate in self.rdates:
rdate = normalize_offset_awareness(rdate, dtstart)
if dtend is not None and rdate < dtend:
rruleset.rdate(rdate)
elif not dtend:
rruleset.rdate(rdate)
if dtend is not None:
rruleset.rdate(dtend)
for exdate in self.exdates:
exdate = normalize_offset_awareness(exdate, dtstart)
if dtend is not None and exdate < dtend:
rruleset.exdate(exdate)
elif not dtend:
rruleset.exdate(exdate)
if cache:
self._cache[dtstart] = rruleset
return rruleset
class Weekday(object):
"""
Representation of a weekday.
A `Weekday` is essentially an integer from 0 to 6, with an
optional `index` which indicates its position in a month. For
example, an `number` of 6 and an `index` of ``-1`` means the last
sunday of the month. Weekday's with a specific index can be
created by calling the existing `MO`, `TU`, `WE`, `TH`, `FR`,
`SA`, `SU` constants::
>>> SU(-1)
-1SU
`Weekday` objects have a smart equality test that can compare
integers, other `Weekday` objects, and string constants as defined
by rfc2445, such as '-1SU'.
"""
def __init__(self, number, index=None):
"""
Create a new weekday constant.
:Parameters:
`number` : int
A number in `range(7)`.
`index` : int
An integer specifying the weekday's position in the
month. A value of ``None`` or ``0`` means the index is
ambiguous and represents all weekdays of that number.
"""
int(number)
if number > 6:
raise ValueError('number must be in range(7)')
self.number = number
self.index = index
def __call__(self, index):
if index == self.index:
return self
else:
return Weekday(self.number, index)
def __hash__(self):
if self.index:
return hash((self.number, self.index))
else:
return hash(self.number)
def __eq__(self, other):
other = to_weekday(other)
return (self.number, self.index) == (other.number, other.index)
def __repr__(self):
if self.index:
return '%s%s' % (self.index, Rule.weekdays[self.number])
else:
return Rule.weekdays[self.number]
weekday = property(lambda self: self.number)
n = property(lambda self: self.index)
MONDAY, TUESDAY, WEDNESDAY, THURSDAY, FRIDAY, SATURDAY, SUNDAY = (
MO, TU, WE, TH, FR, SA, SU) = WEEKDAYS = list(map(lambda n: Weekday(n), range(7)))
def to_weekday(token):
"""
Attempt to convert an object to a `Weekday` constant.
:Parameters:
`token` : str, int, dateutil.rrule.weekday or `Weekday`
Can be values such as `MO`, `SU(-2)`, `"-2SU"`, or an
integer like `1` for Tuesday. dateutil.rrule.weekday`
are returned unchanged.
:Returns:
A `dateutil.rrule.weekday` instance.
"""
if isinstance(token, Weekday):
return token
if isinstance(token, dateutil.rrule.weekday):
return Weekday(token.weekday, token.n)
if isinstance(token, int):
if token > 6:
raise ValueError
return WEEKDAYS[token]
elif not token:
raise ValueError
elif isinstance(token, string_types) and token.isdigit():
if int(token) > 6:
raise ValueError
return WEEKDAYS[int(token)]
elif isinstance(token, string_types):
const = token[-2:].upper()
if const not in Rule.weekdays:
raise ValueError
nth = token[:-2]
if not nth:
return Weekday(list(Rule.weekdays).index(const))
else:
return Weekday(list(Rule.weekdays).index(const), int(nth))
def validate(rule_or_recurrence):
if isinstance(rule_or_recurrence, Rule):
obj = Recurrence(rrules=[rule_or_recurrence])
else:
obj = rule_or_recurrence
try:
if not isinstance(obj, Rule) and not isinstance(obj, Recurrence):
raise exceptions.ValidationError('incompatible object')
except TypeError:
raise exceptions.ValidationError('incompatible object')
def validate_dt(dt):
if not isinstance(dt, datetime.datetime):
raise exceptions.ValidationError('invalid datetime: %r' % dt)
def validate_iterable(rule, param):
try:
[v for v in getattr(rule, param, []) if v]
except TypeError:
# TODO: I'm not sure it's possible to get here - all the
# places we call validate_iterable convert single ints to
# sequences, and other types raise TypeErrors earlier.
raise exceptions.ValidationError(
'%s parameter must be iterable' % param)
def validate_iterable_ints(rule, param, min_value=None, max_value=None):
for value in getattr(rule, param, []):
try:
value = int(value)
if min_value is not None:
if value < min_value:
raise ValueError
if max_value is not None:
if value > max_value:
raise ValueError
except ValueError:
raise exceptions.ValidationError(
'invalid %s parameter: %r' % (param, value))
def validate_rule(rule):
# validate freq
try:
Rule.frequencies[int(rule.freq)]
except IndexError:
raise exceptions.ValidationError(
'invalid freq parameter: %r' % rule.freq)
except ValueError:
raise exceptions.ValidationError(
'invalid freq parameter: %r' % rule.freq)
# validate interval
try:
interval = int(rule.interval)
if interval < 1:
raise ValueError
except ValueError:
raise exceptions.ValidationError(
'invalid interval parameter: %r' % rule.interval)
# validate wkst
if rule.wkst:
try:
to_weekday(rule.wkst)
except ValueError:
raise exceptions.ValidationError(
'invalide wkst parameter: %r' % rule.wkst)
# validate until
if rule.until:
try:
validate_dt(rule.until)
except ValueError:
# TODO: I'm not sure it's possible to get here
# (validate_dt doesn't raise ValueError)
raise exceptions.ValidationError(
'invalid until parameter: %r' % rule.until)
# validate count
if rule.count:
try:
int(rule.count)
except ValueError:
raise exceptions.ValidationError(
'invalid count parameter: %r' % rule.count)
# TODO: Should we check that you haven't specified both
# rule.count and rule.until? Note that we only serialize
# rule.until if there's no rule.count.
# validate byparams
for param in Rule.byparams:
validate_iterable(rule, param)
if param == 'byday':
for value in getattr(rule, 'byday', []):
try:
to_weekday(value)
except ValueError:
raise exceptions.ValidationError(
'invalid byday parameter: %r' % value)
elif param == 'bymonth':
validate_iterable_ints(rule, param, 1, 12)
elif param == 'bymonthday':
validate_iterable_ints(rule, param, -4, 31)
elif param == 'byhour':
validate_iterable_ints(rule, param, 0, 23)
elif param == 'byminute':
validate_iterable_ints(rule, param, 0, 59)
elif param == 'bysecond':
validate_iterable_ints(rule, param, 0, 59)
else:
validate_iterable_ints(rule, param)
if obj.dtstart:
validate_dt(obj.dtstart)
if obj.dtend:
validate_dt(obj.dtend)
if obj.rrules:
list(map(lambda rule: validate_rule(rule), obj.rrules))
if obj.exrules:
list(map(lambda rule: validate_rule(rule), obj.exrules))
if obj.rdates:
list(map(lambda dt: validate_dt(dt), obj.rdates))
if obj.exdates:
list(map(lambda dt: validate_dt(dt), obj.exdates))
def serialize(rule_or_recurrence):
"""
Serialize a `Rule` or `Recurrence` instance.
`Rule` instances are wrapped as an rrule in a `Recurrence`
instance before serialization, and will serialize as the `RRULE`
property.
All `datetime.datetime` objects will be converted and serialized
as UTC.
:Returns:
A rfc2445 formatted unicode string.
"""
def serialize_dt(dt):
if not dt.tzinfo:
dt = localtz().localize(dt)
dt = dt.astimezone(pytz.utc)
return u'%s%s%sT%s%s%sZ' % (
str(dt.year).rjust(4, '0'),
str(dt.month).rjust(2, '0'),
str(dt.day).rjust(2, '0'),
str(dt.hour).rjust(2, '0'),
str(dt.minute).rjust(2, '0'),
str(dt.second).rjust(2, '0'),
)
def serialize_rule(rule):
values = []
values.append((u'FREQ', [Rule.frequencies[rule.freq]]))
if rule.interval != 1:
values.append((u'INTERVAL', [str(int(rule.interval))]))
if rule.wkst:
values.append((u'WKST', [Rule.weekdays[getattr(rule.wkst, 'number', rule.wkst)]]))
if rule.count is not None:
values.append((u'COUNT', [str(rule.count)]))
elif rule.until is not None:
values.append((u'UNTIL', [serialize_dt(rule.until)]))
if rule.byday:
days = []
for d in rule.byday:
d = to_weekday(d)
# TODO - this if/else copies what Weekday's __repr__
# does - perhaps we should refactor it into a __str__
# method on Weekday?
if d.index:
days.append(u'%s%s' % (d.index, Rule.weekdays[d.number]))
else:
days.append(Rule.weekdays[d.number])
values.append((u'BYDAY', days))
remaining_params = list(Rule.byparams)
remaining_params.remove('byday')
for param in remaining_params:
value_list = getattr(rule, param, None)
if value_list:
values.append((param.upper(), [str(n) for n in value_list]))
return u';'.join(u'%s=%s' % (i[0], u','.join(i[1])) for i in values)
if rule_or_recurrence is None:
return None
try:
validate(rule_or_recurrence)
except exceptions.ValidationError as error:
raise exceptions.SerializationError(error.args[0])
obj = rule_or_recurrence
if isinstance(obj, Rule):
obj = Recurrence(rrules=[obj])
items = []
if obj.dtstart:
if obj.dtstart.tzinfo:
dtstart = serialize_dt(obj.dtstart.astimezone(pytz.utc))
else:
dtstart = serialize_dt(
localtz().localize(obj.dtstart).astimezone(pytz.utc))
items.append((u'DTSTART', dtstart))
if obj.dtend:
if obj.dtend.tzinfo:
dtend = serialize_dt(obj.dtend.astimezone(pytz.utc))
else:
dtend = serialize_dt(
localtz().localize(obj.dtend).astimezone(pytz.utc))
items.append((u'DTEND', dtend))
for rrule in obj.rrules:
items.append((u'RRULE', serialize_rule(rrule)))
for exrule in obj.exrules:
items.append((u'EXRULE', serialize_rule(exrule)))
for rdate in obj.rdates:
if rdate.tzinfo:
rdate = rdate.astimezone(pytz.utc)
else:
rdate = localtz().localize(rdate).astimezone(pytz.utc)
items.append((u'RDATE', serialize_dt(rdate)))
for exdate in obj.exdates:
if exdate.tzinfo:
exdate = exdate.astimezone(pytz.utc)
else:
exdate = localtz().localize(exdate).astimezone(pytz.utc)
items.append((u'EXDATE', serialize_dt(exdate)))
return u'\n'.join(u'%s:%s' % i for i in items)
def deserialize(text, include_dtstart=True):
"""
Deserialize a rfc2445 formatted string.
This is a basic parser that is a partial implementation of rfc2445
which pertains to specifying recurring date/times. Limitations
include:
- Only collects `DTSTART`, `DTEND`, `RRULE`, `EXRULE`, `RDATE`,
and `EXDATE` properties.
- Does not capture parameter options (i.e. RDATE;VALUE=PERIOD).
`dateutil.rrule` does not support anything other than
`DATE-TIME` parameter types.
- `VTIMEZONE` and `TZID` can't be specified, so dates without
the 'Z' marker will be localized to
`settings.TIME_ZONE`. `datetime.datetime` objects in
`Recurrence`/`Rrule` objects will be serialized as UTC.
- The `DTSTART`, `DTEND`, `RDATE` and `EXDATE` properties also
only support the `DATE-TIME` type.
:Returns:
A `Recurrence` instance.
"""
def deserialize_dt(text):
try:
year, month, day = int(text[:4]), int(text[4:6]), int(text[6:8])
except ValueError:
raise exceptions.DeserializationError('malformed date-time: %r' % text)
if u'T' in text:
# time is also specified
try:
hour, minute, second = (
int(text[9:11]), int(text[11:13]), int(text[13:15]))
except ValueError:
raise exceptions.DeserializationError('malformed date-time: %r' % text)
else:
# only date is specified, use midnight
hour, minute, second = (0, 0, 0)
if u'Z' in text:
# time is in utc
tzinfo = pytz.utc
else:
# right now there is no support for VTIMEZONE/TZID since
# this is a partial implementation of rfc2445 so we'll
# just use the time zone specified in the Django settings.
tzinfo = localtz()
dt = datetime.datetime(
year, month, day, hour, minute, second, tzinfo=tzinfo)
dt = dt.astimezone(localtz())
# set tz to settings.TIME_ZONE and return offset-naive datetime
return datetime.datetime(
dt.year, dt.month, dt.day, dt.hour, dt.minute, dt.second)
dtstart, dtend, rrules, exrules, rdates, exdates = None, None, [], [], [], []
tokens = re.compile(
u'(DTSTART|DTEND|RRULE|EXRULE|RDATE|EXDATE)[^:]*:(.*)',
re.MULTILINE).findall(text)
if not tokens and text:
raise exceptions.DeserializationError('malformed data')
for label, param_text in tokens:
if not param_text:
raise exceptions.DeserializationError('empty property: %r' % label)
if label in (u'RRULE', u'EXRULE'):
params = {}
param_tokens = filter(lambda p: p, param_text.split(u';'))
for item in param_tokens:
try:
param_name, param_value = map(
lambda i: i.strip(), item.split(u'=', 1))
except ValueError:
raise exceptions.DeserializationError(
'missing parameter value: %r' % item)
params[param_name] = list(map(
lambda i: i.strip(), param_value.split(u',')))
kwargs = {}
for key, value in params.items():
if key == u'FREQ':
try:
kwargs[str(key.lower())] = list(
Rule.frequencies).index(value[0])
except ValueError:
raise exceptions.DeserializationError(
'bad frequency value: %r' % value[0])
elif key == u'INTERVAL':
try:
kwargs[str(key.lower())] = int(value[0])
except ValueError:
raise exceptions.DeserializationError(
'bad interval value: %r' % value[0])
elif key == u'WKST':
try:
kwargs[str(key.lower())] = to_weekday(value[0])
except ValueError:
raise exceptions.DeserializationError(
'bad weekday value: %r' % value[0])
elif key == u'COUNT':
try:
kwargs[str(key.lower())] = int(value[0])
except ValueError:
raise exceptions.DeserializationError(
'bad count value: %r' % value[0])
elif key == u'UNTIL':
kwargs[str(key.lower())] = deserialize_dt(value[0])
elif key == u'BYDAY':
bydays = []
for v in value:
try:
bydays.append(to_weekday(v))
except ValueError:
raise exceptions.DeserializationError(
'bad weekday value: %r' % v)
kwargs[str(key.lower())] = bydays
elif key.lower() in Rule.byparams:
numbers = []
for v in value:
try:
numbers.append(int(v))
except ValueError:
raise exceptions.DeserializationError(
'bad value: %r' % value)
kwargs[str(key.lower())] = numbers
else:
raise exceptions.DeserializationError('bad parameter: %s' % key)
if 'freq' not in kwargs:
raise exceptions.DeserializationError(
'frequency parameter missing from rule')
if label == u'RRULE':
rrules.append(Rule(**kwargs))
else:
exrules.append(Rule(**kwargs))
elif label == u'DTSTART':
dtstart = deserialize_dt(param_text)
elif label == u'DTEND':
dtend = deserialize_dt(param_text)
elif label == u'RDATE':
rdates.append(deserialize_dt(param_text))
elif label == u'EXDATE':
exdates.append(deserialize_dt(param_text))
return Recurrence(dtstart, dtend, rrules, exrules, rdates, exdates, include_dtstart)
def rule_to_text(rule, short=False):
"""
Render the given `Rule` as natural text.
:Parameters:
`short` : bool
Use abbreviated labels, i.e. 'Fri' instead of 'Friday'.
"""
frequencies = (
_('annually'), _('monthly'), _('weekly'), _('daily'),
_('hourly'), _('minutely'), _('secondly'),
)
timeintervals = (
_('years'), _('months'), _('weeks'), _('days'),
_('hours'), _('minutes'), _('seconds'),
)
if short:
positional_display = {
1: _('1st %(weekday)s'),
2: _('2nd %(weekday)s'),
3: _('3rd %(weekday)s'),
-1: _('last %(weekday)s'),
-2: _('2nd last %(weekday)s'),
-3: _('3rd last %(weekday)s'),
}
last_of_month_display = {
-1: _('last'),
-2: _('2nd last'),
-3: _('3rd last'),
-4: _('4th last'),
}
weekdays_display = (
_('Mon'), _('Tue'), _('Wed'),
_('Thu'), _('Fri'), _('Sat'), _('Sun'),
)
months_display = (
_('Jan'), _('Feb'), _('Mar'), _('Apr'),
_p('month name', 'May'), _('Jun'), _('Jul'), _('Aug'),
_('Sep'), _('Oct'), _('Nov'), _('Dec'),
)
else:
positional_display = {
1: _('first %(weekday)s'),
2: _('second %(weekday)s'),
3: _('third %(weekday)s'),
4: _('fourth %(weekday)s'),
-1: _('last %(weekday)s'),
-2: _('second last %(weekday)s'),
-3: _('third last %(weekday)s'),
}
last_of_month_display = {
-1: _('last'),
-2: _('second last'),
-3: _('third last'),
-4: _('fourth last'),
}
weekdays_display = (
_('Monday'), _('Tuesday'), _('Wednesday'),
_('Thursday'), _('Friday'), _('Saturday'), _('Sunday'),
)
months_display = (
_('January'), _('February'), _('March'), _('April'),
_p('month name', 'May'), _('June'), _('July'), _('August'),
_('September'), _('October'), _('November'), _('December'),
)
def get_positional_weekdays(rule):
items = []
if rule.bysetpos and rule.byday:
for setpos in rule.bysetpos:
for byday in rule.byday:
byday = to_weekday(byday)
items.append(
positional_display.get(setpos) % {
'weekday': weekdays_display[byday.number]})
elif rule.byday:
for byday in rule.byday:
byday = to_weekday(byday)
items.append(
positional_display.get(byday.index, '%(weekday)s') % {
'weekday': weekdays_display[byday.number]})
return _(', ').join(items)
parts = []
if rule.interval > 1:
parts.append(
_('every %(number)s %(freq)s') % {
'number': rule.interval,
'freq': timeintervals[rule.freq]
})
else:
parts.append(frequencies[rule.freq])
if rule.freq == YEARLY:
if rule.bymonth:
# bymonths are 1-indexed (January is 1), months_display
# are 0-indexed (January is 0).
items = _(', ').join(
[months_display[month] for month in
[month_index - 1 for month_index in rule.bymonth]])
parts.append(_('each %(items)s') % {'items': items})
if rule.byday or rule.bysetpos:
parts.append(
_('on the %(items)s') % {
'items': get_positional_weekdays(rule)})
if rule.freq == MONTHLY:
if rule.bymonthday:
items = _(', ').join([
dateformat.format(datetime.datetime(1, 1, day), 'jS') if day > 0
else last_of_month_display.get(day, day)
for day in rule.bymonthday])
parts.append(_('on the %(items)s') % {'items': items})
elif rule.byday:
if rule.byday or rule.bysetpos:
parts.append(
_('on the %(items)s') % {
'items': get_positional_weekdays(rule)})
if rule.freq == WEEKLY:
if rule.byday:
items = _(', ').join([
weekdays_display[to_weekday(day).number]
for day in rule.byday])
parts.append(_('each %(items)s') % {'items': items})
# daily freqencies has no additional formatting,
# hour/minute/second formatting not supported
if rule.count:
if rule.count == 1:
parts.append(_('occuring once'))
else:
parts.append(_('occuring %(number)s times') % {
'number': rule.count})
elif rule.until:
parts.append(_('until %(date)s') % {
'date': dateformat.format(rule.until, 'Y-m-d')})
return _(', ').join(parts)
def normalize_offset_awareness(dt, from_dt=None):
"""
Given two `datetime.datetime` objects, return the second object as
timezone offset-aware or offset-naive depending on the existence
of the first object's tzinfo.
If the second object is to be made offset-aware, it is assumed to
be in the local timezone (with the timezone derived from the
TIME_ZONE setting). If it is to be made offset-naive, It is first
converted to the local timezone before being made naive.
:Parameters:
`dt` : `datetime.datetime`
The datetime object to make offset-aware/offset-naive.
`from_dt` : `datetime.datetime`
The datetime object to test the existence of a tzinfo. If
the value is nonzero, it will be understood as
offset-naive
"""
if from_dt and from_dt.tzinfo and dt.tzinfo:
return dt
elif from_dt and from_dt.tzinfo and not dt.tzinfo:
dt = localtz().localize(dt)
elif dt.tzinfo:
dt = dt.astimezone(localtz())
dt = datetime.datetime(
dt.year, dt.month, dt.day,
dt.hour, dt.minute, dt.second)
return dt
def from_dateutil_rrule(rrule):
"""
Convert a `dateutil.rrule.rrule` instance to a `Rule` instance.
:Returns:
A `Rrule` instance.
"""
kwargs = {}
kwargs['freq'] = rrule._freq
kwargs['interval'] = rrule._interval
if rrule._wkst != 0:
kwargs['wkst'] = rrule._wkst
kwargs['bysetpos'] = rrule._bysetpos
if rrule._count is not None:
kwargs['count'] = rrule._count
elif rrule._until is not None:
kwargs['until'] = rrule._until
days = []
if (rrule._byweekday is not None and (
WEEKLY != rrule._freq or len(rrule._byweekday) != 1 or
rrule._dtstart.weekday() != rrule._byweekday[0])):
# ignore byweekday if freq is WEEKLY and day correlates
# with dtstart because it was automatically set by
# dateutil
days.extend(Weekday(n) for n in rrule._byweekday)
if rrule._bynweekday is not None:
days.extend(Weekday(*n) for n in rrule._bynweekday)
if len(days) > 0:
kwargs['byday'] = days
if rrule._bymonthday is not None and len(rrule._bymonthday) > 0:
if not (rrule._freq <= MONTHLY and len(rrule._bymonthday) == 1 and
rrule._bymonthday[0] == rrule._dtstart.day):
# ignore bymonthday if it's generated by dateutil
kwargs['bymonthday'] = list(rrule._bymonthday)
if rrule._bynmonthday is not None and len(rrule._bynmonthday) > 0:
kwargs.setdefault('bymonthday', []).extend(rrule._bynmonthday)
if rrule._bymonth is not None and len(rrule._bymonth) > 0:
if (rrule._byweekday is not None or
len(rrule._bynweekday or ()) > 0 or not (
rrule._freq == YEARLY and len(rrule._bymonth) == 1 and
rrule._bymonth[0] == rrule._dtstart.month)):
# ignore bymonth if it's generated by dateutil
kwargs['bymonth'] = list(rrule._bymonth)
if rrule._byyearday is not None:
kwargs['byyearday'] = list(rrule._byyearday)
if rrule._byweekno is not None:
kwargs['byweekno'] = list(rrule._byweekno)
kwargs['byhour'] = list(rrule._byhour)
kwargs['byminute'] = list(rrule._byminute)
kwargs['bysecond'] = list(rrule._bysecond)
if (rrule._dtstart.hour in rrule._byhour and
rrule._dtstart.minute in rrule._byminute and
rrule._dtstart.second in rrule._bysecond):
# ignore byhour/byminute/bysecond automatically set by
# dateutil from dtstart
kwargs['byhour'].remove(rrule._dtstart.hour)
kwargs['byminute'].remove(rrule._dtstart.minute)
kwargs['bysecond'].remove(rrule._dtstart.second)
return Rule(**kwargs)
def from_dateutil_rruleset(rruleset):
"""
Convert a `dateutil.rrule.rruleset` instance to a `Recurrence`
instance.
:Returns:
A `Recurrence` instance.
"""
rrules = [from_dateutil_rrule(rrule) for rrule in rruleset._rrule]
exrules = [from_dateutil_rrule(exrule) for exrule in rruleset._exrule]
rdates = rruleset._rdate
exdates = rruleset._exdate
dts = [r._dtstart for r in rruleset._rrule] + rruleset._rdate
if len(dts) > 0:
dts.sort()
dtstart = dts[0]
else:
dtstart = None
return Recurrence(dtstart, rrules, exrules, rdates, exdates)