chelsea
ecb79af44e
- Fix bot auth: merge duplicate on_ready handlers so session restore runs (#13) - Fix push notifications: pass Uint8Array directly as applicationServerKey (#6) - Show specific conflict reason on schedule save instead of generic error (#17) - Add inline checkmark button to complete tasks on routines timeline (#18) - Add visibility-change + 60s polling auto-refresh to routines, meds, tasks (#15) - Add every-N-day routine scheduling: schema, API, scheduler, and UI (#16) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
829 lines
33 KiB
Python
829 lines
33 KiB
Python
"""
|
|
Routines API - routine management
|
|
|
|
Routines have ordered steps. Users start sessions to walk through them.
|
|
"""
|
|
|
|
import os
|
|
import uuid
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
import flask
|
|
import jwt
|
|
import core.auth as auth
|
|
import core.postgres as postgres
|
|
import core.routines as routines_core
|
|
import core.tz as tz
|
|
|
|
|
|
def _get_user_uuid(token):
|
|
try:
|
|
payload = jwt.decode(token, os.getenv("JWT_SECRET"), algorithms=["HS256"])
|
|
return payload.get("sub")
|
|
except (jwt.ExpiredSignatureError, jwt.InvalidTokenError):
|
|
return None
|
|
|
|
|
|
def _auth(request):
|
|
"""Extract and verify token. Returns user_uuid or None."""
|
|
header = request.headers.get("Authorization", "")
|
|
if not header.startswith("Bearer "):
|
|
return None
|
|
token = header[7:]
|
|
user_uuid = _get_user_uuid(token)
|
|
if not user_uuid or not auth.verifyLoginToken(token, userUUID=user_uuid):
|
|
return None
|
|
return user_uuid
|
|
|
|
|
|
def _make_aware_utc(dt):
|
|
"""Ensure a datetime is timezone-aware; assume naive datetimes are UTC."""
|
|
if dt.tzinfo is None:
|
|
from datetime import timezone as _tz
|
|
|
|
return dt.replace(tzinfo=_tz.utc)
|
|
return dt
|
|
|
|
|
|
def _record_step_result(session_id, step_id, step_index, result, session):
|
|
"""Record a per-step result (completed or skipped)."""
|
|
try:
|
|
# Compute duration from previous step completion or session start
|
|
prev_results = postgres.select(
|
|
"routine_step_results",
|
|
where={"session_id": session_id},
|
|
order_by="completed_at DESC",
|
|
limit=1,
|
|
)
|
|
now = tz.user_now()
|
|
if prev_results:
|
|
last_completed = prev_results[0].get("completed_at")
|
|
if last_completed:
|
|
if isinstance(last_completed, str):
|
|
last_completed = datetime.fromisoformat(last_completed)
|
|
last_completed = _make_aware_utc(last_completed)
|
|
duration_seconds = max(0, int((now - last_completed).total_seconds()))
|
|
else:
|
|
duration_seconds = None
|
|
else:
|
|
created_at = session.get("created_at")
|
|
if created_at:
|
|
if isinstance(created_at, str):
|
|
created_at = datetime.fromisoformat(created_at)
|
|
created_at = _make_aware_utc(created_at)
|
|
duration_seconds = max(0, int((now - created_at).total_seconds()))
|
|
else:
|
|
duration_seconds = None
|
|
|
|
postgres.insert(
|
|
"routine_step_results",
|
|
{
|
|
"id": str(uuid.uuid4()),
|
|
"session_id": session_id,
|
|
"step_id": step_id,
|
|
"step_index": step_index,
|
|
"result": result,
|
|
"duration_seconds": duration_seconds,
|
|
"completed_at": now.isoformat(),
|
|
},
|
|
)
|
|
except Exception:
|
|
pass # Don't fail the step completion if tracking fails
|
|
|
|
|
|
def _complete_session_with_celebration(session_id, user_uuid, session):
|
|
"""Complete a session and return celebration data."""
|
|
now = tz.user_now()
|
|
created_at = session.get("created_at")
|
|
if created_at:
|
|
if isinstance(created_at, str):
|
|
created_at = datetime.fromisoformat(created_at)
|
|
created_at = _make_aware_utc(created_at)
|
|
duration_minutes = max(0, round((now - created_at).total_seconds() / 60, 1))
|
|
else:
|
|
duration_minutes = 0
|
|
|
|
# Update session as completed with duration — this MUST succeed
|
|
postgres.update(
|
|
"routine_sessions",
|
|
{
|
|
"status": "completed",
|
|
"completed_at": now.isoformat(),
|
|
"actual_duration_minutes": int(duration_minutes),
|
|
},
|
|
{"id": session_id},
|
|
)
|
|
|
|
# Gather celebration stats — failures here should not break completion
|
|
streak_current = 1
|
|
streak_longest = 1
|
|
streak_milestone = None
|
|
steps_completed = 0
|
|
steps_skipped = 0
|
|
total_completions = 1
|
|
|
|
try:
|
|
streak_result = routines_core._update_streak(user_uuid, session["routine_id"])
|
|
streak = postgres.select_one(
|
|
"routine_streaks",
|
|
{
|
|
"user_uuid": user_uuid,
|
|
"routine_id": session["routine_id"],
|
|
},
|
|
)
|
|
if streak:
|
|
streak_current = streak["current_streak"]
|
|
streak_longest = streak["longest_streak"]
|
|
streak_milestone = streak_result.get("milestone") if streak_result else None
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
step_results = postgres.select(
|
|
"routine_step_results", {"session_id": session_id}
|
|
)
|
|
steps_completed = sum(1 for r in step_results if r.get("result") == "completed")
|
|
steps_skipped = sum(1 for r in step_results if r.get("result") == "skipped")
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
all_completed = postgres.select(
|
|
"routine_sessions",
|
|
{
|
|
"routine_id": session["routine_id"],
|
|
"user_uuid": user_uuid,
|
|
"status": "completed",
|
|
},
|
|
)
|
|
total_completions = len(all_completed)
|
|
except Exception:
|
|
pass
|
|
|
|
result = {
|
|
"streak_current": streak_current,
|
|
"streak_longest": streak_longest,
|
|
"session_duration_minutes": duration_minutes,
|
|
"total_completions": total_completions,
|
|
"steps_completed": steps_completed,
|
|
"steps_skipped": steps_skipped,
|
|
}
|
|
if streak_milestone:
|
|
result["streak_milestone"] = streak_milestone
|
|
return result
|
|
|
|
|
|
def register(app):
|
|
# ── Routines CRUD ─────────────────────────────────────────────
|
|
|
|
@app.route("/api/routines", methods=["GET"])
|
|
def api_listRoutines():
|
|
"""List all routines for the logged-in user."""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
routines = postgres.select(
|
|
"routines", where={"user_uuid": user_uuid}, order_by="name"
|
|
)
|
|
return flask.jsonify(routines), 200
|
|
|
|
@app.route("/api/routines", methods=["POST"])
|
|
def api_createRoutine():
|
|
"""Create a new routine. Body: {name, description?, icon?}"""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
data = flask.request.get_json()
|
|
if not data:
|
|
return flask.jsonify({"error": "missing body"}), 400
|
|
if not data.get("name"):
|
|
return flask.jsonify({"error": "missing required field: name"}), 400
|
|
data["id"] = str(uuid.uuid4())
|
|
data["user_uuid"] = user_uuid
|
|
routine = postgres.insert("routines", data)
|
|
return flask.jsonify(routine), 201
|
|
|
|
@app.route("/api/routines/<routine_id>", methods=["GET"])
|
|
def api_getRoutine(routine_id):
|
|
"""Get a routine with its steps."""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
routine = postgres.select_one(
|
|
"routines", {"id": routine_id, "user_uuid": user_uuid}
|
|
)
|
|
if not routine:
|
|
return flask.jsonify({"error": "not found"}), 404
|
|
steps = postgres.select(
|
|
"routine_steps",
|
|
where={"routine_id": routine_id},
|
|
order_by="position",
|
|
)
|
|
return flask.jsonify({"routine": routine, "steps": steps}), 200
|
|
|
|
@app.route("/api/routines/<routine_id>", methods=["PUT"])
|
|
def api_updateRoutine(routine_id):
|
|
"""Update routine details. Body: {name?, description?, icon?}"""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
data = flask.request.get_json()
|
|
if not data:
|
|
return flask.jsonify({"error": "missing body"}), 400
|
|
existing = postgres.select_one(
|
|
"routines", {"id": routine_id, "user_uuid": user_uuid}
|
|
)
|
|
if not existing:
|
|
return flask.jsonify({"error": "not found"}), 404
|
|
allowed = [
|
|
"name",
|
|
"description",
|
|
"icon",
|
|
"location",
|
|
"environment_prompts",
|
|
"habit_stack_after",
|
|
]
|
|
updates = {k: v for k, v in data.items() if k in allowed}
|
|
if not updates:
|
|
return flask.jsonify({"error": "no valid fields to update"}), 400
|
|
result = postgres.update(
|
|
"routines", updates, {"id": routine_id, "user_uuid": user_uuid}
|
|
)
|
|
return flask.jsonify(result[0] if result else {}), 200
|
|
|
|
@app.route("/api/routines/<routine_id>", methods=["DELETE"])
|
|
def api_deleteRoutine(routine_id):
|
|
"""Delete a routine and all its steps/sessions."""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
existing = postgres.select_one(
|
|
"routines", {"id": routine_id, "user_uuid": user_uuid}
|
|
)
|
|
if not existing:
|
|
return flask.jsonify({"error": "not found"}), 404
|
|
postgres.delete("routine_sessions", {"routine_id": routine_id})
|
|
postgres.delete("routine_steps", {"routine_id": routine_id})
|
|
postgres.delete("routine_schedules", {"routine_id": routine_id})
|
|
postgres.delete("routines", {"id": routine_id, "user_uuid": user_uuid})
|
|
return flask.jsonify({"deleted": True}), 200
|
|
|
|
# ── Steps CRUD ────────────────────────────────────────────────
|
|
|
|
@app.route("/api/routines/<routine_id>/steps", methods=["GET"])
|
|
def api_listSteps(routine_id):
|
|
"""List steps for a routine, ordered by position."""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
routine = postgres.select_one(
|
|
"routines", {"id": routine_id, "user_uuid": user_uuid}
|
|
)
|
|
if not routine:
|
|
return flask.jsonify({"error": "not found"}), 404
|
|
steps = postgres.select(
|
|
"routine_steps",
|
|
where={"routine_id": routine_id},
|
|
order_by="position",
|
|
)
|
|
return flask.jsonify(steps), 200
|
|
|
|
@app.route("/api/routines/<routine_id>/steps", methods=["POST"])
|
|
def api_addStep(routine_id):
|
|
"""Add a step to a routine. Body: {name, duration_minutes?, position?}"""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
routine = postgres.select_one(
|
|
"routines", {"id": routine_id, "user_uuid": user_uuid}
|
|
)
|
|
if not routine:
|
|
return flask.jsonify({"error": "not found"}), 404
|
|
data = flask.request.get_json()
|
|
if not data:
|
|
return flask.jsonify({"error": "missing body"}), 400
|
|
if not data.get("name"):
|
|
return flask.jsonify({"error": "missing required field: name"}), 400
|
|
max_pos = postgres.select(
|
|
"routine_steps",
|
|
where={"routine_id": routine_id},
|
|
order_by="position DESC",
|
|
limit=1,
|
|
)
|
|
next_pos = (max_pos[0]["position"] + 1) if max_pos else 1
|
|
step = {
|
|
"id": str(uuid.uuid4()),
|
|
"routine_id": routine_id,
|
|
"name": data["name"],
|
|
"duration_minutes": data.get("duration_minutes"),
|
|
"position": data.get("position", next_pos),
|
|
}
|
|
result = postgres.insert("routine_steps", step)
|
|
return flask.jsonify(result), 201
|
|
|
|
@app.route("/api/routines/<routine_id>/steps/<step_id>", methods=["PUT"])
|
|
def api_updateStep(routine_id, step_id):
|
|
"""Update a step. Body: {name?, duration_minutes?, position?}"""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
routine = postgres.select_one(
|
|
"routines", {"id": routine_id, "user_uuid": user_uuid}
|
|
)
|
|
if not routine:
|
|
return flask.jsonify({"error": "not found"}), 404
|
|
data = flask.request.get_json()
|
|
if not data:
|
|
return flask.jsonify({"error": "missing body"}), 400
|
|
existing = postgres.select_one(
|
|
"routine_steps", {"id": step_id, "routine_id": routine_id}
|
|
)
|
|
if not existing:
|
|
return flask.jsonify({"error": "step not found"}), 404
|
|
allowed = ["name", "duration_minutes", "position"]
|
|
updates = {k: v for k, v in data.items() if k in allowed}
|
|
if not updates:
|
|
return flask.jsonify({"error": "no valid fields to update"}), 400
|
|
result = postgres.update(
|
|
"routine_steps", updates, {"id": step_id, "routine_id": routine_id}
|
|
)
|
|
return flask.jsonify(result[0] if result else {}), 200
|
|
|
|
@app.route("/api/routines/<routine_id>/steps/<step_id>", methods=["DELETE"])
|
|
def api_deleteStep(routine_id, step_id):
|
|
"""Delete a step from a routine."""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
routine = postgres.select_one(
|
|
"routines", {"id": routine_id, "user_uuid": user_uuid}
|
|
)
|
|
if not routine:
|
|
return flask.jsonify({"error": "not found"}), 404
|
|
existing = postgres.select_one(
|
|
"routine_steps", {"id": step_id, "routine_id": routine_id}
|
|
)
|
|
if not existing:
|
|
return flask.jsonify({"error": "step not found"}), 404
|
|
postgres.delete("routine_steps", {"id": step_id, "routine_id": routine_id})
|
|
return flask.jsonify({"deleted": True}), 200
|
|
|
|
@app.route("/api/routines/<routine_id>/steps/reorder", methods=["PUT"])
|
|
def api_reorderSteps(routine_id):
|
|
"""Reorder steps. Body: {step_ids: [ordered list of step UUIDs]}"""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
routine = postgres.select_one(
|
|
"routines", {"id": routine_id, "user_uuid": user_uuid}
|
|
)
|
|
if not routine:
|
|
return flask.jsonify({"error": "not found"}), 404
|
|
data = flask.request.get_json()
|
|
if not data or not data.get("step_ids"):
|
|
return flask.jsonify({"error": "missing step_ids"}), 400
|
|
step_ids = data["step_ids"]
|
|
for i, step_id in enumerate(step_ids):
|
|
postgres.update(
|
|
"routine_steps",
|
|
{"position": i + 1},
|
|
{"id": step_id, "routine_id": routine_id},
|
|
)
|
|
steps = postgres.select(
|
|
"routine_steps",
|
|
where={"routine_id": routine_id},
|
|
order_by="position",
|
|
)
|
|
return flask.jsonify(steps), 200
|
|
|
|
# ── Routine Sessions (active run-through) ─────────────────────
|
|
|
|
@app.route("/api/routines/<routine_id>/start", methods=["POST"])
|
|
def api_startRoutine(routine_id):
|
|
"""Start a routine session. Returns the session with first step."""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
routine = postgres.select_one(
|
|
"routines", {"id": routine_id, "user_uuid": user_uuid}
|
|
)
|
|
if not routine:
|
|
return flask.jsonify({"error": "not found"}), 404
|
|
active = postgres.select_one(
|
|
"routine_sessions", {"user_uuid": user_uuid, "status": "active"}
|
|
)
|
|
if not active:
|
|
active = postgres.select_one(
|
|
"routine_sessions", {"user_uuid": user_uuid, "status": "paused"}
|
|
)
|
|
if active:
|
|
return flask.jsonify(
|
|
{"error": "already have active session", "session_id": active["id"]}
|
|
), 409
|
|
|
|
# Check if starting now would conflict with medication times
|
|
now = tz.user_now()
|
|
current_time = now.strftime("%H:%M")
|
|
current_day = now.strftime("%a").lower()
|
|
routine_dur = _get_routine_duration_minutes(routine_id)
|
|
routine_start = _time_str_to_minutes(current_time)
|
|
|
|
user_meds = postgres.select("medications", {"user_uuid": user_uuid, "active": True})
|
|
for med in user_meds:
|
|
med_times = med.get("times", [])
|
|
if isinstance(med_times, str):
|
|
med_times = json.loads(med_times)
|
|
med_days = med.get("days_of_week", [])
|
|
if isinstance(med_days, str):
|
|
med_days = json.loads(med_days)
|
|
if med_days and current_day not in med_days:
|
|
continue
|
|
for mt in med_times:
|
|
med_start = _time_str_to_minutes(mt)
|
|
if _ranges_overlap(routine_start, routine_dur, med_start, 1):
|
|
return flask.jsonify(
|
|
{"error": f"Starting now would conflict with {med.get('name', 'medication')} at {mt}"}
|
|
), 409
|
|
|
|
steps = postgres.select(
|
|
"routine_steps",
|
|
where={"routine_id": routine_id},
|
|
order_by="position",
|
|
)
|
|
if not steps:
|
|
return flask.jsonify({"error": "no steps in routine"}), 400
|
|
session = {
|
|
"id": str(uuid.uuid4()),
|
|
"routine_id": routine_id,
|
|
"user_uuid": user_uuid,
|
|
"status": "active",
|
|
"current_step_index": 0,
|
|
}
|
|
result = postgres.insert("routine_sessions", session)
|
|
return flask.jsonify({"session": result, "current_step": steps[0]}), 201
|
|
|
|
@app.route("/api/sessions/active", methods=["GET"])
|
|
def api_getActiveSession():
|
|
"""Get the user's currently active routine session, if any."""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
session = postgres.select_one(
|
|
"routine_sessions", {"user_uuid": user_uuid, "status": "active"}
|
|
)
|
|
if not session:
|
|
session = postgres.select_one(
|
|
"routine_sessions", {"user_uuid": user_uuid, "status": "paused"}
|
|
)
|
|
if not session:
|
|
return flask.jsonify({"error": "no active session"}), 404
|
|
routine = postgres.select_one("routines", {"id": session["routine_id"]})
|
|
steps = postgres.select(
|
|
"routine_steps",
|
|
where={"routine_id": session["routine_id"]},
|
|
order_by="position",
|
|
)
|
|
current_step = (
|
|
steps[session["current_step_index"]]
|
|
if session["current_step_index"] < len(steps)
|
|
else None
|
|
)
|
|
return flask.jsonify(
|
|
{"session": session, "routine": routine, "current_step": current_step}
|
|
), 200
|
|
|
|
@app.route("/api/sessions/<session_id>/complete-step", methods=["POST"])
|
|
def api_completeStep(session_id):
|
|
"""Mark current step done, advance to next. Body: {step_id}"""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
session = postgres.select_one(
|
|
"routine_sessions", {"id": session_id, "user_uuid": user_uuid}
|
|
)
|
|
if not session:
|
|
return flask.jsonify({"error": "not found"}), 404
|
|
if session["status"] not in ("active", "paused"):
|
|
return flask.jsonify({"error": "session not active"}), 400
|
|
# Auto-resume if paused
|
|
if session["status"] == "paused":
|
|
postgres.update(
|
|
"routine_sessions",
|
|
{"status": "active", "paused_at": None},
|
|
{"id": session_id},
|
|
)
|
|
data = flask.request.get_json() or {}
|
|
steps = postgres.select(
|
|
"routine_steps",
|
|
where={"routine_id": session["routine_id"]},
|
|
order_by="position",
|
|
)
|
|
current_index = session["current_step_index"]
|
|
current_step = steps[current_index] if current_index < len(steps) else None
|
|
|
|
# Record step result
|
|
if current_step:
|
|
_record_step_result(
|
|
session_id, current_step["id"], current_index, "completed", session
|
|
)
|
|
|
|
next_index = current_index + 1
|
|
if next_index >= len(steps):
|
|
# Session complete — compute celebration data
|
|
celebration = _complete_session_with_celebration(
|
|
session_id, user_uuid, session
|
|
)
|
|
return flask.jsonify(
|
|
{
|
|
"session": {"status": "completed"},
|
|
"next_step": None,
|
|
"celebration": celebration,
|
|
}
|
|
), 200
|
|
postgres.update(
|
|
"routine_sessions", {"current_step_index": next_index}, {"id": session_id}
|
|
)
|
|
return flask.jsonify(
|
|
{
|
|
"session": {"current_step_index": next_index},
|
|
"next_step": steps[next_index],
|
|
}
|
|
), 200
|
|
|
|
@app.route("/api/sessions/<session_id>/skip-step", methods=["POST"])
|
|
def api_skipStep(session_id):
|
|
"""Skip current step, advance to next. Body: {step_id}"""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
session = postgres.select_one(
|
|
"routine_sessions", {"id": session_id, "user_uuid": user_uuid}
|
|
)
|
|
if not session:
|
|
return flask.jsonify({"error": "not found"}), 404
|
|
if session["status"] not in ("active", "paused"):
|
|
return flask.jsonify({"error": "session not active"}), 400
|
|
# Auto-resume if paused
|
|
if session["status"] == "paused":
|
|
postgres.update(
|
|
"routine_sessions",
|
|
{"status": "active", "paused_at": None},
|
|
{"id": session_id},
|
|
)
|
|
steps = postgres.select(
|
|
"routine_steps",
|
|
where={"routine_id": session["routine_id"]},
|
|
order_by="position",
|
|
)
|
|
current_index = session["current_step_index"]
|
|
current_step = steps[current_index] if current_index < len(steps) else None
|
|
|
|
# Record step result as skipped
|
|
if current_step:
|
|
_record_step_result(
|
|
session_id, current_step["id"], current_index, "skipped", session
|
|
)
|
|
|
|
next_index = current_index + 1
|
|
if next_index >= len(steps):
|
|
celebration = _complete_session_with_celebration(
|
|
session_id, user_uuid, session
|
|
)
|
|
return flask.jsonify(
|
|
{
|
|
"session": {"status": "completed"},
|
|
"next_step": None,
|
|
"celebration": celebration,
|
|
}
|
|
), 200
|
|
postgres.update(
|
|
"routine_sessions", {"current_step_index": next_index}, {"id": session_id}
|
|
)
|
|
return flask.jsonify(
|
|
{
|
|
"session": {"current_step_index": next_index},
|
|
"next_step": steps[next_index],
|
|
}
|
|
), 200
|
|
|
|
@app.route("/api/sessions/<session_id>/cancel", methods=["POST"])
|
|
def api_cancelSession(session_id):
|
|
"""Cancel an active routine session."""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
session = postgres.select_one(
|
|
"routine_sessions", {"id": session_id, "user_uuid": user_uuid}
|
|
)
|
|
if not session:
|
|
return flask.jsonify({"error": "not found"}), 404
|
|
postgres.update("routine_sessions", {"status": "cancelled"}, {"id": session_id})
|
|
return flask.jsonify({"session": {"status": "cancelled"}}), 200
|
|
|
|
# ── Routine History / Stats ───────────────────────────────────
|
|
|
|
@app.route("/api/routines/<routine_id>/history", methods=["GET"])
|
|
def api_routineHistory(routine_id):
|
|
"""Get past sessions for a routine. Query: ?days=7"""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
routine = postgres.select_one(
|
|
"routines", {"id": routine_id, "user_uuid": user_uuid}
|
|
)
|
|
if not routine:
|
|
return flask.jsonify({"error": "not found"}), 404
|
|
days = flask.request.args.get("days", 7, type=int)
|
|
sessions = postgres.select(
|
|
"routine_sessions",
|
|
where={"routine_id": routine_id},
|
|
order_by="created_at DESC",
|
|
limit=days,
|
|
)
|
|
return flask.jsonify(sessions), 200
|
|
|
|
# ── Routine Scheduling ────────────────────────────────────────
|
|
|
|
@app.route("/api/routines/schedules", methods=["GET"])
|
|
def api_listAllSchedules():
|
|
"""Get all schedules for the user's routines with routine metadata."""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
routines = postgres.select("routines", where={"user_uuid": user_uuid})
|
|
if not routines:
|
|
return flask.jsonify([]), 200
|
|
|
|
result = []
|
|
for r in routines:
|
|
sched = postgres.select_one("routine_schedules", {"routine_id": r["id"]})
|
|
if not sched:
|
|
continue
|
|
steps = postgres.select("routine_steps", where={"routine_id": r["id"]})
|
|
total_duration = sum(s.get("duration_minutes") or 0 for s in steps)
|
|
entry = {
|
|
"routine_id": r["id"],
|
|
"routine_name": r.get("name", ""),
|
|
"routine_icon": r.get("icon", ""),
|
|
"days": sched.get("days", []),
|
|
"time": sched.get("time"),
|
|
"remind": sched.get("remind", True),
|
|
"total_duration_minutes": total_duration,
|
|
"frequency": sched.get("frequency", "weekly"),
|
|
}
|
|
if sched.get("frequency") == "every_n_days":
|
|
entry["interval_days"] = sched.get("interval_days")
|
|
entry["start_date"] = str(sched.get("start_date")) if sched.get("start_date") else None
|
|
result.append(entry)
|
|
return flask.jsonify(result), 200
|
|
|
|
def _get_routine_duration_minutes(routine_id):
|
|
"""Get total duration of a routine from its steps."""
|
|
steps = postgres.select("routine_steps", where={"routine_id": routine_id})
|
|
total = sum(s.get("duration_minutes", 0) or 0 for s in steps)
|
|
return max(total, 1) # At least 1 minute
|
|
|
|
def _time_str_to_minutes(time_str):
|
|
"""Convert 'HH:MM' to minutes since midnight."""
|
|
parts = time_str.split(":")
|
|
return int(parts[0]) * 60 + int(parts[1])
|
|
|
|
def _ranges_overlap(start1, dur1, start2, dur2):
|
|
"""Check if two time ranges overlap (in minutes since midnight)."""
|
|
end1 = start1 + dur1
|
|
end2 = start2 + dur2
|
|
return start1 < end2 and start2 < end1
|
|
|
|
def _check_schedule_conflicts(user_uuid, new_days, new_time, exclude_routine_id=None, new_routine_id=None):
|
|
"""Check if the proposed schedule conflicts with existing routines or medications.
|
|
Returns (has_conflict, conflict_message) tuple.
|
|
"""
|
|
if not new_days or not new_time:
|
|
return False, None
|
|
|
|
new_start = _time_str_to_minutes(new_time)
|
|
# Get duration of the routine being scheduled
|
|
if new_routine_id:
|
|
new_dur = _get_routine_duration_minutes(new_routine_id)
|
|
else:
|
|
new_dur = 1
|
|
|
|
# Check conflicts with other routines
|
|
user_routines = postgres.select("routines", {"user_uuid": user_uuid})
|
|
for r in user_routines:
|
|
if r["id"] == exclude_routine_id:
|
|
continue
|
|
other_sched = postgres.select_one("routine_schedules", {"routine_id": r["id"]})
|
|
if not other_sched or not other_sched.get("time"):
|
|
continue
|
|
other_days = other_sched.get("days", [])
|
|
if isinstance(other_days, str):
|
|
other_days = json.loads(other_days)
|
|
if not any(d in other_days for d in new_days):
|
|
continue
|
|
other_start = _time_str_to_minutes(other_sched["time"])
|
|
other_dur = _get_routine_duration_minutes(r["id"])
|
|
if _ranges_overlap(new_start, new_dur, other_start, other_dur):
|
|
return True, f"Time conflicts with routine: {r.get('name', 'Unnamed routine')}"
|
|
|
|
# Check conflicts with medications
|
|
user_meds = postgres.select("medications", {"user_uuid": user_uuid, "active": True})
|
|
for med in user_meds:
|
|
med_times = med.get("times", [])
|
|
if isinstance(med_times, str):
|
|
med_times = json.loads(med_times)
|
|
med_days = med.get("days_of_week", [])
|
|
if isinstance(med_days, str):
|
|
med_days = json.loads(med_days)
|
|
# If med has no specific days, it runs every day
|
|
if med_days and not any(d in med_days for d in new_days):
|
|
continue
|
|
for mt in med_times:
|
|
med_start = _time_str_to_minutes(mt)
|
|
# Medication takes ~0 minutes, but check if it falls within routine window
|
|
if _ranges_overlap(new_start, new_dur, med_start, 1):
|
|
return True, f"Time conflicts with medication: {med.get('name', 'Unnamed medication')}"
|
|
|
|
return False, None
|
|
|
|
@app.route("/api/routines/<routine_id>/schedule", methods=["PUT"])
|
|
def api_setRoutineSchedule(routine_id):
|
|
"""Set when this routine should run.
|
|
Body: {days, time, remind, frequency?, interval_days?, start_date?}
|
|
frequency: 'weekly' (default, uses days) or 'every_n_days' (uses interval_days + start_date)
|
|
"""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
routine = postgres.select_one(
|
|
"routines", {"id": routine_id, "user_uuid": user_uuid}
|
|
)
|
|
if not routine:
|
|
return flask.jsonify({"error": "not found"}), 404
|
|
data = flask.request.get_json()
|
|
if not data:
|
|
return flask.jsonify({"error": "missing body"}), 400
|
|
|
|
frequency = data.get("frequency", "weekly")
|
|
|
|
# Check for schedule conflicts (only for weekly — interval conflicts checked at reminder time)
|
|
if frequency == "weekly":
|
|
new_days = data.get("days", [])
|
|
new_time = data.get("time")
|
|
has_conflict, conflict_msg = _check_schedule_conflicts(
|
|
user_uuid, new_days, new_time, exclude_routine_id=routine_id,
|
|
new_routine_id=routine_id,
|
|
)
|
|
if has_conflict:
|
|
return flask.jsonify({"error": conflict_msg}), 409
|
|
|
|
existing = postgres.select_one("routine_schedules", {"routine_id": routine_id})
|
|
schedule_data = {
|
|
"routine_id": routine_id,
|
|
"days": json.dumps(data.get("days", [])),
|
|
"time": data.get("time"),
|
|
"remind": data.get("remind", True),
|
|
"frequency": frequency,
|
|
"interval_days": data.get("interval_days"),
|
|
"start_date": data.get("start_date"),
|
|
}
|
|
if existing:
|
|
result = postgres.update(
|
|
"routine_schedules", schedule_data, {"routine_id": routine_id}
|
|
)
|
|
return flask.jsonify(result[0] if result else {}), 200
|
|
else:
|
|
schedule_data["id"] = str(uuid.uuid4())
|
|
result = postgres.insert("routine_schedules", schedule_data)
|
|
return flask.jsonify(result), 201
|
|
|
|
@app.route("/api/routines/<routine_id>/schedule", methods=["GET"])
|
|
def api_getRoutineSchedule(routine_id):
|
|
"""Get the schedule for a routine."""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
routine = postgres.select_one(
|
|
"routines", {"id": routine_id, "user_uuid": user_uuid}
|
|
)
|
|
if not routine:
|
|
return flask.jsonify({"error": "not found"}), 404
|
|
schedule = postgres.select_one("routine_schedules", {"routine_id": routine_id})
|
|
if not schedule:
|
|
return flask.jsonify({"error": "no schedule set"}), 404
|
|
return flask.jsonify(schedule), 200
|
|
|
|
@app.route("/api/routines/<routine_id>/schedule", methods=["DELETE"])
|
|
def api_deleteRoutineSchedule(routine_id):
|
|
"""Remove the schedule from a routine."""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
routine = postgres.select_one(
|
|
"routines", {"id": routine_id, "user_uuid": user_uuid}
|
|
)
|
|
if not routine:
|
|
return flask.jsonify({"error": "not found"}), 404
|
|
postgres.delete("routine_schedules", {"routine_id": routine_id})
|
|
return flask.jsonify({"deleted": True}), 200
|