Files
nanobot-runtime/skills/remind/scripts/remind_send.py
2026-06-10 10:54:21 +02:00

203 lines
7.1 KiB
Python

#!/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, random_cfg_from_row
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 = random_cfg_from_row(row)
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()