"""Recurrence engine for calendar entries — daily/weekly/monthly/yearly + custom + exceptions.""" from __future__ import annotations from datetime import date, datetime, timedelta from typing import Any WEEKDAYS = ["MO", "TU", "WE", "TH", "FR", "SA", "SU"] def _parse_date(val: Any) -> date | None: """Parse a value into a date object. Accepts date, datetime, or ISO string.""" if val is None: return None if isinstance(val, date) and not isinstance(val, datetime): return val if isinstance(val, datetime): return val.date() if isinstance(val, str): try: return datetime.fromisoformat(val).date() except ValueError: try: return date.fromisoformat(val) except ValueError: return None return None def _parse_datetime(val: Any) -> datetime | None: """Parse a value into a datetime object.""" if val is None: return None if isinstance(val, datetime): return val if isinstance(val, date) and not isinstance(val, datetime): return datetime(val.year, val.month, val.day) if isinstance(val, str): try: return datetime.fromisoformat(val) except ValueError: return None return None def generate_occurrences( recurrence: dict[str, Any], base_start: datetime | None, range_start: date, range_end: date, ) -> list[datetime]: """Generate occurrence start datetimes within [range_start, range_end]. Args: recurrence: {pattern, custom_rule, end_date, exceptions} base_start: the original start_at datetime of the entry range_start: query range start date range_end: query range end date Returns: List of datetime occurrences within the range, excluding exception dates. """ if not recurrence or not base_start: return [base_start] if base_start and range_start <= base_start.date() <= range_end else [] pattern = recurrence.get("pattern", "daily") custom_rule = recurrence.get("custom_rule") end_date = _parse_date(recurrence.get("end_date")) exceptions_raw = recurrence.get("exceptions", []) exception_dates = {_parse_date(d) for d in exceptions_raw if _parse_date(d) is not None} # Max 2 years forward max_end = base_start.date() + timedelta(days=730) effective_end = range_end if end_date and end_date < effective_end: effective_end = end_date if effective_end > max_end: effective_end = max_end if effective_end < range_start: return [] occurrences: list[datetime] = [] if pattern == "daily": interval = 1 if custom_rule and "INTERVAL=" in custom_rule: interval = _extract_int(custom_rule, "INTERVAL") current = base_start while current.date() <= effective_end: if range_start <= current.date() <= range_end and current.date() not in exception_dates: occurrences.append(current) current = current + timedelta(days=interval) elif pattern == "weekly": interval = 1 bydays: list[str] | None = None if custom_rule: interval = _extract_int(custom_rule, "INTERVAL") bydays = _extract_bydays(custom_rule) if bydays is None: bydays = [WEEKDAYS[base_start.weekday()]] weekday_nums = {d: i for i, d in enumerate(WEEKDAYS)} current_week_start = base_start - timedelta(days=base_start.weekday()) while current_week_start.date() <= effective_end: for wd in bydays: wd_num = weekday_nums.get(wd) if wd_num is None: continue occ_date = current_week_start + timedelta(days=wd_num) if occ_date.date() > effective_end: continue if occ_date.date() < base_start.date(): continue if ( range_start <= occ_date.date() <= range_end and occ_date.date() not in exception_dates ): occurrences.append(occ_date) current_week_start = current_week_start + timedelta(weeks=interval) elif pattern == "monthly": interval = 1 if custom_rule and "INTERVAL=" in custom_rule: interval = _extract_int(custom_rule, "INTERVAL") bysetpos: list[int] | None = None if custom_rule and "BYSETPOS=" in custom_rule: bysetpos = _extract_setpos(custom_rule) current = base_start month_offset = 0 while True: year = base_start.year + (base_start.month - 1 + month_offset) // 12 month = (base_start.month - 1 + month_offset) % 12 + 1 if date(year, month, 1) > effective_end: break if bysetpos: for pos in bysetpos: occ_date = _nth_weekday_of_month(year, month, pos, base_start.weekday()) if occ_date is None: continue occ_dt = datetime( occ_date.year, occ_date.month, occ_date.day, base_start.hour, base_start.minute, ) if occ_dt.date() < base_start.date(): continue if ( range_start <= occ_dt.date() <= range_end and occ_dt.date() not in exception_dates ): occurrences.append(occ_dt) else: day = min(base_start.day, _days_in_month(year, month)) occ_date = date(year, month, day) if ( occ_date >= base_start.date() and range_start <= occ_date <= range_end and occ_date not in exception_dates ): occurrences.append( datetime( occ_date.year, occ_date.month, occ_date.day, base_start.hour, base_start.minute, ) ) month_offset += interval elif pattern == "yearly": interval = 1 if custom_rule and "INTERVAL=" in custom_rule: interval = _extract_int(custom_rule, "INTERVAL") current = base_start year_offset = 0 while True: year = base_start.year + year_offset if date(year, base_start.month, 1) > effective_end: break day = min(base_start.day, _days_in_month(year, base_start.month)) occ_date = date(year, base_start.month, day) if ( occ_date >= base_start.date() and range_start <= occ_date <= range_end and occ_date not in exception_dates ): occurrences.append( datetime( occ_date.year, occ_date.month, occ_date.day, base_start.hour, base_start.minute, ) ) year_offset += interval elif pattern == "custom": if custom_rule: interval = _extract_int(custom_rule, "INTERVAL") bydays = _extract_bydays(custom_rule) if bydays: weekday_nums = {d: i for i, d in enumerate(WEEKDAYS)} current_week_start = base_start - timedelta(days=base_start.weekday()) while current_week_start.date() <= effective_end: for wd in bydays: wd_num = weekday_nums.get(wd) if wd_num is None: continue occ_date = current_week_start + timedelta(days=wd_num) if occ_date.date() > effective_end or occ_date.date() < base_start.date(): continue if ( range_start <= occ_date.date() <= range_end and occ_date.date() not in exception_dates ): occurrences.append(occ_date) current_week_start = current_week_start + timedelta(weeks=interval) occurrences.sort() return occurrences def _extract_int(rule: str, key: str) -> int: """Extract an integer value from a rule string like 'INTERVAL=2'.""" parts = rule.split(";") for part in parts: if part.startswith(f"{key}="): try: return int(part.split("=")[1]) except (ValueError, IndexError): return 1 return 1 def _extract_bydays(rule: str) -> list[str] | None: """Extract BYDAY values from a rule string like 'BYDAY=MO,WE,FR'.""" parts = rule.split(";") for part in parts: if part.startswith("BYDAY="): days = part.split("=")[1].split(",") return [d.strip().upper() for d in days if d.strip()] return None def _extract_setpos(rule: str) -> list[int] | None: """Extract BYSETPOS values from a rule string like 'BYSETPOS=2'.""" parts = rule.split(";") for part in parts: if part.startswith("BYSETPOS="): try: return [int(x) for x in part.split("=")[1].split(",")] except (ValueError, IndexError): return None return None def _days_in_month(year: int, month: int) -> int: """Return number of days in a given month.""" if month == 12: return 31 next_month = date(year, month + 1, 1) return (next_month - timedelta(days=1)).day def _nth_weekday_of_month(year: int, month: int, n: int, weekday: int) -> date | None: """Get the nth occurrence of a weekday in a month. n=1 → first, n=2 → second, n=-1 → last, n=-2 → second-to-last. """ if n > 0: first = date(year, month, 1) first_offset = (weekday - first.weekday()) % 7 day = 1 + first_offset + (n - 1) * 7 try: return date(year, month, day) except ValueError: return None else: last_day = _days_in_month(year, month) last = date(year, month, last_day) last_offset = (last.weekday() - weekday) % 7 day = last_day - last_offset + (n + 1) * 7 try: return date(year, month, day) except ValueError: return None