feat(events): recurring expansion, CalDAV pull sync, past search, reminders
- RRULE expansion: list_events now expands recurring events into individual occurrences within the query window using python-dateutil - CalDAV pull sync: new caldav_sync.py + POST /api/events/sync route; imports remote events into the internal store by caldav_uid - Past event search: search_events accepts include_past=true to search historical events; exposed in the LLM tool definition - Internal reminders: migration 0037 adds reminder_minutes + reminder_sent_at columns; event_scheduler.py checks every 5 min and fires push notifications; CalDAV sync job runs hourly - reminder_minutes now stored and returned in create/update routes + tools Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,138 @@
|
||||
"""Scheduler jobs for calendar events.
|
||||
|
||||
- Reminder notifications: checks every 5 minutes for due reminders.
|
||||
- CalDAV pull sync: runs every hour for all users with CalDAV configured.
|
||||
|
||||
Uses the same BackgroundScheduler pattern as briefing_scheduler.py.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from apscheduler.schedulers.background import BackgroundScheduler
|
||||
from apscheduler.triggers.interval import IntervalTrigger
|
||||
from sqlalchemy import select
|
||||
|
||||
from fabledassistant.models import async_session
|
||||
from fabledassistant.models.event import Event
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_scheduler: BackgroundScheduler | None = None
|
||||
_loop: asyncio.AbstractEventLoop | None = None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Reminder job
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def _fire_reminders() -> None:
|
||||
"""Find events with reminders due in the next 5 minutes and fire push notifications."""
|
||||
now = datetime.now(timezone.utc)
|
||||
window_end = now + timedelta(minutes=5)
|
||||
|
||||
async with async_session() as session:
|
||||
result = await session.execute(
|
||||
select(Event).where(
|
||||
Event.reminder_minutes.isnot(None),
|
||||
Event.reminder_sent_at.is_(None),
|
||||
Event.start_dt > now, # event hasn't started yet
|
||||
# reminder fires when now >= start_dt - reminder_minutes
|
||||
# i.e. start_dt <= now + reminder_minutes (approximated by window_end check)
|
||||
)
|
||||
)
|
||||
candidates = list(result.scalars().all())
|
||||
|
||||
to_notify: list[Event] = []
|
||||
for event in candidates:
|
||||
reminder_dt = event.start_dt - timedelta(minutes=event.reminder_minutes)
|
||||
if reminder_dt <= window_end:
|
||||
to_notify.append(event)
|
||||
|
||||
if not to_notify:
|
||||
return
|
||||
|
||||
async with async_session() as session:
|
||||
for event in to_notify:
|
||||
try:
|
||||
from fabledassistant.services.push import send_push_notification # noqa: PLC0415
|
||||
start_local = event.start_dt.strftime("%H:%M")
|
||||
await send_push_notification(
|
||||
user_id=event.user_id,
|
||||
title=f"Reminder: {event.title}",
|
||||
body=f"Starting at {start_local} UTC",
|
||||
)
|
||||
except Exception:
|
||||
logger.warning("Failed to send reminder push for event %d", event.id, exc_info=True)
|
||||
|
||||
# Mark as sent regardless of push success to avoid re-firing
|
||||
result = await session.execute(
|
||||
select(Event).where(Event.id == event.id)
|
||||
)
|
||||
ev = result.scalar_one_or_none()
|
||||
if ev:
|
||||
ev.reminder_sent_at = datetime.now(timezone.utc)
|
||||
await session.commit()
|
||||
|
||||
|
||||
def _run_reminders(loop: asyncio.AbstractEventLoop) -> None:
|
||||
asyncio.run_coroutine_threadsafe(_fire_reminders(), loop)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CalDAV pull sync job
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def _run_caldav_sync() -> None:
|
||||
from fabledassistant.services.caldav_sync import sync_all_users # noqa: PLC0415
|
||||
try:
|
||||
await sync_all_users()
|
||||
except Exception:
|
||||
logger.warning("CalDAV pull sync job failed", exc_info=True)
|
||||
|
||||
|
||||
def _run_caldav_sync_threadsafe(loop: asyncio.AbstractEventLoop) -> None:
|
||||
asyncio.run_coroutine_threadsafe(_run_caldav_sync(), loop)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Lifecycle
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def start_event_scheduler(loop: asyncio.AbstractEventLoop) -> None:
|
||||
global _scheduler, _loop
|
||||
if _scheduler is not None:
|
||||
return
|
||||
_loop = loop
|
||||
_scheduler = BackgroundScheduler()
|
||||
|
||||
# Check reminders every 5 minutes
|
||||
_scheduler.add_job(
|
||||
_run_reminders,
|
||||
trigger=IntervalTrigger(minutes=5),
|
||||
args=[loop],
|
||||
id="event_reminders",
|
||||
replace_existing=True,
|
||||
)
|
||||
|
||||
# CalDAV pull sync every hour
|
||||
_scheduler.add_job(
|
||||
_run_caldav_sync_threadsafe,
|
||||
trigger=IntervalTrigger(hours=1),
|
||||
args=[loop],
|
||||
id="caldav_pull_sync",
|
||||
replace_existing=True,
|
||||
)
|
||||
|
||||
_scheduler.start()
|
||||
logger.info("Event scheduler started (reminders every 5m, CalDAV sync every 1h)")
|
||||
|
||||
|
||||
def stop_event_scheduler() -> None:
|
||||
global _scheduler
|
||||
if _scheduler is not None:
|
||||
_scheduler.shutdown(wait=False)
|
||||
_scheduler = None
|
||||
logger.info("Event scheduler stopped")
|
||||
Reference in New Issue
Block a user