Xavier-Do
e9b62b47dc
Some of the non canononical timezones are not present in Ubuntu Noble, it would be a better practice to only use canonical timezones in data and tests. Note that this is not a real fix for all cases since the database that ran on Ubuntu Jammy and are moved to an ubuntu Noble server will have the issue with timezones already in database. One of the possible fix would be to manage that during upgrades, but this isn't a verry flexible solution since upgrade are meant to manage chyange of version, not change of server. If an old 17.0 versions needs to be moved to a Noble server, this won't work. Another solution would be to install package like tzdata-legacy that may keep the old timezones but it is not the only think since TAI-10 are also in this package. This solution is not ideal because non canonical timezone will still be shown in the dropdown. We would need to filter them. A last solution would be to add the support for those old timezones by monkeypatching the lib. This way, only new timezones would be shown but non canonical one won't crash when used. This is not ideal either because we may need to keep this for a while. But in combination with the upgrade solution, it may work proprely. Part-of: odoo/odoo#160842
601 lines
25 KiB
Python
601 lines
25 KiB
Python
# -*- coding: utf-8 -*-
|
||
# Part of Odoo. See LICENSE file for full copyright and licensing details.
|
||
|
||
from datetime import datetime, time
|
||
import pytz
|
||
|
||
from dateutil import rrule
|
||
from dateutil.relativedelta import relativedelta
|
||
|
||
from odoo import api, fields, models, _
|
||
from odoo.exceptions import UserError
|
||
from odoo.tools.misc import clean_context
|
||
|
||
from odoo.addons.base.models.res_partner import _tz_get
|
||
|
||
|
||
MAX_RECURRENT_EVENT = 720
|
||
|
||
SELECT_FREQ_TO_RRULE = {
|
||
'daily': rrule.DAILY,
|
||
'weekly': rrule.WEEKLY,
|
||
'monthly': rrule.MONTHLY,
|
||
'yearly': rrule.YEARLY,
|
||
}
|
||
|
||
RRULE_FREQ_TO_SELECT = {
|
||
rrule.DAILY: 'daily',
|
||
rrule.WEEKLY: 'weekly',
|
||
rrule.MONTHLY: 'monthly',
|
||
rrule.YEARLY: 'yearly',
|
||
}
|
||
|
||
RRULE_WEEKDAY_TO_FIELD = {
|
||
rrule.MO.weekday: 'mon',
|
||
rrule.TU.weekday: 'tue',
|
||
rrule.WE.weekday: 'wed',
|
||
rrule.TH.weekday: 'thu',
|
||
rrule.FR.weekday: 'fri',
|
||
rrule.SA.weekday: 'sat',
|
||
rrule.SU.weekday: 'sun',
|
||
}
|
||
|
||
RRULE_WEEKDAYS = {'SUN': 'SU', 'MON': 'MO', 'TUE': 'TU', 'WED': 'WE', 'THU': 'TH', 'FRI': 'FR', 'SAT': 'SA'}
|
||
|
||
RRULE_TYPE_SELECTION = [
|
||
('daily', 'Days'),
|
||
('weekly', 'Weeks'),
|
||
('monthly', 'Months'),
|
||
('yearly', 'Years'),
|
||
]
|
||
|
||
END_TYPE_SELECTION = [
|
||
('count', 'Number of repetitions'),
|
||
('end_date', 'End date'),
|
||
('forever', 'Forever'),
|
||
]
|
||
|
||
MONTH_BY_SELECTION = [
|
||
('date', 'Date of month'),
|
||
('day', 'Day of month'),
|
||
]
|
||
|
||
WEEKDAY_SELECTION = [
|
||
('MON', 'Monday'),
|
||
('TUE', 'Tuesday'),
|
||
('WED', 'Wednesday'),
|
||
('THU', 'Thursday'),
|
||
('FRI', 'Friday'),
|
||
('SAT', 'Saturday'),
|
||
('SUN', 'Sunday'),
|
||
]
|
||
|
||
BYDAY_SELECTION = [
|
||
('1', 'First'),
|
||
('2', 'Second'),
|
||
('3', 'Third'),
|
||
('4', 'Fourth'),
|
||
('-1', 'Last'),
|
||
]
|
||
|
||
def freq_to_select(rrule_freq):
|
||
return RRULE_FREQ_TO_SELECT[rrule_freq]
|
||
|
||
|
||
def freq_to_rrule(freq):
|
||
return SELECT_FREQ_TO_RRULE[freq]
|
||
|
||
|
||
def weekday_to_field(weekday_index):
|
||
return RRULE_WEEKDAY_TO_FIELD.get(weekday_index)
|
||
|
||
|
||
class RecurrenceRule(models.Model):
|
||
_name = 'calendar.recurrence'
|
||
_description = 'Event Recurrence Rule'
|
||
|
||
name = fields.Char(compute='_compute_name', store=True)
|
||
base_event_id = fields.Many2one(
|
||
'calendar.event', ondelete='set null', copy=False) # store=False ?
|
||
calendar_event_ids = fields.One2many('calendar.event', 'recurrence_id')
|
||
event_tz = fields.Selection(
|
||
_tz_get, string='Timezone',
|
||
default=lambda self: self.env.context.get('tz') or self.env.user.tz)
|
||
rrule = fields.Char(compute='_compute_rrule', inverse='_inverse_rrule', store=True)
|
||
dtstart = fields.Datetime(compute='_compute_dtstart')
|
||
rrule_type = fields.Selection(RRULE_TYPE_SELECTION, default='weekly')
|
||
end_type = fields.Selection(END_TYPE_SELECTION, default='count')
|
||
interval = fields.Integer(default=1)
|
||
count = fields.Integer(default=1)
|
||
mon = fields.Boolean()
|
||
tue = fields.Boolean()
|
||
wed = fields.Boolean()
|
||
thu = fields.Boolean()
|
||
fri = fields.Boolean()
|
||
sat = fields.Boolean()
|
||
sun = fields.Boolean()
|
||
month_by = fields.Selection(MONTH_BY_SELECTION, default='date')
|
||
day = fields.Integer(default=1)
|
||
weekday = fields.Selection(WEEKDAY_SELECTION, string='Weekday')
|
||
byday = fields.Selection(BYDAY_SELECTION, string='By day')
|
||
until = fields.Date('Repeat Until')
|
||
trigger_id = fields.Many2one('ir.cron.trigger')
|
||
|
||
_sql_constraints = [
|
||
('month_day',
|
||
"CHECK (rrule_type != 'monthly' "
|
||
"OR month_by != 'day' "
|
||
"OR day >= 1 AND day <= 31 "
|
||
"OR weekday in %s AND byday in %s)"
|
||
% (tuple(wd[0] for wd in WEEKDAY_SELECTION), tuple(bd[0] for bd in BYDAY_SELECTION)),
|
||
"The day must be between 1 and 31"),
|
||
]
|
||
|
||
def _get_daily_recurrence_name(self):
|
||
if self.end_type == 'count':
|
||
return _("Every %(interval)s Days for %(count)s events", interval=self.interval, count=self.count)
|
||
if self.end_type == 'end_date':
|
||
return _("Every %(interval)s Days until %(until)s", interval=self.interval, until=self.until)
|
||
return _("Every %(interval)s Days", interval=self.interval)
|
||
|
||
def _get_weekly_recurrence_name(self):
|
||
weekday_selection = dict(self._fields['weekday']._description_selection(self.env))
|
||
weekdays = self._get_week_days()
|
||
# Convert Weekday object
|
||
weekdays = [str(w) for w in weekdays]
|
||
# We need to get the day full name from its three first letters.
|
||
week_map = {v: k for k, v in RRULE_WEEKDAYS.items()}
|
||
weekday_short = [week_map[w] for w in weekdays]
|
||
day_strings = [weekday_selection[day] for day in weekday_short]
|
||
days = ", ".join(day_strings)
|
||
|
||
if self.end_type == 'count':
|
||
return _("Every %(interval)s Weeks on %(days)s for %(count)s events", interval=self.interval, days=days, count=self.count)
|
||
if self.end_type == 'end_date':
|
||
return _("Every %(interval)s Weeks on %(days)s until %(until)s", interval=self.interval, days=days, until=self.until)
|
||
return _("Every %(interval)s Weeks on %(days)s", interval=self.interval, days=days)
|
||
|
||
def _get_monthly_recurrence_name(self):
|
||
if self.month_by == 'day':
|
||
weekday_selection = dict(self._fields['weekday']._description_selection(self.env))
|
||
byday_selection = dict(self._fields['byday']._description_selection(self.env))
|
||
position_label = byday_selection[self.byday]
|
||
weekday_label = weekday_selection[self.weekday]
|
||
|
||
if self.end_type == 'count':
|
||
return _("Every %(interval)s Months on the %(position)s %(weekday)s for %(count)s events", interval=self.interval, position=position_label, weekday=weekday_label, count=self.count)
|
||
if self.end_type == 'end_date':
|
||
return _("Every %(interval)s Months on the %(position)s %(weekday)s until %(until)s", interval=self.interval, position=position_label, weekday=weekday_label, until=self.until)
|
||
return _("Every %(interval)s Months on the %(position)s %(weekday)s", interval=self.interval, position=position_label, weekday=weekday_label)
|
||
else:
|
||
if self.end_type == 'count':
|
||
return _("Every %(interval)s Months day %(day)s for %(count)s events", interval=self.interval, day=self.day, count=self.count)
|
||
if self.end_type == 'end_date':
|
||
return _("Every %(interval)s Months day %(day)s until %(until)s", interval=self.interval, day=self.day, until=self.until)
|
||
return _("Every %(interval)s Months day %(day)s", interval=self.interval, day=self.day)
|
||
|
||
def _get_yearly_recurrence_name(self):
|
||
if self.end_type == 'count':
|
||
return _("Every %(interval)s Years for %(count)s events", interval=self.interval, count=self.count)
|
||
if self.end_type == 'end_date':
|
||
return _("Every %(interval)s Years until %(until)s", interval=self.interval, until=self.until)
|
||
return _("Every %(interval)s Years", interval=self.interval)
|
||
|
||
def get_recurrence_name(self):
|
||
if self.rrule_type == 'daily':
|
||
return self._get_daily_recurrence_name()
|
||
if self.rrule_type == 'weekly':
|
||
return self._get_weekly_recurrence_name()
|
||
if self.rrule_type == 'monthly':
|
||
return self._get_monthly_recurrence_name()
|
||
if self.rrule_type == 'yearly':
|
||
return self._get_yearly_recurrence_name()
|
||
|
||
@api.depends('rrule')
|
||
def _compute_name(self):
|
||
for recurrence in self:
|
||
recurrence.name = recurrence.get_recurrence_name()
|
||
|
||
@api.depends('calendar_event_ids.start')
|
||
def _compute_dtstart(self):
|
||
groups = self.env['calendar.event']._read_group([('recurrence_id', 'in', self.ids)], ['recurrence_id'], ['start:min'])
|
||
start_mapping = {recurrence.id: start_min for recurrence, start_min in groups}
|
||
for recurrence in self:
|
||
recurrence.dtstart = start_mapping.get(recurrence.id)
|
||
|
||
@api.depends(
|
||
'byday', 'until', 'rrule_type', 'month_by', 'interval', 'count', 'end_type',
|
||
'mon', 'tue', 'wed', 'thu', 'fri', 'sat', 'sun', 'day', 'weekday')
|
||
def _compute_rrule(self):
|
||
for recurrence in self:
|
||
current_rule = recurrence._rrule_serialize()
|
||
if recurrence.rrule != current_rule:
|
||
recurrence.write({'rrule': current_rule})
|
||
|
||
def _inverse_rrule(self):
|
||
for recurrence in self:
|
||
if recurrence.rrule:
|
||
values = self._rrule_parse(recurrence.rrule, recurrence.dtstart)
|
||
recurrence.with_context(dont_notify=True).write(values)
|
||
|
||
def _reconcile_events(self, ranges):
|
||
"""
|
||
:param ranges: iterable of tuples (datetime_start, datetime_stop)
|
||
:return: tuple (events of the recurrence already in sync with ranges,
|
||
and ranges not covered by any events)
|
||
"""
|
||
ranges = set(ranges)
|
||
|
||
synced_events = self.calendar_event_ids.filtered(lambda e: e._range() in ranges)
|
||
|
||
existing_ranges = set(event._range() for event in synced_events)
|
||
ranges_to_create = (event_range for event_range in ranges if event_range not in existing_ranges)
|
||
return synced_events, ranges_to_create
|
||
|
||
def _select_new_base_event(self):
|
||
"""
|
||
when the base event is no more available (archived, deleted, etc.), a new one should be selected
|
||
"""
|
||
for recurrence in self:
|
||
recurrence.base_event_id = recurrence._get_first_event()
|
||
|
||
def _apply_recurrence(self, specific_values_creation=None, no_send_edit=False, generic_values_creation=None):
|
||
"""Create missing events in the recurrence and detach events which no longer
|
||
follow the recurrence rules.
|
||
:return: detached events
|
||
"""
|
||
event_vals = []
|
||
keep = self.env['calendar.event']
|
||
if specific_values_creation is None:
|
||
specific_values_creation = {}
|
||
|
||
for recurrence in self.filtered('base_event_id'):
|
||
recurrence.calendar_event_ids |= recurrence.base_event_id
|
||
event = recurrence.base_event_id or recurrence._get_first_event(include_outliers=False)
|
||
duration = event.stop - event.start
|
||
if specific_values_creation:
|
||
ranges = set([(x[1], x[2]) for x in specific_values_creation if x[0] == recurrence.id])
|
||
else:
|
||
ranges = recurrence._range_calculation(event, duration)
|
||
|
||
events_to_keep, ranges = recurrence._reconcile_events(ranges)
|
||
keep |= events_to_keep
|
||
[base_values] = event.copy_data()
|
||
values = []
|
||
for start, stop in ranges:
|
||
value = dict(base_values, start=start, stop=stop, recurrence_id=recurrence.id, follow_recurrence=True)
|
||
if (recurrence.id, start, stop) in specific_values_creation:
|
||
value.update(specific_values_creation[(recurrence.id, start, stop)])
|
||
if generic_values_creation and recurrence.id in generic_values_creation:
|
||
value.update(generic_values_creation[recurrence.id])
|
||
values += [value]
|
||
event_vals += values
|
||
|
||
events = self.calendar_event_ids - keep
|
||
detached_events = self._detach_events(events)
|
||
context = {
|
||
**clean_context(self.env.context),
|
||
**{'no_mail_to_attendees': True, 'mail_create_nolog': True},
|
||
}
|
||
self.env['calendar.event'].with_context(context).create(event_vals)
|
||
return detached_events
|
||
|
||
def _setup_alarms(self, recurrence_update=False):
|
||
""" Schedule cron triggers for future events
|
||
Create one ir.cron.trigger per recurrence.
|
||
:param recurrence_update: boolean: if true, update all recurrences in self, else only the recurrences
|
||
without trigger
|
||
"""
|
||
now = self.env.context.get('date') or fields.Datetime.now()
|
||
# get next events
|
||
self.env['calendar.event'].flush_model(fnames=['recurrence_id', 'start'])
|
||
if not self.calendar_event_ids.ids:
|
||
return
|
||
|
||
self.env.cr.execute("""
|
||
SELECT DISTINCT ON (recurrence_id) id event_id, recurrence_id
|
||
FROM calendar_event
|
||
WHERE start > %s
|
||
AND id IN %s
|
||
ORDER BY recurrence_id,start ASC;
|
||
""", (now, tuple(self.calendar_event_ids.ids)))
|
||
result = self.env.cr.dictfetchall()
|
||
if not result:
|
||
return
|
||
events = self.env['calendar.event'].browse(value['event_id'] for value in result)
|
||
triggers_by_events = events._setup_alarms()
|
||
for vals in result:
|
||
trigger_id = triggers_by_events.get(vals['event_id'])
|
||
if not trigger_id:
|
||
continue
|
||
recurrence = self.env['calendar.recurrence'].browse(vals['recurrence_id'])
|
||
recurrence.trigger_id = trigger_id
|
||
|
||
def _split_from(self, event, recurrence_values=None):
|
||
"""Stops the current recurrence at the given event and creates a new one starting
|
||
with the event.
|
||
:param event: starting point of the new recurrence
|
||
:param recurrence_values: values applied to the new recurrence
|
||
:return: new recurrence
|
||
"""
|
||
if recurrence_values is None:
|
||
recurrence_values = {}
|
||
event.ensure_one()
|
||
if not self:
|
||
return
|
||
[values] = self.copy_data()
|
||
detached_events = self._stop_at(event)
|
||
|
||
count = recurrence_values.get('count', 0) or len(detached_events)
|
||
return self.create({
|
||
**values,
|
||
**recurrence_values,
|
||
'base_event_id': event.id,
|
||
'calendar_event_ids': [(6, 0, detached_events.ids)],
|
||
'count': max(count, 1),
|
||
})
|
||
|
||
def _stop_at(self, event):
|
||
"""Stops the recurrence at the given event. Detach the event and all following
|
||
events from the recurrence.
|
||
|
||
:return: detached events from the recurrence
|
||
"""
|
||
self.ensure_one()
|
||
events = self._get_events_from(event.start)
|
||
detached_events = self._detach_events(events)
|
||
if not self.calendar_event_ids:
|
||
self.with_context(archive_on_error=True).unlink()
|
||
return detached_events
|
||
|
||
if event.allday:
|
||
until = self._get_start_of_period(event.start_date)
|
||
else:
|
||
until_datetime = self._get_start_of_period(event.start)
|
||
until_timezoned = pytz.utc.localize(until_datetime).astimezone(self._get_timezone())
|
||
until = until_timezoned.date()
|
||
self.write({
|
||
'end_type': 'end_date',
|
||
'until': until - relativedelta(days=1),
|
||
})
|
||
return detached_events
|
||
|
||
@api.model
|
||
def _detach_events(self, events):
|
||
events.with_context(dont_notify=True).write({
|
||
'recurrence_id': False,
|
||
'recurrency': True,
|
||
})
|
||
return events
|
||
|
||
def _write_events(self, values, dtstart=None):
|
||
"""
|
||
Write values on events in the recurrence.
|
||
:param values: event values
|
||
:param dstart: if provided, only write events starting from this point in time
|
||
"""
|
||
events = self._get_events_from(dtstart) if dtstart else self.calendar_event_ids
|
||
return events.with_context(no_mail_to_attendees=True, dont_notify=True).write(dict(values, recurrence_update='self_only'))
|
||
|
||
def _rrule_serialize(self):
|
||
"""
|
||
Compute rule string according to value type RECUR of iCalendar
|
||
:return: string containing recurring rule (empty if no rule)
|
||
"""
|
||
if self.interval <= 0:
|
||
raise UserError(_('The interval cannot be negative.'))
|
||
if self.end_type == 'count' and self.count <= 0:
|
||
raise UserError(_('The number of repetitions cannot be negative.'))
|
||
|
||
return str(self._get_rrule()) if self.rrule_type else ''
|
||
|
||
@api.model
|
||
def _rrule_parse(self, rule_str, date_start):
|
||
# LUL TODO clean this mess
|
||
data = {}
|
||
day_list = ['mon', 'tue', 'wed', 'thu', 'fri', 'sat', 'sun']
|
||
|
||
if 'Z' in rule_str and date_start and not date_start.tzinfo:
|
||
date_start = pytz.utc.localize(date_start)
|
||
rule = rrule.rrulestr(rule_str, dtstart=date_start)
|
||
|
||
data['rrule_type'] = freq_to_select(rule._freq)
|
||
data['count'] = rule._count
|
||
data['interval'] = rule._interval
|
||
data['until'] = rule._until
|
||
# Repeat weekly
|
||
if rule._byweekday:
|
||
for weekday in day_list:
|
||
data[weekday] = False # reset
|
||
for weekday_index in rule._byweekday:
|
||
weekday = rrule.weekday(weekday_index)
|
||
data[weekday_to_field(weekday.weekday)] = True
|
||
data['rrule_type'] = 'weekly'
|
||
|
||
# Repeat monthly by nweekday ((weekday, weeknumber), )
|
||
if rule._bynweekday:
|
||
data['weekday'] = day_list[list(rule._bynweekday)[0][0]].upper()
|
||
data['byday'] = str(list(rule._bynweekday)[0][1])
|
||
data['month_by'] = 'day'
|
||
data['rrule_type'] = 'monthly'
|
||
|
||
if rule._bymonthday:
|
||
data['day'] = list(rule._bymonthday)[0]
|
||
data['month_by'] = 'date'
|
||
data['rrule_type'] = 'monthly'
|
||
|
||
# Repeat yearly but for odoo it's monthly, take same information as monthly but interval is 12 times
|
||
if rule._bymonth:
|
||
data['interval'] *= 12
|
||
|
||
if data.get('until'):
|
||
data['end_type'] = 'end_date'
|
||
elif data.get('count'):
|
||
data['end_type'] = 'count'
|
||
else:
|
||
data['end_type'] = 'forever'
|
||
return data
|
||
|
||
def _get_lang_week_start(self):
|
||
lang = self.env['res.lang']._lang_get(self.env.user.lang)
|
||
week_start = int(lang.week_start) # lang.week_start ranges from '1' to '7'
|
||
return rrule.weekday(week_start - 1) # rrule expects an int from 0 to 6
|
||
|
||
def _get_start_of_period(self, dt):
|
||
if self.rrule_type == 'weekly':
|
||
week_start = self._get_lang_week_start()
|
||
start = dt + relativedelta(weekday=week_start(-1))
|
||
elif self.rrule_type == 'monthly':
|
||
start = dt + relativedelta(day=1)
|
||
else:
|
||
start = dt
|
||
# Comparaison of DST (to manage the case of going too far back in time).
|
||
# If we detect a change in the DST between the creation date of an event
|
||
# and the date used for the occurrence period, we use the creation date of the event.
|
||
# This is a hack to avoid duplication of events (for example on google calendar).
|
||
if isinstance(dt, datetime):
|
||
timezone = self._get_timezone()
|
||
dst_dt = timezone.localize(dt).dst()
|
||
dst_start = timezone.localize(start).dst()
|
||
if dst_dt != dst_start:
|
||
start = dt
|
||
return start
|
||
|
||
def _get_first_event(self, include_outliers=False):
|
||
if not self.calendar_event_ids:
|
||
return self.env['calendar.event']
|
||
events = self.calendar_event_ids.sorted('start')
|
||
if not include_outliers:
|
||
events -= self._get_outliers()
|
||
return events[:1]
|
||
|
||
def _get_outliers(self):
|
||
synced_events = self.env['calendar.event']
|
||
for recurrence in self:
|
||
if recurrence.calendar_event_ids:
|
||
start = min(recurrence.calendar_event_ids.mapped('start'))
|
||
starts = set(recurrence._get_occurrences(start))
|
||
synced_events |= recurrence.calendar_event_ids.filtered(lambda e: e.start in starts)
|
||
return self.calendar_event_ids - synced_events
|
||
|
||
def _range_calculation(self, event, duration):
|
||
""" Calculate the range of recurrence when applying the recurrence
|
||
The following issues are taken into account:
|
||
start of period is sometimes in the past (weekly or monthly rule).
|
||
We can easily filter these range values but then the count value may be wrong...
|
||
In that case, we just increase the count value, recompute the ranges and dismiss the useless values
|
||
"""
|
||
self.ensure_one()
|
||
original_count = self.end_type == 'count' and self.count
|
||
ranges = set(self._get_ranges(event.start, duration))
|
||
future_events = set((x, y) for x, y in ranges if x.date() >= event.start.date() and y.date() >= event.start.date())
|
||
if original_count and len(future_events) < original_count:
|
||
# Rise count number because some past values will be dismissed.
|
||
self.count = (2*original_count) - len(future_events)
|
||
ranges = set(self._get_ranges(event.start, duration))
|
||
# We set back the occurrence number to its original value
|
||
self.count = original_count
|
||
# Remove ranges of events occurring in the past
|
||
ranges = set((x, y) for x, y in ranges if x.date() >= event.start.date() and y.date() >= event.start.date())
|
||
return ranges
|
||
|
||
|
||
def _get_ranges(self, start, event_duration):
|
||
starts = self._get_occurrences(start)
|
||
return ((start, start + event_duration) for start in starts)
|
||
|
||
def _get_timezone(self):
|
||
return pytz.timezone(self.event_tz or self.env.context.get('tz') or 'UTC')
|
||
|
||
def _get_occurrences(self, dtstart):
|
||
"""
|
||
Get ocurrences of the rrule
|
||
:param dtstart: start of the recurrence
|
||
:return: iterable of datetimes
|
||
"""
|
||
self.ensure_one()
|
||
dtstart = self._get_start_of_period(dtstart)
|
||
if self._is_allday():
|
||
return self._get_rrule(dtstart=dtstart)
|
||
|
||
timezone = self._get_timezone()
|
||
# Localize the starting datetime to avoid missing the first occurrence
|
||
dtstart = pytz.utc.localize(dtstart).astimezone(timezone)
|
||
# dtstart is given as a naive datetime, but it actually represents a timezoned datetime
|
||
# (rrule package expects a naive datetime)
|
||
occurences = self._get_rrule(dtstart=dtstart.replace(tzinfo=None))
|
||
|
||
# Special timezoning is needed to handle DST (Daylight Saving Time) changes.
|
||
# Given the following recurrence:
|
||
# - monthly
|
||
# - 1st of each month
|
||
# - timezone America/New_York (UTC−05:00)
|
||
# - at 6am America/New_York = 11am UTC
|
||
# - from 2019/02/01 to 2019/05/01.
|
||
# The naive way would be to store:
|
||
# 2019/02/01 11:00 - 2019/03/01 11:00 - 2019/04/01 11:00 - 2019/05/01 11:00 (UTC)
|
||
#
|
||
# But a DST change occurs on 2019/03/10 in America/New_York timezone. America/New_York is now UTC−04:00.
|
||
# From this point in time, 11am (UTC) is actually converted to 7am (America/New_York) instead of the expected 6am!
|
||
# What should be stored is:
|
||
# 2019/02/01 11:00 - 2019/03/01 11:00 - 2019/04/01 10:00 - 2019/05/01 10:00 (UTC)
|
||
# ***** *****
|
||
return (timezone.localize(occurrence, is_dst=False).astimezone(pytz.utc).replace(tzinfo=None) for occurrence in occurences)
|
||
|
||
def _get_events_from(self, dtstart):
|
||
return self.env['calendar.event'].search([
|
||
('id', 'in', self.calendar_event_ids.ids),
|
||
('start', '>=', dtstart)
|
||
])
|
||
|
||
def _get_week_days(self):
|
||
"""
|
||
:return: tuple of rrule weekdays for this recurrence.
|
||
"""
|
||
return tuple(
|
||
rrule.weekday(weekday_index)
|
||
for weekday_index, weekday in {
|
||
rrule.MO.weekday: self.mon,
|
||
rrule.TU.weekday: self.tue,
|
||
rrule.WE.weekday: self.wed,
|
||
rrule.TH.weekday: self.thu,
|
||
rrule.FR.weekday: self.fri,
|
||
rrule.SA.weekday: self.sat,
|
||
rrule.SU.weekday: self.sun,
|
||
}.items() if weekday
|
||
)
|
||
|
||
def _is_allday(self):
|
||
"""Returns whether a majority of events are allday or not (there might be some outlier events)
|
||
"""
|
||
score = sum(1 if e.allday else -1 for e in self.calendar_event_ids)
|
||
return score >= 0
|
||
|
||
def _get_rrule(self, dtstart=None):
|
||
self.ensure_one()
|
||
freq = self.rrule_type
|
||
rrule_params = dict(
|
||
dtstart=dtstart,
|
||
interval=self.interval,
|
||
)
|
||
if freq == 'monthly' and self.month_by == 'date': # e.g. every 15th of the month
|
||
rrule_params['bymonthday'] = self.day
|
||
elif freq == 'monthly' and self.month_by == 'day': # e.g. every 2nd Monday in the month
|
||
rrule_params['byweekday'] = getattr(rrule, RRULE_WEEKDAYS[self.weekday])(int(self.byday)) # e.g. MO(+2) for the second Monday of the month
|
||
elif freq == 'weekly':
|
||
weekdays = self._get_week_days()
|
||
if not weekdays:
|
||
raise UserError(_("You have to choose at least one day in the week"))
|
||
rrule_params['byweekday'] = weekdays
|
||
rrule_params['wkst'] = self._get_lang_week_start()
|
||
|
||
if self.end_type == 'count': # e.g. stop after X occurence
|
||
rrule_params['count'] = min(self.count, MAX_RECURRENT_EVENT)
|
||
elif self.end_type == 'forever':
|
||
rrule_params['count'] = MAX_RECURRENT_EVENT
|
||
elif self.end_type == 'end_date': # e.g. stop after 12/10/2020
|
||
rrule_params['until'] = datetime.combine(self.until, time.max)
|
||
return rrule.rrule(
|
||
freq_to_rrule(freq), **rrule_params
|
||
)
|