Coverage for open_webui/utils/calendar.py: 14%

1""" 

2Calendar utilities. 

3 

4RRULE expansion reusing the automation infra. 

5""" 

6 

7import datetime as dt 

8import logging 

9from zoneinfo import ZoneInfo 

10 

11from dateutil.rrule import rrulestr 

12from open_webui.utils.automations import _resolve_tz 

13 

14log = logging.getLogger(__name__) 

15 

16 

17def expand_recurring_event( 

18 event_dict: dict, 

19 range_start_ns: int, 

20 range_end_ns: int, 

21 tz: str | None = None, 

22 max_instances: int = 5000, 

23) -> list[dict]: 

24 """Expand a recurring event into individual instances within a date range. 

25 

26 Takes an event dict (from CalendarEventModel.model_dump()) and produces 

27 one dict per occurrence, with adjusted start_at / end_at. 

28 """ 

29 rrule_str = event_dict.get('rrule') 

30 if not rrule_str: 

31 return [event_dict] 

32 if 'EXRULE' in rrule_str.upper(): 

33 log.warning(f'EXRULE is not supported for event {event_dict.get("id")}: {rrule_str}') 

34 return [event_dict] 

35 

36 user_timezone = _resolve_tz(tz) 

37 

38 def to_local_datetime(timestamp_ns: int) -> dt.datetime: 

39 return dt.datetime.fromtimestamp(timestamp_ns / 1_000_000_000, tz=user_timezone).replace(tzinfo=None) 

40 

41 range_start = to_local_datetime(range_start_ns) 

42 range_end = to_local_datetime(range_end_ns) 

43 scan_start = range_start - dt.timedelta(days=1) 

44 

45 original_start_ns = event_dict['start_at'] 

46 original_start = to_local_datetime(original_start_ns) 

47 rule_str = '\n'.join(part for part in rrule_str.split() if not part.upper().startswith('DTSTART')) or rrule_str 

48 

49 try: 

50 # Anchor to the event's real start so day-of-week / day-of-month are correct 

51 rule = rrulestr(rule_str, dtstart=original_start, ignoretz=True) 

52 except Exception: 

53 log.warning(f'Failed to parse RRULE for event {event_dict.get("id")}: {rrule_str}') 

54 return [event_dict] 

55 

56 original_end_ns = event_dict.get('end_at') 

57 duration_ns = (original_end_ns - original_start_ns) if original_end_ns else None 

58 

59 instances = [] 

60 previous_start = None 

61 for occurrence_start in rule.xafter(scan_start, count=max_instances, inc=True): 

62 if occurrence_start >= range_end or occurrence_start == previous_start: 

63 break 

64 previous_start = occurrence_start 

65 

66 instance_start_ns = int(occurrence_start.replace(tzinfo=user_timezone).timestamp() * 1_000_000_000) 

67 

68 if instance_start_ns >= range_start_ns: 

69 instance = { 

70 **event_dict, 

71 'start_at': instance_start_ns, 

72 'end_at': (instance_start_ns + duration_ns) if duration_ns else None, 

73 'instance_id': f'{event_dict["id"]}_{instance_start_ns}', 

74 } 

75 instances.append(instance) 

76 

77 return instances 

78 

79 

80def ns_from_date(year: int, month: int, day: int, tz: str | None = None) -> int: 

81 """Create epoch nanoseconds from a date.""" 

82 if tz: 

83 date_time = dt.datetime(year, month, day, tzinfo=ZoneInfo(tz)) 

84 else: 

85 date_time = dt.datetime(year, month, day) 

86 return int(date_time.timestamp() * 1_000_000_000)