Kyle Pope
67456c78dd
- Add ntfy columns to Settings model (server_url, topic, auth_token, enabled, per-type toggles, lead times) - Create NtfySent dedup model to prevent duplicate notifications - Create ntfy service with SSRF validation and async httpx send - Create ntfy_templates service with per-type payload builders - Create APScheduler background dispatch job (60s interval, events/reminders/todos/projects) - Register scheduler in main.py lifespan with max_instances=1 - Update SettingsUpdate with ntfy validators (URL scheme, topic regex, lead time ranges) - Update SettingsResponse with ntfy fields; ntfy_has_token computed, token never exposed - Add POST /api/settings/ntfy/test endpoint - Update GET/PUT settings to use explicit _to_settings_response() helper - Add Alembic migration 022 for ntfy settings columns + ntfy_sent table - Add httpx==0.27.2 and apscheduler==3.10.4 to requirements.txt Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
135 lines
4.2 KiB
Python
135 lines
4.2 KiB
Python
"""
|
|
Notification template builders for ntfy push notifications.
|
|
|
|
Each build_* function returns a dict with keys: title, message, tags, priority.
|
|
These are passed directly to send_ntfy_notification().
|
|
"""
|
|
from datetime import datetime, date
|
|
from typing import Optional
|
|
|
|
|
|
# ── Shared helpers ────────────────────────────────────────────────────────────
|
|
|
|
def urgency_label(target_date: date, today: date) -> str:
|
|
"""Human-readable urgency string relative to today."""
|
|
delta = (target_date - today).days
|
|
if delta < 0:
|
|
return f"OVERDUE ({abs(delta)}d ago)"
|
|
elif delta == 0:
|
|
return "Today"
|
|
elif delta == 1:
|
|
return "Tomorrow"
|
|
elif delta <= 7:
|
|
return f"in {delta} days"
|
|
else:
|
|
return target_date.strftime("%d %b")
|
|
|
|
|
|
def day_str(dt: datetime, today: date) -> str:
|
|
"""Return 'Today', 'Tomorrow', or a short date string."""
|
|
d = dt.date()
|
|
if d == today:
|
|
return "Today"
|
|
delta = (d - today).days
|
|
if delta == 1:
|
|
return "Tomorrow"
|
|
return dt.strftime("%a %d %b")
|
|
|
|
|
|
def time_str(dt: datetime, all_day: bool = False) -> str:
|
|
"""Return 'All day' or HH:MM."""
|
|
if all_day:
|
|
return "All day"
|
|
return dt.strftime("%H:%M")
|
|
|
|
|
|
def _truncate(text: str, max_len: int) -> str:
|
|
"""Truncate with ellipsis if over limit."""
|
|
return (text[:max_len - 3] + "...") if len(text) > max_len else text
|
|
|
|
|
|
# ── Template builders ─────────────────────────────────────────────────────────
|
|
|
|
def build_event_notification(
|
|
title: str,
|
|
start_datetime: datetime,
|
|
all_day: bool,
|
|
today: date,
|
|
location_name: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
is_starred: bool = False,
|
|
) -> dict:
|
|
"""Build notification payload for a calendar event reminder."""
|
|
day = day_str(start_datetime, today)
|
|
time = time_str(start_datetime, all_day)
|
|
loc = f" @ {location_name}" if location_name else ""
|
|
desc = f" — {description[:80]}" if description else ""
|
|
|
|
return {
|
|
"title": _truncate(f"Calendar: {title}", 80),
|
|
"message": _truncate(f"{day} at {time}{loc}{desc}", 200),
|
|
"tags": ["calendar"],
|
|
"priority": 4 if is_starred else 3,
|
|
}
|
|
|
|
|
|
def build_reminder_notification(
|
|
title: str,
|
|
remind_at: datetime,
|
|
today: date,
|
|
description: Optional[str] = None,
|
|
) -> dict:
|
|
"""Build notification payload for a general reminder."""
|
|
day = day_str(remind_at, today)
|
|
time = time_str(remind_at)
|
|
desc = f" — {description[:80]}" if description else ""
|
|
|
|
return {
|
|
"title": _truncate(f"Reminder: {title}", 80),
|
|
"message": _truncate(f"{day} at {time}{desc}", 200),
|
|
"tags": ["bell"],
|
|
"priority": 3,
|
|
}
|
|
|
|
|
|
def build_todo_notification(
|
|
title: str,
|
|
due_date: date,
|
|
today: date,
|
|
priority: str = "medium",
|
|
category: Optional[str] = None,
|
|
) -> dict:
|
|
"""Build notification payload for a todo due date alert."""
|
|
urgency = urgency_label(due_date, today)
|
|
priority_label = priority.capitalize()
|
|
cat = f" [{category}]" if category else ""
|
|
# High priority for today/overdue, default otherwise
|
|
ntfy_priority = 4 if (due_date - today).days <= 0 else 3
|
|
|
|
return {
|
|
"title": _truncate(f"Due {urgency}: {title}", 80),
|
|
"message": _truncate(f"Priority: {priority_label}{cat}", 200),
|
|
"tags": ["white_check_mark"],
|
|
"priority": ntfy_priority,
|
|
}
|
|
|
|
|
|
def build_project_notification(
|
|
name: str,
|
|
due_date: date,
|
|
today: date,
|
|
status: str = "in_progress",
|
|
) -> dict:
|
|
"""Build notification payload for a project deadline alert."""
|
|
urgency = urgency_label(due_date, today)
|
|
# Format status label
|
|
status_label = status.replace("_", " ").title()
|
|
ntfy_priority = 4 if due_date <= today else 3
|
|
|
|
return {
|
|
"title": _truncate(f"Project Deadline: {name}", 80),
|
|
"message": _truncate(f"Due {urgency} — Status: {status_label}", 200),
|
|
"tags": ["briefcase"],
|
|
"priority": ntfy_priority,
|
|
}
|