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

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = ["croniter"]
# ///
"""Deterministic reminder sender backed by SQLite.

Runs every minute from the nanobot user crontab.
Reads reminders from SQLite, finds due fires, sends each directly to Telegram,
logs delivery to reminder.log, and dedups via reminder_fires table.
"""

from __future__ import annotations

import json
import os
import sys
import urllib.parse
import urllib.request
from datetime import date, datetime, timedelta
from pathlib import Path
from zoneinfo import ZoneInfo

from croniter import croniter
from db import get_db, init_db, log_operation
from random_times import compute_fire_times, minutes_to_hhmm

WORKSPACE = Path(__file__).resolve().parent.parent.parent.parent
DEFAULT_DB_PATH = WORKSPACE / "db" / "reminders.sqlite"
DB_PATH = Path(os.environ.get("REMIND_DB", str(DEFAULT_DB_PATH)))
CONFIG = Path.home() / ".nanobot" / "config.json"

TZ = ZoneInfo("Europe/Prague")
TOLERANCE_SECONDS = 60
FALLBACK_CHAT_ID = "8826147089"


def _telegram_config() -> tuple[str, str]:
    """Return (bot token, chat id). Chat id reads channels.telegram.allowFrom[0], with a constant fallback."""
    data = json.loads(CONFIG.read_text(encoding="utf-8"))
    telegram = data["channels"]["telegram"]
    allow_from = telegram.get("allowFrom") or []
    chat_id = str(allow_from[0]) if allow_from else FALLBACK_CHAT_ID
    return telegram["token"], chat_id


def _send_telegram(text: str, token: str, chat_id: str) -> None:
    url = f"https://api.telegram.org/bot{token}/sendMessage"
    payload = urllib.parse.urlencode({"chat_id": chat_id, "text": text}).encode()
    req = urllib.request.Request(url, data=payload, method="POST")
    with urllib.request.urlopen(req, timeout=15) as resp:
        resp.read()


def _now_prague() -> datetime:
    return datetime.now(TZ).replace(tzinfo=None)


def _due_at(conn, now: datetime) -> list[dict]:
    """Find due one-time reminders."""
    since = (now - timedelta(seconds=TOLERANCE_SECONDS)).isoformat(timespec="seconds")
    until = now.isoformat(timespec="seconds")
    rows = conn.execute(
        """
        SELECT r.id, r.text, sa.id AS schedule_id, sa.at_datetime AS fire_time
        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 <= ?
          AND NOT EXISTS (
              SELECT 1 FROM reminder_fires rf
              WHERE rf.reminder_id = r.id AND rf.schedule_id = sa.id
                AND rf.schedule_type = 'at' AND rf.fire_time = sa.at_datetime
                AND rf.status = 'delivered'
          )
        """,
        (since, until),
    ).fetchall()
    return [{**dict(r), "schedule_type": "at"} for r in rows]


def _due_cron(conn, now: datetime) -> list[dict]:
    """Find due cron reminders."""
    rows = conn.execute(
        """
        SELECT r.id, r.text, sc.id AS schedule_id, 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()
    due = []
    for row in rows:
        prev = croniter(row["cron_expr"], now + timedelta(seconds=1)).get_prev(datetime)
        if 0 <= (now - prev).total_seconds() < TOLERANCE_SECONDS:
            fire_iso = prev.isoformat(timespec="seconds")
            already = conn.execute(
                """
                SELECT 1 FROM reminder_fires
                WHERE reminder_id = ? AND schedule_id = ? AND schedule_type = 'cron'
                  AND fire_time = ? AND status = 'delivered'
                """,
                (row["id"], row["schedule_id"], fire_iso),
            ).fetchone()
            if not already:
                due.append({
                    "id": row["id"],
                    "text": row["text"],
                    "schedule_id": row["schedule_id"],
                    "fire_time": fire_iso,
                    "schedule_type": "cron",
                })
    return due


def _due_random(conn, now: datetime) -> list[dict]:
    """Find due random reminders."""
    rows = conn.execute(
        """
        SELECT r.id, r.text, sr.id AS schedule_id, sr.times_per_day, sr.window_start, sr.window_end,
               sr.days_filter, sr.from_date, sr.until_date
        FROM reminders r
        JOIN schedule_random sr ON sr.reminder_id = r.id
        WHERE r.enabled = 1 AND r.deleted_at IS NULL
        """
    ).fetchall()
    due = []
    for row in rows:
        cfg = {
            "times_per_day": row["times_per_day"],
            "window": f"{minutes_to_hhmm(row['window_start'])}-{minutes_to_hhmm(row['window_end'])}",
        }
        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"]
        try:
            fires = compute_fire_times(now.date(), row["text"], cfg)
        except ValueError as exc:
            print(f"remind_send: bad random config for {row['text']!r}: {exc}", file=sys.stderr)
            continue
        for ft in fires:
            if 0 <= (now - ft).total_seconds() < TOLERANCE_SECONDS:
                fire_iso = ft.isoformat(timespec="seconds")
                already = conn.execute(
                    """
                    SELECT 1 FROM reminder_fires
                    WHERE reminder_id = ? AND schedule_id = ? AND schedule_type = 'random'
                      AND fire_time = ? AND status = 'delivered'
                    """,
                    (row["id"], row["schedule_id"], fire_iso),
                ).fetchone()
                if not already:
                    due.append({
                        "id": row["id"],
                        "text": row["text"],
                        "schedule_id": row["schedule_id"],
                        "fire_time": fire_iso,
                        "schedule_type": "random",
                    })
    return due


def _record_fire(conn, reminder_id: int, schedule_id: int, schedule_type: str, fire_time: str, status: str, error: str | None = None) -> None:
    now = _now_prague().isoformat(timespec="seconds")
    conn.execute(
        """
        INSERT INTO reminder_fires (reminder_id, schedule_id, schedule_type, fire_time, delivered_at, status, error_message)
        VALUES (?, ?, ?, ?, ?, ?, ?)
        """,
        (reminder_id, schedule_id, schedule_type, fire_time, now if status == "delivered" else None, status, error),
    )


def main() -> None:
    if not DB_PATH.exists():
        init_db(DB_PATH)

    conn = get_db(DB_PATH)
    try:
        now = _now_prague()
        due = _due_at(conn, now) + _due_cron(conn, now) + _due_random(conn, now)
        if not due:
            return

        token, chat_id = _telegram_config()
        for fire in due:
            text = fire["text"]
            rid = fire["id"]
            sid = fire["schedule_id"]
            ft = fire["fire_time"]
            schedule_type = fire["schedule_type"]

            try:
                _send_telegram(f"⏰ Reminder: {text}", token, chat_id)
            except Exception as e:
                print(f"remind_send: delivery failed for {text!r}: {e}", file=sys.stderr)
                _record_fire(conn, rid, sid, schedule_type, ft, "failed", str(e))
                continue

            _record_fire(conn, rid, sid, schedule_type, ft, "delivered")
            log_operation("DELIVER", rid, f'text="{text}"')
    finally:
        conn.close()


if __name__ == "__main__":
    main()