commit 817cf8fd95b2ffd333bdbbf134e2b284f76646cf Author: Danny Date: Tue Mar 24 15:50:05 2026 +0100 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 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