agentbox
09eda19fc4
CI / test-and-package (push) Successful in 42s
- ntfy.py: switch from JSON body to HTTP headers (Title, Priority, Tags, Actions, Click) with plain message body. ntfy servers treat the POST body as the message, so JSON body fields were ignored and displayed as raw JSON. - ntfy.py: replace em dashes in headers with hyphens (HTTP headers must be latin-1 encodable). - meals.py: add /auto-login/<token> route that logs users in via their ntfy callback token so clicking a notification opens the dashboard without re-entering credentials. - ntfy.py: update Click URL to /auto-login/<token>?date=...
118 lines
3.6 KiB
Python
118 lines
3.6 KiB
Python
"""Ntfy notification helpers — send meal reminders with action buttons."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import urllib.request
|
|
import urllib.error
|
|
from datetime import date
|
|
|
|
|
|
def _ntfy_headers(token: str, extra: dict[str, str] | None = None) -> dict[str, str]:
|
|
"""Build headers for ntfy request, optionally with Authorization."""
|
|
headers = dict(extra) if extra else {}
|
|
if token.strip():
|
|
headers["Authorization"] = f"Bearer {token.strip()}"
|
|
return headers
|
|
|
|
|
|
def send_meal_reminder(
|
|
ntfy_url: str,
|
|
ntfy_token: str,
|
|
callback_token: str,
|
|
meal_type: str,
|
|
for_date: date,
|
|
app_base_url: str,
|
|
) -> bool:
|
|
"""Send a meal reminder notification with Home/Out action buttons.
|
|
|
|
Args:
|
|
ntfy_url: Full topic URL (e.g. https://ntfy.sh/mytopic).
|
|
ntfy_token: Bearer token for auth (empty for public topics).
|
|
callback_token: Per-user token for the webhook callback.
|
|
meal_type: 'lunch' or 'dinner'.
|
|
for_date: The meal date.
|
|
app_base_url: Base URL of this app (e.g. https://meals.ct.cozytren.ch).
|
|
|
|
Returns:
|
|
True if the notification was sent successfully.
|
|
"""
|
|
date_str = for_date.isoformat()
|
|
meal_label = "Lunch" if meal_type == "lunch" else "Dinner"
|
|
|
|
callback_url = f"{app_base_url.rstrip('/')}/api/ntfy-callback"
|
|
|
|
topic = ntfy_url.rstrip("/").rsplit("/", 1)[-1]
|
|
message = f"Will you be home for {meal_label.lower()} on {date_str}?"
|
|
|
|
home_body = json.dumps({
|
|
"token": callback_token,
|
|
"meal_type": meal_type,
|
|
"date": date_str,
|
|
"status": "yes",
|
|
})
|
|
out_body = json.dumps({
|
|
"token": callback_token,
|
|
"meal_type": meal_type,
|
|
"date": date_str,
|
|
"status": "no",
|
|
})
|
|
|
|
actions = json.dumps([
|
|
{
|
|
"action": "http",
|
|
"label": "Home",
|
|
"method": "POST",
|
|
"url": callback_url,
|
|
"headers": {"Content-Type": "application/json"},
|
|
"body": home_body,
|
|
},
|
|
{
|
|
"action": "http",
|
|
"label": "Out",
|
|
"method": "POST",
|
|
"url": callback_url,
|
|
"headers": {"Content-Type": "application/json"},
|
|
"body": out_body,
|
|
},
|
|
])
|
|
|
|
headers = _ntfy_headers(ntfy_token, {
|
|
"Title": f"{meal_label} - Meal Reminder",
|
|
"Priority": "4",
|
|
"Tags": "plate_with_cutlery",
|
|
"Actions": actions,
|
|
"Click": f"{app_base_url.rstrip('/')}/auto-login/{callback_token}?date={date_str}",
|
|
})
|
|
|
|
data = message.encode("utf-8")
|
|
req = urllib.request.Request(ntfy_url, data=data, headers=headers)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
return 200 <= resp.status < 300
|
|
except urllib.error.URLError as e:
|
|
print(f"[ntfy] Failed to send to {ntfy_url}: {e}")
|
|
return False
|
|
|
|
|
|
def send_test_notification(ntfy_url: str, ntfy_token: str) -> bool:
|
|
"""Send a simple test notification to verify ntfy configuration."""
|
|
topic = ntfy_url.rstrip("/").rsplit("/", 1)[-1]
|
|
|
|
headers = _ntfy_headers(ntfy_token, {
|
|
"Title": "Meal Tracker - Test",
|
|
"Priority": "3",
|
|
"Tags": "white_check_mark",
|
|
})
|
|
|
|
data = "Your ntfy integration is working! You'll receive meal reminders here.".encode("utf-8")
|
|
req = urllib.request.Request(ntfy_url, data=data, headers=headers)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
return 200 <= resp.status < 300
|
|
except urllib.error.URLError as e:
|
|
print(f"[ntfy] Test failed for {ntfy_url}: {e}")
|
|
return False
|