From 6fb6207041fb2316b4f1a02a87b57164c570db8f Mon Sep 17 00:00:00 2001 From: Danny Date: Mon, 13 Apr 2026 20:41:37 +0200 Subject: [PATCH] feat(tg-fitness-bot): soft delete, edit workouts, notes via API Deleted workouts are marked with deleted_at instead of being removed. All queries filter on deleted_at IS NULL. New update_workout() does soft-delete + recreate preserving the original timestamp. PUT endpoint at /api/workouts/{id}. POST /api/workouts now accepts a note field. Co-Authored-By: Claude Opus 4.6 (1M context) --- db.py | 102 ++++++++++++++++++++++++++++++++++-------------------- server.py | 25 +++++++++++-- 2 files changed, 88 insertions(+), 39 deletions(-) diff --git a/db.py b/db.py index 5920603..5f424cd 100644 --- a/db.py +++ b/db.py @@ -66,21 +66,40 @@ def init_db(): 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") + if "deleted_at" not in cols: + conn.execute("ALTER TABLE workouts ADD COLUMN deleted_at 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_exercises(conn, workout_id: int, superset_groups: list[list[dict]]): + """Insert superset groups and exercises for a workout.""" + for group_pos, group in enumerate(superset_groups): + cur = conn.execute( + "INSERT INTO superset_groups (workout_id, position) VALUES (?, ?)", + (workout_id, group_pos), + ) + group_id = cur.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, sets_detail) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", + (group_id, ex_pos, ex["name"], ex.get("machine_id"), + ex["sets"], ex["reps"], ex["weight_kg"], ex.get("raw_line"), + sets_detail_json), + ) + + 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, sets_detail} - raw_text: the full original message text, stored verbatim. - - Returns the workout id. + Save a parsed workout. Returns the workout id. """ with get_db() as conn: cur = conn.execute( @@ -88,47 +107,57 @@ def save_workout(user_id: int, timestamp: datetime, superset_groups: list[list[d (user_id, timestamp.isoformat(), note, raw_text), ) workout_id = cur.lastrowid - - for group_pos, group in enumerate(superset_groups): - cur2 = conn.execute( - "INSERT INTO superset_groups (workout_id, position) VALUES (?, ?)", - (workout_id, group_pos), - ) - 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, sets_detail) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", - (group_id, ex_pos, ex["name"], ex.get("machine_id"), - ex["sets"], ex["reps"], ex["weight_kg"], ex.get("raw_line"), - sets_detail_json), - ) - + _save_exercises(conn, workout_id, superset_groups) return workout_id +def update_workout(user_id: int, workout_id: int, superset_groups: list[list[dict]], note: str | None = None) -> int | None: + """ + Update a workout by soft-deleting the old one and creating a new one + with the same timestamp. Returns the new workout id, or None if not found. + """ + with get_db() as conn: + # Fetch the original workout + row = conn.execute( + "SELECT timestamp, raw_text FROM workouts WHERE id = ? AND user_id = ? AND deleted_at IS NULL", + (workout_id, user_id), + ).fetchone() + if not row: + return None + + # Soft-delete the old workout + conn.execute( + "UPDATE workouts SET deleted_at = datetime('now') WHERE id = ?", + (workout_id,), + ) + + # Create new workout with original timestamp + cur = conn.execute( + "INSERT INTO workouts (user_id, timestamp, note, raw_text) VALUES (?, ?, ?, ?)", + (user_id, row["timestamp"], note, row["raw_text"]), + ) + new_id = cur.lastrowid + _save_exercises(conn, new_id, superset_groups) + return new_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.""" + """Soft-delete a workout. Returns True if found, False otherwise.""" with get_db() as conn: cur = conn.execute( - "DELETE FROM workouts WHERE id = ? AND user_id = ?", + "UPDATE workouts SET deleted_at = datetime('now') WHERE id = ? AND user_id = ? AND deleted_at IS NULL", (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.""" + """Fetch recent non-deleted workouts for a user, newest first.""" with get_db() as conn: rows = conn.execute( """SELECT id, timestamp, note, raw_text, created_at FROM workouts - WHERE user_id = ? + WHERE user_id = ? AND deleted_at IS NULL ORDER BY timestamp DESC LIMIT ? OFFSET ?""", (user_id, limit, offset), @@ -154,7 +183,6 @@ def get_workouts(user_id: int, limit: int = 10, offset: int = 0) -> list[dict]: if gp not in superset_groups: superset_groups[gp] = [] 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"]) @@ -173,7 +201,7 @@ def get_workouts(user_id: int, limit: int = 10, offset: int = 0) -> list[dict]: def get_workout_count(user_id: int) -> int: with get_db() as conn: row = conn.execute( - "SELECT COUNT(*) as cnt FROM workouts WHERE user_id = ?", (user_id,) + "SELECT COUNT(*) as cnt FROM workouts WHERE user_id = ? AND deleted_at IS NULL", (user_id,) ).fetchone() return row["cnt"] @@ -190,7 +218,7 @@ def get_stats_sql(user_id: int) -> dict: 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 = ? + WHERE w.user_id = ? AND w.deleted_at IS NULL """, (user_id,)).fetchone() return { @@ -202,7 +230,7 @@ def get_stats_sql(user_id: int) -> dict: def export_workouts(user_id: int) -> list[dict]: - """Export all workouts as flat records for CSV/JSON export.""" + """Export all non-deleted workouts as flat records for CSV/JSON export.""" with get_db() as conn: rows = conn.execute(""" SELECT @@ -213,7 +241,7 @@ def export_workouts(user_id: int) -> list[dict]: 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 = ? + WHERE w.user_id = ? AND w.deleted_at IS NULL ORDER BY w.timestamp DESC, sg.position, e.position """, (user_id,)).fetchall() diff --git a/server.py b/server.py index 5fc62da..ae88f51 100644 --- a/server.py +++ b/server.py @@ -13,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, get_stats_sql, delete_workout, export_workouts +from db import init_db, get_db, save_workout, get_workouts, get_workout_count, get_stats_sql, delete_workout, update_workout, export_workouts from parser import parse_workout, format_workout logging.basicConfig( @@ -112,6 +112,7 @@ async def api_save_workout(request: web.Request): body = await request.json() raw_text = body.get("raw_text", "") superset_groups = body.get("superset_groups") + note = body.get("note") or None if superset_groups: # Structured input from the Mini App UI @@ -121,6 +122,7 @@ async def api_save_workout(request: web.Request): timestamp=datetime.now(timezone.utc), superset_groups=superset_groups, raw_text=raw_text or None, + note=note, ) elif raw_text: # Text-based input (same format as sending a message to the bot) @@ -138,6 +140,7 @@ async def api_save_workout(request: web.Request): timestamp=datetime.now(timezone.utc), superset_groups=superset_dicts, raw_text=raw_text, + note=note, ) else: return web.json_response( @@ -147,9 +150,26 @@ async def api_save_workout(request: web.Request): return web.json_response({"workout_id": workout_id}, status=201) +@require_auth +async def api_update_workout(request: web.Request): + """Update a workout — soft-deletes old, creates new with same timestamp.""" + workout_id = int(request.match_info["workout_id"]) + body = await request.json() + superset_groups = body.get("superset_groups") + note = body.get("note") or None + + if not superset_groups: + return web.json_response({"error": "Provide superset_groups"}, status=400) + + new_id = update_workout(request["user_id"], workout_id, superset_groups, note=note) + if new_id is None: + return web.json_response({"error": "Not found"}, status=404) + return web.json_response({"workout_id": new_id}) + + @require_auth async def api_delete_workout(request: web.Request): - """Delete a workout by ID.""" + """Soft-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}) @@ -213,6 +233,7 @@ 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_put("/api/workouts/{workout_id}", api_update_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)