Coverage for app/backend/src/couchers/event_recurrence.py: 100%

1""" 

2Pure logic for computing recurring event occurrence dates from an RRULE. 

3 

4No database or session dependency. 

5""" 

6 

7from collections.abc import Generator 

8from datetime import date, datetime, time, timedelta 

9 

10from dateutil.rrule import DAILY, MONTHLY, WEEKLY, rrule, rrulestr 

11 

12from couchers.utils import now 

13 

14 

15def make_every_nth_week_rrule(start_date: date, n: int, end_date: date | None = None) -> str: 

16 """Build an RRULE for recurring every nth week. Unbounded if no `end_date`.""" 

17 until = datetime.combine(end_date, time.min) if end_date is not None else None 

18 return str(rrule(WEEKLY, dtstart=datetime.combine(start_date, time.min), interval=n, until=until)) 

19 

20 

21def make_daily_rrule(start_date: date, end_date: date | None = None) -> str: 

22 """Build an RRULE for recurring daily. Unbounded if no `end_date`.""" 

23 until = datetime.combine(end_date, time.min) if end_date is not None else None 

24 return str(rrule(DAILY, dtstart=datetime.combine(start_date, time.min), until=until)) 

25 

26 

27def make_monthly_rrule(start_date: date, end_date: date | None = None) -> str: 

28 """Build an RRULE for recurring monthly. Unbounded if no `end_date`.""" 

29 until = datetime.combine(end_date, time.min) if end_date is not None else None 

30 return str(rrule(MONTHLY, dtstart=datetime.combine(start_date, time.min), until=until)) 

31 

32 

33def get_future_occurrences(from_date: date, rrule: str) -> Generator[date]: 

34 """ 

35 Yields the date of each future occurrence of an RRULE, starting on or after `from_date`. 

36 This may be unbounded depending on the RRULE. 

37 """ 

38 rule = rrulestr(rrule) 

39 from_datetime = datetime.combine(from_date, time.min) 

40 for occurrence in rule.xafter(from_datetime, inc=True): 

41 yield occurrence.date() 

42 

43 

44def schedule_occurrences( 

45 *, 

46 rrule: str, 

47 schedule_window: timedelta, 

48 min_occurrences: int, 

49 last_scheduled_date: date | None = None, 

50 today: date | None = None, 

51) -> list[date]: 

52 """ 

53 Determine which occurrence dates should be scheduled right now. 

54 

55 Takes every future occurrence within `schedule_window` of `today`, or at least 

56 `min_occurrences` occurrences if fewer fall in that window, then drops any on or before 

57 `last_scheduled_date` (already scheduled). If `last_scheduled_date` is None, nothing has 

58 been scheduled yet, so nothing is dropped. 

59 """ 

60 today = today if today is not None else now().date() 

61 window_end = today + schedule_window 

62 

63 occurrences: list[date] = [] 

64 for occurrence_date in get_future_occurrences(today, rrule): 

65 if len(occurrences) >= min_occurrences and occurrence_date > window_end: 

66 break 

67 occurrences.append(occurrence_date) 

68 

69 if last_scheduled_date is None: 

70 return occurrences 

71 return [occurrence_date for occurrence_date in occurrences if occurrence_date > last_scheduled_date]