centralcloud/oncall-engine
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Improve update shift logic using rotation start and until dates

Browse source
This commit is contained in:
Julia 2022-08-16 15:38:52 +03:00
parent 6f8ee39026
commit a0efa4e025
4 changed files with 61 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
engine/apps/schedules/constants.py Normal file
View file

@ -0,0 +1,14 @@
import re
ICAL_DATETIME_START = "DTSTART"
ICAL_DATETIME_END = "DTEND"
ICAL_DATETIME_STAMP = "DTSTAMP"
ICAL_SUMMARY = "SUMMARY"
ICAL_DESCRIPTION = "DESCRIPTION"
ICAL_ATTENDEE = "ATTENDEE"
ICAL_UID = "UID"
ICAL_RRULE = "RRULE"
ICAL_UNTIL = "UNTIL"
RE_PRIORITY = re.compile(r"^\[L(\d)\]")
RE_EVENT_UID_V1 = re.compile(r"amixr-([\w\d-]+)-U(\d+)-E(\d+)-S(\d+)")
RE_EVENT_UID_V2 = re.compile(r"oncall-([\w\d-]+)-PK([\w\d]+)-U(\d+)-E(\d+)-S(\d+)")

18
engine/apps/schedules/ical_events/adapter/amixr_recurring_ical_events_adapter.py
View file

@ -6,6 +6,7 @@ from django.utils import timezone
from icalendar import Calendar, Event
from recurring_ical_events import UnfoldableCalendar, compare_greater, is_event, time_span_contains_event
from apps.schedules.constants import ICAL_DATETIME_END, ICAL_DATETIME_STAMP, ICAL_DATETIME_START, ICAL_RRULE, ICAL_UNTIL
from apps.schedules.ical_events.proxy.ical_proxy import IcalService
EXTRA_LOOKUP_DAYS = 16
@ -19,6 +20,17 @@ class AmixrUnfoldableCalendar(UnfoldableCalendar):
So i took part of code from 0.1.20b0 but leave 0.1.16b in requirements.
"""
class RepeatedEvent(UnfoldableCalendar.RepeatedEvent):
class Repetition(UnfoldableCalendar.RepeatedEvent.Repetition):
"""
A repetition of an event. Overridden version of
recurring_ical_events.UnfoldableCalendar.RepeatedEvent.Repetition. This is overridden to remove the 'RRULE'
param from ATTRIBUTES_TO_DELETE_ON_COPY, because the 'UNTIL' param must be stored in repetition events to
calculate its end date.
"""
ATTRIBUTES_TO_DELETE_ON_COPY = ["RDATE", "EXDATE"]
def between(self, start, stop):
"""Return events at a time between start (inclusive) and end (inclusive)"""
span_start = self.to_datetime(start)
@ -83,6 +95,10 @@ class AmixrRecurringIcalEventsAdapter(IcalService):
)
def filter_extra_days(event):
return time_span_contains_event(start_date, end_date, event["DTSTART"].dt, event["DTEND"].dt)
event_start = max(event[ICAL_DATETIME_START].dt, event[ICAL_DATETIME_STAMP].dt)
event_end = event[ICAL_DATETIME_END].dt
if event.get(ICAL_RRULE, {}).get(ICAL_UNTIL):
event_end = min(event[ICAL_RRULE][ICAL_UNTIL][0], event[ICAL_DATETIME_END].dt)
return time_span_contains_event(start_date, end_date, event_start, event_end)
return list(filter(filter_extra_days, events))

34
engine/apps/schedules/ical_utils.py
View file

@ -13,6 +13,20 @@ from django.db.models import Q
from django.utils import timezone
from icalendar import Calendar
from apps.schedules.constants import (
ICAL_ATTENDEE,
ICAL_DATETIME_END,
ICAL_DATETIME_STAMP,
ICAL_DATETIME_START,
ICAL_DESCRIPTION,
ICAL_RRULE,
ICAL_SUMMARY,
ICAL_UID,
ICAL_UNTIL,
RE_EVENT_UID_V1,
RE_EVENT_UID_V2,
RE_PRIORITY,
)
from apps.schedules.ical_events import ical_events
from common.constants.role import Role
from common.utils import timed_lru_cache
@ -68,15 +82,6 @@ def memoized_users_in_ical(usernames_from_ical, organization):
return users_in_ical(usernames_from_ical, organization)
ICAL_DATETIME_START = "DTSTART"
ICAL_DATETIME_END = "DTEND"
ICAL_SUMMARY = "SUMMARY"
ICAL_DESCRIPTION = "DESCRIPTION"
ICAL_ATTENDEE = "ATTENDEE"
ICAL_UID = "UID"
RE_PRIORITY = re.compile(r"^\[L(\d)\]")
RE_EVENT_UID_V1 = re.compile(r"amixr-([\w\d-]+)-U(\d+)-E(\d+)-S(\d+)")
RE_EVENT_UID_V2 = re.compile(r"oncall-([\w\d-]+)-PK([\w\d]+)-U(\d+)-E(\d+)-S(\d+)")
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
@ -171,8 +176,10 @@ def get_shifts_dict(calendar, calendar_type, schedule, datetime_start, datetime_
# Define on-call shift out of ical event that has the actual user
if len(users) > 0 or with_empty_shifts:
if type(event[ICAL_DATETIME_START].dt) == datetime.date:
start = event[ICAL_DATETIME_START].dt
start = max(event[ICAL_DATETIME_START].dt, event[ICAL_DATETIME_STAMP].dt.date())
end = event[ICAL_DATETIME_END].dt
if event.get(ICAL_RRULE, {}).get(ICAL_UNTIL):
end = min(event[ICAL_DATETIME_END].dt, event[ICAL_RRULE][ICAL_UNTIL][0].date())
if start <= date < end:
result_date.append(
{
@ -187,8 +194,13 @@ def get_shifts_dict(calendar, calendar_type, schedule, datetime_start, datetime_
}
)
else:
start = event[ICAL_DATETIME_START].dt.astimezone(pytz.UTC)
start = max(
event[ICAL_DATETIME_START].dt.astimezone(pytz.UTC),
event[ICAL_DATETIME_STAMP].dt.astimezone(pytz.UTC),
)
end = event[ICAL_DATETIME_END].dt.astimezone(pytz.UTC)
if event.get(ICAL_RRULE, {}).get(ICAL_UNTIL):
end = min(event[ICAL_DATETIME_END].dt.astimezone(pytz.UTC), event[ICAL_RRULE][ICAL_UNTIL][0])
result_datetime.append(
{

13
engine/apps/schedules/models/custom_on_call_shift.py
View file

@ -280,7 +280,7 @@ class CustomOnCallShift(models.Model):
# rolling_users shift converts to several ical events
if self.type in (CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, CustomOnCallShift.TYPE_OVERRIDE):
# generate initial iCal for counting rotation start date
event_ical = self.generate_ical(self.start, user_counter=0)
event_ical = self.generate_ical(self.start)
rotations_created = 0
all_rotation_checked = False
@ -301,13 +301,14 @@ class CustomOnCallShift(models.Model):
if not start: # means that rotation ends before next event starts
all_rotation_checked = True
break
elif start >= self.rotation_start: # event has already started, generate iCal for each user
elif start + self.duration > self.rotation_start:
# event has already started, generate iCal for each user
for user_counter, user in enumerate(users, start=1):
event_ical = self.generate_ical(start, user_counter, user, counter, time_zone)
result += event_ical
rotations_created += 1
else: # generate default iCal to calculate the date for the next rotation
event_ical = self.generate_ical(start, user_counter=0)
event_ical = self.generate_ical(start)
if rotations_created == len(users_queue): # means that we generated iCal for every user group
all_rotation_checked = True
@ -319,14 +320,14 @@ class CustomOnCallShift(models.Model):
result += self.generate_ical(self.start, user_counter, user, time_zone=time_zone)
return result
def generate_ical(self, start, user_counter, user=None, counter=1, time_zone="UTC"):
def generate_ical(self, start, user_counter=0, user=None, counter=1, time_zone="UTC"):
event = Event()
event["uid"] = f"oncall-{self.uuid}-PK{self.public_primary_key}-U{user_counter}-E{counter}-S{self.source}"
if user:
event.add("summary", self.get_summary_with_user_for_ical(user))
event.add("dtstart", self.convert_dt_to_schedule_timezone(start, time_zone))
event.add("dtend", self.convert_dt_to_schedule_timezone(start + self.duration, time_zone))
event.add("dtstamp", timezone.now())
event.add("dtstamp", self.rotation_start)
if self.event_ical_rules:
event.add("rrule", self.event_ical_rules)
try:
@ -407,7 +408,7 @@ class CustomOnCallShift(models.Model):
for event in ical_iter:
if end_date: # end_date exists for long events with frequency weekly and monthly
if end_date >= event.start >= next_event_start:
if event.start >= self.rotation_start:
if event.stop > self.rotation_start:
next_event = event
break
else: