From 817cf8fd95b2ffd333bdbbf134e2b284f76646cf Mon Sep 17 00:00:00 2001 From: Danny Date: Tue, 24 Mar 2026 15:50:05 +0100 Subject: [PATCH] feat(tg-fitness-bot): initial Telegram workout tracker bot Python bot that parses workout messages (Exercise: SetsxRepsxWeight), detects supersets from consecutive lines, extracts machine IDs, stores both raw message text and parsed data in SQLite, and reads original timestamps from forwarded Saved Messages. Co-Authored-By: Claude Opus 4.6 --- .env.example | 2 + .gitignore | 5 ++ bot.py | 211 +++++++++++++++++++++++++++++++++++++++++++++++ db.py | 148 +++++++++++++++++++++++++++++++++ flake.lock | 61 ++++++++++++++ flake.nix | 40 +++++++++ parser.py | 121 +++++++++++++++++++++++++++ requirements.txt | 2 + 8 files changed, 590 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 bot.py create mode 100644 db.py create mode 100644 flake.lock create mode 100644 flake.nix create mode 100644 parser.py create mode 100644 requirements.txt diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..bd21342 --- /dev/null +++ b/.env.example @@ -0,0 +1,2 @@ +# Get your bot token from @BotFather on Telegram +BOT_TOKEN=your-token-here diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e93c787 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +.env +workouts.db +__pycache__/ +*.pyc +result diff --git a/bot.py b/bot.py new file mode 100644 index 0000000..2e3d99b --- /dev/null +++ b/bot.py @@ -0,0 +1,211 @@ +"""Telegram Fitness Bot — track your workouts.""" + +import logging +import os +from datetime import datetime, timezone + +from dotenv import load_dotenv +from telegram import Update +from telegram.constants import ParseMode +from telegram.ext import ( + ApplicationBuilder, + CommandHandler, + ContextTypes, + MessageHandler, + filters, +) + +from db import init_db, save_workout, get_workouts, get_workout_count +from parser import parse_workout, format_workout + +load_dotenv() + +logging.basicConfig( + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", + level=logging.INFO, +) +logger = logging.getLogger(__name__) + +# Token resolution: secrets file → .env / environment variable +SECRETS_FILE = os.path.expanduser("~/.secrets/bigbiggerbiggestbot") + + +def _load_token() -> str: + # 1. Try the secrets file + if os.path.isfile(SECRETS_FILE): + token = open(SECRETS_FILE).read().strip() + if token: + return token + # 2. Fall back to env var (set via .env or shell) + token = os.environ.get("BOT_TOKEN", "").strip() + if token: + return token + raise RuntimeError( + f"No bot token found. Put it in {SECRETS_FILE} or set BOT_TOKEN env var." + ) + + +BOT_TOKEN = _load_token() + + +# ── Helpers ────────────────────────────────────────────────────────────────── + + +def extract_timestamp(update: Update) -> tuple[datetime, bool]: + """ + Get the best timestamp for a workout message. + + In python-telegram-bot v21+, forwarded message info lives on + message.forward_origin (a MessageOrigin object) with a .date attribute. + + Returns (timestamp, is_forwarded). + """ + msg = update.effective_message + + # v21+: forward_origin is set when a user forwards a message + origin = getattr(msg, "forward_origin", None) + if origin is not None and hasattr(origin, "date"): + return origin.date.replace(tzinfo=timezone.utc), True + + return msg.date.replace(tzinfo=timezone.utc), False + + +# ── Command handlers ──────────────────────────────────────────────────────── + + +async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE): + await update.message.reply_text( + "💪 Fitness Tracker Bot\n\n" + "Send me your workout and I'll save it!\n\n" + "Format:\n" + "Bench press: 4x8x35\n" + "Lateral raise: 4x8x4\n\n" + "Tri Press rom: 3x10x45\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 — " + "I'll use the original timestamp.\n\n" + "Commands:\n" + "/history — view recent workouts\n" + "/stats — quick summary", + parse_mode=ParseMode.HTML, + ) + + +async def cmd_history(update: Update, context: ContextTypes.DEFAULT_TYPE): + user_id = update.effective_user.id + workouts = get_workouts(user_id, limit=5) + + if not workouts: + await update.message.reply_text("No workouts saved yet. Send me one!") + return + + parts = [] + for w in workouts: + ts = datetime.fromisoformat(w["timestamp"]) + header = f"📅 {ts.strftime('%a %d %b %Y, %H:%M')}" + body = format_workout(w["superset_groups"]) + parts.append(f"{header}\n{body}") + + text = "\n\n───────────────\n\n".join(parts) + total = get_workout_count(user_id) + text += f"\n\nShowing latest 5 of {total} workouts." + + await update.message.reply_text(text, parse_mode=ParseMode.HTML) + + +async def cmd_stats(update: Update, context: ContextTypes.DEFAULT_TYPE): + user_id = update.effective_user.id + total = get_workout_count(user_id) + + if total == 0: + await update.message.reply_text("No workouts yet — 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", + parse_mode=ParseMode.HTML, + ) + + +# ── Message handler (workout parsing) ─────────────────────────────────────── + + +async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE): + """Parse any text message as a potential workout.""" + text = update.effective_message.text + if not text: + return + + groups = parse_workout(text) + if not groups: + # Not a workout message — silently ignore so the bot isn't noisy + return + + user_id = update.effective_user.id + timestamp, is_forwarded = extract_timestamp(update) + + superset_dicts = [[ex.to_dict() for ex in group] for group in groups] + workout_id = save_workout(user_id, timestamp, superset_dicts, raw_text=text) + + # Count totals for the confirmation + total_exercises = sum(len(g) for g in groups) + total_sets = sum(ex.sets for g in groups for ex in g) + supersets = sum(1 for g in groups if len(g) > 1) + + 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", + ] + if supersets: + confirm_parts.append(f"🔗 {supersets} superset(s)") + + confirm_parts.append(f"\n{format_workout(superset_dicts)}") + + await update.message.reply_text( + "\n".join(confirm_parts), + parse_mode=ParseMode.HTML, + ) + + +# ── Main ───────────────────────────────────────────────────────────────────── + + +def main(): + init_db() + + app = ApplicationBuilder().token(BOT_TOKEN).build() + + app.add_handler(CommandHandler("start", cmd_start)) + app.add_handler(CommandHandler("history", cmd_history)) + app.add_handler(CommandHandler("stats", cmd_stats)) + + # Handle all text messages (including forwarded ones) + app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message)) + + logger.info("Bot started — polling…") + app.run_polling(allowed_updates=Update.ALL_TYPES) + + +if __name__ == "__main__": + main() diff --git a/db.py b/db.py new file mode 100644 index 0000000..664c6cb --- /dev/null +++ b/db.py @@ -0,0 +1,148 @@ +"""Database layer for the fitness bot.""" + +import sqlite3 +from datetime import datetime +from pathlib import Path +from contextlib import contextmanager + +DB_PATH = Path(__file__).parent / "workouts.db" + + +def get_connection() -> sqlite3.Connection: + conn = sqlite3.connect(DB_PATH) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA foreign_keys=ON") + return conn + + +@contextmanager +def get_db(): + conn = get_connection() + try: + yield conn + conn.commit() + finally: + conn.close() + + +def init_db(): + """Create tables if they don't exist.""" + with get_db() as conn: + conn.executescript(""" + CREATE TABLE IF NOT EXISTS workouts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id INTEGER NOT NULL, + timestamp TEXT NOT NULL, -- ISO-8601, original workout time + created_at TEXT NOT NULL DEFAULT (datetime('now')), + note TEXT -- optional free-text note + ); + + CREATE TABLE IF NOT EXISTS superset_groups ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + workout_id INTEGER NOT NULL REFERENCES workouts(id) ON DELETE CASCADE, + position INTEGER NOT NULL -- ordering within the workout + ); + + CREATE TABLE IF NOT EXISTS exercises ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + 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" + sets INTEGER NOT NULL, + reps INTEGER NOT NULL, + weight_kg REAL NOT NULL, + raw_line TEXT -- the original line as typed + ); + + CREATE INDEX IF NOT EXISTS idx_workouts_user + ON workouts(user_id, timestamp); + """) + + # Migration: add raw_text column if it doesn't exist yet + 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") + + +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} + raw_text: the full original message text, stored verbatim. + + Returns the workout id. + """ + with get_db() as conn: + cur = conn.execute( + "INSERT INTO workouts (user_id, timestamp, note, raw_text) VALUES (?, ?, ?, ?)", + (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): + conn.execute( + """INSERT INTO exercises + (superset_group_id, position, name, machine_id, sets, reps, weight_kg, raw_line) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", + (group_id, ex_pos, ex["name"], ex.get("machine_id"), + ex["sets"], ex["reps"], ex["weight_kg"], ex.get("raw_line")), + ) + + return workout_id + + +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: + rows = conn.execute( + """SELECT id, timestamp, note, raw_text, created_at + FROM workouts + WHERE user_id = ? + ORDER BY timestamp DESC + LIMIT ? OFFSET ?""", + (user_id, limit, offset), + ).fetchall() + + workouts = [] + for row in rows: + 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 + FROM superset_groups sg + JOIN exercises e ON e.superset_group_id = sg.id + WHERE sg.workout_id = ? + ORDER BY sg.position, e.position""", + (row["id"],), + ).fetchall() + + superset_groups = {} + for g in groups: + gp = g["group_pos"] + if gp not in superset_groups: + superset_groups[gp] = [] + superset_groups[gp].append(dict(g)) + + workout["superset_groups"] = [superset_groups[k] for k in sorted(superset_groups)] + workouts.append(workout) + + return workouts + + +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,) + ).fetchone() + return row["cnt"] diff --git a/flake.lock b/flake.lock new file mode 100644 index 0000000..7b63514 --- /dev/null +++ b/flake.lock @@ -0,0 +1,61 @@ +{ + "nodes": { + "flake-utils": { + "inputs": { + "systems": "systems" + }, + "locked": { + "lastModified": 1731533236, + "narHash": "sha256-l0KFg5HjrsfsO/JpG+r7fRrqm12kzFHyUHqHCVpMMbI=", + "owner": "numtide", + "repo": "flake-utils", + "rev": "11707dc2f618dd54ca8739b309ec4fc024de578b", + "type": "github" + }, + "original": { + "owner": "numtide", + "repo": "flake-utils", + "type": "github" + } + }, + "nixpkgs": { + "locked": { + "lastModified": 1774243609, + "narHash": "sha256-6sB2IYqXYwoQS11Ev0u3b0lpAleTpvNv5iv4iqiCCR8=", + "owner": "NixOS", + "repo": "nixpkgs", + "rev": "4724d5647207377bede08da3212f809cbd94a648", + "type": "github" + }, + "original": { + "owner": "NixOS", + "ref": "nixpkgs-unstable", + "repo": "nixpkgs", + "type": "github" + } + }, + "root": { + "inputs": { + "flake-utils": "flake-utils", + "nixpkgs": "nixpkgs" + } + }, + "systems": { + "locked": { + "lastModified": 1681028828, + "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=", + "owner": "nix-systems", + "repo": "default", + "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e", + "type": "github" + }, + "original": { + "owner": "nix-systems", + "repo": "default", + "type": "github" + } + } + }, + "root": "root", + "version": 7 +} diff --git a/flake.nix b/flake.nix new file mode 100644 index 0000000..d20eb4d --- /dev/null +++ b/flake.nix @@ -0,0 +1,40 @@ +{ + description = "BigBiggerBiggestBot — Telegram fitness tracker"; + + inputs = { + nixpkgs.url = "github:NixOS/nixpkgs/nixpkgs-unstable"; + flake-utils.url = "github:numtide/flake-utils"; + }; + + outputs = { self, nixpkgs, flake-utils }: + flake-utils.lib.eachDefaultSystem (system: + let + pkgs = nixpkgs.legacyPackages.${system}; + + python = pkgs.python3; + + pythonEnv = python.withPackages (ps: with ps; [ + python-telegram-bot + python-dotenv + ]); + in + { + # `nix develop` — drop into a shell with everything available + devShells.default = pkgs.mkShell { + packages = [ pythonEnv ]; + shellHook = '' + echo "💪 BigBiggerBiggestBot dev shell" + echo " Run: python bot.py" + ''; + }; + + # `nix run` — start the bot from the current directory + apps.default = { + type = "app"; + program = toString (pkgs.writeShellScript "run-bot" '' + exec ${pythonEnv}/bin/python "$PWD/bot.py" + ''); + }; + } + ); +} diff --git a/parser.py b/parser.py new file mode 100644 index 0000000..c28dcb1 --- /dev/null +++ b/parser.py @@ -0,0 +1,121 @@ +""" +Parse workout messages into structured data. + +Format per line: + Exercise Name (optional_machine_id): SETSxREPSxWEIGHT + +Lines with no blank line between them form a superset group. +Blank lines separate superset groups. +""" + +import re +from dataclasses import dataclass + + +@dataclass +class Exercise: + name: str + machine_id: str | None + sets: int + reps: int + weight_kg: float + raw_line: str + + def to_dict(self) -> dict: + return { + "name": self.name, + "machine_id": self.machine_id, + "sets": self.sets, + "reps": self.reps, + "weight_kg": self.weight_kg, + "raw_line": self.raw_line, + } + + +# 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*$", + re.IGNORECASE, +) + + +def parse_exercise_line(line: str) -> Exercise | None: + """Parse a single exercise line. Returns None if it doesn't match.""" + line = line.strip() + if not line: + return None + + m = EXERCISE_RE.match(line) + if not m: + 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")), + raw_line=line, + ) + + +def parse_workout(text: str) -> list[list[Exercise]]: + """ + 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. + """ + lines = text.strip().splitlines() + groups: list[list[Exercise]] = [] + current_group: list[Exercise] = [] + + for line in lines: + stripped = line.strip() + + if not stripped: + # blank line → end current group + if current_group: + groups.append(current_group) + current_group = [] + continue + + exercise = parse_exercise_line(stripped) + if exercise: + current_group.append(exercise) + # non-matching lines are silently skipped (e.g. notes, headers) + + # flush last group + if current_group: + groups.append(current_group) + + return groups + + +def format_workout(superset_groups: list[list[dict]], include_raw: bool = False) -> str: + """Format structured workout data back into readable text.""" + parts = [] + for i, group in enumerate(superset_groups): + if i > 0: + parts.append("") # blank line between groups + + is_superset = len(group) > 1 + if is_superset: + parts.append("🔗 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" + parts.append(line) + + return "\n".join(parts) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..84250d7 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +python-telegram-bot>=21.0 +python-dotenv>=1.0