#!/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())