"""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, get_household_by_public_token, generate_public_token, set_public_enabled, ) 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/", 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/") 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) # ── Public sharing ─────────────────────────────────────────────────────────── @meals_bp.route("/public/") def public_dashboard(token): """Public, read-only dashboard showing the household's meal status. No login required — the token is a hard-to-guess secret. Query params: - date: YYYY-MM-DD (default: today) """ household = get_household_by_public_token(token) if household is None: return render_template("public-not-found.html"), 404 today = date.today() date_str = request.args.get("date") if date_str: try: today = datetime.strptime(date_str, "%Y-%m-%d").date() except ValueError: pass data = get_dashboard_data(today, household["id"]) return render_template( "public.html", dashboard=data, household=household, viewing_date=today, today=date.today(), prev_date=today - timedelta(days=1), next_date=today + timedelta(days=1), ) @meals_bp.route("/admin/public-link", methods=["POST"]) @login_required def admin_public_link(): """Admin enables/disables or regenerates the public sharing link.""" if not session["is_admin"]: flash(t("Only admins can manage the public link."), "error") return redirect(url_for("meals.dashboard")) action = request.form.get("action", "").strip() household_id = session["household_id"] if action == "generate": token = generate_public_token(household_id) flash(t("Public link generated!"), "success") elif action == "disable": set_public_enabled(household_id, False) flash(t("Public link disabled."), "success") elif action == "enable": set_public_enabled(household_id, True) flash(t("Public link enabled."), "success") else: flash(t("Unknown action."), "error") return redirect(url_for("meals.dashboard")) # ── 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