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 <noreply@anthropic.com>
This commit is contained in:
Danny 2026-03-24 15:50:05 +01:00
commit 817cf8fd95
8 changed files with 590 additions and 0 deletions

2
.env.example Normal file
View file

@ -0,0 +1,2 @@
# Get your bot token from @BotFather on Telegram
BOT_TOKEN=your-token-here

5
.gitignore vendored Normal file
View file

@ -0,0 +1,5 @@
.env
workouts.db
__pycache__/
*.pyc
result

211
bot.py Normal file
View file

@ -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(
"💪 <b>Fitness Tracker Bot</b>\n\n"
"Send me your workout and I'll save it!\n\n"
"<b>Format:</b>\n"
"<code>Bench press: 4x8x35</code>\n"
"<code>Lateral raise: 4x8x4</code>\n\n"
"<code>Tri Press rom: 3x10x45</code>\n\n"
"Lines without a blank line between them = superset.\n"
"Machine IDs go in parentheses: <code>Lat pulldown (500): 3x5x45</code>\n\n"
"You can also <b>forward</b> messages from Saved Messages — "
"I'll use the original timestamp.\n\n"
"<b>Commands:</b>\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"📅 <b>{ts.strftime('%a %d %b %Y, %H:%M')}</b>"
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\n<i>Showing latest 5 of {total} workouts.</i>"
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"📊 <b>Your Stats</b>\n\n"
f" • Workouts logged: <b>{total}</b>\n"
f" • Unique exercises: <b>{len(exercise_names)}</b>\n"
f" • Total sets: <b>{total_sets}</b>\n"
f" • Total volume: <b>{total_volume:,.0f} kg</b>",
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"✅ <b>Workout #{workout_id} saved!</b>",
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()

148
db.py Normal file
View file

@ -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"]

61
flake.lock generated Normal file
View file

@ -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
}

40
flake.nix Normal file
View file

@ -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"
'');
};
}
);
}

121
parser.py Normal file
View file

@ -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<name>.+?)" # exercise name (lazy)
r"(?:\s*\((?P<machine>\d+)\))?" # optional (machine_id)
r"\s*:\s*" # colon separator
r"(?P<sets>\d+)\s*x\s*" # sets
r"(?P<reps>\d+)\s*x\s*" # reps
r"(?P<weight>[\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("🔗 <b>Superset:</b>")
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)

2
requirements.txt Normal file
View file

@ -0,0 +1,2 @@
python-telegram-bot>=21.0
python-dotenv>=1.0