Add ntfy push and Listmonk newsletter with lightweight Docker deploy tooling.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/ntfy_push.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+import re
+from datetime import datetime, time
+from typing import Any
+from zoneinfo import ZoneInfo
+
+import httpx
+
+from .settings import get_settings
+
+
+def ntfy_topic_for_user(user_id: str) -> str:
+    clean = re.sub(r"[^a-zA-Z0-9_-]", "", user_id.strip())[:32]
+    return f"nomadcna-{clean or 'guest'}"
+
+
+def ntfy_subscribe_url(user_id: str) -> str:
+    settings = get_settings()
+    topic = ntfy_topic_for_user(user_id)
+    return f"{settings.ntfy_url}/{topic}"
+
+
+def _in_quiet_hours(preferences: dict[str, Any]) -> bool:
+    start_raw = str(preferences.get("quietHoursStart") or "22:00")
+    end_raw = str(preferences.get("quietHoursEnd") or "08:00")
+    try:
+        start = time.fromisoformat(start_raw[:5])
+        end = time.fromisoformat(end_raw[:5])
+    except ValueError:
+        return False
+    now = datetime.now(ZoneInfo("Asia/Shanghai")).time()
+    if start <= end:
+        return start <= now <= end
+    return now >= start or now <= end
+
+
+def _should_push(preferences: dict[str, Any], category: str, priority: str) -> bool:
+    channels = preferences.get("channels") or {}
+    categories = preferences.get("categories") or {}
+    if not channels.get("push"):
+        return False
+    if not categories.get(category, True):
+        return False
+    if priority == "high":
+        return True
+    return not _in_quiet_hours(preferences)
+
+
+async def push_ntfy_message(
+    *,
+    topic: str,
+    title: str,
+    message: str,
+    click: str = "",
+    priority: str = "default",
+    tags: str = "",
+) -> bool:
+    settings = get_settings()
+    if not settings.ntfy_enabled:
+        return False
+    url = f"{settings.ntfy_url.rstrip('/')}/{topic}"
+    headers: dict[str, str] = {"Title": title[:250]}
+    if click:
+        headers["Click"] = click[:1000]
+    if tags:
+        headers["Tags"] = tags[:100]
+    if priority in {"min", "low", "default", "high", "max", "urgent"}:
+        headers["Priority"] = priority
+    auth = settings.ntfy_auth_token.strip()
+    if auth:
+        headers["Authorization"] = f"Bearer {auth}"
+    try:
+        async with httpx.AsyncClient(timeout=12.0) as client:
+            response = await client.post(url, content=message[:4000], headers=headers)
+            return response.status_code < 400
+    except httpx.HTTPError:
+        return False
+
+
+async def dispatch_ntfy_for_user(
+    *,
+    user_id: str,
+    preferences: dict[str, Any],
+    title: str,
+    body: str,
+    category: str = "system",
+    priority: str = "normal",
+    action_url: str = "",
+    icon: str = "🔔",
+) -> bool:
+    if not user_id or not _should_push(preferences, category, priority):
+        return False
+    settings = get_settings()
+    site = settings.site_public_url.rstrip("/")
+    click = action_url if action_url.startswith("http") else f"{site}{action_url}" if action_url else site
+    ntfy_priority = "high" if priority == "high" else "default"
+    return await push_ntfy_message(
+        topic=ntfy_topic_for_user(user_id),
+        title=title,
+        message=body or title,
+        click=click,
+        priority=ntfy_priority,
+        tags=icon,
+    )