726 lines
24 KiB
Python
Executable File
726 lines
24 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import calendar
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import signal
|
|
import sys
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
|
|
|
import requests
|
|
|
|
LOG = logging.getLogger("reminder-bot")
|
|
|
|
TELEGRAM_API_BASE = "https://api.telegram.org"
|
|
MAX_SEND_ATTEMPTS = 5
|
|
LOOP_TICK_SECONDS = 60.0 # wake up at least this often: schedule + config check
|
|
MAX_ONESHOT_RETRIES = 3 # loop-level retries of a one-shot after a failed send
|
|
|
|
UNIT_SECONDS = {"s": 1.0, "m": 60.0, "h": 3600.0, "d": 86400.0}
|
|
INTERVAL_PART_RE = re.compile(r"(\d+(?:\.\d+)?)\s*([smhd])")
|
|
REPEAT_UNITS = ("yearly", "monthly")
|
|
|
|
|
|
class ConfigError(Exception):
|
|
"""Raised when the JSON config is invalid."""
|
|
|
|
|
|
# Parsing
|
|
|
|
|
|
def parse_interval(value: Any) -> float:
|
|
"""Convert an interval to seconds.
|
|
|
|
Accepts a number (seconds) or a string like "90s", "5m", "2h", "1d" or "1h30m".
|
|
"""
|
|
if isinstance(value, bool) or not isinstance(value, (int, float, str)):
|
|
raise ConfigError(
|
|
f"invalid interval {value!r}: expected a number of seconds "
|
|
"or a string like '1h30m'"
|
|
)
|
|
|
|
if isinstance(value, (int, float)):
|
|
seconds = float(value)
|
|
else:
|
|
text = value.strip().lower()
|
|
seconds = 0.0
|
|
pos = 0
|
|
for match in INTERVAL_PART_RE.finditer(text):
|
|
if text[pos : match.start()].strip():
|
|
raise ConfigError(f"invalid interval {value!r}")
|
|
seconds += float(match.group(1)) * UNIT_SECONDS[match.group(2)]
|
|
pos = match.end()
|
|
if pos == 0 or text[pos:].strip():
|
|
raise ConfigError(
|
|
f"invalid interval {value!r}: use a number of seconds "
|
|
"or a string like '1h30m'"
|
|
)
|
|
|
|
if seconds <= 0:
|
|
raise ConfigError(f"invalid interval {value!r}: must be positive")
|
|
return seconds
|
|
|
|
|
|
def parse_datetime(value: Any, default_tz: ZoneInfo, field: str) -> datetime:
|
|
"""Parse an ISO 8601 date/time; naive values are interpreted in `default_tz`."""
|
|
if not isinstance(value, str):
|
|
raise ConfigError(f"{field}: expected an ISO 8601 string, got {value!r}")
|
|
try:
|
|
dt = datetime.fromisoformat(value.strip())
|
|
except ValueError as exc:
|
|
raise ConfigError(f"{field}: invalid date/time {value!r}: {exc}") from exc
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=default_tz)
|
|
return dt
|
|
|
|
|
|
# Config
|
|
|
|
|
|
@dataclass
|
|
class Reminder:
|
|
message: str
|
|
chat_id: int | str
|
|
interval: float | None = None
|
|
at: datetime | None = None
|
|
start_at: datetime | None = None
|
|
repeat: str | None = None
|
|
next_fire: datetime | None = None
|
|
failures: int = 0
|
|
occurrence: int = 0
|
|
|
|
|
|
@dataclass
|
|
class AppConfig:
|
|
path: Path
|
|
token: str
|
|
chat_id: int | str | None
|
|
timezone: ZoneInfo
|
|
reminders: list[Reminder]
|
|
|
|
|
|
def _normalize_chat_id(value: Any, field: str) -> int | str:
|
|
if isinstance(value, bool) or not isinstance(value, (int, str)):
|
|
raise ConfigError(
|
|
f"{field}: chat_id must be an integer or a string, got {value!r}"
|
|
)
|
|
if isinstance(value, str):
|
|
stripped = value.strip()
|
|
if not stripped:
|
|
raise ConfigError(f"{field}: chat_id must not be empty")
|
|
if stripped.lstrip("-").isdigit():
|
|
return int(stripped)
|
|
return stripped # e.g. "@channelname"
|
|
return value
|
|
|
|
|
|
def _parse_reminder(
|
|
item: Any, index: int, default_chat: Any, tz: ZoneInfo
|
|
) -> Reminder | None:
|
|
where = f"reminders[{index}]"
|
|
if not isinstance(item, dict):
|
|
raise ConfigError(f"{where}: expected an object")
|
|
if not item.get("enabled", True):
|
|
LOG.info("%s: disabled, skipped", where)
|
|
return None
|
|
|
|
message = item.get("message")
|
|
if not isinstance(message, str) or not message.strip():
|
|
raise ConfigError(f"{where}: 'message' must be a non-empty string")
|
|
|
|
chat = item.get("chat_id", default_chat)
|
|
if chat is None:
|
|
raise ConfigError(
|
|
f"{where}: no chat_id — set it here or globally in the config"
|
|
)
|
|
chat = _normalize_chat_id(chat, where)
|
|
|
|
has_interval = "interval" in item
|
|
has_at = "at" in item
|
|
if has_interval == has_at:
|
|
raise ConfigError(f"{where}: exactly one of 'interval' or 'at' is required")
|
|
|
|
interval = parse_interval(item["interval"]) if has_interval else None
|
|
at = parse_datetime(item["at"], tz, f"{where}.at") if has_at else None
|
|
|
|
repeat = item.get("repeat")
|
|
if repeat is not None:
|
|
if not isinstance(repeat, str) or repeat.lower() not in REPEAT_UNITS:
|
|
raise ConfigError(
|
|
f"{where}: 'repeat' must be one of: {', '.join(REPEAT_UNITS)}"
|
|
)
|
|
if not has_at:
|
|
raise ConfigError(f"{where}: 'repeat' only makes sense together with 'at'")
|
|
repeat = repeat.lower()
|
|
|
|
start_at = None
|
|
if "start_at" in item:
|
|
if not has_interval:
|
|
raise ConfigError(
|
|
f"{where}: 'start_at' only makes sense together with 'interval'"
|
|
)
|
|
start_at = parse_datetime(item["start_at"], tz, f"{where}.start_at")
|
|
|
|
return Reminder(
|
|
message=message,
|
|
chat_id=chat,
|
|
interval=interval,
|
|
at=at,
|
|
start_at=start_at,
|
|
repeat=repeat,
|
|
)
|
|
|
|
|
|
def load_config(path: Path) -> AppConfig:
|
|
try:
|
|
raw = json.loads(path.read_text(encoding="utf-8"))
|
|
except FileNotFoundError as exc:
|
|
raise ConfigError(f"config file not found: {path}") from exc
|
|
except OSError as exc:
|
|
raise ConfigError(f"cannot read config file {path}: {exc}") from exc
|
|
except json.JSONDecodeError as exc:
|
|
raise ConfigError(f"invalid JSON in {path}: {exc}") from exc
|
|
|
|
if not isinstance(raw, dict):
|
|
raise ConfigError("config root must be a JSON object")
|
|
|
|
tz_name = raw.get("timezone", "UTC")
|
|
if not isinstance(tz_name, str):
|
|
raise ConfigError("'timezone' must be a string like 'Europe/Moscow'")
|
|
try:
|
|
tz = ZoneInfo(tz_name)
|
|
except (ZoneInfoNotFoundError, ValueError) as exc:
|
|
raise ConfigError(f"unknown timezone {tz_name!r}") from exc
|
|
|
|
token = raw.get("bot_token")
|
|
if not isinstance(token, str) or not token.strip():
|
|
raise ConfigError(
|
|
"'bot_token' is required: put the Telegram bot token in the config"
|
|
)
|
|
|
|
default_chat = raw.get("chat_id")
|
|
if default_chat is not None:
|
|
default_chat = _normalize_chat_id(default_chat, "chat_id")
|
|
|
|
reminders_raw = raw.get("reminders")
|
|
if not isinstance(reminders_raw, list) or not reminders_raw:
|
|
raise ConfigError("'reminders' must be a non-empty array")
|
|
|
|
reminders: list[Reminder] = []
|
|
for i, item in enumerate(reminders_raw):
|
|
reminder = _parse_reminder(item, i, default_chat, tz)
|
|
if reminder is not None:
|
|
reminders.append(reminder)
|
|
|
|
if not reminders:
|
|
LOG.warning("all reminders are disabled")
|
|
|
|
return AppConfig(
|
|
path=path, token=token, chat_id=default_chat, timezone=tz, reminders=reminders
|
|
)
|
|
|
|
|
|
# Scheduling
|
|
|
|
|
|
def _is_recurring(reminder: Reminder) -> bool:
|
|
"""True if the reminder fires more than once."""
|
|
return reminder.interval is not None or reminder.repeat is not None
|
|
|
|
|
|
def _add_years(dt: datetime, years: int) -> datetime:
|
|
try:
|
|
return dt.replace(year=dt.year + years)
|
|
except ValueError:
|
|
# Feb 29 -> Feb 28 in non-leap years
|
|
return dt.replace(year=dt.year + years, day=28)
|
|
|
|
|
|
def _add_months(dt: datetime, months: int) -> datetime:
|
|
year, month = divmod(dt.year * 12 + (dt.month - 1) + months, 12)
|
|
month += 1
|
|
day = min(dt.day, calendar.monthrange(year, month)[1])
|
|
return dt.replace(year=year, month=month, day=day)
|
|
|
|
|
|
def _occurrence(anchor: datetime, repeat: str, index: int) -> datetime:
|
|
"""Return the index-th occurrence counted from the anchor date."""
|
|
if repeat == "yearly":
|
|
return _add_years(anchor, index)
|
|
return _add_months(anchor, index)
|
|
|
|
|
|
def _next_occurrence_index(anchor: datetime, repeat: str, now: datetime) -> int:
|
|
"""Return the smallest index whose occurrence is not before `now`."""
|
|
if repeat == "yearly":
|
|
index = max(0, now.year - anchor.year)
|
|
else:
|
|
index = max(0, (now.year - anchor.year) * 12 + (now.month - anchor.month))
|
|
while _occurrence(anchor, repeat, index) < now:
|
|
index += 1
|
|
while index > 0 and _occurrence(anchor, repeat, index - 1) >= now:
|
|
index -= 1
|
|
return index
|
|
|
|
|
|
def build_schedule(reminders: list[Reminder], now: datetime) -> None:
|
|
"""Set the first fire time for every reminder.
|
|
|
|
Calendar-repeating reminders (`at` + `repeat`) are moved to their next
|
|
occurrence. One-shot reminders already in the past are skipped with a
|
|
warning. Interval reminders fire immediately unless `start_at` says otherwise.
|
|
"""
|
|
for reminder in reminders:
|
|
if reminder.at is not None:
|
|
if reminder.repeat is not None:
|
|
reminder.occurrence = _next_occurrence_index(
|
|
reminder.at, reminder.repeat, now
|
|
)
|
|
reminder.next_fire = _occurrence(
|
|
reminder.at, reminder.repeat, reminder.occurrence
|
|
)
|
|
elif reminder.at <= now:
|
|
LOG.warning(
|
|
"skipping one-shot reminder %r: %s is already in the past",
|
|
reminder.message,
|
|
reminder.at.isoformat(),
|
|
)
|
|
reminder.next_fire = None
|
|
else:
|
|
reminder.next_fire = reminder.at
|
|
elif reminder.interval is not None:
|
|
if reminder.start_at is not None:
|
|
next_fire = reminder.start_at
|
|
if next_fire <= now:
|
|
step = timedelta(seconds=reminder.interval)
|
|
missed = int((now - next_fire) // step) + 1
|
|
next_fire += missed * step
|
|
else:
|
|
# no start_at: send right away, then every interval
|
|
next_fire = now
|
|
reminder.next_fire = next_fire
|
|
|
|
|
|
def _scheduled_fire(reminder: Reminder) -> datetime:
|
|
"""Return the next fire time of an active reminder.
|
|
|
|
Active reminders always have one; this helper makes that invariant
|
|
explicit for the type checker and fails loudly if it is ever broken.
|
|
"""
|
|
if reminder.next_fire is None:
|
|
raise RuntimeError(
|
|
f"internal error: reminder {reminder.message!r} has no scheduled time"
|
|
)
|
|
return reminder.next_fire
|
|
|
|
|
|
def _advance(reminder: Reminder, now: datetime) -> datetime | None:
|
|
"""Return the next fire time after a send, or None if the reminder is done."""
|
|
if reminder.interval is not None:
|
|
step = timedelta(seconds=reminder.interval)
|
|
next_fire = _scheduled_fire(reminder) + step
|
|
if next_fire <= now:
|
|
# the loop was delayed (sleep/suspend): skip missed occurrences
|
|
# instead of sending a burst of messages
|
|
missed = int((now - next_fire) // step) + 1
|
|
next_fire += missed * step
|
|
return next_fire
|
|
if reminder.repeat is not None and reminder.at is not None:
|
|
reminder.occurrence += 1
|
|
next_fire = _occurrence(reminder.at, reminder.repeat, reminder.occurrence)
|
|
while next_fire <= now:
|
|
# skip occurrences missed while the loop was not running
|
|
reminder.occurrence += 1
|
|
next_fire = _occurrence(reminder.at, reminder.repeat, reminder.occurrence)
|
|
return next_fire
|
|
return None
|
|
|
|
|
|
# Reload
|
|
|
|
|
|
def _config_stamp(path: Path) -> tuple[int, int, int] | None:
|
|
"""Return (mtime_ns, size, inode) of the config, or None if it is unreadable."""
|
|
try:
|
|
stat = path.stat()
|
|
except OSError:
|
|
return None
|
|
return (stat.st_mtime_ns, stat.st_size, stat.st_ino)
|
|
|
|
|
|
def _schedule_key(reminder: Reminder) -> tuple[Any, ...]:
|
|
"""Identity used to carry the schedule of a reminder over a config reload.
|
|
|
|
The message text is deliberately not part of the key: fixing a typo in a
|
|
message must not restart the schedule (or trigger an immediate send).
|
|
"""
|
|
return (
|
|
reminder.chat_id,
|
|
reminder.interval,
|
|
reminder.at,
|
|
reminder.start_at,
|
|
reminder.repeat,
|
|
)
|
|
|
|
|
|
def _try_load(path: Path) -> AppConfig | None:
|
|
"""Load the config for a live reload; on error keep the previous one."""
|
|
try:
|
|
return load_config(path)
|
|
except ConfigError as exc:
|
|
LOG.error("config reload failed, keeping the previous config: %s", exc)
|
|
return None
|
|
|
|
|
|
def _apply_reload(
|
|
active: list[Reminder],
|
|
new_config: AppConfig,
|
|
completed: set[tuple[Any, ...]],
|
|
now: datetime,
|
|
) -> list[Reminder]:
|
|
"""Build the new active list, preserving schedules of unchanged reminders."""
|
|
pools: dict[tuple[Any, ...], list[Reminder]] = {}
|
|
for reminder in active:
|
|
pools.setdefault(_schedule_key(reminder), []).append(reminder)
|
|
|
|
kept: list[Reminder] = []
|
|
added: list[Reminder] = []
|
|
for reminder in new_config.reminders:
|
|
pool = pools.get(_schedule_key(reminder))
|
|
if pool:
|
|
old = pool.pop(0)
|
|
reminder.next_fire = old.next_fire
|
|
reminder.failures = old.failures
|
|
reminder.occurrence = old.occurrence
|
|
kept.append(reminder)
|
|
if old.message != reminder.message:
|
|
LOG.info("updated message: %r -> %r", old.message, reminder.message)
|
|
else:
|
|
added.append(reminder)
|
|
|
|
removed = [r for pool in pools.values() for r in pool]
|
|
# one-shots that already fired (or were skipped as past) are not rescheduled
|
|
added = [
|
|
r for r in added if not (r.at is not None and _schedule_key(r) in completed)
|
|
]
|
|
build_schedule(added, now)
|
|
for reminder in added:
|
|
if reminder.at is not None and reminder.next_fire is None:
|
|
completed.add(_schedule_key(reminder))
|
|
|
|
new_active = kept + [r for r in added if r.next_fire is not None]
|
|
LOG.info(
|
|
"config reloaded: %d kept, %d added, %d removed",
|
|
len(kept),
|
|
len(added),
|
|
len(removed),
|
|
)
|
|
for reminder in added:
|
|
next_send = reminder.next_fire.isoformat() if reminder.next_fire else "-"
|
|
LOG.info(
|
|
"added: %r (chat %s), next send at %s",
|
|
reminder.message,
|
|
reminder.chat_id,
|
|
next_send,
|
|
)
|
|
for reminder in removed:
|
|
LOG.info("removed: %r (chat %s)", reminder.message, reminder.chat_id)
|
|
|
|
return new_active
|
|
|
|
|
|
# Sending
|
|
|
|
|
|
def _retry_after(response: requests.Response) -> float:
|
|
try:
|
|
params = response.json().get("parameters") or {}
|
|
return float(params.get("retry_after", 5))
|
|
except (ValueError, TypeError, AttributeError):
|
|
return 5.0
|
|
|
|
|
|
def send_message(
|
|
token: str,
|
|
chat_id: int | str,
|
|
text: str,
|
|
*,
|
|
dry_run: bool = False,
|
|
stop_event: threading.Event | None = None,
|
|
) -> bool:
|
|
"""Send one message via the Telegram Bot API, retrying transient failures."""
|
|
if dry_run:
|
|
LOG.info("[dry-run] -> chat %s: %s", chat_id, text)
|
|
return True
|
|
|
|
url = f"{TELEGRAM_API_BASE}/bot{token}/sendMessage"
|
|
payload = {"chat_id": chat_id, "text": text}
|
|
|
|
for attempt in range(1, MAX_SEND_ATTEMPTS + 1):
|
|
wait = float(min(2**attempt, 60))
|
|
try:
|
|
response = requests.post(url, json=payload, timeout=(10, 30))
|
|
except requests.RequestException as exc:
|
|
LOG.warning(
|
|
"sending to chat %s failed (attempt %d/%d): %s",
|
|
chat_id,
|
|
attempt,
|
|
MAX_SEND_ATTEMPTS,
|
|
exc,
|
|
)
|
|
else:
|
|
if response.status_code == 200:
|
|
try:
|
|
ok = bool(response.json().get("ok"))
|
|
except ValueError:
|
|
ok = False
|
|
if ok:
|
|
LOG.info("sent to chat %s: %s", chat_id, text)
|
|
return True
|
|
LOG.error("unexpected Telegram response: %s", response.text[:300])
|
|
return False
|
|
if response.status_code == 429:
|
|
wait = min(_retry_after(response), 300.0)
|
|
LOG.warning("rate limited by Telegram, retrying in %.0f s", wait)
|
|
elif response.status_code >= 500:
|
|
LOG.warning(
|
|
"Telegram server error %d (attempt %d/%d)",
|
|
response.status_code,
|
|
attempt,
|
|
MAX_SEND_ATTEMPTS,
|
|
)
|
|
else:
|
|
LOG.error(
|
|
"Telegram rejected the message (HTTP %d): %s",
|
|
response.status_code,
|
|
response.text[:300],
|
|
)
|
|
return False # bad token/chat/etc. — retrying won't help
|
|
|
|
if attempt < MAX_SEND_ATTEMPTS:
|
|
if stop_event is not None:
|
|
if stop_event.wait(wait):
|
|
return False
|
|
else:
|
|
time.sleep(wait)
|
|
|
|
LOG.error(
|
|
"gave up sending to chat %s after %d attempts", chat_id, MAX_SEND_ATTEMPTS
|
|
)
|
|
return False
|
|
|
|
|
|
# Main loop
|
|
|
|
|
|
def run(
|
|
config: AppConfig,
|
|
*,
|
|
dry_run: bool,
|
|
stop_event: threading.Event,
|
|
wake_event: threading.Event,
|
|
reload_requested: threading.Event,
|
|
) -> int:
|
|
active = [r for r in config.reminders if r.next_fire is not None]
|
|
# one-shots that were skipped as "in the past" must not be re-warned on reload
|
|
completed: set[tuple[Any, ...]] = {
|
|
_schedule_key(r)
|
|
for r in config.reminders
|
|
if r.at is not None and r.next_fire is None
|
|
}
|
|
|
|
if not active:
|
|
LOG.warning("no active reminders (all are disabled or already in the past)")
|
|
return 0
|
|
|
|
for reminder in active:
|
|
LOG.info(
|
|
"scheduled: %r (chat %s), next send at %s",
|
|
reminder.message,
|
|
reminder.chat_id,
|
|
_scheduled_fire(reminder).isoformat(),
|
|
)
|
|
|
|
stamp = _config_stamp(config.path)
|
|
|
|
while not stop_event.is_set():
|
|
if reload_requested.is_set() or _config_stamp(config.path) != stamp:
|
|
reload_requested.clear()
|
|
stamp = _config_stamp(config.path)
|
|
new_config = _try_load(config.path)
|
|
if new_config is not None:
|
|
active = _apply_reload(active, new_config, completed, datetime.now(UTC))
|
|
config = new_config
|
|
|
|
now = datetime.now(UTC)
|
|
for reminder in [r for r in active if _scheduled_fire(r) <= now]:
|
|
sent = send_message(
|
|
config.token,
|
|
reminder.chat_id,
|
|
reminder.message,
|
|
dry_run=dry_run,
|
|
stop_event=stop_event,
|
|
)
|
|
if sent or _is_recurring(reminder):
|
|
next_fire = _advance(reminder, now)
|
|
if next_fire is None:
|
|
active.remove(reminder)
|
|
completed.add(_schedule_key(reminder))
|
|
LOG.info("reminder %r is done", reminder.message)
|
|
else:
|
|
reminder.next_fire = next_fire
|
|
else:
|
|
# a one-shot send failed: retry on the next tick, then give up
|
|
reminder.failures += 1
|
|
if reminder.failures >= MAX_ONESHOT_RETRIES:
|
|
LOG.error(
|
|
"dropping one-shot reminder %r after %d failed attempts",
|
|
reminder.message,
|
|
reminder.failures,
|
|
)
|
|
active.remove(reminder)
|
|
|
|
if not active:
|
|
LOG.info("no active reminders left, exiting")
|
|
return 0
|
|
|
|
next_fire = min(_scheduled_fire(r) for r in active)
|
|
delay = (next_fire - datetime.now(UTC)).total_seconds()
|
|
wake_event.wait(min(max(delay, 0.0), LOOP_TICK_SECONDS))
|
|
wake_event.clear()
|
|
|
|
LOG.info("stopped")
|
|
return 0
|
|
|
|
|
|
def install_signal_handlers(
|
|
stop_event: threading.Event,
|
|
wake_event: threading.Event,
|
|
reload_requested: threading.Event,
|
|
) -> None:
|
|
def on_stop(signum: int, _frame: Any) -> None:
|
|
LOG.info("received %s, shutting down", signal.Signals(signum).name)
|
|
stop_event.set()
|
|
wake_event.set()
|
|
|
|
def on_reload(signum: int, _frame: Any) -> None:
|
|
LOG.info("received %s, reloading config", signal.Signals(signum).name)
|
|
reload_requested.set()
|
|
wake_event.set()
|
|
|
|
for sig in (signal.SIGINT, signal.SIGTERM):
|
|
signal.signal(sig, on_stop)
|
|
if hasattr(signal, "SIGHUP"):
|
|
signal.signal(signal.SIGHUP, on_reload)
|
|
|
|
|
|
# cli
|
|
|
|
|
|
def _format_dt(dt: datetime) -> str:
|
|
return dt.isoformat(sep=" ", timespec="seconds")
|
|
|
|
|
|
def print_plan(config: AppConfig) -> None:
|
|
print(f"Config: {config.path}")
|
|
print(f"Timezone: {config.timezone} (used for dates without an offset)")
|
|
default_chat = config.chat_id if config.chat_id is not None else "not set"
|
|
print(f"Default chat_id: {default_chat}")
|
|
print()
|
|
print(f"{len(config.reminders)} reminder(s):")
|
|
for i, reminder in enumerate(config.reminders, 1):
|
|
if reminder.interval is not None:
|
|
kind = f"every {reminder.interval:g}s"
|
|
elif reminder.repeat is not None:
|
|
kind = reminder.repeat
|
|
else:
|
|
kind = "one-shot"
|
|
print(f" {i}. [{kind}, chat {reminder.chat_id}] {reminder.message!r}")
|
|
if reminder.next_fire is None:
|
|
print(" -> will not fire (time is in the past)")
|
|
continue
|
|
fires = [reminder.next_fire]
|
|
if reminder.interval is not None:
|
|
step = timedelta(seconds=reminder.interval)
|
|
for _ in range(2):
|
|
fires.append(fires[-1] + step)
|
|
elif reminder.repeat is not None and reminder.at is not None:
|
|
index = reminder.occurrence
|
|
for _ in range(2):
|
|
index += 1
|
|
fires.append(_occurrence(reminder.at, reminder.repeat, index))
|
|
print(" -> next: " + ", ".join(_format_dt(f) for f in fires))
|
|
|
|
|
|
def parse_args(argv: list[str] | None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Telegram reminder bot: sends messages on a schedule from a JSON config"
|
|
)
|
|
parser.add_argument(
|
|
"-c",
|
|
"--config",
|
|
default=os.environ.get("CONFIG_PATH", "config.json"),
|
|
help="path to the JSON config (default: $CONFIG_PATH or ./config.json)",
|
|
)
|
|
parser.add_argument(
|
|
"--check",
|
|
action="store_true",
|
|
help="validate the config, print the schedule and exit",
|
|
)
|
|
parser.add_argument(
|
|
"--dry-run",
|
|
action="store_true",
|
|
help="run normally but log messages instead of sending them",
|
|
)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
# Main
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
logging.basicConfig(
|
|
level=os.environ.get("LOG_LEVEL", "INFO").upper(),
|
|
format="%(asctime)s %(levelname)-7s %(message)s",
|
|
)
|
|
|
|
try:
|
|
config = load_config(Path(args.config))
|
|
except ConfigError as exc:
|
|
LOG.error("config error: %s", exc)
|
|
return 2
|
|
|
|
build_schedule(config.reminders, datetime.now(UTC))
|
|
|
|
if args.check:
|
|
print_plan(config)
|
|
return 0
|
|
|
|
stop_event = threading.Event()
|
|
wake_event = threading.Event()
|
|
reload_requested = threading.Event()
|
|
install_signal_handlers(stop_event, wake_event, reload_requested)
|
|
LOG.info("reminder bot starting (config: %s)", config.path)
|
|
return run(
|
|
config,
|
|
dry_run=args.dry_run,
|
|
stop_event=stop_event,
|
|
wake_event=wake_event,
|
|
reload_requested=reload_requested,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|