NestQuest/custom_components/nestquest/core/recurrence.py

"""The ScheduleRule model: rule types, validation, and serialization.

A :class:`ScheduleRule` describes WHEN a recurring chore fires — a pure
frozen dataclass with no Home Assistant and no database imports.  The
evaluation engine (Feature 04) consumes it; the DAO layer (Feature 02)
serializes its dict form into the ``schedule_rules`` table.

Type model: one enum member per recurrence shape.  ``MONTHLY_DAY``
("the 15th of every month") and ``MONTHLY_WEEKDAY`` ("the second
Tuesday of every month") are distinct types rather than one monthly
type with optional fields, so validation is total: a rule either
constructs, or raises :class:`RuleValidationError` naming the offending
field.  Every type carries ONLY the fields its shape needs; carrying
any other shape's field is a construction error.

Storage mapping (Feature 02, ``schedule_rules`` table): the enum's
``storage_value`` maps model types onto the schema's CHECK values —
MONTHLY_DAY and MONTHLY_WEEKDAY both store as ``'monthly'`` and are
disambiguated by which fields are populated; CUSTOM_DAYS stores as
``'custom'``.  ``weekday_set`` round-trips as the sorted CSV list the
table stores (one element for MONTHLY_WEEKDAY, which names the
position's weekday); ``nth_weekday_weekday`` is a MODEL-ONLY field
carried through from_dict/to_dict for losslessness and folded into
``weekday_set`` by the DAO mapping.  ``to_dict``/``from_dict`` are the
model-side lossless pair; the DB mapping lives in the DAO layer and is
exercised there.

Validation policy (fail fast at construction — the schema's CHECK
constraints are a second line of defense, but invalid rules must never
reach storage):

- ``interval`` >= 1 for every rule.
- ``weekday_set`` is required for WEEKLY and CUSTOM_DAYS, forbidden
  elsewhere; entries are 0 (Monday) .. 6 (Sunday), no booleans.
- ``day_of_month`` 1..31 is required for MONTHLY_DAY.
- ``nth_weekday`` 1..5 or -1 (0 is meaningless as a position and is
  rejected) with a weekday 0..6 for MONTHLY_WEEKDAY.
- ``month`` 1..12 is required for YEARLY.
- ``start_date`` is a strict ISO date; ``end_date`` is optional and,
  when present, on or after ``start_date``.

Month-end policy (implemented in :func:`_occurs_monthly_day` and
documented for Feature 04): in each interval-eligible month, a
MONTHLY_DAY rule for a day that does not exist in that month — the
31st in April, the 30th or 29th in February, February 29th in a
non-leap year — fires on the LAST day of the month instead of being
skipped.  A monthly chore still happens in every eligible month,
pinned to the month's end when its nominal date is absent; exactly one
day per eligible month matches the rule (nominal day when it exists,
month end when it does not).  The interval anchors on nominal months
elapsed from start_date's month, so clamping never shifts the cadence.
"""
from __future__ import annotations

import datetime
import enum
from dataclasses import dataclass, field

#: Date-shape policy: strict YYYY-MM-DD, matching the DAO layers.
_DATE_FORMAT = "%Y-%m-%d"

#: Weekday-name policy: 0 = Monday .. 6 = Sunday (ISO order).


class RuleType(enum.Enum):
    """The six recurrence shapes the engine supports.

    ``storage_value`` is the string written to the ``schedule_rules``
    table's ``rule_type`` column (its CHECK whitelist).  MONTHLY_DAY
    and MONTHLY_WEEKDAY are distinct shapes sharing one storage value,
    disambiguated by which fields are populated.
    """

    DAILY = "daily"
    WEEKLY = "weekly"
    MONTHLY_DAY = "monthly_day"
    MONTHLY_WEEKDAY = "monthly_weekday"
    YEARLY = "yearly"
    CUSTOM_DAYS = "custom_days"

    @property
    def storage_value(self) -> str:
        """The schema-level rule_type string this model type maps to."""
        return {
            RuleType.DAILY: "daily",
            RuleType.WEEKLY: "weekly",
            RuleType.MONTHLY_DAY: "monthly",
            RuleType.MONTHLY_WEEKDAY: "monthly",
            RuleType.YEARLY: "yearly",
            RuleType.CUSTOM_DAYS: "custom",
        }[self]


class RuleValidationError(ValueError):
    """Raised when a ScheduleRule field combination is invalid."""


def _validate_date(value: str, field_name: str) -> None:
    """Raise unless ``value`` is a strict YYYY-MM-DD calendar date."""
    try:
        parsed = datetime.datetime.strptime(value, _DATE_FORMAT).date()
    except (TypeError, ValueError):
        raise RuleValidationError(
            f"{field_name} must be an ISO date (YYYY-MM-DD), got {value!r}"
        ) from None
    if parsed.isoformat() != value:
        raise RuleValidationError(
            f"{field_name} must be a strict YYYY-MM-DD date, got {value!r}"
        )


def _validate_weekday_set(weekday_set) -> frozenset[int]:
    """Return a normalized frozenset or raise with a clear message.

    Sequence entries are validated BEFORE set() coercion: a list like
    [0, False] must not collapse the boolean into 0, and unhashable
    entries must raise RuleValidationError, never a raw TypeError.
    """
    if isinstance(weekday_set, (list, tuple)):
        for entry in weekday_set:
            # _is_plain_int: bool and int subclasses (potentially
            # unhashable) are rejected before set() coercion.
            if not _is_plain_int(entry):
                raise RuleValidationError(
                    f"weekday_set entries must be plain integers, got "
                    f"{entry!r}"
                )
        weekday_set = set(weekday_set)
    if not isinstance(weekday_set, (set, frozenset)):
        raise RuleValidationError(
            f"weekday_set must be a set of weekday integers, got "
            f"{type(weekday_set).__name__}"
        )
    if not weekday_set:
        raise RuleValidationError(
            "weekday_set must not be empty for weekly/custom rules"
        )
    for entry in weekday_set:
        if not _is_plain_int(entry):
            raise RuleValidationError(
                f"weekday_set entries must be plain integers, got "
                f"{entry!r}"
            )
        if not 0 <= entry <= 6:
            raise RuleValidationError(
                f"weekday_set entries must be 0 (Monday)..6 (Sunday), "
                f"got {entry!r}"
            )
    return frozenset(weekday_set)


def _is_plain_int(value) -> bool:
    """True only for real ints: bool excluded AND int subclasses too
    (an unhashable int subclass would break the frozen-hashable model)."""
    return type(value) is int


def _validate_monthly_weekday(nth_weekday, weekday) -> None:
    """Validate the MONTHLY_WEEKDAY position fields."""
    if nth_weekday is None or not _is_plain_int(nth_weekday):
        raise RuleValidationError(
            "nth_weekday must be an integer for monthly_weekday rules, "
            f"got {nth_weekday!r}"
        )
    if not (-1 <= nth_weekday <= 5) or nth_weekday == 0:
        raise RuleValidationError(
            f"nth_weekday must be -1 (last) or 1..5, got {nth_weekday!r}"
        )
    if weekday is None or not _is_plain_int(weekday) or not 0 <= weekday <= 6:
        raise RuleValidationError(
            f"monthly_weekday rules require a weekday 0 (Monday)..6 "
            f"(Sunday), got {weekday!r}"
        )





@dataclass(frozen=True)
class ScheduleRule:
    """An immutable, validated recurrence rule (see module docstring).

    Construct with keyword arguments; invalid combinations raise
    :class:`RuleValidationError` at construction time.  Fields map
    one-to-one onto the ``schedule_rules`` table columns, with
    ``nth_weekday_weekday`` as the model-only weekday companion to
    ``nth_weekday`` (serialized inside ``weekday_set`` at the DAO
    boundary).
    """

    rule_type: RuleType
    interval: int = 1
    weekday_set: frozenset[int] | None = None
    day_of_month: int | None = None
    nth_weekday: int | None = None
    nth_weekday_weekday: int | None = None
    month: int | None = None
    start_date: str = "1970-01-01"
    end_date: str | None = None

    def __post_init__(self) -> None:
        # Normalize a string rule_type into the enum; anything else
        # that is not already a RuleType is rejected.
        if isinstance(self.rule_type, str):
            # EXACT model names only: no whitespace stripping, no storage
            # aliases.  The DAO maps storage strings to model fields when
            # reading rows; the model boundary stays unambiguous.
            candidates = {rt.name.lower(): rt for rt in RuleType}
            if self.rule_type in candidates:
                object.__setattr__(
                    self, "rule_type", candidates[self.rule_type]
                )
            else:
                raise RuleValidationError(
                    f"rule_type must be a RuleType or one of "
                    f"{sorted(candidates)}, got {self.rule_type!r}"
                )
        if not isinstance(self.rule_type, RuleType):
            raise RuleValidationError(
                f"rule_type must be a RuleType or one of its string "
                f"values, got {self.rule_type!r}"
            )
        if not _is_plain_int(self.interval):
            raise RuleValidationError(
                f"interval must be an integer >= 1, got {self.interval!r}"
            )
        if self.interval < 1:
            raise RuleValidationError(
                f"interval must be >= 1, got {self.interval!r}"
            )

        _validate_date(self.start_date, "start_date")
        if self.end_date is not None:
            _validate_date(self.end_date, "end_date")
            if self.end_date < self.start_date:
                raise RuleValidationError(
                    f"end_date {self.end_date!r} must be on or after "
                    f"start_date {self.start_date!r}"
                )

        # Type-exclusive field policy, enforced for every type: build
        # the dict of present fields, then reject every field the shape
        # must not carry (weekday_set present is detectable via None vs
        # set even after normalization below).
        shape = self.rule_type
        if shape is RuleType.DAILY:
            forbidden_fields = (
                "weekday_set",
                "day_of_month",
                "nth_weekday",
                "nth_weekday_weekday",
                "month",
            )
        elif shape is RuleType.WEEKLY:
            forbidden_fields = (
                "day_of_month",
                "nth_weekday",
                "nth_weekday_weekday",
                "month",
            )
        elif shape is RuleType.CUSTOM_DAYS:
            forbidden_fields = (
                "day_of_month",
                "nth_weekday",
                "nth_weekday_weekday",
                "month",
            )
        elif shape is RuleType.MONTHLY_DAY:
            forbidden_fields = ("weekday_set", "nth_weekday",
                                "nth_weekday_weekday", "month")
        elif shape is RuleType.MONTHLY_WEEKDAY:
            forbidden_fields = ("weekday_set", "day_of_month", "month")
        elif shape is RuleType.YEARLY:
            forbidden_fields = (
                "weekday_set",
                "nth_weekday",
                "nth_weekday_weekday",
            )
        present_forbidden = [
            name
            for name in forbidden_fields
            if getattr(self, name) is not None
        ]
        if present_forbidden:
            raise RuleValidationError(
                f"{shape.value} rules must not set "
                f"{', '.join(sorted(present_forbidden))}"
            )

        # Required fields per shape.
        if shape is RuleType.WEEKLY or shape is RuleType.CUSTOM_DAYS:
            normalized = _validate_weekday_set(self.weekday_set)
            object.__setattr__(self, "weekday_set", normalized)
        elif shape is RuleType.MONTHLY_DAY:
            if self.day_of_month is None or not _is_plain_int(
                self.day_of_month
            ):
                raise RuleValidationError(
                    "monthly_day rules require day_of_month (1..31)"
                )
            if not 1 <= self.day_of_month <= 31:
                raise RuleValidationError(
                    f"day_of_month must be 1..31, got {self.day_of_month!r}"
                )
        elif shape is RuleType.MONTHLY_WEEKDAY:
            _validate_monthly_weekday(
                self.nth_weekday, self.nth_weekday_weekday
            )
        elif shape is RuleType.YEARLY:
            if self.month is None or not _is_plain_int(self.month):
                raise RuleValidationError(
                    "yearly rules require month (1..12)"
                )
            if not 1 <= self.month <= 12:
                raise RuleValidationError(
                    f"month must be 1..12, got {self.month!r}"
                )
            if self.day_of_month is not None and not (
                _is_plain_int(self.day_of_month)
                and 1 <= self.day_of_month <= 31
            ):
                raise RuleValidationError(
                    f"day_of_month must be 1..31 for yearly rules, got "
                    f"{self.day_of_month!r}"
                )

    def to_dict(self) -> dict:
        """Return the lossless dict form (model-side storage shape).

        ``weekday_set`` serializes as a sorted list; MONTHLY_WEEKDAY's
        weekday travels inside it (appended) plus
        ``nth_weekday_weekday`` so the pair round-trips without
        ambiguity at the model layer.
        """
        return {
            "rule_type": self.rule_type.name.lower(),  # model name
            "interval": self.interval,
            "weekday_set": (
                sorted(self.weekday_set) if self.weekday_set is not None
                else None
            ),
            "day_of_month": self.day_of_month,
            "nth_weekday": self.nth_weekday,
            "nth_weekday_weekday": self.nth_weekday_weekday,
            "month": self.month,
            "start_date": self.start_date,
            "end_date": self.end_date,
        }

    @classmethod
    def from_dict(cls, data: dict) -> "ScheduleRule":
        """Rebuild a rule from :meth:`to_dict` output.

        Raises :class:`RuleValidationError` on unknown keys, missing
        keys, or invalid combinations — never a raw TypeError.
        """
        if not isinstance(data, dict):
            raise RuleValidationError(
                f"rule data must be a dict, got {type(data).__name__}"
            )
        allowed = {
            "rule_type",
            "interval",
            "weekday_set",
            "day_of_month",
            "nth_weekday",
            "nth_weekday_weekday",
            "month",
            "start_date",
            "end_date",
        }
        unknown = [key for key in data if key not in allowed]
        if unknown:
            raise RuleValidationError(
                f"unknown rule fields: {sorted(unknown, key=repr)}"
            )
        if "rule_type" not in data:
            raise RuleValidationError("missing required rule field: rule_type")
        weekday_set = data.get("weekday_set")
        if isinstance(weekday_set, (list, tuple)):
            # Validate entries as plain ints BEFORE set() coercion, so a
            # list like [0, False] cannot collapse False into 0 and an
            # unhashable/int-subclass entry cannot leak a raw TypeError.
            for entry in weekday_set:
                if not _is_plain_int(entry):
                    raise RuleValidationError(
                        f"weekday_set entries must be plain integers, got "
                        f"{entry!r}"
                    )
            weekday_set = set(weekday_set)
        return cls(
            rule_type=data["rule_type"],
            interval=data.get("interval", 1),
            weekday_set=weekday_set,
            day_of_month=data.get("day_of_month"),
            nth_weekday=data.get("nth_weekday"),
            nth_weekday_weekday=data.get("nth_weekday_weekday"),
            month=data.get("month"),
            start_date=data.get("start_date", "1970-01-01"),
            end_date=data.get("end_date"),
        )


# ---------------------------------------------------------------------------
# Evaluation engine (Feature 04): pure date arithmetic on ScheduleRule
# ---------------------------------------------------------------------------


def _parse_date(value: str) -> datetime.date:
    """Parse a strict ISO date already validated by the model."""
    return datetime.datetime.strptime(value, _DATE_FORMAT).date()


def _anchor_offset(rule: ScheduleRule, target: datetime.date) -> int:
    """Day offset from the rule's anchor date to ``target``.

    The anchor for DAILY/WEEKLY-interval arithmetic is start_date: the
    first day the rule CAN fire.  A negative offset means the target
    precedes the rule's window entirely.
    """
    return (target - _parse_date(rule.start_date)).days


def _within_window(rule: ScheduleRule, target: datetime.date) -> bool:
    """True when ``target`` is inside the rule's [start, end] window."""
    if target < _parse_date(rule.start_date):
        return False
    if rule.end_date is not None and target > _parse_date(rule.end_date):
        return False
    return True


def _month_end(year: int, month: int) -> int:
    """The last calendar day of the given month (leap-aware)."""
    if month == 12:
        return 31
    return (datetime.date(year, month + 1, 1)
            - datetime.timedelta(days=1)).day


def _occurs_monthly_day(rule: ScheduleRule, target: datetime.date) -> bool:
    """MONTHLY_DAY evaluation with the documented month-end policy.

    Policy: in each interval-eligible month, a rule for a day-of-month
    that does not exist in that month (the 31st in April, the 30th in
    February, the 29th of February in a non-leap year) fires on the
    LAST day of that month instead of being skipped — the chore still
    happens that month, pinned to the month's end when its nominal date
    is absent.

    The interval anchors on months ELAPSED from start_date's month: the
    rule fires when the months between the anchor month and the target's
    month form a multiple of the interval, keeping "every 2 months on
    the 15th" aligned to the anchor month even when the firing lands on
    a clamped month-end day.
    """
    anchor = _parse_date(rule.start_date)
    months_elapsed = (
        (target.year - anchor.year) * 12 + (target.month - anchor.month)
    )
    if months_elapsed % rule.interval != 0:
        return False
    month_end = _month_end(target.year, target.month)
    return target.day == min(rule.day_of_month, month_end)


def _nth_weekday_of_month(
    year: int, month: int, weekday: int, nth: int
) -> int | None:
    """The day-of-month of the ``nth`` ``weekday``, or None when the
    month has no such occurrence (``nth`` 5 beyond four occurrences).

    Positive ``nth`` counts forward (1 = first); ``-1`` counts from the
    end (last occurrence, whatever the total).  Pure calendar math.
    """
    if nth == -1:
        last = datetime.date(year, month, _month_end(year, month))
        back = (last.weekday() - weekday) % 7
        return last.day - back
    first = datetime.date(year, month, 1)
    first_offset = (weekday - first.weekday()) % 7
    day = 1 + first_offset + (nth - 1) * 7
    if day > _month_end(year, month):
        return None
    return day


def _occurs_monthly_weekday(
    rule: ScheduleRule, target: datetime.date
) -> bool:
    """MONTHLY_WEEKDAY evaluation.

    Fires when the target is the ``nth`` occurrence of
    ``nth_weekday_weekday`` in its month (1..4 forward, -1 last).  A
    month with fewer occurrences than ``nth`` (5 on a four-occurrence
    month) does not fire — n 5 is accepted by the model but skips such
    months, matching the done-condition.  The interval anchors on
    nominal months elapsed from start_date's month.
    """
    anchor = _parse_date(rule.start_date)
    months_elapsed = (
        (target.year - anchor.year) * 12 + (target.month - anchor.month)
    )
    if months_elapsed % rule.interval != 0:
        return False
    if target.weekday() != rule.nth_weekday_weekday:
        return False
    nth_day = _nth_weekday_of_month(
        target.year, target.month, rule.nth_weekday_weekday, rule.nth_weekday
    )
    return nth_day is not None and target.day == nth_day


def _occurs_yearly(rule: ScheduleRule, target: datetime.date) -> bool:
    """YEARLY evaluation with the documented leap-day clamp policy.

    Policy: in each interval-eligible year, the rule fires on
    ``day_of_month`` of ``month`` when that day exists.  A YEARLY rule
    for February 29th fires on FEBRUARY 28TH in a non-leap year (the
    same clamp-as-month-end policy as MONTHLY_DAY: the annual chore
    still happens every year, pinned to February's end when its
    nominal date is absent) rather than silently vanishing.  A YEARLY
    rule with NO day_of_month fires on the first day of ``month``.
    The interval anchors on years elapsed from start_date's year.
    """
    anchor = _parse_date(rule.start_date)
    years_elapsed = target.year - anchor.year
    if years_elapsed % rule.interval != 0:
        return False
    if target.month != rule.month:
        return False
    if rule.day_of_month is None:
        # Day-less yearly rules fire on the first of the month.
        return target.day == 1
    month_end = _month_end(target.year, target.month)
    return target.day == min(rule.day_of_month, month_end)


def occurs_on(rule: ScheduleRule, target: datetime.date) -> bool:
    """Return True when ``rule`` fires on ``target``.

    Pure function: no I/O, no timezone awareness, no mutation.  Dates
    are plain calendar dates (Feature 04 guardrail); the caller decides
    the timezone that produced them.
    """
    # Dispatch FIRST: an unimplemented shape must raise loudly even for
    # dates outside the window, never silently evaluate to False.
    shape = rule.rule_type
    if shape not in (
        RuleType.DAILY, RuleType.WEEKLY, RuleType.CUSTOM_DAYS,
        RuleType.MONTHLY_DAY, RuleType.MONTHLY_WEEKDAY, RuleType.YEARLY,
    ):
        raise NotImplementedError(
            f"occurs_on for {shape} lands with its own engine task "
            "(Features 04 task sequence)"
        )
    if not _within_window(rule, target):
        return False
    if shape is RuleType.DAILY:
        offset = (target - _parse_date(rule.start_date)).days
        return offset % rule.interval == 0
    if shape is RuleType.MONTHLY_DAY:
        return _occurs_monthly_day(rule, target)
    if shape is RuleType.MONTHLY_WEEKDAY:
        return _occurs_monthly_weekday(rule, target)
    if shape is RuleType.YEARLY:
        return _occurs_yearly(rule, target)
    # WEEKLY / CUSTOM_DAYS: whole-week anchoring from start_date, never
    # ISO week numbers (Feature 04/05 guardrail: 53-week years flip ISO
    # parity on January 1st; anchor arithmetic does not).
    anchor = _parse_date(rule.start_date)
    if shape is RuleType.CUSTOM_DAYS:
        interval = 1
    else:
        interval = rule.interval
    week_offset = (target - anchor).days // 7
    if week_offset % interval != 0:
        return False
    return target.weekday() in rule.weekday_set


def occurrences_between(
    rule: ScheduleRule, start: str, end: str
) -> list[str]:
    """Return every date string in [start, end] the rule fires on.

    Inclusive on both ends.  Raises RuleValidationError on malformed or
    inverted bounds; a window entirely outside [rule.start_date,
    rule.end_date] yields an empty list.
    """
    if rule.rule_type not in (
        RuleType.DAILY, RuleType.WEEKLY, RuleType.CUSTOM_DAYS,
        RuleType.MONTHLY_DAY, RuleType.MONTHLY_WEEKDAY, RuleType.YEARLY,
    ):
        raise NotImplementedError(
            f"occurrences_between for {rule.rule_type} lands with its own "
            "engine task (Features 04 task sequence)"
        )
    _validate_date(start, "start")
    _validate_date(end, "end")
    start_date = _parse_date(start)
    end_date = _parse_date(end)
    if end_date < start_date:
        raise RuleValidationError(
            f"end {end!r} must be on or after start {start!r}"
        )
    rule_start = _parse_date(rule.start_date)
    if end_date < rule_start:
        return []
    # Clamp the walk to the rule's own window and the queried range.
    effective_start = max(start_date, rule_start)
    effective_end = end_date
    if rule.end_date is not None:
        effective_end = min(effective_end, _parse_date(rule.end_date))
    results: list[str] = []
    cursor = effective_start
    while cursor <= effective_end:
        if occurs_on(rule, cursor):
            results.append(cursor.isoformat())
        cursor += datetime.timedelta(days=1)
    return results