291 lines
10 KiB
Python
291 lines
10 KiB
Python
|
"""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
|