"""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, ) from i18n import t from ntfy import send_test_notification 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/", 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 = "" set_user_ntfy(session["user_id"], ntfy_url, ntfy_token, callback_token) 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