diff --git a/bot.py b/bot.py index 88e5893..b891517 100644 --- a/bot.py +++ b/bot.py @@ -16,7 +16,7 @@ from telegram.ext import ( filters, ) -from db import init_db, save_workout, get_workouts, get_workout_count +from db import init_db, save_workout, get_workouts, get_workout_count, get_stats_sql, delete_workout from parser import parse_workout, format_workout load_dotenv() @@ -48,7 +48,7 @@ def _load_token() -> str: BOT_TOKEN = _load_token() -# Mini App URL β€” set automatically by start.py via localtunnel +# Mini App URL β€” set automatically by start.py via tunnel WEBAPP_URL = os.environ.get("WEBAPP_URL", "") @@ -79,19 +79,21 @@ def extract_timestamp(update: Update) -> tuple[datetime, bool]: async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE): text = ( - "πŸ’ͺ Fitness Tracker Bot\n\n" + "\U0001f4aa Fitness Tracker Bot\n\n" "Send me your workout and I'll save it!\n\n" - "Format:\n" + "Formats:\n" "Bench press: 4x8x35\n" - "Lateral raise: 4x8x4\n\n" - "Tri Press rom: 3x10x45\n\n" + "Pull-ups: 3x10 (bodyweight)\n" + "Shoulder press (3032): 8x25, 5x35, 6x40\n\n" "Lines without a blank line between them = superset.\n" - "Machine IDs go in parentheses: Lat pulldown (500): 3x5x45\n\n" - "You can also forward messages from Saved Messages β€” " + "Machine IDs go in parentheses.\n\n" + "You can also forward messages from Saved Messages \u2014 " "I'll use the original timestamp.\n\n" "Commands:\n" - "/history β€” view recent workouts\n" - "/stats β€” quick summary" + "/history \u2014 view recent workouts\n" + "/stats \u2014 quick summary\n" + "/delete <id> \u2014 delete a workout\n" + "/export \u2014 export all data as JSON" ) if WEBAPP_URL: @@ -118,11 +120,11 @@ async def cmd_history(update: Update, context: ContextTypes.DEFAULT_TYPE): parts = [] for w in workouts: ts = datetime.fromisoformat(w["timestamp"]) - header = f"πŸ“… {ts.strftime('%a %d %b %Y, %H:%M')}" + header = f"\U0001f4c5 {ts.strftime('%a %d %b %Y, %H:%M')} (#{w['id']})" body = format_workout(w["superset_groups"]) parts.append(f"{header}\n{body}") - text = "\n\n───────────────\n\n".join(parts) + text = "\n\n\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n\n".join(parts) total = get_workout_count(user_id) text += f"\n\nShowing latest 5 of {total} workouts." @@ -131,35 +133,70 @@ async def cmd_history(update: Update, context: ContextTypes.DEFAULT_TYPE): async def cmd_stats(update: Update, context: ContextTypes.DEFAULT_TYPE): user_id = update.effective_user.id - total = get_workout_count(user_id) + stats = get_stats_sql(user_id) - if total == 0: - await update.message.reply_text("No workouts yet β€” send me your first one!") + if stats["total_workouts"] == 0: + await update.message.reply_text("No workouts yet \u2014 send me your first one!") return - workouts = get_workouts(user_id, limit=1000) - - # Collect all unique exercise names - exercise_names = set() - total_sets = 0 - total_volume = 0.0 - for w in workouts: - for group in w["superset_groups"]: - for ex in group: - exercise_names.add(ex["name"].lower()) - total_sets += ex["sets"] - total_volume += ex["sets"] * ex["reps"] * ex["weight_kg"] - await update.message.reply_text( - f"πŸ“Š Your Stats\n\n" - f" β€’ Workouts logged: {total}\n" - f" β€’ Unique exercises: {len(exercise_names)}\n" - f" β€’ Total sets: {total_sets}\n" - f" β€’ Total volume: {total_volume:,.0f} kg", + f"\U0001f4ca Your Stats\n\n" + f" \u2022 Workouts logged: {stats['total_workouts']}\n" + f" \u2022 Unique exercises: {stats['unique_exercises']}\n" + f" \u2022 Total sets: {stats['total_sets']}\n" + f" \u2022 Total volume: {stats['total_volume']:,.0f} kg", parse_mode=ParseMode.HTML, ) +async def cmd_delete(update: Update, context: ContextTypes.DEFAULT_TYPE): + user_id = update.effective_user.id + + if not context.args: + await update.message.reply_text( + "Usage: /delete <workout_id>\n" + "Use /history to see workout IDs.", + parse_mode=ParseMode.HTML, + ) + return + + try: + workout_id = int(context.args[0]) + except ValueError: + await update.message.reply_text("Workout ID must be a number.") + return + + if delete_workout(user_id, workout_id): + await update.message.reply_text(f"\U0001f5d1 Workout #{workout_id} deleted.") + else: + await update.message.reply_text( + f"Workout #{workout_id} not found (or not yours)." + ) + + +async def cmd_export(update: Update, context: ContextTypes.DEFAULT_TYPE): + """Send all workout data as a JSON file.""" + import json + import io + from db import export_workouts + + user_id = update.effective_user.id + data = export_workouts(user_id) + + if not data: + await update.message.reply_text("No workouts to export.") + return + + content = json.dumps(data, indent=2, ensure_ascii=False) + buf = io.BytesIO(content.encode("utf-8")) + buf.name = "workouts_export.json" + + await update.message.reply_document( + document=buf, + caption=f"\U0001f4e6 Exported {len(data)} exercise records.", + ) + + # ── Message handler (workout parsing) ─────────────────────────────────────── @@ -169,9 +206,24 @@ async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE): if not text: return - groups = parse_workout(text) - if not groups: - # Not a workout message β€” silently ignore so the bot isn't noisy + groups, errors = parse_workout(text) + + if not groups and not errors: + # Doesn't look like a workout at all β€” silently ignore + return + + if not groups and errors: + # Looks like they tried but every line failed + error_lines = "\n".join(f" \u2022 {e.line}" for e in errors) + await update.message.reply_text( + f"\u26a0\ufe0f Could not parse workout. Check your format:\n\n" + f"{error_lines}\n\n" + f"Expected formats:\n" + f"Exercise: 4x8x35\n" + f"Exercise: 3x10 (bodyweight)\n" + f"Exercise: 8x25, 5x35, 6x40", + parse_mode=ParseMode.HTML, + ) return user_id = update.effective_user.id @@ -188,12 +240,17 @@ async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE): ts_str = timestamp.strftime("%a %d %b %Y, %H:%M") confirm_parts = [ - f"βœ… Workout #{workout_id} saved!", - f"πŸ“… {ts_str}" + (" (from forwarded message)" if is_forwarded else ""), - f"πŸ‹οΈ {total_exercises} exercises, {total_sets} total sets", + f"\u2705 Workout #{workout_id} saved!", + f"\U0001f4c5 {ts_str}" + (" (from forwarded message)" if is_forwarded else ""), + f"\U0001f3cb\ufe0f {total_exercises} exercises, {total_sets} total sets", ] if supersets: - confirm_parts.append(f"πŸ”— {supersets} superset(s)") + confirm_parts.append(f"\U0001f517 {supersets} superset(s)") + + # Show errors for partially parsed workouts + if errors: + skipped = "\n".join(f" \u2022 {e.line}" for e in errors) + confirm_parts.append(f"\n\u26a0\ufe0f Skipped {len(errors)} unparseable line(s):\n{skipped}") confirm_parts.append(f"\n{format_workout(superset_dicts)}") @@ -231,11 +288,13 @@ def main(): app.add_handler(CommandHandler("start", cmd_start)) app.add_handler(CommandHandler("history", cmd_history)) app.add_handler(CommandHandler("stats", cmd_stats)) + app.add_handler(CommandHandler("delete", cmd_delete)) + app.add_handler(CommandHandler("export", cmd_export)) # Handle all text messages (including forwarded ones) app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message)) - logger.info("Bot started β€” polling…") + logger.info("Bot started \u2014 polling\u2026") app.run_polling(allowed_updates=Update.ALL_TYPES) diff --git a/db.py b/db.py index 664c6cb..5920603 100644 --- a/db.py +++ b/db.py @@ -1,5 +1,6 @@ """Database layer for the fitness bot.""" +import json import sqlite3 from datetime import datetime from pathlib import Path @@ -49,29 +50,34 @@ def init_db(): superset_group_id INTEGER NOT NULL REFERENCES superset_groups(id) ON DELETE CASCADE, position INTEGER NOT NULL, -- ordering within the superset group name TEXT NOT NULL, - machine_id TEXT, -- e.g. "500", "620" + machine_id TEXT, -- e.g. "3032", "5014" sets INTEGER NOT NULL, reps INTEGER NOT NULL, weight_kg REAL NOT NULL, - raw_line TEXT -- the original line as typed + raw_line TEXT, -- the original line as typed + sets_detail TEXT -- JSON array of {reps, weight_kg} per set ); CREATE INDEX IF NOT EXISTS idx_workouts_user ON workouts(user_id, timestamp); """) - # Migration: add raw_text column if it doesn't exist yet + # Migrations cols = {r[1] for r in conn.execute("PRAGMA table_info(workouts)").fetchall()} if "raw_text" not in cols: conn.execute("ALTER TABLE workouts ADD COLUMN raw_text TEXT") + ex_cols = {r[1] for r in conn.execute("PRAGMA table_info(exercises)").fetchall()} + if "sets_detail" not in ex_cols: + conn.execute("ALTER TABLE exercises ADD COLUMN sets_detail TEXT") + def save_workout(user_id: int, timestamp: datetime, superset_groups: list[list[dict]], raw_text: str | None = None, note: str | None = None) -> int: """ Save a parsed workout. superset_groups: list of groups, each group is a list of exercise dicts: - {name, machine_id, sets, reps, weight_kg, raw_line} + {name, machine_id, sets, reps, weight_kg, raw_line, sets_detail} raw_text: the full original message text, stored verbatim. Returns the workout id. @@ -91,17 +97,31 @@ def save_workout(user_id: int, timestamp: datetime, superset_groups: list[list[d group_id = cur2.lastrowid for ex_pos, ex in enumerate(group): + sets_detail_json = None + if ex.get("sets_detail"): + sets_detail_json = json.dumps(ex["sets_detail"]) conn.execute( """INSERT INTO exercises - (superset_group_id, position, name, machine_id, sets, reps, weight_kg, raw_line) - VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", + (superset_group_id, position, name, machine_id, sets, reps, weight_kg, raw_line, sets_detail) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", (group_id, ex_pos, ex["name"], ex.get("machine_id"), - ex["sets"], ex["reps"], ex["weight_kg"], ex.get("raw_line")), + ex["sets"], ex["reps"], ex["weight_kg"], ex.get("raw_line"), + sets_detail_json), ) return workout_id +def delete_workout(user_id: int, workout_id: int) -> bool: + """Delete a workout by ID. Returns True if deleted, False if not found or not owned.""" + with get_db() as conn: + cur = conn.execute( + "DELETE FROM workouts WHERE id = ? AND user_id = ?", + (workout_id, user_id), + ) + return cur.rowcount > 0 + + def get_workouts(user_id: int, limit: int = 10, offset: int = 0) -> list[dict]: """Fetch recent workouts for a user, newest first.""" with get_db() as conn: @@ -119,7 +139,8 @@ def get_workouts(user_id: int, limit: int = 10, offset: int = 0) -> list[dict]: workout = dict(row) groups = conn.execute( """SELECT sg.id as group_id, sg.position as group_pos, - e.name, e.machine_id, e.sets, e.reps, e.weight_kg, e.raw_line, e.position as ex_pos + e.name, e.machine_id, e.sets, e.reps, e.weight_kg, + e.raw_line, e.position as ex_pos, e.sets_detail FROM superset_groups sg JOIN exercises e ON e.superset_group_id = sg.id WHERE sg.workout_id = ? @@ -132,7 +153,16 @@ def get_workouts(user_id: int, limit: int = 10, offset: int = 0) -> list[dict]: gp = g["group_pos"] if gp not in superset_groups: superset_groups[gp] = [] - superset_groups[gp].append(dict(g)) + ex_dict = dict(g) + # Deserialize sets_detail JSON + if ex_dict.get("sets_detail"): + try: + ex_dict["sets_detail"] = json.loads(ex_dict["sets_detail"]) + except json.JSONDecodeError: + ex_dict["sets_detail"] = [] + else: + ex_dict["sets_detail"] = [] + superset_groups[gp].append(ex_dict) workout["superset_groups"] = [superset_groups[k] for k in sorted(superset_groups)] workouts.append(workout) @@ -146,3 +176,45 @@ def get_workout_count(user_id: int) -> int: "SELECT COUNT(*) as cnt FROM workouts WHERE user_id = ?", (user_id,) ).fetchone() return row["cnt"] + + +def get_stats_sql(user_id: int) -> dict: + """Compute stats entirely in SQL.""" + with get_db() as conn: + row = conn.execute(""" + SELECT + COUNT(DISTINCT w.id) as total_workouts, + COUNT(DISTINCT LOWER(e.name)) as unique_exercises, + COALESCE(SUM(e.sets), 0) as total_sets, + COALESCE(SUM(e.sets * e.reps * e.weight_kg), 0) as total_volume + FROM workouts w + JOIN superset_groups sg ON sg.workout_id = w.id + JOIN exercises e ON e.superset_group_id = sg.id + WHERE w.user_id = ? + """, (user_id,)).fetchone() + + return { + "total_workouts": row["total_workouts"], + "unique_exercises": row["unique_exercises"], + "total_sets": row["total_sets"], + "total_volume": round(row["total_volume"], 1), + } + + +def export_workouts(user_id: int) -> list[dict]: + """Export all workouts as flat records for CSV/JSON export.""" + with get_db() as conn: + rows = conn.execute(""" + SELECT + w.id as workout_id, w.timestamp, w.created_at, w.raw_text, + sg.position as group_pos, + e.name, e.machine_id, e.sets, e.reps, e.weight_kg, + e.raw_line, e.sets_detail + FROM workouts w + JOIN superset_groups sg ON sg.workout_id = w.id + JOIN exercises e ON e.superset_group_id = sg.id + WHERE w.user_id = ? + ORDER BY w.timestamp DESC, sg.position, e.position + """, (user_id,)).fetchall() + + return [dict(r) for r in rows] diff --git a/parser.py b/parser.py index c28dcb1..db93933 100644 --- a/parser.py +++ b/parser.py @@ -1,25 +1,39 @@ """ Parse workout messages into structured data. -Format per line: - Exercise Name (optional_machine_id): SETSxREPSxWEIGHT +Supported formats per line: + Exercise Name: SETSxREPSxWEIGHT β€” e.g. Bench press: 4x8x35 + Exercise Name: SETSxREPS β€” bodyweight, e.g. Pull-ups: 3x10 + Exercise Name: REPSxWEIGHT, REPSxWEIGHT β€” per-set, e.g. Shoulder press: 8x25, 5x35, 6x40 + Exercise Name: REPS, REPS, REPS β€” bodyweight per-set, e.g. Pull-ups: 12, 10, 8 + Exercise Name (machine_id): ... β€” optional machine ID in parentheses Lines with no blank line between them form a superset group. Blank lines separate superset groups. """ import re -from dataclasses import dataclass +from dataclasses import dataclass, field + + +@dataclass +class SetDetail: + reps: int + weight_kg: float + + def to_dict(self) -> dict: + return {"reps": self.reps, "weight_kg": self.weight_kg} @dataclass class Exercise: name: str machine_id: str | None - sets: int - reps: int - weight_kg: float + sets: int # total number of sets + reps: int # reps of first set (for backward compat / simple display) + weight_kg: float # weight of first set (for backward compat / simple display) raw_line: str + sets_detail: list[SetDetail] = field(default_factory=list) def to_dict(self) -> dict: return { @@ -29,21 +43,32 @@ class Exercise: "reps": self.reps, "weight_kg": self.weight_kg, "raw_line": self.raw_line, + "sets_detail": [s.to_dict() for s in self.sets_detail], } -# Matches lines like: -# Bench press: 4x8x35 -# Lat pulldown (500): 3x5x45 -# Russian Twists: 3x15x0 -EXERCISE_RE = re.compile( - r"^(?P.+?)" # exercise name (lazy) - r"(?:\s*\((?P\d+)\))?" # optional (machine_id) - r"\s*:\s*" # colon separator - r"(?P\d+)\s*x\s*" # sets - r"(?P\d+)\s*x\s*" # reps - r"(?P[\d.]+)" # weight - r"\s*$", +# Header pattern: captures exercise name and optional machine ID +HEADER_RE = re.compile( + r"^(?P.+?)" + r"(?:\s*\((?P[^)]+)\))?" + r"\s*:\s*" + r"(?P.+)$", + re.IGNORECASE, +) + +# Classic format: SETSxREPSxWEIGHT (e.g. 4x8x35, 3x10) +CLASSIC_RE = re.compile( + r"^(?P\d+)\s*[x*]\s*(?P\d+)" + r"(?:\s*[x*]\s*(?P[\d.]+))?" + r"$", + re.IGNORECASE, +) + +# Per-set entry: REPSxWEIGHT or just REPS (e.g. 8x25, or 12) +SET_ENTRY_RE = re.compile( + r"^(?P\d+)" + r"(?:\s*[x*]\s*(?P[\d.]+))?" + r"$", re.IGNORECASE, ) @@ -54,37 +79,80 @@ def parse_exercise_line(line: str) -> Exercise | None: if not line: return None - m = EXERCISE_RE.match(line) + m = HEADER_RE.match(line) if not m: return None + name = m.group("name").strip() + machine_id = m.group("machine").strip() if m.group("machine") else None + rest = m.group("rest").strip() + + # Try classic format first: SETSxREPSxWEIGHT or SETSxREPS + classic = CLASSIC_RE.match(rest) + if classic: + sets = int(classic.group("sets")) + reps = int(classic.group("reps")) + weight = float(classic.group("weight")) if classic.group("weight") else 0.0 + details = [SetDetail(reps=reps, weight_kg=weight)] * sets + return Exercise( + name=name, + machine_id=machine_id, + sets=sets, + reps=reps, + weight_kg=weight, + raw_line=line, + sets_detail=details, + ) + + # Try comma-separated per-set format: 8x25, 5x35, 6x40 or 12, 10, 8 + entries = [e.strip() for e in rest.split(",")] + details = [] + for entry in entries: + em = SET_ENTRY_RE.match(entry) + if not em: + return None # one bad entry invalidates the line + reps = int(em.group("reps")) + weight = float(em.group("weight")) if em.group("weight") else 0.0 + details.append(SetDetail(reps=reps, weight_kg=weight)) + + if not details: + return None + return Exercise( - name=m.group("name").strip(), - machine_id=m.group("machine"), - sets=int(m.group("sets")), - reps=int(m.group("reps")), - weight_kg=float(m.group("weight")), + name=name, + machine_id=machine_id, + sets=len(details), + reps=details[0].reps, + weight_kg=details[0].weight_kg, raw_line=line, + sets_detail=details, ) -def parse_workout(text: str) -> list[list[Exercise]]: +class ParseError: + """Represents a line that looks like a workout entry but failed to parse.""" + def __init__(self, line: str, reason: str): + self.line = line + self.reason = reason + + +def parse_workout(text: str) -> tuple[list[list[Exercise]], list[ParseError]]: """ Parse a full workout message into superset groups. - Returns a list of groups, where each group is a list of Exercises. - Consecutive non-blank lines form a superset group. - Blank lines separate groups. + Returns (groups, errors): + - groups: list of superset groups, each a list of Exercises + - errors: list of ParseError for lines that looked like exercises but failed """ lines = text.strip().splitlines() groups: list[list[Exercise]] = [] current_group: list[Exercise] = [] + errors: list[ParseError] = [] for line in lines: stripped = line.strip() if not stripped: - # blank line β†’ end current group if current_group: groups.append(current_group) current_group = [] @@ -93,13 +161,20 @@ def parse_workout(text: str) -> list[list[Exercise]]: exercise = parse_exercise_line(stripped) if exercise: current_group.append(exercise) - # non-matching lines are silently skipped (e.g. notes, headers) + elif ":" in stripped: + # Has a colon β€” likely an attempted exercise line that failed to parse + errors.append(ParseError(stripped, "could not parse sets/reps/weight after colon")) + # Lines without colons are silently skipped (notes, headers, etc.) - # flush last group if current_group: groups.append(current_group) - return groups + return groups, errors + + +def _fmt_weight(w: float) -> str: + """Format weight: 70.0 β†’ '70', 22.5 β†’ '22.5'.""" + return str(int(w)) if w == int(w) else str(w) def format_workout(superset_groups: list[list[dict]], include_raw: bool = False) -> str: @@ -107,15 +182,34 @@ def format_workout(superset_groups: list[list[dict]], include_raw: bool = False) parts = [] for i, group in enumerate(superset_groups): if i > 0: - parts.append("") # blank line between groups + parts.append("") is_superset = len(group) > 1 if is_superset: - parts.append("πŸ”— Superset:") + parts.append("\U0001f517 Superset:") for ex in group: machine = f" ({ex['machine_id']})" if ex.get("machine_id") else "" - line = f" β€’ {ex['name']}{machine}: {ex['sets']}x{ex['reps']}x{ex['weight_kg']}kg" + details = ex.get("sets_detail", []) + if details and not all( + d["reps"] == details[0]["reps"] and d["weight_kg"] == details[0]["weight_kg"] + for d in details + ): + # Varying sets β€” show each + set_strs = [] + for d in details: + if d["weight_kg"]: + set_strs.append(f"{d['reps']}x{_fmt_weight(d['weight_kg'])}kg") + else: + set_strs.append(f"{d['reps']}") + line = f" \u2022 {ex['name']}{machine}: {', '.join(set_strs)}" + else: + # Uniform sets β€” compact format + w = ex.get('weight_kg', 0) + if w: + line = f" \u2022 {ex['name']}{machine}: {ex['sets']}x{ex['reps']}x{_fmt_weight(w)}kg" + else: + line = f" \u2022 {ex['name']}{machine}: {ex['sets']}x{ex['reps']}" parts.append(line) return "\n".join(parts) diff --git a/server.py b/server.py index 963ea78..5fc62da 100644 --- a/server.py +++ b/server.py @@ -2,8 +2,10 @@ API + static file server for the Telegram Mini App. Serves webapp/ and REST endpoints, using the existing db.py layer. """ +import csv import hashlib import hmac +import io import json import logging import os @@ -11,7 +13,7 @@ from urllib.parse import parse_qs from aiohttp import web -from db import init_db, get_db, save_workout, get_workouts, get_workout_count +from db import init_db, get_db, save_workout, get_workouts, get_workout_count, get_stats_sql, delete_workout, export_workouts from parser import parse_workout, format_workout logging.basicConfig( @@ -122,9 +124,13 @@ async def api_save_workout(request: web.Request): ) elif raw_text: # Text-based input (same format as sending a message to the bot) - groups = parse_workout(raw_text) + groups, errors = parse_workout(raw_text) if not groups: - return web.json_response({"error": "Could not parse workout text"}, status=400) + error_lines = [e.line for e in errors] if errors else [] + return web.json_response( + {"error": "Could not parse workout text", "failed_lines": error_lines}, + status=400, + ) from datetime import datetime, timezone superset_dicts = [[ex.to_dict() for ex in group] for group in groups] workout_id = save_workout( @@ -141,6 +147,15 @@ async def api_save_workout(request: web.Request): return web.json_response({"workout_id": workout_id}, status=201) +@require_auth +async def api_delete_workout(request: web.Request): + """Delete a workout by ID.""" + workout_id = int(request.match_info["workout_id"]) + if delete_workout(request["user_id"], workout_id): + return web.json_response({"deleted": True}) + return web.json_response({"error": "Not found"}, status=404) + + @require_auth async def api_get_exercise_names(request: web.Request): """Return unique exercise names this user has logged (for autocomplete).""" @@ -160,31 +175,33 @@ async def api_get_exercise_names(request: web.Request): @require_auth async def api_get_stats(request: web.Request): """Return summary stats for the user.""" - user_id = request["user_id"] - total = get_workout_count(user_id) - if total == 0: - return web.json_response({ - "total_workouts": 0, "unique_exercises": 0, - "total_sets": 0, "total_volume": 0, - }) + stats = get_stats_sql(request["user_id"]) + return web.json_response(stats) - workouts = get_workouts(user_id, limit=10000) - exercise_names = set() - total_sets = 0 - total_volume = 0.0 - for w in workouts: - for group in w["superset_groups"]: - for ex in group: - exercise_names.add(ex["name"].lower()) - total_sets += ex["sets"] - total_volume += ex["sets"] * ex["reps"] * ex["weight_kg"] - return web.json_response({ - "total_workouts": total, - "unique_exercises": len(exercise_names), - "total_sets": total_sets, - "total_volume": round(total_volume, 1), - }) +@require_auth +async def api_export_json(request: web.Request): + """Export all workouts as JSON.""" + data = export_workouts(request["user_id"]) + return web.json_response({"records": data, "count": len(data)}) + + +@require_auth +async def api_export_csv(request: web.Request): + """Export all workouts as CSV.""" + data = export_workouts(request["user_id"]) + + output = io.StringIO() + if data: + writer = csv.DictWriter(output, fieldnames=data[0].keys()) + writer.writeheader() + writer.writerows(data) + + return web.Response( + text=output.getvalue(), + content_type="text/csv", + headers={"Content-Disposition": "attachment; filename=workouts.csv"}, + ) # ── App setup ──────────────────────────────────────────────────── @@ -196,8 +213,11 @@ def create_app() -> web.Application: app.router.add_get("/api/workouts", api_get_workouts) app.router.add_post("/api/workouts", api_save_workout) + app.router.add_delete("/api/workouts/{workout_id}", api_delete_workout) app.router.add_get("/api/exercises", api_get_exercise_names) app.router.add_get("/api/stats", api_get_stats) + app.router.add_get("/api/export/json", api_export_json) + app.router.add_get("/api/export/csv", api_export_csv) # Serve the webapp/ folder import pathlib