agentbox
bda54eda3a
- New /change-password route: validates current password, enforces min 4 chars, requires confirmation match - update_password() helper in models.py - Dashboard UI split into two collapsible sections: 👤 Account Settings (password, ntfy, delete account) 🏠 Household Settings (admin only: public link, timezone, delete household) - 14 new i18n keys with fr/de translations - Restored accidentally dropped 'Delete My Account' key
523 lines
17 KiB
Python
523 lines
17 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, generate_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,
|
|
update_password,
|
|
)
|
|
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("/change-password", methods=["POST"])
|
|
@login_required
|
|
def change_password():
|
|
"""Change the logged-in user's password."""
|
|
current = request.form.get("current_password", "")
|
|
new = request.form.get("new_password", "")
|
|
confirm = request.form.get("confirm_password", "")
|
|
|
|
user = get_user_by_id(session["user_id"])
|
|
|
|
if not check_password_hash(user["password_hash"], current):
|
|
flash(t("Current password is incorrect."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
if len(new) < 4:
|
|
flash(t("New password must be at least 4 characters."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
if new != confirm:
|
|
flash(t("New passwords do not match."), "error")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
update_password(session["user_id"], generate_password_hash(new))
|
|
flash(t("Password changed successfully."), "success")
|
|
return redirect(url_for("meals.dashboard"))
|
|
|
|
|
|
@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)
|
|
|
|
|
|
# ── Public sharing ───────────────────────────────────────────────────────────
|
|
|
|
@meals_bp.route("/public/<token>")
|
|
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
|