nanobot-runtime/skills/remind/scripts/random_times.py

"""Deterministic random fire-time computation for reminders.

Shared by remind_send.py (runtime) and remind_cli.py (validation). Stdlib only,
so it imports cleanly regardless of the caller's uv/PEP 723 environment.

A reminder's `random` block produces `times_per_day` fire times inside a daily
`window`, spaced at least MIN_GAP_MIN apart. The times are random but fully
determined by (date, text): any script computing them for the same day gets the
same result, so no state needs to be persisted.
"""

from __future__ import annotations

import random
from datetime import date, datetime, time, timedelta

MIN_GAP_MIN = 15  # minimum gap between fire times in minutes; tune here
DAYS_PER_WEEK = 7


def compute_fire_times(target_date: date, text: str, cfg: dict) -> list[datetime]:
    """Deterministic fire times for one day.

    With period 'day' (default) the count is per day; with 'week' it is per week,
    spread across distinct days. Returns [] when the day falls outside the
    days/from/until filters. Raises ValueError on a malformed config (bad window,
    days, dates, or an infeasible count) — these are structural and validated
    before any date filter, so the same call validates a config regardless of the
    date passed in.
    """
    if cfg.get("period", "day") == "week":
        return _weekly_fire_times(target_date, text, cfg)

    count = _parse_count(cfg.get("times_per_day"))
    start, end = parse_window(cfg.get("window"))
    day_set = _parse_days(cfg["days"]) if cfg.get("days") is not None else None
    from_date = _parse_date(cfg["from"]) if cfg.get("from") is not None else None
    until_date = _parse_date(cfg["until"]) if cfg.get("until") is not None else None

    total = end - start
    required = (count - 1) * MIN_GAP_MIN
    if required > total:
        raise ValueError(
            f"{count} times with a {MIN_GAP_MIN}-min gap need {required} min, "
            f"but the window is only {total} min wide"
        )

    if from_date is not None and target_date < from_date:
        return []
    if until_date is not None and target_date > until_date:
        return []
    if day_set is not None and _cron_weekday(target_date) not in day_set:
        return []

    slack = total - required
    rnd = random.Random(f"{target_date.isoformat()}|{text}")
    offsets = sorted(rnd.randint(0, slack) for _ in range(count))
    minutes = [start + offset + index * MIN_GAP_MIN for index, offset in enumerate(offsets)]
    return [datetime.combine(target_date, _minute_to_time(m)) for m in minutes]


def _weekly_fire_times(target_date: date, text: str, cfg: dict) -> list[datetime]:
    """Deterministic fire times for target_date within a weekly schedule.

    Picks `count` distinct days (Mon–Sun week) eligible under the days/from/until
    filters, one random time per chosen day inside the window. Seeded by the
    week, not the day, so every day of the same week computes the identical plan
    and this returns only the slice landing on target_date.
    """
    count = _parse_count(cfg.get("times_per_day"))
    start, end = parse_window(cfg.get("window"))
    day_set = _parse_days(cfg["days"]) if cfg.get("days") is not None else None
    from_date = _parse_date(cfg["from"]) if cfg.get("from") is not None else None
    until_date = _parse_date(cfg["until"]) if cfg.get("until") is not None else None

    week_capacity = len(day_set) if day_set is not None else DAYS_PER_WEEK
    if count > week_capacity:
        raise ValueError(
            f"{count} times per week need {count} eligible days, but only {week_capacity} match the filter"
        )

    week_start = target_date - timedelta(days=target_date.weekday())
    eligible = [
        day
        for offset in range(DAYS_PER_WEEK)
        for day in [week_start + timedelta(days=offset)]
        if (from_date is None or day >= from_date)
        and (until_date is None or day <= until_date)
        and (day_set is None or _cron_weekday(day) in day_set)
    ]
    if not eligible:
        return []

    rnd = random.Random(f"{week_start.isoformat()}|{text}|week")
    chosen = sorted(rnd.sample(eligible, min(count, len(eligible))))
    fires = [datetime.combine(day, _minute_to_time(start + rnd.randint(0, end - start))) for day in chosen]
    return [fire for fire in fires if fire.date() == target_date]


def _parse_count(raw: object) -> int:
    if not isinstance(raw, int) or isinstance(raw, bool) or raw < 1:
        raise ValueError(f"times_per_day must be an int >= 1, got {raw!r}")
    return raw


def parse_window(raw: object) -> tuple[int, int]:
    if not isinstance(raw, str) or "-" not in raw:
        raise ValueError(f"window must be 'HH:MM-HH:MM', got {raw!r}")
    start_str, end_str = raw.split("-", 1)
    start = _hhmm_to_minutes(start_str.strip())
    end = _hhmm_to_minutes(end_str.strip())
    if start >= end:
        raise ValueError(f"window start must be before end: {raw!r}")
    return start, end


def _hhmm_to_minutes(value: str) -> int:
    parts = value.split(":")
    if len(parts) != 2:
        raise ValueError(f"invalid time {value!r}, expected HH:MM")
    hours, minutes = int(parts[0]), int(parts[1])
    if not (0 <= hours < 24 and 0 <= minutes < 60):
        raise ValueError(f"time out of range: {value!r}")
    return hours * 60 + minutes


def minutes_to_hhmm(total: int) -> str:
    """Format a minute offset (e.g. 570) as HH:MM (e.g. '09:30')."""
    return f"{total // 60:02d}:{total % 60:02d}"


def random_cfg_from_row(row) -> dict:
    """Build a compute_fire_times config from a schedule_random DB row."""
    cfg = {
        "times_per_day": row["times_per_day"],
        "window": f"{minutes_to_hhmm(row['window_start'])}-{minutes_to_hhmm(row['window_end'])}",
        "period": row["period"],
    }
    if row["days_filter"]:
        cfg["days"] = row["days_filter"]
    if row["from_date"]:
        cfg["from"] = row["from_date"]
    if row["until_date"]:
        cfg["until"] = row["until_date"]
    return cfg


def _minute_to_time(total_minutes: int) -> time:
    return time(total_minutes // 60, total_minutes % 60)


def _parse_date(raw: object) -> date:
    if isinstance(raw, datetime):
        return raw.date()
    if isinstance(raw, date):
        return raw
    return date.fromisoformat(str(raw))


def _parse_days(spec: object) -> set[int]:
    """Parse a cron day-of-week spec into a set of cron weekdays (0/7=Sun, 1=Mon..6=Sat)."""
    text = str(spec).strip()
    if text == "*":
        return set(range(7))
    result: set[int] = set()
    for part in text.split(","):
        part = part.strip()
        if "-" in part:
            low_str, high_str = part.split("-", 1)
            low, high = int(low_str), int(high_str)
            result.update(_normalize_dow(day) for day in range(low, high + 1))
        else:
            result.add(_normalize_dow(int(part)))
    return result


def _normalize_dow(value: int) -> int:
    """cron allows 7 for Sunday; normalize it to 0."""
    if not 0 <= value <= 7:
        raise ValueError(f"day-of-week out of range (0-7): {value}")
    return 0 if value == 7 else value


def _cron_weekday(target: date) -> int:
    """Map Python weekday (Mon=0..Sun=6) to cron weekday (Sun=0, Mon=1..Sat=6)."""
    return (target.weekday() + 1) % 7