diff --git a/telegram-fitness-bot/.envrc b/telegram-fitness-bot/.envrc new file mode 100644 index 0000000..3550a30 --- /dev/null +++ b/telegram-fitness-bot/.envrc @@ -0,0 +1 @@ +use flake diff --git a/telegram-fitness-bot/.gitignore b/telegram-fitness-bot/.gitignore new file mode 100644 index 0000000..35ff47a --- /dev/null +++ b/telegram-fitness-bot/.gitignore @@ -0,0 +1,19 @@ +# Nix +result +result-* +.direnv/ + +# Python +__pycache__/ +*.pyc + +# Database +*.db +*.db-journal +*.db-wal + +# Secrets — never commit +.env + +# Node (localtunnel cache) +node_modules/ diff --git a/telegram-fitness-bot/README.md b/telegram-fitness-bot/README.md new file mode 100644 index 0000000..dcfa3e0 --- /dev/null +++ b/telegram-fitness-bot/README.md @@ -0,0 +1,108 @@ +# Telegram Fitness Bot — Mini App + +A Telegram bot + Mini App for tracking gym workouts. Log exercises, sets, reps, and weight right inside Telegram. + +## Quick Start + +### 1. Get a bot token + +Message [@BotFather](https://t.me/BotFather) on Telegram and create a new bot. + +### 2. Create a `.env` file + +```bash +echo 'BOT_TOKEN=your-token-here' > .env +``` + +### 3. Run + +```bash +nix run +``` + +That's it. The app will: +- Load your `BOT_TOKEN` from `.env` +- Start the API server on port 8080 +- Open a localtunnel to get a public HTTPS URL +- Start the bot with that URL wired in +- Create `fitness.db` in the current directory + +Open your bot in Telegram and tap the **Workout** menu button. + +## Architecture + +``` +┌───────────────────────┐ ┌─────────────────────────┐ +│ Telegram Bot │ │ API Server (aiohttp) │ +│ (python-telegram- │ │ │ +│ bot, polling) │ │ GET/POST /api/* │ +│ │ │ Static /webapp/* │ +│ /start /workout │ │ │ +│ /finish /history │ │ ← Telegram initData │ +└───────────┬───────────┘ │ validation (HMAC) │ + │ └────────────┬────────────┘ + │ │ + └──────────┬───────────────────┘ + │ + ┌──────┴──────┐ + │ SQLite │ + │ fitness.db │ + └─────────────┘ +``` + +## Project Structure + +``` +├── flake.nix # Nix flake — `nix run` entry point +├── start.py # Orchestrator: loads .env, starts server + tunnel + bot +├── bot.py # Telegram bot (commands, reminders) +├── server.py # aiohttp API + static file server +├── database.py # SQLite data layer +├── config.py # Environment-based config +├── .env # Your BOT_TOKEN (not committed) +├── .envrc # direnv — auto-activates nix develop +└── webapp/ + ├── index.html # Mini App entry point + ├── style.css # Telegram-native themed styles + └── app.js # Frontend logic +``` + +## Development + +```bash +# Enter dev shell with all dependencies +nix develop + +# Or with direnv +direnv allow + +# Run directly +python start.py +``` + +## Environment Variables + +| Variable | Default | Description | +|----------|---------|-------------| +| `BOT_TOKEN` | — | Telegram bot token (required, loaded from `.env`) | +| `API_PORT` | `8080` | API server port | +| `DB_PATH` | `./fitness.db` | SQLite database file path | + +`WEBAPP_URL` is set automatically by the localtunnel — you never need to touch it. + +## API Endpoints + +| Method | Path | Description | +|--------|------|-------------| +| GET | `/api/exercises` | List user's exercises | +| POST | `/api/exercises` | Create exercise `{name}` | +| DELETE | `/api/exercises/:id` | Delete exercise | +| GET | `/api/workouts` | Recent workouts with summaries | +| GET | `/api/workouts/active` | Current active workout | +| POST | `/api/workouts` | Start new workout | +| POST | `/api/workouts/:id/finish` | Finish workout | +| GET | `/api/workouts/:id/sets` | Sets in a workout | +| POST | `/api/workouts/:id/sets` | Log a set `{exercise_id, reps, weight}` | +| DELETE | `/api/sets/:id` | Delete a set | + +All API requests require the `X-Telegram-Init-Data` header for authentication. diff --git a/telegram-fitness-bot/bot.py b/telegram-fitness-bot/bot.py new file mode 100644 index 0000000..083d1f9 --- /dev/null +++ b/telegram-fitness-bot/bot.py @@ -0,0 +1,198 @@ +""" +Telegram Fitness Bot — handles chat commands, reminders, and launches the Mini App. +""" +import logging + +from telegram import ( + Update, + WebAppInfo, + InlineKeyboardButton, + InlineKeyboardMarkup, + MenuButtonWebApp, +) +from telegram.ext import ( + Application, + CommandHandler, + ContextTypes, + MessageHandler, + filters, +) + +import database as db +from config import BOT_TOKEN, WEBAPP_URL + +logging.basicConfig( + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + level=logging.INFO, +) +logger = logging.getLogger(__name__) + + +# ── Helpers ────────────────────────────────────────────────────── + +def ensure_user(update: Update) -> dict: + """Create or update the user record from the Telegram message.""" + tg_user = update.effective_user + return db.upsert_user( + telegram_id=tg_user.id, + first_name=tg_user.first_name or "", + username=tg_user.username or "", + ) + + +def format_summary(summary: dict) -> str: + """Format a workout summary dict into a nice chat message.""" + if not summary: + return "No workout data found." + + lines = [f"*Workout Summary*"] + lines.append(f"Started: {summary['started_at']}") + if summary.get("finished_at"): + lines.append(f"Finished: {summary['finished_at']}") + lines.append("") + + for exercise_name, sets in summary["exercises"].items(): + lines.append(f"*{exercise_name}*") + for i, s in enumerate(sets, 1): + lines.append(f" Set {i}: {s['reps']} reps × {s['weight']} kg") + lines.append("") + + lines.append(f"Total sets: {summary['total_sets']}") + lines.append(f"Total volume: {summary['total_volume']} kg") + return "\n".join(lines) + + +# ── Command handlers ───────────────────────────────────────────── + +async def cmd_start(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + """Greet the user and show the Mini App button.""" + ensure_user(update) + webapp_btn = InlineKeyboardButton( + text="Open Workout Tracker", + web_app=WebAppInfo(url=WEBAPP_URL), + ) + keyboard = InlineKeyboardMarkup([[webapp_btn]]) + + await update.message.reply_text( + "Hey! I'm your fitness tracker bot.\n\n" + "Tap the button below to open the workout logger, " + "or use these commands:\n" + "/workout — start a new workout via chat\n" + "/history — see your recent workouts\n" + "/help — list all commands", + reply_markup=keyboard, + ) + + +async def cmd_help(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + await update.message.reply_text( + "*Commands*\n" + "/start — show the Mini App button\n" + "/workout — quick-start a new workout\n" + "/finish — finish the current workout\n" + "/history — recent workout summaries\n" + "/help — this message", + parse_mode="Markdown", + ) + + +async def cmd_workout(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + """Start a new workout from chat.""" + user = ensure_user(update) + active = db.get_active_workout(user["telegram_id"]) + if active: + await update.message.reply_text( + "You already have an active workout! " + "Use /finish to end it, or open the Mini App to keep logging." + ) + return + + workout = db.start_workout(user["telegram_id"]) + webapp_btn = InlineKeyboardButton( + text="Log Sets", + web_app=WebAppInfo(url=WEBAPP_URL), + ) + keyboard = InlineKeyboardMarkup([[webapp_btn]]) + await update.message.reply_text( + f"Workout #{workout['id']} started! Open the app to log your sets.", + reply_markup=keyboard, + ) + + +async def cmd_finish(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + """Finish the active workout and send a summary.""" + user = ensure_user(update) + active = db.get_active_workout(user["telegram_id"]) + if not active: + await update.message.reply_text("No active workout to finish.") + return + + db.finish_workout(active["id"], user["telegram_id"]) + summary = db.get_workout_summary(active["id"]) + text = format_summary(summary) + await update.message.reply_text(text, parse_mode="Markdown") + + +async def cmd_history(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + """Show recent workout summaries.""" + user = ensure_user(update) + workouts = db.get_recent_workouts(user["telegram_id"], limit=5) + if not workouts: + await update.message.reply_text("No workouts yet! Tap /workout to start one.") + return + + for w in workouts: + summary = db.get_workout_summary(w["id"]) + text = format_summary(summary) + await update.message.reply_text(text, parse_mode="Markdown") + + +# ── Web App data handler ──────────────────────────────────────── + +async def handle_web_app_data(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + """Handle data sent from the Mini App via Telegram.WebApp.sendData().""" + data = update.effective_message.web_app_data.data + logger.info("Received web app data: %s", data) + await update.message.reply_text("Got it! Your workout has been saved.") + + +# ── Post-init: set the menu button ────────────────────────────── + +async def post_init(app: Application): + """Set the bot's menu button to open the Mini App.""" + 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) + + +# ── Main ───────────────────────────────────────────────────────── + +def main(): + db.init_db() + + app = ( + Application.builder() + .token(BOT_TOKEN) + .post_init(post_init) + .build() + ) + + app.add_handler(CommandHandler("start", cmd_start)) + app.add_handler(CommandHandler("help", cmd_help)) + app.add_handler(CommandHandler("workout", cmd_workout)) + app.add_handler(CommandHandler("finish", cmd_finish)) + app.add_handler(CommandHandler("history", cmd_history)) + app.add_handler( + MessageHandler(filters.StatusUpdate.WEB_APP_DATA, handle_web_app_data) + ) + + logger.info("Bot starting...") + app.run_polling(drop_pending_updates=True) + + +if __name__ == "__main__": + main() diff --git a/telegram-fitness-bot/config.py b/telegram-fitness-bot/config.py new file mode 100644 index 0000000..5b39981 --- /dev/null +++ b/telegram-fitness-bot/config.py @@ -0,0 +1,14 @@ +import os + +# Telegram Bot Token — loaded from .env automatically by start.py +BOT_TOKEN = os.environ.get("BOT_TOKEN", "") + +# Public HTTPS URL — set automatically by start.py via localtunnel +WEBAPP_URL = os.environ.get("WEBAPP_URL", "http://localhost:8080") + +# API server settings +API_HOST = os.environ.get("API_HOST", "0.0.0.0") +API_PORT = int(os.environ.get("API_PORT", "8080")) + +# Database path — defaults to fitness.db in the working directory +DB_PATH = os.environ.get("DB_PATH", "fitness.db") diff --git a/telegram-fitness-bot/database.py b/telegram-fitness-bot/database.py new file mode 100644 index 0000000..f2093b8 --- /dev/null +++ b/telegram-fitness-bot/database.py @@ -0,0 +1,245 @@ +import sqlite3 +import json +from datetime import datetime, timezone +from contextlib import contextmanager + +from config import DB_PATH + + +@contextmanager +def get_db(): + conn = sqlite3.connect(DB_PATH) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA foreign_keys=ON") + try: + yield conn + conn.commit() + except Exception: + conn.rollback() + raise + finally: + conn.close() + + +def init_db(): + with get_db() as db: + db.executescript(""" + CREATE TABLE IF NOT EXISTS users ( + id INTEGER PRIMARY KEY, + telegram_id INTEGER UNIQUE NOT NULL, + first_name TEXT NOT NULL DEFAULT '', + username TEXT DEFAULT '', + created_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + + CREATE TABLE IF NOT EXISTS exercises ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id INTEGER NOT NULL, + name TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + FOREIGN KEY (user_id) REFERENCES users(telegram_id), + UNIQUE(user_id, name) + ); + + CREATE TABLE IF NOT EXISTS workouts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id INTEGER NOT NULL, + started_at TEXT NOT NULL DEFAULT (datetime('now')), + finished_at TEXT, + notes TEXT DEFAULT '', + FOREIGN KEY (user_id) REFERENCES users(telegram_id) + ); + + CREATE TABLE IF NOT EXISTS sets ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + workout_id INTEGER NOT NULL, + exercise_id INTEGER NOT NULL, + set_order INTEGER NOT NULL DEFAULT 0, + reps INTEGER NOT NULL, + weight REAL NOT NULL, + logged_at TEXT NOT NULL DEFAULT (datetime('now')), + FOREIGN KEY (workout_id) REFERENCES workouts(id), + FOREIGN KEY (exercise_id) REFERENCES exercises(id) + ); + """) + + +# ── User operations ────────────────────────────────────────────── + +def upsert_user(telegram_id: int, first_name: str, username: str = "") -> dict: + with get_db() as db: + db.execute( + """INSERT INTO users (telegram_id, first_name, username) + VALUES (?, ?, ?) + ON CONFLICT(telegram_id) DO UPDATE SET + first_name = excluded.first_name, + username = excluded.username""", + (telegram_id, first_name, username), + ) + row = db.execute( + "SELECT * FROM users WHERE telegram_id = ?", (telegram_id,) + ).fetchone() + return dict(row) + + +# ── Exercise operations ────────────────────────────────────────── + +def add_exercise(user_id: int, name: str) -> dict: + with get_db() as db: + db.execute( + "INSERT INTO exercises (user_id, name) VALUES (?, ?)", + (user_id, name.strip()), + ) + row = db.execute( + "SELECT * FROM exercises WHERE user_id = ? AND name = ?", + (user_id, name.strip()), + ).fetchone() + return dict(row) + + +def get_exercises(user_id: int) -> list[dict]: + with get_db() as db: + rows = db.execute( + "SELECT * FROM exercises WHERE user_id = ? ORDER BY name", + (user_id,), + ).fetchall() + return [dict(r) for r in rows] + + +def delete_exercise(user_id: int, exercise_id: int) -> bool: + with get_db() as db: + cur = db.execute( + "DELETE FROM exercises WHERE id = ? AND user_id = ?", + (exercise_id, user_id), + ) + return cur.rowcount > 0 + + +# ── Workout operations ─────────────────────────────────────────── + +def start_workout(user_id: int) -> dict: + with get_db() as db: + cur = db.execute( + "INSERT INTO workouts (user_id) VALUES (?)", (user_id,) + ) + row = db.execute( + "SELECT * FROM workouts WHERE id = ?", (cur.lastrowid,) + ).fetchone() + return dict(row) + + +def finish_workout(workout_id: int, user_id: int) -> dict | None: + with get_db() as db: + db.execute( + """UPDATE workouts SET finished_at = datetime('now') + WHERE id = ? AND user_id = ?""", + (workout_id, user_id), + ) + row = db.execute( + "SELECT * FROM workouts WHERE id = ?", (workout_id,) + ).fetchone() + return dict(row) if row else None + + +def get_active_workout(user_id: int) -> dict | None: + with get_db() as db: + row = db.execute( + """SELECT * FROM workouts + WHERE user_id = ? AND finished_at IS NULL + ORDER BY started_at DESC LIMIT 1""", + (user_id,), + ).fetchone() + return dict(row) if row else None + + +def get_recent_workouts(user_id: int, limit: int = 10) -> list[dict]: + with get_db() as db: + rows = db.execute( + """SELECT * FROM workouts + WHERE user_id = ? + ORDER BY started_at DESC LIMIT ?""", + (user_id, limit), + ).fetchall() + return [dict(r) for r in rows] + + +# ── Set operations ─────────────────────────────────────────────── + +def add_set(workout_id: int, exercise_id: int, reps: int, weight: float) -> dict: + with get_db() as db: + # figure out next set_order for this exercise in this workout + row = db.execute( + """SELECT COALESCE(MAX(set_order), 0) + 1 AS next_order + FROM sets WHERE workout_id = ? AND exercise_id = ?""", + (workout_id, exercise_id), + ).fetchone() + next_order = row["next_order"] + + cur = db.execute( + """INSERT INTO sets (workout_id, exercise_id, set_order, reps, weight) + VALUES (?, ?, ?, ?, ?)""", + (workout_id, exercise_id, next_order, reps, weight), + ) + new_row = db.execute( + "SELECT * FROM sets WHERE id = ?", (cur.lastrowid,) + ).fetchone() + return dict(new_row) + + +def get_workout_sets(workout_id: int) -> list[dict]: + with get_db() as db: + rows = db.execute( + """SELECT s.*, e.name AS exercise_name + FROM sets s + JOIN exercises e ON e.id = s.exercise_id + WHERE s.workout_id = ? + ORDER BY s.exercise_id, s.set_order""", + (workout_id,), + ).fetchall() + return [dict(r) for r in rows] + + +def delete_set(set_id: int) -> bool: + with get_db() as db: + cur = db.execute("DELETE FROM sets WHERE id = ?", (set_id,)) + return cur.rowcount > 0 + + +# ── Summary helpers ────────────────────────────────────────────── + +def get_workout_summary(workout_id: int) -> dict: + """Return a human-friendly summary of a finished workout.""" + with get_db() as db: + workout = db.execute( + "SELECT * FROM workouts WHERE id = ?", (workout_id,) + ).fetchone() + if not workout: + return {} + + sets = db.execute( + """SELECT e.name, s.reps, s.weight, s.set_order + FROM sets s JOIN exercises e ON e.id = s.exercise_id + WHERE s.workout_id = ? + ORDER BY s.exercise_id, s.set_order""", + (workout_id,), + ).fetchall() + + exercises = {} + for s in sets: + name = s["name"] + if name not in exercises: + exercises[name] = [] + exercises[name].append({"reps": s["reps"], "weight": s["weight"]}) + + total_sets = len(sets) + total_volume = sum(s["reps"] * s["weight"] for s in sets) + + return { + "workout_id": workout_id, + "started_at": workout["started_at"], + "finished_at": workout["finished_at"], + "exercises": exercises, + "total_sets": total_sets, + "total_volume": round(total_volume, 1), + } diff --git a/telegram-fitness-bot/flake.nix b/telegram-fitness-bot/flake.nix new file mode 100644 index 0000000..06d4885 --- /dev/null +++ b/telegram-fitness-bot/flake.nix @@ -0,0 +1,87 @@ +{ + description = "Telegram Fitness Bot — Mini App for tracking gym workouts"; + + inputs = { + nixpkgs.url = "github:NixOS/nixpkgs/nixos-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 + aiohttp + ]); + + # localtunnel via npm + localtunnel = pkgs.buildNpmPackage { + pname = "localtunnel"; + version = "2.0.2"; + src = pkgs.fetchFromGitHub { + owner = "localtunnel"; + repo = "localtunnel"; + rev = "v2.0.2"; + hash = "sha256-deKDwCjGT+0YjeW/AM2J6IH+hEoQrESmKKM23n0JLWY="; + }; + npmDepsHash = "sha256-R9FYkEe93oGF+dR7i1MxwzEW3EM3SasH/B6LLC2CNXM="; + dontNpmBuild = true; + }; + + runtimePath = pkgs.lib.makeBinPath [ + pythonEnv + localtunnel + ]; + + in { + packages.default = pkgs.stdenv.mkDerivation { + pname = "telegram-fitness-bot"; + version = "0.1.0"; + + src = pkgs.lib.cleanSource ./.; + + nativeBuildInputs = [ pkgs.makeWrapper ]; + + installPhase = '' + mkdir -p $out/lib/telegram-fitness-bot + cp -r start.py bot.py server.py database.py config.py webapp $out/lib/telegram-fitness-bot/ + + mkdir -p $out/bin + + # Main entry point: `nix run` → start.py + makeWrapper ${pythonEnv}/bin/python $out/bin/telegram-fitness-bot \ + --prefix PATH : ${runtimePath} \ + --add-flags "$out/lib/telegram-fitness-bot/start.py" + ''; + + meta = with pkgs.lib; { + description = "Telegram Mini App for tracking gym workouts"; + license = licenses.mit; + mainProgram = "telegram-fitness-bot"; + }; + }; + + # `nix develop` — interactive dev shell + devShells.default = pkgs.mkShell { + packages = [ + pythonEnv + localtunnel + ]; + + shellHook = '' + echo "" + echo " Fitness Bot dev shell" + echo " python: $(python --version)" + echo " lt: $(lt --version)" + echo "" + echo " Run: python start.py" + echo "" + ''; + }; + } + ); +} diff --git a/telegram-fitness-bot/requirements.txt b/telegram-fitness-bot/requirements.txt new file mode 100644 index 0000000..906d091 --- /dev/null +++ b/telegram-fitness-bot/requirements.txt @@ -0,0 +1,2 @@ +python-telegram-bot>=21.0 +aiohttp>=3.9 diff --git a/telegram-fitness-bot/run.sh b/telegram-fitness-bot/run.sh new file mode 100755 index 0000000..eefed9d --- /dev/null +++ b/telegram-fitness-bot/run.sh @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +# Convenience wrapper for running without nix. +# With nix, just use: nix run +set -euo pipefail +cd "$(dirname "${BASH_SOURCE[0]}")" +exec python start.py "$@" diff --git a/telegram-fitness-bot/server.py b/telegram-fitness-bot/server.py new file mode 100644 index 0000000..8cf1abc --- /dev/null +++ b/telegram-fitness-bot/server.py @@ -0,0 +1,247 @@ +""" +API + static file server for the Telegram Mini App. +Run alongside bot.py — serves the webapp/ folder and REST endpoints. +""" +import hashlib +import hmac +import json +import logging +from urllib.parse import parse_qs + +from aiohttp import web + +import database as db +from config import BOT_TOKEN, API_HOST, API_PORT + +logging.basicConfig( + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + level=logging.INFO, +) +logger = logging.getLogger(__name__) + + +# ── Telegram initData validation ───────────────────────────────── + +def validate_init_data(init_data: str) -> dict | None: + """ + Validate the Telegram WebApp initData string. + Returns the parsed user dict if valid, None otherwise. + See: https://core.telegram.org/bots/webapps#validating-data-received-via-the-mini-app + """ + if not init_data: + return None + + parsed = parse_qs(init_data, keep_blank_values=True) + received_hash = parsed.get("hash", [None])[0] + if not received_hash: + return None + + # Build the data-check-string: sorted key=value pairs, excluding "hash" + data_pairs = [] + for key, values in parsed.items(): + if key == "hash": + continue + data_pairs.append(f"{key}={values[0]}") + data_pairs.sort() + data_check_string = "\n".join(data_pairs) + + # HMAC-SHA256 with secret = HMAC-SHA256("WebAppData", bot_token) + secret_key = hmac.new( + b"WebAppData", BOT_TOKEN.encode(), hashlib.sha256 + ).digest() + computed_hash = hmac.new( + secret_key, data_check_string.encode(), hashlib.sha256 + ).hexdigest() + + if not hmac.compare_digest(computed_hash, received_hash): + logger.warning("Invalid initData hash") + return None + + # Parse the user JSON + user_json = parsed.get("user", [None])[0] + if not user_json: + return None + + try: + user = json.loads(user_json) + return user + except json.JSONDecodeError: + return None + + +# ── Auth middleware ─────────────────────────────────────────────── + +def get_user_id(request: web.Request) -> int | None: + """Extract and validate the user from the request headers.""" + init_data = request.headers.get("X-Telegram-Init-Data", "") + + # In production, always validate. For local dev, allow a fallback. + user = validate_init_data(init_data) + if user: + # Upsert user record + db.upsert_user( + telegram_id=user["id"], + first_name=user.get("first_name", ""), + username=user.get("username", ""), + ) + return user["id"] + + # DEV FALLBACK: if token is placeholder, allow X-Dev-User-Id header + if BOT_TOKEN == "YOUR_BOT_TOKEN_HERE": + dev_id = request.headers.get("X-Dev-User-Id") + if dev_id: + return int(dev_id) + + return None + + +def require_auth(handler): + """Decorator that rejects unauthenticated requests.""" + async def wrapper(request: web.Request): + user_id = get_user_id(request) + if not user_id: + return web.json_response({"error": "Unauthorized"}, status=401) + request["user_id"] = user_id + return await handler(request) + return wrapper + + +# ── API Routes ─────────────────────────────────────────────────── + +# Exercises + +@require_auth +async def get_exercises(request: web.Request): + exercises = db.get_exercises(request["user_id"]) + return web.json_response({"exercises": exercises}) + + +@require_auth +async def create_exercise(request: web.Request): + body = await request.json() + name = body.get("name", "").strip() + if not name: + return web.json_response({"error": "Name is required"}, status=400) + try: + exercise = db.add_exercise(request["user_id"], name) + return web.json_response({"exercise": exercise}, status=201) + except Exception as e: + return web.json_response({"error": str(e)}, status=400) + + +@require_auth +async def delete_exercise(request: web.Request): + exercise_id = int(request.match_info["id"]) + ok = db.delete_exercise(request["user_id"], exercise_id) + if not ok: + return web.json_response({"error": "Not found"}, status=404) + return web.json_response({"ok": True}) + + +# Workouts + +@require_auth +async def get_workouts(request: web.Request): + workouts = db.get_recent_workouts(request["user_id"]) + # Attach summaries + result = [] + for w in workouts: + w["summary"] = db.get_workout_summary(w["id"]) + result.append(w) + return web.json_response({"workouts": result}) + + +@require_auth +async def get_active_workout(request: web.Request): + workout = db.get_active_workout(request["user_id"]) + return web.json_response({"workout": workout}) + + +@require_auth +async def create_workout(request: web.Request): + # Check if there's already an active one + active = db.get_active_workout(request["user_id"]) + if active: + return web.json_response({"workout": active}) + workout = db.start_workout(request["user_id"]) + return web.json_response({"workout": workout}, status=201) + + +@require_auth +async def finish_workout(request: web.Request): + workout_id = int(request.match_info["id"]) + workout = db.finish_workout(workout_id, request["user_id"]) + if not workout: + return web.json_response({"error": "Not found"}, status=404) + return web.json_response({"workout": workout}) + + +# Sets + +@require_auth +async def get_workout_sets(request: web.Request): + workout_id = int(request.match_info["id"]) + sets = db.get_workout_sets(workout_id) + return web.json_response({"sets": sets}) + + +@require_auth +async def create_set(request: web.Request): + workout_id = int(request.match_info["id"]) + body = await request.json() + + exercise_id = body.get("exercise_id") + reps = body.get("reps") + weight = body.get("weight", 0) + + if not exercise_id or not reps: + return web.json_response( + {"error": "exercise_id and reps are required"}, status=400 + ) + + new_set = db.add_set(workout_id, exercise_id, int(reps), float(weight)) + return web.json_response({"set": new_set}, status=201) + + +@require_auth +async def delete_set(request: web.Request): + set_id = int(request.match_info["id"]) + ok = db.delete_set(set_id) + if not ok: + return web.json_response({"error": "Not found"}, status=404) + return web.json_response({"ok": True}) + + +# ── App setup ──────────────────────────────────────────────────── + +def create_app() -> web.Application: + db.init_db() + + app = web.Application() + + # API routes + app.router.add_get("/api/exercises", get_exercises) + app.router.add_post("/api/exercises", create_exercise) + app.router.add_delete("/api/exercises/{id}", delete_exercise) + + app.router.add_get("/api/workouts", get_workouts) + app.router.add_get("/api/workouts/active", get_active_workout) + app.router.add_post("/api/workouts", create_workout) + app.router.add_post("/api/workouts/{id}/finish", finish_workout) + + app.router.add_get("/api/workouts/{id}/sets", get_workout_sets) + app.router.add_post("/api/workouts/{id}/sets", create_set) + app.router.add_delete("/api/sets/{id}", delete_set) + + # Serve the webapp/ folder for the Mini App + import pathlib + webapp_dir = pathlib.Path(__file__).parent / "webapp" + app.router.add_static("/", webapp_dir, show_index=True) + + return app + + +if __name__ == "__main__": + app = create_app() + logger.info("Server starting on %s:%s", API_HOST, API_PORT) + web.run_app(app, host=API_HOST, port=API_PORT) diff --git a/telegram-fitness-bot/start.py b/telegram-fitness-bot/start.py new file mode 100644 index 0000000..1f752c6 --- /dev/null +++ b/telegram-fitness-bot/start.py @@ -0,0 +1,188 @@ +#!/usr/bin/env python3 +""" +Orchestrator — the single entry point for `nix run`. + 1. Loads BOT_TOKEN from .env in the current directory + 2. Starts the API server + 3. Starts a localtunnel to get a public HTTPS URL + 4. Starts the Telegram bot with that URL + 5. Cleans up everything on Ctrl+C +""" +import os +import re +import signal +import subprocess +import sys +import time +import pathlib + +SCRIPT_DIR = pathlib.Path(__file__).resolve().parent + + +def load_dotenv(): + """Load .env from the working directory (where the user ran `nix run`).""" + env_file = pathlib.Path.cwd() / ".env" + if not env_file.exists(): + # Also check next to the script (for non-nix usage) + env_file = SCRIPT_DIR / ".env" + if not env_file.exists(): + return + + print(f"Loading secrets from {env_file}") + with open(env_file) as f: + for line in f: + line = line.strip() + if not line or line.startswith("#"): + continue + if "=" in line: + key, _, value = line.partition("=") + key = key.strip() + value = value.strip().strip("\"'") + os.environ.setdefault(key, value) + + +def check_token(): + token = os.environ.get("BOT_TOKEN", "") + if not token or token == "YOUR_BOT_TOKEN_HERE": + print("\n No BOT_TOKEN found!\n") + print(" Create a .env file in this directory with:") + print(" BOT_TOKEN=your-token-from-@BotFather\n") + sys.exit(1) + # Mask token in logs + masked = token[:5] + "..." + token[-4:] + print(f" BOT_TOKEN: {masked}") + return token + + +def start_server(port: int) -> subprocess.Popen: + """Start the aiohttp API server.""" + env = {**os.environ, "API_PORT": str(port)} + return subprocess.Popen( + [sys.executable, str(SCRIPT_DIR / "server.py")], + env=env, + cwd=pathlib.Path.cwd(), # DB writes go to user's working directory + ) + + +def start_tunnel(port: int) -> tuple[subprocess.Popen, str]: + """Start localtunnel and return (process, public_url).""" + print(f" Starting tunnel to port {port}...") + + proc = subprocess.Popen( + ["lt", "--port", str(port)], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + ) + + # localtunnel prints "your url is: https://xxx.loca.lt" + url = None + deadline = time.time() + 30 + while time.time() < deadline: + line = proc.stdout.readline() + if not line: + if proc.poll() is not None: + print(" Tunnel process exited unexpectedly.") + break + continue + line = line.strip() + print(f" [tunnel] {line}") + match = re.search(r"https?://\S+", line) + if match: + url = match.group(0) + break + + if not url: + proc.kill() + print("\n Could not get a tunnel URL.") + print(" Make sure localtunnel is working: lt --port 8080\n") + sys.exit(1) + + return proc, url + + +def start_bot(webapp_url: str) -> subprocess.Popen: + """Start the Telegram bot with the tunnel URL.""" + env = {**os.environ, "WEBAPP_URL": webapp_url} + return subprocess.Popen( + [sys.executable, str(SCRIPT_DIR / "bot.py")], + env=env, + cwd=pathlib.Path.cwd(), + ) + + +def main(): + port = int(os.environ.get("API_PORT", "8080")) + procs: list[subprocess.Popen] = [] + + def cleanup(sig=None, frame=None): + print("\nShutting down...") + for p in procs: + try: + p.terminate() + except OSError: + pass + for p in procs: + try: + p.wait(timeout=5) + except subprocess.TimeoutExpired: + p.kill() + sys.exit(0) + + signal.signal(signal.SIGINT, cleanup) + signal.signal(signal.SIGTERM, cleanup) + + print() + print(" ==========================================") + print(" Telegram Fitness Bot") + print(" ==========================================") + print() + + # 1. Load .env + load_dotenv() + check_token() + + # 2. Database will be created in the working directory + db_path = os.environ.setdefault("DB_PATH", str(pathlib.Path.cwd() / "fitness.db")) + print(f" Database: {db_path}") + + # 3. Start API server + print(f"\n Starting API server on port {port}...") + server = start_server(port) + procs.append(server) + time.sleep(1) # Give it a moment to bind + + if server.poll() is not None: + print(" Server failed to start!") + sys.exit(1) + + # 4. Start tunnel + tunnel, webapp_url = start_tunnel(port) + procs.append(tunnel) + + # 5. Start bot + print(f"\n WEBAPP_URL: {webapp_url}") + print(f" Starting bot...\n") + bot = start_bot(webapp_url) + procs.append(bot) + + print(" ==========================================") + print(f" All systems go!") + print(f" Mini App: {webapp_url}") + print(f" API: http://localhost:{port}") + print(f" Press Ctrl+C to stop") + print(" ==========================================") + print() + + # Wait for any process to exit + while True: + for p in procs: + ret = p.poll() + if ret is not None: + name = {id(server): "Server", id(tunnel): "Tunnel", id(bot): "Bot"}.get(id(p), "?") + print(f"\n {name} exited with code {ret}") + cleanup() + time.sleep(1) + + +if __name__ == "__main__": + main() diff --git a/telegram-fitness-bot/webapp/app.js b/telegram-fitness-bot/webapp/app.js new file mode 100644 index 0000000..3f8dfaa --- /dev/null +++ b/telegram-fitness-bot/webapp/app.js @@ -0,0 +1,361 @@ +// ── Telegram Web App init ─────────────────────────────────────── +const tg = window.Telegram.WebApp; +tg.ready(); +tg.expand(); + +const API = window.location.origin + "/api"; +const userId = tg.initDataUnsafe?.user?.id; + +if (!userId) { + document.getElementById("app").innerHTML = + '
Please open this app from Telegram.
No active workout
+ +No workouts yet
+