recurring_ical_events.py

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.
"""Calculate repetitions of icalendar components."""

from __future__ import annotations

import contextlib
import datetime
import re
from abc import ABC, abstractmethod
from collections import defaultdict
from functools import wraps
from typing import TYPE_CHECKING, Callable, Generator, Optional, Sequence, Union

import x_wr_timezone
from dateutil.rrule import rruleset, rrulestr
from icalendar.prop import vDDDTypes

if TYPE_CHECKING:
    from icalendar.cal import Component

    Time = Union[datetime.date, datetime.datetime]
    DateArgument = tuple[int] | Time | str | int
    from dateutil.rrule import rrule

    UID = str
    ComponentID = tuple[str, UID, Time]
    Timestamp = float
    RecurrenceID = datetime.datetime
    RecurrenceIDs = tuple[RecurrenceID]


# The minimum value accepted as date (pytz + zoneinfo)
DATE_MIN = (1970, 1, 1)
DATE_MIN_DT = datetime.date(*DATE_MIN)
# The maximum value accepted as date (pytz + zoneinfo)
DATE_MAX = (2038, 1, 1)
DATE_MAX_DT = datetime.date(*DATE_MAX)


class InvalidCalendar(ValueError):
    """Exception thrown for bad icalendar content."""

    def __init__(self, message: str):
        """Create a new error with a message."""
        self._message = message
        super().__init__(self.message)

    @property
    def message(self) -> str:
        """The error message."""
        return self._message


class PeriodEndBeforeStart(InvalidCalendar):
    """An event or component starts before it ends."""

    def __init__(self, message: str, start: Time, end: Time):
        """Create a new PeriodEndBeforeStart error."""
        super().__init__(message)
        self._start = start
        self._end = end

    @property
    def start(self) -> Time:
        """The start of the component's period."""
        return self._start

    @property
    def end(self) -> Time:
        """The end of the component's period."""
        return self._end


class BadRuleStringFormat(InvalidCalendar):
    """An iCal rule string is badly formatted."""

    def __init__(self, message: str, rule: str):
        """Create an error with a bad rule string."""
        super().__init__(message + ": " + rule)
        self._rule = rule

    @property
    def rule(self) -> str:
        """The malformed rule string"""
        return self._rule


def timestamp(dt: datetime.datetime) -> Timestamp:
    """Return the time stamp of a datetime"""
    return dt.timestamp()


NEGATIVE_RRULE_COUNT_REGEX = re.compile(r"COUNT=-\d+;?")


def convert_to_date(date: Time) -> datetime.date:
    """Converts a date or datetime to a date"""
    return datetime.date(date.year, date.month, date.day)


def convert_to_datetime(date: Time, tzinfo: Optional[datetime.tzinfo]):  # noqa: UP007
    """Converts a date to a datetime.

    Dates are converted to datetimes with tzinfo.
    Datetimes loose their timezone if tzinfo is None.
    Datetimes receive tzinfo as a timezone if they do not have a timezone.
    Datetimes retain their timezone if they have one already (tzinfo is not None).
    """
    if isinstance(date, datetime.datetime):
        if date.tzinfo is None:
            if tzinfo is not None:
                if is_pytz(tzinfo):
                    return tzinfo.localize(date)
                return date.replace(tzinfo=tzinfo)
        elif tzinfo is None:
            return normalize_pytz(date).replace(tzinfo=None)
        return date
    if isinstance(date, datetime.date):
        return datetime.datetime(date.year, date.month, date.day, tzinfo=tzinfo)
    return date


def time_span_contains_event(
    span_start: Time,
    span_stop: Time,
    event_start: Time,
    event_stop: Time,
    comparable: bool = False,  # noqa: FBT001
):
    """Return whether an event should is included within a time span.

    - span_start and span_stop define the time span
    - event_start and event_stop define the event time
    - comparable indicates whether the dates can be compared.
        You can set it to True if you are sure you have timezones and
        date/datetime correctly or used make_comparable() before.

    Note that the stops are exlusive but the starts are inclusive.

    This is an essential function of the module. It should be tested in
    test/test_time_span_contains_event.py.

    This raises a PeriodEndBeforeStart exception if a start is after an end.
    """
    if not comparable:
        span_start, span_stop, event_start, event_stop = make_comparable(
            (span_start, span_stop, event_start, event_stop)
        )
    if event_start > event_stop:
        raise PeriodEndBeforeStart(
            (
                "the event must start before it ends"
                f"(start: {event_start} end: {event_stop})"
            ),
            event_start,
            event_stop,
        )
    if span_start > span_stop:
        raise PeriodEndBeforeStart(
            (
                "the time span must start before it ends"
                f"(start: {span_start} end: {span_stop})"
            ),
            span_start,
            span_stop,
        )
    if event_start == event_stop:
        if span_start == span_stop:
            return event_start == span_start
        return span_start <= event_start < span_stop
    if span_start == span_stop:
        return event_start <= span_start < event_stop
    return event_start < span_stop and span_start < event_stop


def make_comparable(dates: Sequence[Time]) -> list[Time]:
    """Make an list or tuple of dates comparable.

    Returns an list.
    """
    tzinfo = None
    for date in dates:
        tzinfo = getattr(date, "tzinfo", None)
        if tzinfo is not None:
            break
    return [convert_to_datetime(date, tzinfo) for date in dates]


def compare_greater(date1: Time, date2: Time) -> bool:
    """Compare two dates if date1 > date2 and make them comparable before."""
    date1, date2 = make_comparable((date1, date2))
    return date1 > date2


def cmp(date1: Time, date2: Time) -> int:
    """Compare two dates, like cmp().

    Returns
    -------
        -1 if date1 < date2
         0 if date1 = date2
         1 if date1 > date2

    """
    # credits: https://www.geeksforgeeks.org/python-cmp-function/
    # see https://stackoverflow.com/a/22490617/1320237
    date1, date2 = make_comparable((date1, date2))
    return (date1 > date2) - (date1 < date2)


def is_pytz(tzinfo: datetime.tzinfo | Time):
    """Whether the time zone requires localize() and normalize().

    pytz requires these funtions to be used in order to correctly use the
    time zones after operations.
    """
    return hasattr(tzinfo, "localize")


def is_pytz_dt(time: Time):
    """Whether the time requires localize() and normalize().

    pytz requires these funtions to be used in order to correctly use the
    time zones after operations.
    """
    return isinstance(time, datetime.datetime) and is_pytz(time.tzinfo)


def normalize_pytz(time: Time):
    """We have to normalize the time after a calculation if we use pytz."""
    if is_pytz_dt(time):
        return time.tzinfo.normalize(time)
    return time


def cached_property(func: Callable):
    """Cache the property value for speed up."""
    name = f"_cached_{func.__name__}"
    not_found = object()

    @property
    @wraps(func)
    def cached_property(self: object):
        value = self.__dict__.get(name, not_found)
        if value is not_found:
            self.__dict__[name] = value = func(self)
        return value

    return cached_property


def to_recurrence_ids(time: Time) -> RecurrenceIDs:
    """Convert the time to a recurrence id so it can be hashed and recognized.

    The first value should be used to identify a component as it is a datetime in UTC.
    The other values can be used to look the component up.
    """
    # We are inside the Series calculation with this and want to identify
    # a date. It is fair to assume that the timezones are the same now.
    if not isinstance(time, datetime.datetime):
        return (convert_to_datetime(time, None),)
    if time.tzinfo is None:
        return (time,)
    return (
        time.astimezone(datetime.timezone.utc).replace(tzinfo=None),
        time.replace(tzinfo=None),
    )


def is_date(time: Time):
    """Whether this is a date and not a datetime."""
    return isinstance(time, datetime.date) and not isinstance(time, datetime.datetime)


def with_highest_sequence(
    adapter1: ComponentAdapter | None, adapter2: ComponentAdapter | None
):
    """Return the one with the highest sequence."""
    return max(
        adapter1,
        adapter2,
        key=lambda adapter: -1e10 if adapter is None else adapter.sequence,
    )


def get_any(dictionary: dict, keys: Sequence[object], default: object = None):
    """Get any item from the keys and return it."""
    result = default
    for key in keys:
        result = dictionary.get(key, result)
    return result


class Series:
    """Base class for components that result in a series of occurrences."""

    def __init__(self, components: Sequence[ComponentAdapter]):
        """Create an component which may have repetitions in it."""
        if len(components) == 0:
            raise ValueError("No components given to calculate a series.")
        # We identify recurrences with a timestamp as all recurrence values
        # should be the same in UTC either way and we want to omit
        # inequality because of timezone implementation mismatches.
        self.recurrence_id_to_modification: dict[
            RecurrenceID, ComponentAdapter
        ] = {}  # RECURRENCE-ID -> adapter
        core: ComponentAdapter | None = None
        for component in components:
            if component.is_modification():
                for recurrence_id in component.recurrence_ids:
                    self.recurrence_id_to_modification[recurrence_id] = (
                        with_highest_sequence(
                            self.recurrence_id_to_modification.get(recurrence_id),
                            component,
                        )
                    )
            else:
                core = with_highest_sequence(core, component)
        self.modifications: set[ComponentAdapter] = set(
            self.recurrence_id_to_modification.values()
        )
        del component
        if core is None:
            raise InvalidCalendar(
                f"The event definition for {components[0].uid} "
                "only contains modifications."
            )
        self.core = core
        # Setup complete. We create the attribtues
        self.start = self.original_start = self.core.start
        self.end = self.original_end = self.core.end
        self.exdates: set[Time] = set()
        self.check_exdates_datetime: set[RecurrenceID] = set()  # should be in UTC
        self.check_exdates_date: set[datetime.date] = set()  # should be in UTC
        self.rdates: set[Time] = set()
        self.replace_ends: dict[RecurrenceID, Time] = {}  # for periods, in UTC
        # fill the attributes
        for exdate in self.core.exdates:
            self.exdates.add(exdate)
            self.check_exdates_datetime.update(to_recurrence_ids(exdate))
            if is_date(exdate):
                self.check_exdates_date.add(exdate)
        for rdate in self.core.rdates:
            if isinstance(rdate, tuple):
                # we have a period as rdate
                self.rdates.add(rdate[0])
                for recurrence_id in to_recurrence_ids(rdate[0]):
                    self.replace_ends[recurrence_id] = rdate[1]
            else:
                # we have a date/datetime
                self.rdates.add(rdate)

        # We make sure that all dates and times mentioned here are either:
        # - a date
        # - a datetime with None is tzinfo
        # - a datetime with a timezone
        self.make_all_dates_comparable()

        # Calculate the rules with the same timezones
        self.rule_set = rruleset(cache=True)
        self.rrules = []
        last_until: Time | None = None
        for rrule_string in self.core.rrules:
            rule = self.create_rule_with_start(rrule_string)
            self.rrules.append(rule)
            if rule.until and (
                not last_until or compare_greater(rule.until, last_until)
            ):
                last_until = rule.until

        for exdate in self.exdates:
            self.check_exdates_datetime.add(exdate)
        for rdate in self.rdates:
            self.rule_set.rdate(rdate)

        if not last_until or not compare_greater(self.start, last_until):
            self.rule_set.rdate(self.start)

    def create_rule_with_start(self, rule_string) -> rrule:
        """Helper to create an rrule from a rule_string

        The rrule is starting at the start of the component.
        Since the creation is a bit more complex, this function handles special cases.
        """
        try:
            return self.rrulestr(rule_string)
        except ValueError:
            # string: FREQ=WEEKLY;UNTIL=20191023;BYDAY=TH;WKST=SU
            # start: 2019-08-01 14:00:00+01:00
            # ValueError: RRULE UNTIL values must be specified in UTC
            # when DTSTART is timezone-aware
            rule_list = rule_string.split(";UNTIL=")
            if len(rule_list) != 2:
                raise BadRuleStringFormat(
                    "UNTIL parameter is missing", rule_string
                ) from None
            date_end_index = rule_list[1].find(";")
            if date_end_index == -1:
                date_end_index = len(rule_list[1])
            until_string = rule_list[1][:date_end_index]
            if self.is_all_dates:
                until_string = until_string[:8]
            elif self.tzinfo is None:
                # remove the Z from the time zone
                until_string = until_string[:-1]
            else:
                # we assume the start is timezone aware but the until value
                # is not, see the comment above
                if len(until_string) == 8:
                    until_string += "T000000"
                if len(until_string) != 15:
                    raise BadRuleStringFormat(
                        "UNTIL parameter has a bad format", rule_string
                    ) from None
                until_string += "Z"  # https://stackoverflow.com/a/49991809
            new_rule_string = (
                rule_list[0] + rule_list[1][date_end_index:] + ";UNTIL=" + until_string
            )
            return self.rrulestr(new_rule_string)

    def rrulestr(self, rule_string) -> rrule:
        """Return an rrulestr with a start. This might fail."""
        rule_string = NEGATIVE_RRULE_COUNT_REGEX.sub("", rule_string)  # Issue 128
        rule = rrulestr(rule_string, dtstart=self.start, cache=True)
        rule.string = rule_string
        rule.until = until = self._get_rrule_until(rule)
        if is_pytz(self.start.tzinfo) and rule.until:
            # when starting in a time zone that is one hour off to the end,
            # we might miss the last occurrence
            # see issue 107 and test/test_issue_107_omitting_last_event.py
            rule = rule.replace(until=rule.until + datetime.timedelta(hours=1))
            rule.until = until
        return rule

    def _get_rrule_until(self, rrule) -> None | Time:
        """Return the UNTIL datetime of the rrule or None if absent."""
        rule_list = rrule.string.split(";UNTIL=")
        if len(rule_list) == 1:
            return None
        if len(rule_list) != 2:
            raise BadRuleStringFormat("There should be only one UNTIL", rrule)
        date_end_index = rule_list[1].find(";")
        if date_end_index == -1:
            date_end_index = len(rule_list[1])
        until_string = rule_list[1][:date_end_index]
        return vDDDTypes.from_ical(until_string)

    def make_all_dates_comparable(self):
        """Make sure we can use all dates with eachother.

        Dates may be mixed and we have many of them.
        - date
        - datetime without timezone
        - datetime with timezone
        These three are not comparable but can be converted.
        """
        self.tzinfo = None
        dates = [self.start, self.end, *self.exdates, *self.rdates]
        self.is_all_dates = not any(
            isinstance(date, datetime.datetime) for date in dates
        )
        for date in dates:
            if isinstance(date, datetime.datetime) and date.tzinfo is not None:
                self.tzinfo = date.tzinfo
                break
        self.start = convert_to_datetime(self.start, self.tzinfo)

        self.end = convert_to_datetime(self.end, self.tzinfo)
        self.rdates = {convert_to_datetime(rdate, self.tzinfo) for rdate in self.rdates}
        self.exdates = {
            convert_to_datetime(exdate, self.tzinfo) for exdate in self.exdates
        }

    def rrule_between(self, span_start: Time, span_stop: Time) -> Generator[Time]:
        """Recalculate the rrules so that minor mistakes are corrected."""
        yield from self.rule_set
        for rule in self.rrules:
            for start in rule.between(span_start, span_stop, inc=True):
                if is_pytz_dt(start):
                    # update the time zone in case of summer/winter time change
                    start = start.tzinfo.localize(start.replace(tzinfo=None))  # noqa: PLW2901
                # We could now well be out of bounce of the end of the UNTIL
                # value. This is tested by test/test_issue_20_exdate_ignored.py.
                if rule.until is None or not compare_greater(start, rule.until):
                    yield start

    def between(self, span_start: Time, span_stop: Time) -> Generator[Occurrence]:
        """components between the start (inclusive) and end (exclusive)"""
        # make dates comparable, rrule converts them to datetimes
        span_start_dt = convert_to_datetime(span_start, self.tzinfo)
        span_stop_dt = convert_to_datetime(span_stop, self.tzinfo)
        if compare_greater(span_start_dt, self.start):
            # do not exclude an component if it spans across the time span
            span_start_dt -= self.core.duration
        # we have to account for pytz timezones not being properly calculated
        # at the timezone changes. This is a heuristic: most changes are only 1 hour.
        # This will still create problems at the fringes of timezone definition changes.
        if is_pytz(self.tzinfo):
            span_start_dt = normalize_pytz(span_start_dt - datetime.timedelta(hours=1))
            span_stop_dt = normalize_pytz(span_stop_dt + datetime.timedelta(hours=1))
        returned_starts: set[Time] = set()
        returned_modifications: set[ComponentAdapter] = set()
        # NOTE: If in the following line, we get an error, datetime and date
        # may still be mixed because RDATE, EXDATE, start and rule.
        for start in self.rrule_between(span_start_dt, span_stop_dt):
            recurrence_ids = to_recurrence_ids(start)
            if (
                start in returned_starts
                or convert_to_date(start) in self.check_exdates_date
                or self.check_exdates_datetime & set(recurrence_ids)
            ):
                continue
            adapter: ComponentAdapter = get_any(
                self.recurrence_id_to_modification, recurrence_ids, self.core
            )
            if adapter is self.core:
                stop = get_any(
                    self.replace_ends,
                    recurrence_ids,
                    normalize_pytz(start + self.core.duration),
                )
                occurrence = Occurrence(
                    self.core,
                    self.convert_to_original_type(start),
                    self.convert_to_original_type(stop),
                )
                returned_starts.add(start)
            else:
                # We found a modification
                if adapter in returned_modifications:
                    continue
                returned_modifications.add(adapter)
                occurrence = Occurrence(adapter)
            if occurrence.is_in_span(span_start, span_stop):
                yield occurrence
        for modification in self.modifications:
            # we assume that the modifications are actually included
            if (
                modification in returned_modifications
                or self.check_exdates_datetime & set(modification.recurrence_ids)
            ):
                continue
            if modification.is_in_span(span_start, span_stop):
                returned_modifications.add(modification)
                yield Occurrence(modification)

    def convert_to_original_type(self, date):
        """Convert a date back if this is possible.

        Dates may get converted to datetimes to make calculations possible.
        This reverts the process where possible so that Repetitions end
        up with the type (date/datetime) that was specified in the icalendar
        component.
        """
        if not isinstance(self.original_start, datetime.datetime) and not isinstance(
            self.original_end,
            datetime.datetime,
        ):
            return convert_to_date(date)
        return date

    @property
    def uid(self):
        """The UID that identifies this series."""
        return getattr(self.core, "uid", "invalid")

    def __repr__(self):
        """A string representation."""
        return (
            f"<{self.__class__.__name__} uid={self.uid} "
            f"modifications:{len(self.recurrence_id_to_modification)}>"
        )


class ComponentAdapter(ABC):
    """A unified interface to work with icalendar components."""

    ATTRIBUTES_TO_DELETE_ON_COPY = ["RRULE", "RDATE", "EXDATE"]

    @staticmethod
    @abstractmethod
    def component_name() -> str:
        """The icalendar component name."""

    def __init__(self, component: Component):
        """Create a new adapter."""
        self._component = component

    @property
    def end_property(self) -> str | None:
        """The name of the end property."""
        return None

    @property
    @abstractmethod
    def start(self) -> Time:
        """The start time."""

    @property
    @abstractmethod
    def end(self) -> Time:
        """The end time."""

    @property
    def uid(self) -> UID:
        """The UID of a component.

        UID is required by RFC5545.
        If the UID is absent, we use the Python ID.
        """
        return self._component.get("UID", str(id(self._component)))

    @classmethod
    def collect_components(
        cls, source: Component, suppress_errors: tuple[Exception]
    ) -> Sequence[Series]:
        """Collect all components from the source component.

        suppress_errors - a list of errors that should be suppressed.
            A Series of events with such an error is removed from all results.
        """
        components: dict[str, list[Component]] = defaultdict(list)  # UID -> components
        for component in source.walk(cls.component_name()):
            adapter = cls(component)
            components[adapter.uid].append(adapter)
        result = []
        for components in components.values():
            with contextlib.suppress(suppress_errors):
                result.append(Series(components))
        return result

    def as_component(self, start: Time, stop: Time, keep_recurrence_attributes: bool):  # noqa: FBT001
        """Create a shallow copy of the source event and modify some attributes."""
        copied_component = self._component.copy()
        copied_component["DTSTART"] = vDDDTypes(start)
        if self.end_property is not None:
            copied_component[self.end_property] = vDDDTypes(stop)
        if not keep_recurrence_attributes:
            for attribute in self.ATTRIBUTES_TO_DELETE_ON_COPY:
                if attribute in copied_component:
                    del copied_component[attribute]
        for subcomponent in self._component.subcomponents:
            copied_component.add_component(subcomponent)
        return copied_component

    @cached_property
    def recurrence_ids(self) -> RecurrenceIDs:
        """The recurrence ids of the component that might be used to identify it."""
        recurrence_id = self._component.get("RECURRENCE-ID")
        if recurrence_id is None:
            return ()
        return to_recurrence_ids(recurrence_id.dt)

    def is_modification(self) -> bool:
        """Whether the adapter is a modification."""
        return bool(self.recurrence_ids)

    @cached_property
    def sequence(self) -> int:
        """The sequence in the history of modification.

        The sequence is negative if none was found.
        """
        return self._component.get("SEQUENCE", -1)

    def __repr__(self) -> str:
        """Debug representation with more info."""
        return (
            f"<{self.__class__.__name__} UID={self.uid} start={self.start} "
            f"recurrence_ids={self.recurrence_ids} sequence={self.sequence} "
            f"end={self.end}>"
        )

    @cached_property
    def exdates(self) -> list[Time]:
        """A list of exdates."""
        result: list[Time] = []
        exdates = self._component.get("EXDATE", [])
        for exdates in (exdates,) if not isinstance(exdates, list) else exdates:
            result.extend(exdate.dt for exdate in exdates.dts)
        return result

    @cached_property
    def rrules(self) -> set[str]:
        """A list of rrules of this component."""
        rules = self._component.get("RRULE", None)
        if not rules:
            return set()
        return {
            rrule.to_ical().decode()
            for rrule in (rules if isinstance(rules, list) else [rules])
        }

    @cached_property
    def rdates(self) -> list[Time, tuple[Time, Time]]:
        """A list of rdates, possibly a period."""
        rdates = self._component.get("RDATE", [])
        result = []
        for rdates in (rdates,) if not isinstance(rdates, list) else rdates:
            result.extend(rdate.dt for rdate in rdates.dts)
        return result

    @cached_property
    def duration(self) -> datetime.timedelta:
        """The duration of the component."""
        return self.end - self.start

    def is_in_span(self, span_start: Time, span_stop: Time) -> bool:
        """Return whether the component is in the span."""
        return time_span_contains_event(span_start, span_stop, self.start, self.end)


class EventAdapter(ComponentAdapter):
    """An icalendar event adapter."""

    @staticmethod
    def component_name() -> str:
        """The icalendar component name."""
        return "VEVENT"

    @property
    def end_property(self) -> str:
        """DTEND"""
        return "DTEND"

    @cached_property
    def start(self) -> Time:
        """Return DTSTART"""
        # Arguably, it may be considered a feature that this breaks
        # if no DTSTART is set
        return self._component["DTSTART"].dt

    @cached_property
    def end(self) -> Time:
        """Yield DTEND or calculate the end of the event based on
        DTSTART and DURATION.
        """
        ## an even may have DTEND or DURATION, but not both
        end = self._component.get("DTEND")
        if end is not None:
            return end.dt
        duration = self._component.get("DURATION")
        if duration is not None:
            return normalize_pytz(self._component["DTSTART"].dt + duration.dt)
        return self._component["DTSTART"].dt


class TodoAdapter(ComponentAdapter):
    """Unified access to TODOs."""

    @staticmethod
    def component_name() -> str:
        """The icalendar component name."""
        return "VTODO"

    @property
    def end_property(self) -> str:
        """DUE"""
        return "DUE"

    @cached_property
    def start(self) -> Time:
        """Return DTSTART if it set, do not panic if it's not set."""
        ## easy case - DTSTART set
        start = self._component.get("DTSTART")
        if start is not None:
            return start.dt
        ## Tasks may have DUE set, but no DTSTART.
        ## Let's assume 0 duration and return the DUE
        due = self._component.get("DUE")
        if due is not None:
            return due.dt

        ## Assume infinite time span if neither is given
        ## (see the comments under _get_event_end)
        return DATE_MIN_DT

    @cached_property
    def end(self) -> Time:
        """Return DUE or DTSTART+DURATION or something"""
        ## Easy case - DUE is set
        end = self._component.get("DUE")
        if end is not None:
            return end.dt

        dtstart = self._component.get("DTSTART")

        ## DURATION can be specified instead of DUE.
        duration = self._component.get("DURATION")
        ## It is no requirement that DTSTART is set.
        ## Perhaps duration is a time estimate rather than an indirect
        ## way to set DUE.
        if duration is not None and dtstart is not None:
            return normalize_pytz(self._component["DTSTART"].dt + duration.dt)

        ## According to the RFC, a VEVENT without an end/duration
        ## is to be considered to have zero duration.  Assuming the
        ## same applies to VTODO.
        if dtstart:
            return dtstart.dt

        ## The RFC says this about VTODO:
        ## > A "VTODO" calendar component without the "DTSTART" and "DUE" (or
        ## > "DURATION") properties specifies a to-do that will be associated
        ## > with each successive calendar date, until it is completed.
        ## It can be interpreted in different ways, though probably it may
        ## be considered equivalent with a DTSTART in the infinite past and DUE
        ## in the infinite future?
        return DATE_MAX_DT


class JournalAdapter(ComponentAdapter):
    """Apdater for journal entries."""

    @staticmethod
    def component_name() -> str:
        """The icalendar component name."""
        return "VJOURNAL"

    @property
    def end_property(self) -> str:
        """There is no end property"""
        return None

    @cached_property
    def start(self) -> Time:
        """Return DTSTART if it set, do not panic if it's not set."""
        ## according to the specification, DTSTART in a VJOURNAL is optional
        dtstart = self._component.get("DTSTART")
        if dtstart is not None:
            return dtstart.dt
        return DATE_MIN_DT

    @cached_property
    def end(self) -> Time:
        """The end time is the same as the start."""
        ## VJOURNAL cannot have a DTEND.  We should consider a VJOURNAL to
        ## describe one day if DTSTART is a date, and we can probably
        ## consider it to have zero duration if a timestamp is given.
        return self.start


class Occurrence:
    """A repetition of an event."""

    ATTRIBUTES_TO_DELETE_ON_COPY = ["RRULE", "RDATE", "EXDATE"]

    def __init__(
        self,
        adapter: ComponentAdapter,
        start: Time | None = None,
        end: Time | None = None,
    ):
        """Create an event repetition.

        - source - the icalendar Event
        - start - the start date/datetime to replace
        - stop - the end date/datetime to replace
        """
        self._adapter = adapter
        self.start = adapter.start if start is None else start
        self.end = adapter.end if end is None else end

    def as_component(self, keep_recurrence_attributes: bool) -> Component:  # noqa: FBT001
        """Create a shallow copy of the source component and modify some attributes."""
        return self._adapter.as_component(
            self.start, self.end, keep_recurrence_attributes
        )

    def is_in_span(self, span_start: Time, span_stop: Time) -> bool:
        """Return whether the component is in the span."""
        return time_span_contains_event(span_start, span_stop, self.start, self.end)

    def __lt__(self, other: Occurrence) -> bool:
        """Compare two occurrences for sorting.

        See https://stackoverflow.com/a/4010558/1320237
        """
        self_start, other_start = make_comparable((self.start, other.start))
        return self_start < other_start

    @cached_property
    def id(self) -> ComponentID:
        """The id of the component."""
        recurrence_id = ((*self._adapter.recurrence_ids, self.start))[0]
        return (
            self._adapter.component_name(),
            self._adapter.uid,
            recurrence_id,
        )

    def __hash__(self) -> int:
        """Hash this for an occurrence."""
        return hash(self.id)

    def __eq__(self, other: Occurrence) -> bool:
        """self == other"""
        return self.id == other.id


class CalendarQuery:
    """A calendar that can unfold its events at a certain time.

    Functions like at(), between() and after() can be used to query the
    selected components. If any malformed icalendar information is found,
    an InvalidCalendar exception is raised. For other bad arguments, you
    should expect a ValueError.

    suppressed_errors - a list of errors to suppress when
        skip_bad_series is True

    component_adapters - a list of component adapters
    """

    component_adapters = [EventAdapter, TodoAdapter, JournalAdapter]

    @cached_property
    def _component_adapters(self) -> dict[str : type[ComponentAdapter]]:
        """A mapping of component adapters."""
        return {
            adapter.component_name(): adapter for adapter in self.component_adapters
        }

    suppressed_errors = [BadRuleStringFormat, PeriodEndBeforeStart]

    def __init__(
        self,
        calendar: Component,
        keep_recurrence_attributes: bool = False,  # noqa: FBT001
        components: Sequence[str | type[ComponentAdapter]] = ("VEVENT",),
        skip_bad_series: bool = False,  # noqa: FBT001
    ):
        """Create an unfoldable calendar from a given calendar.

        calendar - an icalendar component - probably a calendar -
            from which occurrences will be calculated
        keep_recurrence_attributes - whether to keep values
            in the results that are used for calculation
        skip_bad_events - whether to skip a series of components that
            contains errors. This skips self.suppressed_errors.
        """
        self.keep_recurrence_attributes = keep_recurrence_attributes
        if calendar.get("CALSCALE", "GREGORIAN") != "GREGORIAN":
            # https://www.kanzaki.com/docs/ical/calscale.html
            raise InvalidCalendar("Only Gregorian calendars are supported.")

        self.series: list[Series] = []  # component
        self._skip_errors = tuple(self.suppressed_errors) if skip_bad_series else ()
        for component_adapter_id in components:
            if isinstance(component_adapter_id, str):
                if component_adapter_id not in self._component_adapters:
                    raise ValueError(
                        f'"{component_adapter_id}" is an unknown name for a '
                        'recurring component. '
                        f"I only know these: { ', '.join(self._component_adapters)}."
                    )
                component_adapter = self._component_adapters[component_adapter_id]
            else:
                component_adapter = component_adapter_id
            self.series.extend(
                component_adapter.collect_components(calendar, self._skip_errors)
            )

    @staticmethod
    def to_datetime(date: DateArgument):
        """Convert date inputs of various sorts into a datetime object."""
        if isinstance(date, int):
            date = (date,)
        if isinstance(date, tuple):
            date += (1,) * (3 - len(date))
            return datetime.datetime(*date)  # noqa: DTZ001
        if isinstance(date, str):
            # see https://docs.python.org/2/library/datetime.html#strftime-strptime-behavior
            if len(date) == 8:
                return datetime.datetime.strptime(date, "%Y%m%d")  # noqa: DTZ007
            return datetime.datetime.strptime(date, "%Y%m%dT%H%M%SZ")  # noqa: DTZ007
        return date

    def all(self) -> Generator[Component]:
        """Generate all Components.

        The Components are sorted from the first to the last Occurrence.
        Calendars can contain millions of Occurrences. This iterates
        safely across all of them
        """
        # MAX and MIN values may change in the future
        return self.after(DATE_MIN_DT)

    _DELTAS = [
        datetime.timedelta(days=1),
        datetime.timedelta(hours=1),
        datetime.timedelta(minutes=1),
        datetime.timedelta(seconds=1),
    ]

    def at(self, date: DateArgument):
        """Return all events within the next 24 hours of starting at the given day.

        - date can be a year like (2019,) or 2099
        - a month like (2019, 1) for January of 2019
        - a day like (2019, 1, 1) for the first of January 2019
        - a day with hours, (2019, 1, 1, 1)
        - a day with minutes, (2019, 1, 1, 1, 1)
        - a day with seconds, (2019, 1, 1, 1, 1, 1)
        """
        if isinstance(date, int):
            date = (date,)
        if isinstance(date, str):
            if len(date) != 8 or not date.isdigit():
                raise ValueError(f"Format yyyymmdd expected for {date!r}.")
            date = (int(date[:4], 10), int(date[4:6], 10), int(date[6:]))
        if isinstance(date, datetime.datetime):
            return self.between(date, date)
        if isinstance(date, datetime.date):
            return self.between(date, date + datetime.timedelta(days=1))
        if len(date) == 1:
            return self.between((date[0], 1, 1), (date[0] + 1, 1, 1))
        if len(date) == 2:
            year, month = date
            if month == 12:
                return self.between((year, 12, 1), (year + 1, 1, 1))
            return self.between((year, month, 1), (year, month + 1, 1))
        dt = self.to_datetime(date)
        return self._between(dt, dt + self._DELTAS[len(date) - 3])

    def between(self, start: DateArgument, stop: DateArgument | datetime.timedelta):
        """Return events at a time between start (inclusive) and end (inclusive)"""
        start = self.to_datetime(start)
        stop = (
            start + stop
            if isinstance(stop, datetime.timedelta)
            else self.to_datetime(stop)
        )
        return self._between(start, stop)

    def _occurrences_to_components(
        self, occurrences: list[Occurrence]
    ) -> list[Component]:
        """Map occurrences to components."""
        return [
            occurrence.as_component(self.keep_recurrence_attributes)
            for occurrence in occurrences
        ]

    def _between(self, start: Time, end: Time) -> list[Component]:
        """Return the occurrences between the start and the end."""
        return self._occurrences_to_components(self._occurrences_between(start, end))

    def _occurrences_between(self, start: Time, end: Time) -> list[Occurrence]:
        """Return the components between the start and the end."""
        occurrences: list[Occurrence] = []
        for series in self.series:
            with contextlib.suppress(self._skip_errors):
                occurrences.extend(series.between(start, end))
        return occurrences

    def after(self, earliest_end: DateArgument) -> Generator[Component]:
        """
        Return an iterator over the next events that happen during or after
        the earliest_end.
        """
        time_span = datetime.timedelta(days=1)
        min_time_span = datetime.timedelta(minutes=15)
        earliest_end = self.to_datetime(earliest_end)
        done = False
        result_ids: set[ComponentID] = set()

        while not done:
            try:
                next_end = earliest_end + time_span
            except OverflowError:
                # We ran to the end
                next_end = DATE_MAX_DT
                if compare_greater(earliest_end, next_end):
                    return  # we might run too far
                done = True
            occurrences = self._occurrences_between(earliest_end, next_end)
            occurrences.sort()
            for occurrence in occurrences:
                if occurrence.id not in result_ids:
                    yield occurrence.as_component(self.keep_recurrence_attributes)
                    result_ids.add(occurrence.id)
            # prepare next query
            time_span = max(
                time_span / 2 if occurrences else time_span * 2,
                min_time_span,
            )  # binary search to improve speed
            earliest_end = next_end

    def count(self) -> int:
        """Return the amount of recurring components in this calendar."""
        i = 0
        for _ in self.all():
            i += 1
        return i


def of(
    a_calendar: Component,
    keep_recurrence_attributes=False,
    components: Sequence[str | type[ComponentAdapter]] = ("VEVENT",),
    skip_bad_series: bool = False,  # noqa: FBT001
) -> CalendarQuery:
    """Unfold recurring events of a_calendar

    - a_calendar is an icalendar VCALENDAR component or something like that.
    - keep_recurrence_attributes - whether to keep attributes that are only used
      to calculate the recurrence.
    - components is a list of component type names of which the recurrences
      should be returned.
    """
    a_calendar = x_wr_timezone.to_standard(a_calendar)
    return CalendarQuery(
        a_calendar, keep_recurrence_attributes, components, skip_bad_series
    )


__all__ = [
    "of",
    "InvalidCalendar",
    "PeriodEndBeforeStart",
    "BadRuleStringFormat",
    "is_pytz",
    "DATE_MIN",
    "DATE_MAX",
    "CalendarQuery",
    "ComponentAdapter",
    "EventAdapter",
    "TodoAdapter",
    "JournalAdapter",
    "Time",
    "RecurrenceID",
    "RecurrenceIDs",
]