Source code for schedium.triggers.sugar.weekly

from __future__ import annotations

from datetime import time

from schedium.triggers.base import BaseTrigger
from schedium.triggers.on import On
from schedium.triggers.sugar._time import parse_at
from schedium.triggers.sugar.tick import Tick
from schedium.types.granularity import Granularity

_WEEKDAY_ALIASES_TO_ISO = {
    "mon": 1,
    "tue": 2,
    "wed": 3,
    "thu": 4,
    "fri": 5,
    "sat": 6,
    "sun": 7,
}


def _parse_weekday_to_on_value(day: str | int) -> int:
    if isinstance(day, int):
        if 1 <= day <= 7:
            return day
        raise ValueError("weekday integer must be 1..7 (Mon..Sun)")

    key = day.strip()[:3].lower()

    if key in _WEEKDAY_ALIASES_TO_ISO:
        return _WEEKDAY_ALIASES_TO_ISO[key]

    raise ValueError(
        "weekday must be one of: mon/tue/wed/thu/fri/sat/sun (or full names), or 1..7"
    )




def Weekly(
    day: str | int, *, at: str | time | None = None, force_0_minute: bool = False
) -> BaseTrigger:
    """
    Convenience trigger: run weekly on a specific weekday, optionally at a time.

    Parameters
    ----------
    day : str | int
        Weekday to run on. Can be a string like "mon"/"monday" (case-insensitive,
        only first 3 letters are considered) or an integer in ISO format where
        Monday=1 and Sunday=7.
    at : str | datetime.time, optional
        Time of day to run at. If not provided, it will not constrain the time.
        Can be a :class:`datetime.time` object, or a string in "HH:MM" or
        "HH:MM:SS[.mmm]" format.
    force_0_minute : bool, default False
        By default, if `at` is provided without a minute component (e.g. "09:00" or
        `time(9, 0)`), the trigger does not constrain the minute (i.e., it can run
        any minute during the 9 o'clock hour). Setting `force_0_minute=True` makes
        it so that the minute is constrained to 0, meaning it will only run at the top
        of the hour (e.g. 09:00:37).

    Notes
    -----
    This helper composes triggers to express a weekly schedule.

    - `Tick(Granularity.WEEK)` provides a WEEK "bucket" for deduplication.
    - `On(day_of_week=...)` and, if `at` is provided, hour/minute
        (and optionally second/millisecond).

    Using `Tick(WEEK)` (instead of `Every(unit="week", interval=1)`) avoids forcing
    alignment to week boundaries (e.g. Monday 00:00). That alignment can make
    `next_window()` for AND-combinations like "Monday at 09:30" converge
    poorly by repeatedly jumping to week boundaries.

    Examples
    --------
    >>> Weekly("monday")

    Specify a time to run at

    >>> Weekly("mon", at="09:30")
    >>> from datetime import time
    >>> Weekly("thursday", at=time(9, 30))

    Don't run if overdue by more than a minute (e.g., scheduler was down)

    >>> Weekly("thursday", at=time(9, 0), force_0_minute=True)
    """

    weekday_value = _parse_weekday_to_on_value(day)

    trigger: BaseTrigger = Tick(Granularity.WEEK) & On(
        unit="day_of_week", value=weekday_value
    )

    if at is None:
        return trigger

    hour, minute, second, millisecond = parse_at(at)

    trigger = trigger & On(unit="hour_of_day", value=hour)
    if minute != 0 or force_0_minute:
        trigger = trigger & On(unit="minute_of_hour", value=minute)

    if second is not None:
        trigger = trigger & On(unit="second_of_minute", value=second)
    if millisecond is not None:
        trigger = trigger & On(unit="millisecond_of_second", value=millisecond)

    return trigger