HomeAssistantVS/custom_components/maintenance_supporter/helpers/dates.py

"""Calendar-aware interval arithmetic (no external dependencies).

Used by the scheduling model so an interval can be expressed in days, weeks,
months, or years. Month/year steps are calendar-aware and clamp the day to the
target month's last day (e.g. Jan 31 + 1 month -> Feb 28/29), avoiding the
drift you get from approximating "monthly" as 30 days.
"""

from __future__ import annotations

import calendar
from collections.abc import Iterator
from datetime import UTC, date, datetime, time, timedelta
from typing import Any

from homeassistant.util import dt as dt_util

from ..const import TIME_HHMMSS_PATTERN
from .workday import is_business_day

INTERVAL_UNITS = ("days", "weeks", "months", "years")


def parse_hhmm(value: Any) -> time | None:
    """``"HH:MM"`` or ``"HH:MM:SS"`` (HA's TimeSelector) → ``time`` to the
    minute; ``None`` when absent or malformed (hours 0–23, minutes 0–59).

    The one parser behind the status sensor, the calendar entity and the
    model's sub-day overdue refinement — three hand-copied ``split(":")``
    blocks before — and behind :func:`normalize_hhmm`.
    """
    if not isinstance(value, str):
        return None
    text = value.strip()
    if not TIME_HHMMSS_PATTERN.match(text):
        return None
    hours, minutes = text.split(":")[:2]
    return time(int(hours), int(minutes))


def local_date_from_iso(value: Any) -> str | None:
    """The calendar date (``YYYY-MM-DD``) of an ISO timestamp *in HA's local
    zone*. ``value[:10]`` is right for stamps written locally but off by one
    around midnight for imported ``+00:00`` stamps; this is the one place
    that does it properly. ``None`` when the value is not a timestamp."""
    if not isinstance(value, str) or not value:
        return None
    try:
        parsed = datetime.fromisoformat(value)
    except ValueError:
        return value[:10] if len(value) >= 10 and value[4] == "-" else None
    if parsed.tzinfo is None:
        return parsed.date().isoformat()
    return dt_util.as_local(parsed).date().isoformat()


def normalize_hhmm(value: Any) -> str | None:
    """Canonical ``"HH:MM"`` for a stored time, dropping TimeSelector seconds;
    ``None`` when the value does not parse (the caller drops the field)."""
    parsed = parse_hhmm(value)
    return f"{parsed.hour:02d}:{parsed.minute:02d}" if parsed is not None else None


def parse_iso_date(value: str | None) -> date | None:
    """ISO date string → ``date``; ``None`` when absent or malformed."""
    if not value:
        return None
    try:
        return date.fromisoformat(value)
    except (ValueError, TypeError):
        return None


def parse_persisted_utc(value: str | None) -> datetime | None:
    """Persisted ISO timestamp → TZ-aware ``datetime``; ``None`` when absent
    or malformed.

    Live writes use ``dt_util.utcnow().isoformat()`` (TZ-aware), but older
    payloads may be naive — those are assumed UTC, because subtracting a naive
    datetime from ``dt_util.utcnow()`` raises ``TypeError``. One home for the
    block the threshold and state-change triggers each hand-rolled (and the
    runtime trigger FORGOT — a naive ``on_since`` crashed its elapsed math).
    """
    if not value:
        return None
    try:
        parsed = datetime.fromisoformat(value)
    except (ValueError, TypeError):
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=UTC)
    return parsed


def _add_months(anchor: date, months: int) -> date:
    """Advance ``anchor`` by ``months`` calendar months, clamping the day."""
    total = anchor.month - 1 + months
    year = anchor.year + total // 12
    month = total % 12 + 1
    last_day = calendar.monthrange(year, month)[1]
    return date(year, month, min(anchor.day, last_day))


def add_interval(anchor: date, n: int, unit: str = "days") -> date:
    """Return ``anchor`` advanced by ``n`` of ``unit``.

    ``unit`` is one of days / weeks / months / years; anything else (or None)
    falls back to days, which keeps existing day-based tasks working unchanged.
    """
    if unit == "weeks":
        return anchor + timedelta(weeks=n)
    if unit == "months":
        return _add_months(anchor, n)
    if unit == "years":
        return _add_months(anchor, n * 12)
    return anchor + timedelta(days=n)


def try_add_interval(anchor: date, n: int, unit: str = "days") -> date | None:
    """:func:`add_interval`, or ``None`` when the result would lie past the
    calendar's end (year 9999).

    ``add_interval`` raises OverflowError (day/week steps) or ValueError
    (month/year steps) there. A far-future reset date or an absurd interval
    reached it from inside the coordinator refresh and took the whole object
    down on every refresh and every restart (bug audit 2026-09-27) — the
    schedule reads such a date as "no due date" instead.
    """
    try:
        return add_interval(anchor, n, unit)
    except (OverflowError, ValueError):
        return None


def nth_weekday_of_month(year: int, month: int, nth: int, weekday: int) -> date | None:
    """The ``nth`` ``weekday`` of ``(year, month)``; ``None`` if it doesn't exist.

    ``weekday`` is 0=Mon … 6=Sun (``date.weekday()``). ``nth`` is 1..5, or -1 for
    the last occurrence. A 5th occurrence that the month doesn't have → ``None``.
    """
    last_day = calendar.monthrange(year, month)[1]
    if nth == -1:
        anchor = date(year, month, last_day)
        return anchor - timedelta(days=(anchor.weekday() - weekday) % 7)
    first = date(year, month, 1)
    day = 1 + (weekday - first.weekday()) % 7 + (nth - 1) * 7
    if day > last_day:
        return None
    return date(year, month, day)


def next_weekday_in_set(ref: date, weekdays: tuple[int, ...], *, inclusive: bool) -> date | None:
    """Next date on/after ``ref`` whose weekday is in ``weekdays`` (0=Mon…6=Sun).

    ``inclusive`` includes ``ref`` itself; otherwise the search starts the next
    day. ``None`` when ``weekdays`` is empty.
    """
    if not weekdays:
        return None
    start = ref if inclusive else ref + timedelta(days=1)
    for offset in range(7):
        candidate = start + timedelta(days=offset)
        if candidate.weekday() in weekdays:
            return candidate
    return None  # pragma: no cover  (unreachable: a non-empty weekday set always matches within 7 days)


def _iter_months(year: int, month: int, limit: int = 60) -> Iterator[tuple[int, int]]:
    """Yield ``(year, month)`` forward from the given month, ``limit`` times."""
    for _ in range(limit):
        yield year, month
        month += 1
        if month > 12:
            month = 1
            year += 1


def next_nth_weekday(
    ref: date,
    nth: int,
    weekday: int,
    months: tuple[int, ...] | None = None,
    *,
    inclusive: bool,
) -> date | None:
    """Next ``nth``-``weekday``-of-month occurrence on/after ``ref``.

    Optionally restricted to ``months`` (1=Jan…12=Dec). ``None`` if no occurrence
    is found within a bounded horizon (e.g. a 5th weekday restricted to a month
    that never has one).
    """
    for year, month in _iter_months(ref.year, ref.month):
        if months and month not in months:
            continue
        occ = nth_weekday_of_month(year, month, nth, weekday)
        if occ is not None and (occ >= ref if inclusive else occ > ref):
            return occ
    return None


def next_day_of_month(
    ref: date,
    day: int,
    months: tuple[int, ...] | None = None,
    *,
    inclusive: bool,
) -> date | None:
    """Next ``day``-of-month occurrence on/after ``ref`` (day clamped to month).

    ``day`` 1..31 is clamped to the month's length (e.g. 31 → Feb 28/29);
    ``day == -1`` means the LAST day of the month (#83). Optionally restricted
    to ``months``.
    """
    for year, month in _iter_months(ref.year, ref.month):
        if months and month not in months:
            continue
        month_len = calendar.monthrange(year, month)[1]
        clamped = month_len if day == -1 else min(day, month_len)
        candidate = date(year, month, clamped)
        if candidate >= ref if inclusive else candidate > ref:
            return candidate
    return None


def roll_back_to_business_day(d: date) -> date:
    """Roll a date back to the preceding business day (business days unchanged).

    Used by the day-of-month schedule's ``business`` flag (#83): "last business
    day of the month" = last day rolled back past non-business days. What
    counts as a business day comes from :mod:`.workday`: plain Mon-Fri by
    default, or the user's Workday integration configuration (public holidays,
    custom working weekdays) when one is set up.

    Bounded: a pathological provider that never yields a business day within
    two weeks returns *d* unchanged rather than walking off into the past.
    """
    candidate = d
    for _ in range(14):
        if is_business_day(candidate):
            return candidate
        candidate -= timedelta(days=1)
    return d


def interval_span_days(n: int | None, unit: str = "days") -> int:
    """Length of one ``n``-``unit`` interval in days, calendar-aware.

    Measured as the distance from a fixed reference date to that date advanced
    by the interval, so months/years reflect real calendar lengths instead of
    the raw count. Used so things like the warning window aren't capped by the
    bare interval *count* for month/year tasks (issue #58/#59 — a 6-month task
    must not collapse a 14-day warning to ``min(14, 6)``). Returns 0 when there
    is no positive interval.
    """
    if not n or n <= 0:
        return 0
    ref = date(2001, 1, 1)  # common (non-leap) year; representative span
    end = try_add_interval(ref, n, unit)
    # An interval reaching past year 9999 (read-path garbage — the schedule
    # clamps what it stores) spans "everything left", never a crash in the
    # warning-window math of every refresh (bug audit 2026-09-27).
    return ((end or date.max) - ref).days