118 lines
4.4 KiB
Python
118 lines
4.4 KiB
Python
"""Forecast which reminders will fire in a time window.
|
|
|
|
A pure projection of the schedule — answers "what goes out today / tomorrow /
|
|
this week". Shared by remind_cli.py. Unlike remind_send.py it applies no
|
|
tolerance and no reminder_fires dedup: `upcoming` shows the plan, `delivered`
|
|
shows what actually happened.
|
|
|
|
Times are Prague local (naive), matching how schedules are stored.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from datetime import date, datetime, time, timedelta
|
|
|
|
from croniter import croniter
|
|
from random_times import compute_fire_times, random_cfg_from_row
|
|
|
|
|
|
def window_for(now: datetime, date_str: str | None, days: int | None) -> tuple[datetime, datetime]:
|
|
"""Resolve the [start, end) window from CLI args.
|
|
|
|
--date: the whole given day. --days N: now through midnight after N
|
|
calendar days (including today). Neither: the rest of today (= --days 1).
|
|
"""
|
|
if date_str is not None and days is not None:
|
|
raise ValueError("--date and --days are mutually exclusive")
|
|
if date_str is not None:
|
|
start = datetime.combine(date.fromisoformat(date_str), time.min)
|
|
return start, start + timedelta(days=1)
|
|
span = 1 if days is None else days
|
|
if span < 1:
|
|
raise ValueError(f"--days must be >= 1, got {span}")
|
|
end = datetime.combine(now.date(), time.min) + timedelta(days=span)
|
|
return now, end
|
|
|
|
|
|
def fires_in_window(conn, start: datetime, end: datetime) -> list[dict]:
|
|
"""All scheduled fires in [start, end) across active reminders, sorted by time."""
|
|
fires = _at_fires(conn, start, end) + _cron_fires(conn, start, end) + _random_fires(conn, start, end)
|
|
fires.sort(key=lambda f: (f["fire_time"], f["id"]))
|
|
return fires
|
|
|
|
|
|
def format_upcoming(fires: list[dict], id_to_display: dict[int, int]) -> list[str]:
|
|
if not fires:
|
|
return ["(nothing scheduled in this window)"]
|
|
return [
|
|
f"{f['fire_time']:%Y-%m-%d %H:%M} #{id_to_display[f['id']]} {f['text']} ({f['schedule_type']})"
|
|
for f in fires
|
|
]
|
|
|
|
|
|
def _at_fires(conn, start: datetime, end: datetime) -> list[dict]:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT r.id, r.text, sa.at_datetime
|
|
FROM reminders r
|
|
JOIN schedule_at sa ON sa.reminder_id = r.id
|
|
WHERE r.enabled = 1 AND r.deleted_at IS NULL
|
|
AND sa.at_datetime >= ? AND sa.at_datetime < ?
|
|
""",
|
|
(start.isoformat(timespec="seconds"), end.isoformat(timespec="seconds")),
|
|
).fetchall()
|
|
return [
|
|
{"fire_time": datetime.fromisoformat(r["at_datetime"]), "id": r["id"], "text": r["text"], "schedule_type": "at"}
|
|
for r in rows
|
|
]
|
|
|
|
|
|
def _cron_fires(conn, start: datetime, end: datetime) -> list[dict]:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT r.id, r.text, sc.cron_expr
|
|
FROM reminders r
|
|
JOIN schedule_cron sc ON sc.reminder_id = r.id
|
|
WHERE r.enabled = 1 AND r.deleted_at IS NULL
|
|
"""
|
|
).fetchall()
|
|
fires = []
|
|
for row in rows:
|
|
it = croniter(row["cron_expr"], start - timedelta(seconds=1))
|
|
while True:
|
|
fire_time = it.get_next(datetime)
|
|
if fire_time >= end:
|
|
break
|
|
if fire_time >= start:
|
|
fires.append({"fire_time": fire_time, "id": row["id"], "text": row["text"], "schedule_type": "cron"})
|
|
return fires
|
|
|
|
|
|
def _random_fires(conn, start: datetime, end: datetime) -> list[dict]:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT r.id, r.text, sr.times_per_day, sr.window_start, sr.window_end,
|
|
sr.days_filter, sr.from_date, sr.until_date, sr.period
|
|
FROM reminders r
|
|
JOIN schedule_random sr ON sr.reminder_id = r.id
|
|
WHERE r.enabled = 1 AND r.deleted_at IS NULL
|
|
"""
|
|
).fetchall()
|
|
fires = []
|
|
last_day = (end - timedelta(seconds=1)).date()
|
|
for row in rows:
|
|
cfg = random_cfg_from_row(row)
|
|
day = start.date()
|
|
while day <= last_day:
|
|
try:
|
|
day_fires = compute_fire_times(day, row["text"], cfg)
|
|
except ValueError as exc:
|
|
print(f"forecast: bad random config for {row['text']!r}: {exc}", file=sys.stderr)
|
|
break
|
|
for fire_time in day_fires:
|
|
if start <= fire_time < end:
|
|
fires.append({"fire_time": fire_time, "id": row["id"], "text": row["text"], "schedule_type": "random"})
|
|
day += timedelta(days=1)
|
|
return fires
|