221 lines
7.3 KiB
Python
221 lines
7.3 KiB
Python
from collections import OrderedDict
|
|
from datetime import datetime, timezone
|
|
|
|
from flask import Blueprint, jsonify, request, session
|
|
from app.config import SYDNEY_TZ
|
|
from app.auth import login_required, privacy_guard
|
|
from app.db import query
|
|
from app.utils import calculate_weight_change
|
|
|
|
bp = Blueprint("api", __name__, url_prefix="/api")
|
|
|
|
# Distinct hues for up to 12 users; cycles if more
|
|
_CHART_HUES = [210, 160, 270, 30, 340, 190, 50, 120, 300, 80, 240, 10]
|
|
|
|
|
|
@bp.route("/progress-over-time")
|
|
@login_required
|
|
def progress_over_time():
|
|
"""Return per-user % weight lost over time for the leaderboard line chart."""
|
|
# --- parse optional filters ---
|
|
start = request.args.get("start") # ISO date string
|
|
end = request.args.get("end")
|
|
user_ids_raw = request.args.get("user_ids", "")
|
|
|
|
# Build WHERE fragments
|
|
where_clauses = ["u.is_private = FALSE"]
|
|
params = []
|
|
|
|
if start:
|
|
where_clauses.append("c.checked_in_at >= %s")
|
|
params.append(start)
|
|
if end:
|
|
where_clauses.append("c.checked_in_at < (%s::date + interval '1 day')")
|
|
params.append(end)
|
|
if user_ids_raw:
|
|
try:
|
|
uid_list = [int(x) for x in user_ids_raw.split(",") if x.strip()]
|
|
except ValueError:
|
|
uid_list = []
|
|
if uid_list:
|
|
placeholders = ",".join(["%s"] * len(uid_list))
|
|
where_clauses.append(f"u.id IN ({placeholders})")
|
|
params.extend(uid_list)
|
|
|
|
where_sql = " AND ".join(where_clauses)
|
|
|
|
# Use CTE for first_weight instead of correlated subquery
|
|
rows = query(f"""
|
|
WITH first_weights AS (
|
|
SELECT DISTINCT ON (user_id) user_id, weight_kg AS first_weight
|
|
FROM checkins
|
|
ORDER BY user_id, checked_in_at ASC
|
|
)
|
|
SELECT
|
|
u.id AS user_id,
|
|
u.display_name,
|
|
u.username,
|
|
u.starting_weight_kg,
|
|
fw.first_weight,
|
|
c.weight_kg,
|
|
c.checked_in_at
|
|
FROM checkins c
|
|
JOIN users u ON u.id = c.user_id
|
|
LEFT JOIN first_weights fw ON fw.user_id = u.id
|
|
WHERE {where_sql}
|
|
ORDER BY u.id, c.checked_in_at ASC
|
|
""", params)
|
|
|
|
# Group rows by user
|
|
users_map = OrderedDict()
|
|
for r in rows:
|
|
uid = r["user_id"]
|
|
if uid not in users_map:
|
|
start_w = float(r["starting_weight_kg"] or r["first_weight"] or 0)
|
|
users_map[uid] = {
|
|
"id": uid,
|
|
"name": r["display_name"] or r["username"],
|
|
"start_w": start_w,
|
|
"data": [],
|
|
}
|
|
entry = users_map[uid]
|
|
w = float(r["weight_kg"])
|
|
dt = r["checked_in_at"]
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
entry["data"].append({
|
|
"date": dt.astimezone(SYDNEY_TZ).strftime("%Y-%m-%d"),
|
|
"weight": round(w, 1),
|
|
})
|
|
|
|
# Build response with best-fit
|
|
result = []
|
|
for idx, (uid, entry) in enumerate(users_map.items()):
|
|
hue = _CHART_HUES[idx % len(_CHART_HUES)]
|
|
color = f"hsl({hue}, 70%, 55%)"
|
|
|
|
# Simple linear regression (x = day index, y = weight)
|
|
points = entry["data"]
|
|
best_fit = {"slope": 0, "intercept": 0}
|
|
if len(points) >= 2:
|
|
base = datetime.strptime(points[0]["date"], "%Y-%m-%d")
|
|
xs = [(datetime.strptime(p["date"], "%Y-%m-%d") - base).days for p in points]
|
|
ys = [p["weight"] for p in points]
|
|
n = len(xs)
|
|
sum_x = sum(xs)
|
|
sum_y = sum(ys)
|
|
sum_xy = sum(x * y for x, y in zip(xs, ys))
|
|
sum_x2 = sum(x * x for x in xs)
|
|
denom = n * sum_x2 - sum_x * sum_x
|
|
if denom != 0:
|
|
slope = (n * sum_xy - sum_x * sum_y) / denom
|
|
intercept = (sum_y - slope * sum_x) / n
|
|
best_fit = {"slope": round(slope, 4), "intercept": round(intercept, 4)}
|
|
|
|
result.append({
|
|
"id": uid,
|
|
"name": entry["name"],
|
|
"color": color,
|
|
"data": points,
|
|
"best_fit": best_fit,
|
|
})
|
|
|
|
return jsonify({"users": result})
|
|
|
|
|
|
@bp.route("/chart-data/<int:user_id>")
|
|
@login_required
|
|
@privacy_guard
|
|
def chart_data(user_id):
|
|
"""Return weight & BMI over time for Chart.js."""
|
|
checkins = query(
|
|
"""SELECT weight_kg, bmi, checked_in_at
|
|
FROM checkins WHERE user_id = %s
|
|
ORDER BY checked_in_at ASC""",
|
|
(user_id,),
|
|
)
|
|
|
|
labels = [c["checked_in_at"].replace(tzinfo=timezone.utc).astimezone(SYDNEY_TZ).strftime("%d %b") for c in checkins]
|
|
weights = [float(c["weight_kg"]) for c in checkins]
|
|
bmis = [float(c["bmi"]) if c["bmi"] else None for c in checkins]
|
|
|
|
return jsonify({
|
|
"labels": labels,
|
|
"weights": weights,
|
|
"bmis": bmis,
|
|
})
|
|
|
|
|
|
@bp.route("/comparison")
|
|
@login_required
|
|
def comparison():
|
|
"""Return all-user comparison data for bar charts."""
|
|
# Use CTE with window functions instead of correlated subqueries
|
|
users = query("""
|
|
WITH user_weights AS (
|
|
SELECT
|
|
user_id,
|
|
FIRST_VALUE(weight_kg) OVER (PARTITION BY user_id ORDER BY checked_in_at ASC) AS first_weight,
|
|
FIRST_VALUE(weight_kg) OVER (PARTITION BY user_id ORDER BY checked_in_at DESC) AS current_weight,
|
|
ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY checked_in_at DESC) AS rn
|
|
FROM checkins
|
|
)
|
|
SELECT
|
|
u.id,
|
|
u.display_name,
|
|
u.username,
|
|
u.starting_weight_kg,
|
|
uw.first_weight,
|
|
uw.current_weight
|
|
FROM users u
|
|
JOIN user_weights uw ON uw.user_id = u.id AND uw.rn = 1
|
|
WHERE u.is_private = FALSE
|
|
ORDER BY u.display_name
|
|
""")
|
|
|
|
names = []
|
|
pct_lost = []
|
|
kg_lost = []
|
|
|
|
for u in users:
|
|
start_w = float(u["starting_weight_kg"] or u["first_weight"] or 0)
|
|
current_w = float(u["current_weight"] or start_w)
|
|
lost, pct = calculate_weight_change(start_w, current_w)
|
|
names.append(u["display_name"] or u["username"])
|
|
pct_lost.append(pct)
|
|
kg_lost.append(lost)
|
|
|
|
return jsonify({
|
|
"names": names,
|
|
"pct_lost": pct_lost,
|
|
"kg_lost": kg_lost,
|
|
})
|
|
|
|
|
|
@bp.route("/weekly-change/<int:user_id>")
|
|
@login_required
|
|
@privacy_guard
|
|
def weekly_change(user_id):
|
|
"""Return weekly weight changes for bar chart."""
|
|
checkins = query(
|
|
"""SELECT weight_kg, checked_in_at
|
|
FROM checkins WHERE user_id = %s
|
|
ORDER BY checked_in_at ASC""",
|
|
(user_id,),
|
|
)
|
|
|
|
if len(checkins) < 2:
|
|
return jsonify({"labels": [], "changes": []})
|
|
|
|
labels = []
|
|
changes = []
|
|
for i in range(1, len(checkins)):
|
|
prev_w = float(checkins[i - 1]["weight_kg"])
|
|
curr_w = float(checkins[i]["weight_kg"])
|
|
change = round(curr_w - prev_w, 1)
|
|
label = checkins[i]["checked_in_at"].replace(tzinfo=timezone.utc).astimezone(SYDNEY_TZ).strftime("%d %b")
|
|
labels.append(label)
|
|
changes.append(change)
|
|
|
|
return jsonify({"labels": labels, "changes": changes})
|