dannydannydanny/bigbiggerbiggestbot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: drop the slash-command bot — Mini App is the only interface

Browse source 
bot.py is gone. /start, /history, /stats, /delete, /export,
/feedback, and the text-message workout parser are no longer
exposed. Everything those commands did is already available in
the Mini App (history listing, stats, edit/delete, JSON export
via /api/export/json, etc.).

Why: prod runs behind a Mini App URL, and shipyard staging is a
tenant under the existing shipyard_poc_bot which polls Telegram
itself. A second polling process on the same token would 409. By
removing polling entirely, prod and staging share one
architecture: a pure HTTP server validated against whatever
BOT_TOKEN is provided.

Changes:
- delete bot.py
- start.py: stop spawning the bot subprocess; load token, start
  server, optionally start cloudflared. WEBAPP_URL still skips
  the tunnel.
- flake.nix / requirements.txt: drop python-telegram-bot.
- README: rewrite to reflect Mini-App-only architecture.

The prod systemd unit doesn't need to change — its ExecStart is
`python start.py`, which now boots only the server (+ no tunnel
since WEBAPP_URL is set in the unit env).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Danny 2026-05-10 13:40:28 +02:00
parent aa43e492c3
commit 17248e239b
5 changed files with 53 additions and 406 deletions
Download patch file Download diff file Expand all files Collapse all files

52
README.md
View file

@ -1,14 +1,13 @@
# BigBiggerBiggestBot 💪
A Telegram bot for logging gym workouts, with an embedded Mini App.
Send workouts as plain text, forward them from Saved Messages, or tap
through a structured log form inside Telegram. History, stats, notes,
A Telegram **Mini App** for logging gym workouts. History, stats, notes,
edit & delete, JSON/CSV export — all per-user, all in SQLite.
## Format
The slash-command bot was removed: the Mini App is the only interface.
A Telegram bot identity (token) is still required so the Mini App can
validate user sessions via `initData` HMAC.
Send messages like:
## Workout text format (still supported via "Paste as text" in the Mini App)
```
Bench press: 4x8x35
@ -23,15 +22,6 @@ Pull-ups: 3x10
- Blank line separates superset groups; consecutive lines form a superset
- Both `,` and `.` work as decimal separators
## Commands
- `/start` — help & open Mini App
- `/history` — recent workouts
- `/stats` — summary (total workouts, sets, volume)
- `/delete <id>` — soft-delete a workout
- `/export` — download all data as JSON
- `/feedback <text>` — send feedback to the bot author
## Run locally
```bash
@ -39,9 +29,9 @@ nix run
```
This launches:
- API server (port 8080)
- cloudflared tunnel for the Mini App
- Telegram bot (polling)
- API + Mini App server (port 8080)
- cloudflared Quick Tunnel for a public HTTPS URL (skipped if `WEBAPP_URL`
is already set in the environment, e.g. fronted by a reverse proxy)
Put your bot token (from [@BotFather](https://t.me/BotFather)) in
`~/.secrets/bigbiggerbiggestbot` or a `.env` file:
@ -63,10 +53,12 @@ nix develop --command pytest tests/ -v
Two environments share one host (`sunken-ship`):
- **Production** — `fitness-bot.service`, working dir `/home/danny/tg_fitness_bot`,
watches `origin/main`, served behind a stable URL via the VPS Caddy.
watches `origin/main`, served behind a stable URL via the VPS Caddy at
`https://bbbot.dannydannydanny.me`.
- **Shipyard staging** — `fitness-bot-shipyard.service`, working dir
`/home/danny/tg_fitness_bot_shipyard`, watches `origin/staging`, separate
bot token, ephemeral cloudflared URL each restart.
`/home/danny/tg_fitness_bot_shipyard`, watches `origin/staging`, served
by the shared `shipyard_poc_bot` Telegram bot (B3Bot beta is the active
POC tenant).
Each has its own pull timer that fetches every ~15 minutes and restarts
the service when its branch has new commits.
@ -74,7 +66,6 @@ the service when its branch has new commits.
**Workflow:**
```
# 1. land changes on a working branch (or main locally)
git push origin <branch>:staging # → shipyard auto-deploys, test there
git push origin <branch>:main # → production auto-deploys
```
@ -84,13 +75,16 @@ so testing on shipyard never touches production data.
## Architecture
- `bot.py` — Telegram command handlers, polling, message parsing
- `server.py` — aiohttp REST API + static file server for the Mini App
- `db.py` — SQLite data layer (workouts, supersets, exercises, feedback; soft delete)
- `parser.py` — workout text → structured data
- `webapp/` — Mini App (HTML/CSS/vanilla JS, Telegram WebApp SDK)
- `start.py` — orchestrator: starts server + tunnel + bot, wires up the Mini App URL
- `tests/` — pytest suite for parser + db
- `server.py` — aiohttp REST API + static file server for the Mini App;
validates Telegram `initData` HMACs against the bot token.
- `db.py` — SQLite data layer (workouts, supersets, exercises, feedback,
events, settings; soft delete).
- `parser.py` — workout text → structured data (used by the Mini App's
"Paste as text" path).
- `webapp/` — Mini App (HTML/CSS/vanilla JS, Telegram WebApp SDK).
- `start.py` — orchestrator: loads token, starts server, optionally starts
cloudflared.
- `tests/` — pytest suite for parser + db.
## License

336
bot.py
View file

@ -1,336 +0,0 @@
"""Telegram Fitness Bot — track your workouts."""
import logging
import os
from datetime import datetime, timezone
from dotenv import load_dotenv
from telegram import Update, WebAppInfo, InlineKeyboardButton, InlineKeyboardMarkup, MenuButtonWebApp
from telegram.constants import ParseMode
from telegram.ext import (
Application,
ApplicationBuilder,
CommandHandler,
ContextTypes,
MessageHandler,
filters,
)
from db import init_db, save_workout, get_workouts, get_workout_count, get_stats_sql, delete_workout, save_feedback, get_user_workout_number, resolve_user_number, log_event
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: BOT_TOKEN env var → secrets file
# Env var wins so multiple instances on the same host (e.g. prod + shipyard
# staging) can each point to a different token without sharing a secrets file.
SECRETS_FILE = os.path.expanduser("~/.secrets/bigbiggerbiggestbot")
def _load_token() -> str:
# 1. Env var (set by systemd EnvironmentFile in multi-instance setups)
token = os.environ.get("BOT_TOKEN", "").strip()
if token:
return token
# 2. Default secrets file
if os.path.isfile(SECRETS_FILE):
token = open(SECRETS_FILE).read().strip()
if token:
return token
raise RuntimeError(
f"No bot token found. Set BOT_TOKEN env var or put it in {SECRETS_FILE}."
)
BOT_TOKEN = _load_token()
# Mini App URL — set automatically by start.py via tunnel
WEBAPP_URL = os.environ.get("WEBAPP_URL", "")
# ── 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):
log_event(update.effective_user.id, "cmd.start")
text = (
"\U0001f4aa <b>Fitness Tracker Bot</b>\n\n"
"Send me your workout and I'll save it!\n\n"
"<b>Formats:</b>\n"
"<code>Bench press: 4x8x35</code>\n"
"<code>Pull-ups: 3x10</code> (bodyweight)\n"
"<code>Shoulder press (3032): 8x25, 5x35, 6x40</code>\n\n"
"Lines without a blank line between them = superset.\n"
"Machine IDs go in parentheses.\n\n"
"You can also <b>forward</b> messages from Saved Messages \u2014 "
"I'll use the original timestamp.\n\n"
"<b>Commands:</b>\n"
"/history \u2014 view recent workouts\n"
"/stats \u2014 quick summary\n"
"/delete &lt;number&gt; \u2014 delete a workout (see /history)\n"
"/export \u2014 export all data as JSON\n"
"/feedback &lt;text&gt; \u2014 send feedback"
)
if WEBAPP_URL:
btn = InlineKeyboardButton(
text="Open Workout Tracker",
web_app=WebAppInfo(url=WEBAPP_URL),
)
await update.message.reply_text(
text, parse_mode=ParseMode.HTML,
reply_markup=InlineKeyboardMarkup([[btn]]),
)
else:
await update.message.reply_text(text, parse_mode=ParseMode.HTML)
async def cmd_history(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.effective_user.id
log_event(user_id, "cmd.history")
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"\U0001f4c5 <b>{ts.strftime('%a %d %b %Y, %H:%M')}</b> (#{w['user_number']})"
body = format_workout(w["superset_groups"])
parts.append(f"{header}\n{body}")
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\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
log_event(user_id, "cmd.stats")
stats = get_stats_sql(user_id)
if stats["total_workouts"] == 0:
await update.message.reply_text("No workouts yet \u2014 send me your first one!")
return
await update.message.reply_text(
f"\U0001f4ca <b>Your Stats</b>\n\n"
f" \u2022 Workouts logged: <b>{stats['total_workouts']}</b>\n"
f" \u2022 Unique exercises: <b>{stats['unique_exercises']}</b>\n"
f" \u2022 Total sets: <b>{stats['total_sets']}</b>\n"
f" \u2022 Total volume: <b>{stats['total_volume']:,.0f} kg</b>",
parse_mode=ParseMode.HTML,
)
async def cmd_delete(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.effective_user.id
log_event(user_id, "cmd.delete", {"args": context.args or []})
if not context.args:
await update.message.reply_text(
"Usage: /delete &lt;number&gt;\n"
"Use /history to see workout numbers.",
parse_mode=ParseMode.HTML,
)
return
try:
user_number = int(context.args[0])
except ValueError:
await update.message.reply_text("Workout number must be a number.")
return
workout_id = resolve_user_number(user_id, user_number)
if workout_id is not None and delete_workout(user_id, workout_id):
log_event(user_id, "workout.delete", {"workout_id": workout_id, "user_number": user_number})
await update.message.reply_text(f"\U0001f5d1 Workout #{user_number} deleted.")
else:
await update.message.reply_text(
f"Workout #{user_number} not found."
)
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
log_event(user_id, "cmd.export")
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.",
)
async def cmd_feedback(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.effective_user.id
text = " ".join(context.args) if context.args else ""
if not text:
await update.message.reply_text(
"Usage: /feedback &lt;your feedback&gt;",
parse_mode=ParseMode.HTML,
)
return
save_feedback(user_id, text)
log_event(user_id, "cmd.feedback")
await update.message.reply_text("\U0001f4dd Feedback saved, thanks!")
# ── 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, 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 <code>{e.line}</code>" 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"<b>Expected formats:</b>\n"
f"<code>Exercise: 4x8x35</code>\n"
f"<code>Exercise: 3x10</code> (bodyweight)\n"
f"<code>Exercise: 8x25, 5x35, 6x40</code>",
parse_mode=ParseMode.HTML,
)
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)
user_number = get_user_workout_number(user_id, workout_id) or workout_id
log_event(user_id, "workout.save", {
"source": "text",
"workout_id": workout_id,
"user_number": user_number,
"forwarded": is_forwarded,
})
# 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"\u2705 <b>Workout #{user_number} saved!</b>",
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"\U0001f517 {supersets} superset(s)")
# Show errors for partially parsed workouts
if errors:
skipped = "\n".join(f" \u2022 <code>{e.line}</code>" 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)}")
await update.message.reply_text(
"\n".join(confirm_parts),
parse_mode=ParseMode.HTML,
)
# ── Main ─────────────────────────────────────────────────────────────────────
async def post_init(app: Application):
"""Set the bot's menu button to open the Mini App (if URL is available)."""
if WEBAPP_URL:
await app.bot.set_chat_menu_button(
menu_button=MenuButtonWebApp(
text="Workout",
web_app=WebAppInfo(url=WEBAPP_URL),
)
)
logger.info("Menu button set to Mini App at %s", WEBAPP_URL)
else:
logger.info("No WEBAPP_URL — menu button not set")
def main():
init_db()
builder = ApplicationBuilder().token(BOT_TOKEN)
if WEBAPP_URL:
builder = builder.post_init(post_init)
app = builder.build()
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))
app.add_handler(CommandHandler("feedback", cmd_feedback))
# Handle all text messages (including forwarded ones)
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
logger.info("Bot started \u2014 polling\u2026")
app.run_polling(allowed_updates=Update.ALL_TYPES)
if __name__ == "__main__":
main()

6
flake.nix
View file

@ -14,7 +14,8 @@
python = pkgs.python3;
pythonEnv = python.withPackages (ps: with ps; [
python-telegram-bot
# python-telegram-bot was used by the now-deleted bot.py polling
# loop; the Mini App backend doesn't need it. Kept off this list.
python-dotenv
aiohttp
pytest
@ -27,8 +28,7 @@
packages = [ pythonEnv pkgs.cloudflared ];
shellHook = ''
echo "💪 BigBiggerBiggestBot dev shell"
echo " Run: python start.py (server + tunnel + bot)"
echo " Run: python bot.py (bot only, no mini app)"
echo " Run: python start.py (server + tunnel)"
'';
};

2
requirements.txt
View file

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

63
start.py
View file

@ -1,11 +1,15 @@
#!/usr/bin/env python3
"""
Orchestrator — single entry point for `nix run`.
1. Loads BOT_TOKEN from ~/.secrets or .env (same as bot.py)
2. Starts the API server
3. Starts localtunnel to get a public HTTPS URL
4. Starts the Telegram bot with WEBAPP_URL set
5. Cleans up everything on Ctrl+C
1. Loads BOT_TOKEN (env var → ~/.secrets → .env)
2. Starts the API + Mini App server
3. Optionally starts a cloudflared Quick Tunnel for a public HTTPS URL
(skipped if WEBAPP_URL is already set, e.g. fronted by a reverse proxy)
4. Cleans up on Ctrl+C
The slash-command bot was removed: the Mini App is the only interface.
The bot's identity (token) only matters now for validating Telegram WebApp
initData HMACs and for the menu-button URL — both handled in server.py.
"""
import os
import re
@ -23,8 +27,8 @@ SECRETS_FILE = pathlib.Path.home() / ".secrets" / "bigbiggerbiggestbot"
def load_token() -> str:
"""Load bot token: BOT_TOKEN env → secrets file → .env in cwd.
Env var wins so multiple instances on the same host (prod + shipyard)
can each get a distinct token via systemd EnvironmentFile.
Env var wins so multiple instances on the same host (prod + shipyard
staging) can each get a distinct token via systemd EnvironmentFile.
"""
# 1. Already in environment (systemd EnvironmentFile sets this for staging)
token = os.environ.get("BOT_TOKEN", "").strip()
@ -32,7 +36,7 @@ def load_token() -> str:
print(" Token loaded from BOT_TOKEN env var")
return token
# 2. Secrets file (same path as bot.py uses)
# 2. Default secrets file
if SECRETS_FILE.is_file():
token = SECRETS_FILE.read_text().strip()
if token:
@ -102,21 +106,13 @@ def start_tunnel(port: int) -> tuple[subprocess.Popen, str]:
# Keep draining cloudflared output so its pipe buffer doesn't fill up
# (which would block the process and kill the tunnel)
def _drain():
for line in proc.stdout:
for _line in proc.stdout:
pass
threading.Thread(target=_drain, daemon=True).start()
return proc, url
def start_bot(bot_token: str, webapp_url: str) -> subprocess.Popen:
env = {**os.environ, "BOT_TOKEN": bot_token, "WEBAPP_URL": webapp_url}
return subprocess.Popen(
[sys.executable, str(SCRIPT_DIR / "bot.py")],
env=env,
)
def main():
port = int(os.environ.get("API_PORT", "8080"))
procs: list[subprocess.Popen] = []
@ -140,11 +136,11 @@ def main():
print()
print(" ==========================================")
print(" BigBiggerBiggest — Fitness Tracker")
print(" BigBiggerBiggest — Mini App backend")
print(" ==========================================")
print()
# 1. Load token
# 1. Load token (used by server.py to validate initData HMACs)
bot_token = load_token()
masked = bot_token[:5] + "..." + bot_token[-4:]
print(f" BOT_TOKEN: {masked}")
@ -159,11 +155,10 @@ def main():
print(" Server failed to start!")
sys.exit(1)
# 3. Tunnel — skipped if WEBAPP_URL is already provided (e.g. the bot
# is fronted by an external reverse proxy that terminates TLS and
# proxies back to localhost:$API_PORT over a private network).
# 3. Tunnel — skipped if WEBAPP_URL is already provided (e.g. fronted
# by an external reverse proxy that terminates TLS and proxies back to
# localhost:$API_PORT over a private network).
env_webapp_url = os.environ.get("WEBAPP_URL", "").strip()
tunnel = None
if env_webapp_url:
webapp_url = env_webapp_url
print(f" WEBAPP_URL from environment: {webapp_url} (skipping cloudflared)")
@ -171,27 +166,21 @@ def main():
tunnel, webapp_url = start_tunnel(port)
procs.append(tunnel)
# 4. Start bot
print(f"\n WEBAPP_URL: {webapp_url}")
print(" Starting bot...\n")
bot = start_bot(bot_token, webapp_url)
procs.append(bot)
print(" ==========================================")
print(f" All systems go!")
print("\n ==========================================")
print(" All systems go!")
print(f" Mini App: {webapp_url}")
print(f" API: http://localhost:{port}")
print(f" Press Ctrl+C to stop")
print(" ==========================================")
print()
print(" Press Ctrl+C to stop")
print(" ==========================================\n")
proc_names = {id(server): "Server"}
if procs[-1] is not server:
proc_names[id(procs[-1])] = "Tunnel"
while True:
for p in procs:
ret = p.poll()
if ret is not None:
proc_names = {id(server): "Server", id(bot): "Bot"}
if tunnel is not None:
proc_names[id(tunnel)] = "Tunnel"
name = proc_names.get(id(p), "?")
print(f"\n {name} exited with code {ret}")
cleanup()