432 lines
16 KiB
Python
Executable File
432 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# /// script
|
|
# requires-python = ">=3.11"
|
|
# dependencies = ["croniter", "pyyaml"]
|
|
# ///
|
|
"""Deterministic CRUD for reminders backed by SQLite.
|
|
|
|
CLI tool for LLM skills to create, list, edit, enable, disable, and remove reminders.
|
|
All mutations are atomic SQLite transactions with audit logging.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from croniter import croniter
|
|
from db import get_db, init_db, log_operation
|
|
from random_times import compute_fire_times
|
|
|
|
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)))
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
|
|
|
|
def _ensure_db() -> None:
|
|
if not DB_PATH.exists():
|
|
init_db(DB_PATH)
|
|
|
|
|
|
def _build_random(args: argparse.Namespace) -> dict | None:
|
|
"""Assemble and validate the random schedule block, or None if no --random-* flag given."""
|
|
fields = {
|
|
"times_per_day": args.random_times_per_day,
|
|
"window": args.random_window,
|
|
"days": args.random_days,
|
|
"from": args.random_from,
|
|
"until": args.random_until,
|
|
}
|
|
if all(value is None for value in fields.values()):
|
|
return None
|
|
if fields["times_per_day"] is None or fields["window"] is None:
|
|
raise ValueError("random schedule needs --random-times-per-day and --random-window")
|
|
|
|
cfg = {key: value for key, value in fields.items() if value is not None}
|
|
compute_fire_times(__import__("datetime").date(2000, 1, 1), "validation", cfg)
|
|
return cfg
|
|
|
|
|
|
def _parse_window(window: str) -> tuple[int, int]:
|
|
start_str, end_str = window.split("-", 1)
|
|
start = _hhmm_to_minutes(start_str.strip())
|
|
end = _hhmm_to_minutes(end_str.strip())
|
|
return start, end
|
|
|
|
|
|
def _hhmm_to_minutes(value: str) -> int:
|
|
h, m = value.split(":")
|
|
return int(h) * 60 + int(m)
|
|
|
|
|
|
def _insert_schedules(conn, reminder_id: int, args: argparse.Namespace) -> None:
|
|
if args.at:
|
|
for at_str in args.at:
|
|
conn.execute(
|
|
"INSERT INTO schedule_at (reminder_id, at_datetime) VALUES (?, ?)",
|
|
(reminder_id, at_str),
|
|
)
|
|
if args.cron:
|
|
for expr in args.cron:
|
|
conn.execute(
|
|
"INSERT INTO schedule_cron (reminder_id, cron_expr) VALUES (?, ?)",
|
|
(reminder_id, expr),
|
|
)
|
|
random_cfg = _build_random(args)
|
|
if random_cfg:
|
|
start, end = _parse_window(random_cfg["window"])
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO schedule_random
|
|
(reminder_id, times_per_day, window_start, window_end, days_filter, from_date, until_date)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
reminder_id,
|
|
random_cfg["times_per_day"],
|
|
start,
|
|
end,
|
|
random_cfg.get("days"),
|
|
random_cfg.get("from"),
|
|
random_cfg.get("until"),
|
|
),
|
|
)
|
|
|
|
|
|
def _fetch_reminder(conn, reminder_id: int) -> dict:
|
|
row = conn.execute(
|
|
"SELECT id, text, enabled, timezone, created_at, updated_at, deleted_at FROM reminders WHERE id = ?",
|
|
(reminder_id,),
|
|
).fetchone()
|
|
if row is None:
|
|
raise ValueError(f"reminder {reminder_id} not found")
|
|
reminder = dict(row)
|
|
reminder["at"] = [
|
|
dict(r) for r in conn.execute(
|
|
"SELECT id, at_datetime FROM schedule_at WHERE reminder_id = ?", (reminder_id,)
|
|
).fetchall()
|
|
]
|
|
reminder["cron"] = [
|
|
dict(r) for r in conn.execute(
|
|
"SELECT id, cron_expr FROM schedule_cron WHERE reminder_id = ?", (reminder_id,)
|
|
).fetchall()
|
|
]
|
|
reminder["random"] = [
|
|
dict(r) for r in conn.execute(
|
|
"SELECT id, times_per_day, window_start, window_end, days_filter, from_date, until_date FROM schedule_random WHERE reminder_id = ?",
|
|
(reminder_id,),
|
|
).fetchall()
|
|
]
|
|
return reminder
|
|
|
|
|
|
def _find_by_keyword(conn, keyword: str) -> list[dict]:
|
|
rows = conn.execute(
|
|
"SELECT id, text, enabled, timezone, created_at, updated_at, deleted_at FROM reminders WHERE text LIKE ? AND deleted_at IS NULL",
|
|
(f"%{keyword}%",),
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
def cmd_list(_args: argparse.Namespace) -> int:
|
|
_ensure_db()
|
|
conn = get_db(DB_PATH)
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT id, text, enabled, timezone, created_at, updated_at, deleted_at FROM reminders WHERE deleted_at IS NULL ORDER BY id"
|
|
).fetchall()
|
|
reminders = []
|
|
for row in rows:
|
|
reminder = dict(row)
|
|
rid = reminder["id"]
|
|
reminder["at"] = [
|
|
dict(r) for r in conn.execute(
|
|
"SELECT id, at_datetime FROM schedule_at WHERE reminder_id = ?", (rid,)
|
|
).fetchall()
|
|
]
|
|
reminder["cron"] = [
|
|
dict(r) for r in conn.execute(
|
|
"SELECT id, cron_expr FROM schedule_cron WHERE reminder_id = ?", (rid,)
|
|
).fetchall()
|
|
]
|
|
reminder["random"] = [
|
|
dict(r) for r in conn.execute(
|
|
"SELECT id, times_per_day, window_start, window_end, days_filter, from_date, until_date FROM schedule_random WHERE reminder_id = ?",
|
|
(rid,),
|
|
).fetchall()
|
|
]
|
|
reminders.append(reminder)
|
|
print(json.dumps({"reminders": reminders}, ensure_ascii=False))
|
|
return 0
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def cmd_add(args: argparse.Namespace) -> int:
|
|
text = (args.text or "").strip()
|
|
if not text:
|
|
print(json.dumps({"error": "text must not be empty"}), file=sys.stderr)
|
|
return 1
|
|
|
|
try:
|
|
random_cfg = _build_random(args)
|
|
except ValueError as exc:
|
|
print(json.dumps({"error": str(exc)}), file=sys.stderr)
|
|
return 1
|
|
|
|
if not args.at and not args.cron and not random_cfg:
|
|
print(json.dumps({"error": "provide --cron, --at, or --random-* options"}), file=sys.stderr)
|
|
return 1
|
|
|
|
if args.at:
|
|
for at_str in args.at:
|
|
try:
|
|
datetime.fromisoformat(at_str)
|
|
except ValueError as exc:
|
|
print(json.dumps({"error": f"invalid --at datetime: {exc}"}), file=sys.stderr)
|
|
return 1
|
|
|
|
if args.cron:
|
|
for expr in args.cron:
|
|
if not croniter.is_valid(expr):
|
|
print(json.dumps({"error": f"invalid cron expression: {expr!r}"}), file=sys.stderr)
|
|
return 1
|
|
|
|
_ensure_db()
|
|
conn = get_db(DB_PATH)
|
|
try:
|
|
conn.execute("BEGIN")
|
|
now = _now()
|
|
cur = conn.execute(
|
|
"INSERT INTO reminders (text, enabled, timezone, created_at, updated_at) VALUES (?, 1, 'Europe/Prague', ?, ?)",
|
|
(text, now, now),
|
|
)
|
|
reminder_id = cur.lastrowid
|
|
_insert_schedules(conn, reminder_id, args)
|
|
conn.execute("COMMIT")
|
|
reminder = _fetch_reminder(conn, reminder_id)
|
|
log_operation("ADD", reminder_id, f'text="{text}"')
|
|
print(json.dumps({"added": reminder}, ensure_ascii=False))
|
|
return 0
|
|
except Exception as exc:
|
|
conn.execute("ROLLBACK")
|
|
print(json.dumps({"error": str(exc)}), file=sys.stderr)
|
|
return 1
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def cmd_remove(args: argparse.Namespace) -> int:
|
|
keyword = (args.keyword or "").strip().lower()
|
|
if not keyword:
|
|
print(json.dumps({"error": "keyword must not be empty"}), file=sys.stderr)
|
|
return 1
|
|
|
|
_ensure_db()
|
|
conn = get_db(DB_PATH)
|
|
try:
|
|
matches = _find_by_keyword(conn, keyword)
|
|
if len(matches) == 0:
|
|
print(json.dumps({"error": "no match", "keyword": args.keyword}), file=sys.stderr)
|
|
return 1
|
|
if len(matches) > 1:
|
|
print(
|
|
json.dumps({"error": "ambiguous", "matches": [{"text": m["text"]} for m in matches]}, ensure_ascii=False),
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
rid = matches[0]["id"]
|
|
conn.execute("BEGIN")
|
|
conn.execute("UPDATE reminders SET deleted_at = ?, updated_at = ? WHERE id = ?", (_now(), _now(), rid))
|
|
conn.execute("COMMIT")
|
|
log_operation("REMOVE", rid, f'text="{matches[0]["text"]}"')
|
|
reminder = _fetch_reminder(conn, rid)
|
|
print(json.dumps({"removed": reminder}, ensure_ascii=False))
|
|
return 0
|
|
except Exception as exc:
|
|
conn.execute("ROLLBACK")
|
|
print(json.dumps({"error": str(exc)}), file=sys.stderr)
|
|
return 1
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def cmd_edit(args: argparse.Namespace) -> int:
|
|
keyword = (args.keyword or "").strip().lower()
|
|
if not keyword:
|
|
print(json.dumps({"error": "keyword must not be empty"}), file=sys.stderr)
|
|
return 1
|
|
|
|
_ensure_db()
|
|
conn = get_db(DB_PATH)
|
|
try:
|
|
matches = _find_by_keyword(conn, keyword)
|
|
if len(matches) == 0:
|
|
print(json.dumps({"error": "no match", "keyword": args.keyword}), file=sys.stderr)
|
|
return 1
|
|
if len(matches) > 1:
|
|
print(
|
|
json.dumps({"error": "ambiguous", "matches": [{"text": m["text"]} for m in matches]}, ensure_ascii=False),
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
rid = matches[0]["id"]
|
|
conn.execute("BEGIN")
|
|
now = _now()
|
|
|
|
if args.text:
|
|
new_text = args.text.strip()
|
|
if not new_text:
|
|
print(json.dumps({"error": "text must not be empty"}), file=sys.stderr)
|
|
return 1
|
|
conn.execute("UPDATE reminders SET text = ?, updated_at = ? WHERE id = ?", (new_text, now, rid))
|
|
log_operation("EDIT", rid, f'text="{new_text}"')
|
|
|
|
if args.replace_schedules:
|
|
conn.execute("DELETE FROM schedule_at WHERE reminder_id = ?", (rid,))
|
|
conn.execute("DELETE FROM schedule_cron WHERE reminder_id = ?", (rid,))
|
|
conn.execute("DELETE FROM schedule_random WHERE reminder_id = ?", (rid,))
|
|
_insert_schedules(conn, rid, args)
|
|
conn.execute("UPDATE reminders SET updated_at = ? WHERE id = ?", (now, rid))
|
|
log_operation("EDIT", rid, "schedules replaced")
|
|
|
|
conn.execute("COMMIT")
|
|
reminder = _fetch_reminder(conn, rid)
|
|
print(json.dumps({"edited": reminder}, ensure_ascii=False))
|
|
return 0
|
|
except Exception as exc:
|
|
conn.execute("ROLLBACK")
|
|
print(json.dumps({"error": str(exc)}), file=sys.stderr)
|
|
return 1
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def cmd_enable(args: argparse.Namespace) -> int:
|
|
keyword = (args.keyword or "").strip().lower()
|
|
if not keyword:
|
|
print(json.dumps({"error": "keyword must not be empty"}), file=sys.stderr)
|
|
return 1
|
|
|
|
_ensure_db()
|
|
conn = get_db(DB_PATH)
|
|
try:
|
|
matches = _find_by_keyword(conn, keyword)
|
|
if len(matches) == 0:
|
|
print(json.dumps({"error": "no match", "keyword": args.keyword}), file=sys.stderr)
|
|
return 1
|
|
if len(matches) > 1:
|
|
print(
|
|
json.dumps({"error": "ambiguous", "matches": [{"text": m["text"]} for m in matches]}, ensure_ascii=False),
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
rid = matches[0]["id"]
|
|
conn.execute("UPDATE reminders SET enabled = 1, updated_at = ? WHERE id = ?", (_now(), rid))
|
|
log_operation("ENABLE", rid, None)
|
|
reminder = _fetch_reminder(conn, rid)
|
|
print(json.dumps({"enabled": reminder}, ensure_ascii=False))
|
|
return 0
|
|
except Exception as exc:
|
|
print(json.dumps({"error": str(exc)}), file=sys.stderr)
|
|
return 1
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def cmd_disable(args: argparse.Namespace) -> int:
|
|
keyword = (args.keyword or "").strip().lower()
|
|
if not keyword:
|
|
print(json.dumps({"error": "keyword must not be empty"}), file=sys.stderr)
|
|
return 1
|
|
|
|
_ensure_db()
|
|
conn = get_db(DB_PATH)
|
|
try:
|
|
matches = _find_by_keyword(conn, keyword)
|
|
if len(matches) == 0:
|
|
print(json.dumps({"error": "no match", "keyword": args.keyword}), file=sys.stderr)
|
|
return 1
|
|
if len(matches) > 1:
|
|
print(
|
|
json.dumps({"error": "ambiguous", "matches": [{"text": m["text"]} for m in matches]}, ensure_ascii=False),
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
rid = matches[0]["id"]
|
|
conn.execute("UPDATE reminders SET enabled = 0, updated_at = ? WHERE id = ?", (_now(), rid))
|
|
log_operation("DISABLE", rid, None)
|
|
reminder = _fetch_reminder(conn, rid)
|
|
print(json.dumps({"disabled": reminder}, ensure_ascii=False))
|
|
return 0
|
|
except Exception as exc:
|
|
print(json.dumps({"error": str(exc)}), file=sys.stderr)
|
|
return 1
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="CRUD for reminders (SQLite backed)")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
sub.add_parser("list", help="List all active reminders as JSON")
|
|
|
|
add_p = sub.add_parser("add", help="Add a new reminder")
|
|
add_p.add_argument("--text", required=True, help="Reminder text")
|
|
add_p.add_argument("--cron", action="append", metavar="EXPR", help="Cron expression (repeatable)")
|
|
add_p.add_argument("--at", action="append", metavar="ISO_DATETIME", help="One-time datetime ISO 8601 (repeatable)")
|
|
add_p.add_argument("--random-times-per-day", type=int, dest="random_times_per_day", metavar="N", help="Random schedule: fires per day")
|
|
add_p.add_argument("--random-window", dest="random_window", metavar="HH:MM-HH:MM", help="Random schedule: daily time window")
|
|
add_p.add_argument("--random-days", dest="random_days", metavar="DOW", help="Random schedule: cron day-of-week filter")
|
|
add_p.add_argument("--random-from", dest="random_from", metavar="YYYY-MM-DD", help="Random schedule: start date")
|
|
add_p.add_argument("--random-until", dest="random_until", metavar="YYYY-MM-DD", help="Random schedule: end date")
|
|
|
|
remove_p = sub.add_parser("remove", help="Remove a reminder by keyword (soft delete)")
|
|
remove_p.add_argument("--keyword", required=True, help="Substring to match against reminder text")
|
|
|
|
edit_p = sub.add_parser("edit", help="Edit a reminder by keyword")
|
|
edit_p.add_argument("--keyword", required=True, help="Substring to match against reminder text")
|
|
edit_p.add_argument("--text", help="New reminder text")
|
|
edit_p.add_argument("--replace-schedules", action="store_true", help="Replace all schedules with new ones")
|
|
edit_p.add_argument("--cron", action="append", metavar="EXPR", help="Cron expression (repeatable)")
|
|
edit_p.add_argument("--at", action="append", metavar="ISO_DATETIME", help="One-time datetime (repeatable)")
|
|
edit_p.add_argument("--random-times-per-day", type=int, dest="random_times_per_day", metavar="N")
|
|
edit_p.add_argument("--random-window", dest="random_window", metavar="HH:MM-HH:MM")
|
|
edit_p.add_argument("--random-days", dest="random_days", metavar="DOW")
|
|
edit_p.add_argument("--random-from", dest="random_from", metavar="YYYY-MM-DD")
|
|
edit_p.add_argument("--random-until", dest="random_until", metavar="YYYY-MM-DD")
|
|
|
|
enable_p = sub.add_parser("enable", help="Enable a reminder by keyword")
|
|
enable_p.add_argument("--keyword", required=True)
|
|
|
|
disable_p = sub.add_parser("disable", help="Disable a reminder by keyword")
|
|
disable_p.add_argument("--keyword", required=True)
|
|
|
|
args = parser.parse_args()
|
|
dispatch = {
|
|
"list": cmd_list,
|
|
"add": cmd_add,
|
|
"remove": cmd_remove,
|
|
"edit": cmd_edit,
|
|
"enable": cmd_enable,
|
|
"disable": cmd_disable,
|
|
}
|
|
return dispatch[args.command](args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|