"""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