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=...
433 lines
14 KiB
Python
433 lines
14 KiB
Python
"""Meals blueprint — dashboard, responses, history."""
|
|
|
|
from datetime import date, datetime, timedelta
|
|
|
|
from flask import (
|
|
Blueprint,
|
|
render_template,
|
|
request,
|
|
redirect,
|
|
url_for,
|
|
session,
|
|
flash,
|
|
)
|
|
|
|
from werkzeug.security import check_password_hash
|
|
|
|
from models import (
|
|
get_dashboard_data,
|
|
upsert_response,
|
|
ensure_meal_periods,
|
|
get_user_by_id,
|
|
delete_user,
|
|
delete_household,
|
|
count_household_users,
|
|
set_user_ntfy,
|
|
get_user_ntfy,
|
|
get_user_by_callback_token,
|
|
set_household_timezone,
|
|
get_household_by_id,
|
|
get_all_households,
|
|
get_users_with_ntfy,
|
|
update_reminder_sent,
|
|
)
|
|
from i18n import t
|
|
from ntfy import send_test_notification, send_meal_reminder
|
|
import os
|
|
from zoneinfo import ZoneInfo
|
|
|
|
meals_bp = Blueprint("meals", __name__)
|
|
|
|
|
|
def login_required(f):
|
|
"""Decorator: redirect to login if not authenticated or session is stale."""
|
|
from functools import wraps
|
|
|
|
@wraps(f)
|
|
def wrapper(*args, **kwargs):
|
|
if "user_id" not in session or "household_id" not in session:
|
|
session.clear()
|
|
return redirect(url_for("auth.login"))
|
|
return f(*args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
@meals_bp.route("/")
|
|
def index():
|
|
if "user_id" in session:
|
|
return redirect(url_for("meals.dashboard"))
|
|
return redirect(url_for("auth.login"))
|
|
|
|
|
|
@meals_bp.route("/dashboard")
|
|
@login_required
|
|
def dashboard():
|
|
today = date.today()
|
|
household_id = session["household_id"]
|
|
|
|
# Allow viewing other dates via query param
|
|
date_str = request.args.get("date")
|
|
if date_str:
|
|
try:
|
|
today = datetime.strptime(date_str, "%Y-%m-%d").date()
|
|
except ValueError:
|
|
flash(t("Invalid date format. Use YYYY-MM-DD."), "error")
|
|
|
|
if household_id is None:
|
|
session.clear()
|
|
return redirect(url_for("auth.login"))
|
|
|
|
data = get_dashboard_data(today, household_id)
|
|
ntfy = get_user_ntfy(session["user_id"])
|
|
household = get_household_by_id(household_id)
|
|
return render_template(
|
|
"dashboard.html",
|
|
dashboard=data,
|
|
viewing_date=today,
|
|
today=date.today(),
|
|
prev_date=today - timedelta(days=1),
|
|
next_date=today + timedelta(days=1),
|
|
current_user_id=session["user_id"],
|
|
is_admin=session["is_admin"],
|
|
ntfy=ntfy,
|
|
household=household,
|
|
)
|
|
|
|
|
|
@meals_bp.route("/respond", methods=["POST"])
|
|
@login_required
|
|
def respond():
|
|
meal_type = request.form.get("meal_type", "").strip()
|
|
status = request.form.get("status", "").strip()
|
|
date_str = request.form.get("date", date.today().isoformat())
|
|
|
|
if meal_type not in ("lunch", "dinner"):
|
|
flash(t("Invalid meal type."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
if status not in ("yes", "no"):
|
|
flash(t("Invalid status."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
try:
|
|
target_date = datetime.strptime(date_str, "%Y-%m-%d").date()
|
|
except ValueError:
|
|
flash(t("Invalid date."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
if target_date < date.today():
|
|
flash(t("Cannot change responses for past dates."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
periods = ensure_meal_periods(target_date)
|
|
period = next((p for p in periods if p["meal_type"] == meal_type), None)
|
|
if period is None:
|
|
flash(t("Meal period not found."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
upsert_response(session["user_id"], period["id"], status)
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
|
|
@meals_bp.route("/history")
|
|
@login_required
|
|
def history():
|
|
date_str = request.args.get("date", date.today().isoformat())
|
|
try:
|
|
target_date = datetime.strptime(date_str, "%Y-%m-%d").date()
|
|
except ValueError:
|
|
target_date = date.today()
|
|
|
|
data = get_dashboard_data(target_date, session["household_id"])
|
|
ntfy = get_user_ntfy(session["user_id"])
|
|
household = get_household_by_id(session["household_id"])
|
|
return render_template(
|
|
"dashboard.html",
|
|
dashboard=data,
|
|
viewing_date=target_date,
|
|
today=date.today(),
|
|
prev_date=target_date - timedelta(days=1),
|
|
next_date=target_date + timedelta(days=1),
|
|
current_user_id=session["user_id"],
|
|
is_admin=session["is_admin"],
|
|
history_mode=True,
|
|
ntfy=ntfy,
|
|
household=household,
|
|
)
|
|
|
|
|
|
# ── Account / household management ────────────────────────────────────────────
|
|
|
|
@meals_bp.route("/delete-account", methods=["POST"])
|
|
@login_required
|
|
def delete_account():
|
|
"""Delete the currently logged-in user's own account."""
|
|
password = request.form.get("password", "")
|
|
user = get_user_by_id(session["user_id"])
|
|
|
|
if not check_password_hash(user["password_hash"], password):
|
|
flash(t("Incorrect password. Account not deleted."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
household_id = session["household_id"]
|
|
is_admin = session["is_admin"]
|
|
user_count = count_household_users(household_id)
|
|
|
|
# If admin is the last user, delete the household too
|
|
if is_admin and user_count == 1:
|
|
delete_household(household_id)
|
|
flash(t("Your account and household have been deleted."), "success")
|
|
else:
|
|
delete_user(session["user_id"])
|
|
flash(t("Your account has been deleted."), "success")
|
|
|
|
# Clear auth keys but keep the session so flash survives the redirect
|
|
for key in ("user_id", "username", "is_admin", "household_id", "household_name"):
|
|
session.pop(key, None)
|
|
return redirect(url_for("auth.login"))
|
|
|
|
|
|
@meals_bp.route("/admin/remove-user/<int:user_id>", methods=["POST"])
|
|
@login_required
|
|
def admin_remove_user(user_id):
|
|
"""Admin removes a user from the household."""
|
|
if not session["is_admin"]:
|
|
flash(t("Only admins can remove users."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
user = get_user_by_id(user_id)
|
|
if user is None or user["household_id"] != session["household_id"]:
|
|
flash(t("User not found in your household."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
if user["is_admin"]:
|
|
flash(t("Cannot remove the admin. Transfer admin role first or delete the household."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
delete_user(user_id)
|
|
flash(t("User '{name}' has been removed.", name=user["username"]), "success")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
|
|
@meals_bp.route("/admin/delete-household", methods=["POST"])
|
|
@login_required
|
|
def admin_delete_household():
|
|
"""Admin deletes the entire household."""
|
|
if not session["is_admin"]:
|
|
flash(t("Only admins can delete the household."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
password = request.form.get("password", "")
|
|
user = get_user_by_id(session["user_id"])
|
|
|
|
if not check_password_hash(user["password_hash"], password):
|
|
flash(t("Incorrect password. Household not deleted."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
delete_household(session["household_id"])
|
|
flash(t("Household and all members have been deleted."), "success")
|
|
for key in ("user_id", "username", "is_admin", "household_id", "household_name"):
|
|
session.pop(key, None)
|
|
return redirect(url_for("auth.login"))
|
|
|
|
|
|
@meals_bp.route("/admin/timezone", methods=["POST"])
|
|
@login_required
|
|
def admin_set_timezone():
|
|
"""Admin sets the household timezone."""
|
|
if not session["is_admin"]:
|
|
flash(t("Only admins can change the timezone."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
timezone = request.form.get("timezone", "").strip()
|
|
if not timezone:
|
|
flash(t("Please select a timezone."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
set_household_timezone(session["household_id"], timezone)
|
|
flash(t("Timezone updated."), "success")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
|
|
# ── Ntfy integration ─────────────────────────────────────────────────────────
|
|
|
|
@meals_bp.route("/ntfy-settings", methods=["POST"])
|
|
@login_required
|
|
def ntfy_settings():
|
|
"""Save ntfy URL and token for the current user."""
|
|
import secrets
|
|
|
|
ntfy_url = request.form.get("ntfy_url", "").strip()
|
|
ntfy_token = request.form.get("ntfy_token", "").strip()
|
|
|
|
existing = get_user_ntfy(session["user_id"])
|
|
callback_token = existing.get("callback_token") if existing else ""
|
|
|
|
# Generate a new callback token if user is enabling ntfy for the first time
|
|
if ntfy_url and not callback_token:
|
|
callback_token = secrets.token_urlsafe(32)
|
|
|
|
# If clearing ntfy, also clear the callback token
|
|
if not ntfy_url:
|
|
callback_token = ""
|
|
ntfy_token = ""
|
|
|
|
# Capture the base URL from the user's browser request
|
|
base_url = request.host_url.rstrip('/') if ntfy_url else ""
|
|
|
|
set_user_ntfy(session["user_id"], ntfy_url, ntfy_token, callback_token, base_url)
|
|
flash(t("Ntfy settings saved."), "success")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
|
|
@meals_bp.route("/ntfy-test", methods=["POST"])
|
|
@login_required
|
|
def ntfy_test():
|
|
"""Send a test ntfy notification to verify the configuration."""
|
|
ntfy_url = request.form.get("ntfy_url", "").strip()
|
|
ntfy_token = request.form.get("ntfy_token", "").strip()
|
|
|
|
if not ntfy_url:
|
|
flash(t("Please enter an ntfy URL first."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
ok = send_test_notification(ntfy_url, ntfy_token)
|
|
if ok:
|
|
flash(t("Test notification sent! Check your device."), "success")
|
|
else:
|
|
flash(t("Failed to send test notification. Check your ntfy URL and token."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
|
|
@meals_bp.route("/api/ntfy-callback", methods=["POST"])
|
|
def ntfy_callback():
|
|
"""Receive ntfy action callbacks (no login — authenticated via callback_token).
|
|
|
|
Expects JSON body: {"token": "...", "meal_type": "lunch|dinner",
|
|
"date": "YYYY-MM-DD", "status": "yes|no"}
|
|
"""
|
|
data = request.get_json(silent=True)
|
|
if not data:
|
|
return {"error": "Invalid JSON"}, 400
|
|
|
|
token = data.get("token", "")
|
|
meal_type = data.get("meal_type", "")
|
|
date_str = data.get("date", "")
|
|
status = data.get("status", "")
|
|
|
|
user = get_user_by_callback_token(token)
|
|
if user is None:
|
|
return {"error": "Invalid token"}, 403
|
|
|
|
if meal_type not in ("lunch", "dinner"):
|
|
return {"error": "Invalid meal_type"}, 400
|
|
if status not in ("yes", "no"):
|
|
return {"error": "Invalid status"}, 400
|
|
|
|
try:
|
|
target_date = datetime.strptime(date_str, "%Y-%m-%d").date()
|
|
except (ValueError, TypeError):
|
|
return {"error": "Invalid date"}, 400
|
|
|
|
if target_date < date.today():
|
|
return {"error": "Cannot change past meals"}, 400
|
|
|
|
periods = ensure_meal_periods(target_date)
|
|
period = next((p for p in periods if p["meal_type"] == meal_type), None)
|
|
if period is None:
|
|
return {"error": "Meal period not found"}, 400
|
|
|
|
upsert_response(user["id"], period["id"], status)
|
|
return {"message": "Response recorded", "status": status}, 200
|
|
|
|
|
|
# ── Auto-login from ntfy notification click ──────────────────────────────────
|
|
|
|
@meals_bp.route("/auto-login/<token>")
|
|
def auto_login(token):
|
|
"""Log in a user via their ntfy callback token, then redirect to dashboard.
|
|
|
|
Clicking an ntfy notification opens this URL. The callback token
|
|
identifies the user, so they don't need to re-enter credentials.
|
|
"""
|
|
user = get_user_by_callback_token(token)
|
|
if user is None:
|
|
flash(t("Invalid login link."), "error")
|
|
return redirect(url_for("auth.login"))
|
|
|
|
session["user_id"] = user["id"]
|
|
session["username"] = user["username"]
|
|
session["is_admin"] = bool(user["is_admin"])
|
|
session["household_id"] = user["household_id"]
|
|
session["household_name"] = user["household_name"] or ""
|
|
|
|
date_param = request.args.get("date", "")
|
|
target = url_for("meals.dashboard")
|
|
if date_param:
|
|
target += f"?date={date_param}"
|
|
return redirect(target)
|
|
|
|
|
|
# ── Cron tick (called by meal-cron container) ─────────────────────────────────
|
|
|
|
REMINDER_HOURS = {"lunch": 10, "dinner": 16}
|
|
|
|
|
|
def _resolve_timezone(tz_str: str) -> ZoneInfo:
|
|
try:
|
|
return ZoneInfo(tz_str)
|
|
except Exception:
|
|
return ZoneInfo("UTC")
|
|
|
|
|
|
@meals_bp.route("/api/cron-tick", methods=["POST"])
|
|
def cron_tick():
|
|
"""Called by the cron container every 60s.
|
|
|
|
Checks each household's local time and sends ntfy reminders
|
|
at 10:00 (lunch) and 16:00 (dinner). Protected by X-Cron-Secret header.
|
|
"""
|
|
expected = os.environ.get("CRON_SECRET", "")
|
|
if expected and request.headers.get("X-Cron-Secret") != expected:
|
|
return {"error": "unauthorized"}, 403
|
|
|
|
now_utc = datetime.now(ZoneInfo("UTC"))
|
|
sent = 0
|
|
|
|
for hh in get_all_households():
|
|
tz = _resolve_timezone(hh.get("timezone", "UTC"))
|
|
local_now = now_utc.astimezone(tz)
|
|
today_str = local_now.strftime("%Y-%m-%d")
|
|
hour = local_now.hour
|
|
minute = local_now.minute
|
|
|
|
for meal_type, trigger_hour in REMINDER_HOURS.items():
|
|
if hour != trigger_hour or minute >= 5:
|
|
continue
|
|
|
|
last_col = f"last_{meal_type}_reminder"
|
|
if hh.get(last_col) == today_str:
|
|
continue
|
|
|
|
users = get_users_with_ntfy(hh["id"])
|
|
if not users:
|
|
continue
|
|
|
|
for user in users:
|
|
base_url = user.get("base_url", "") or request.host_url.rstrip("/")
|
|
ok = send_meal_reminder(
|
|
ntfy_url=user["ntfy_url"],
|
|
ntfy_token=user["ntfy_token"],
|
|
callback_token=user["callback_token"],
|
|
meal_type=meal_type,
|
|
for_date=local_now.date(),
|
|
app_base_url=base_url,
|
|
)
|
|
if ok:
|
|
sent += 1
|
|
|
|
update_reminder_sent(hh["id"], meal_type, today_str)
|
|
|
|
return {"sent": sent}, 200
|