106 lines
3.1 KiB
Python
106 lines
3.1 KiB
Python
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,
|
|
)
|