diff --git a/bot.py b/bot.py index 2e3d99b..88e5893 100644 --- a/bot.py +++ b/bot.py @@ -5,9 +5,10 @@ import os from datetime import datetime, timezone from dotenv import load_dotenv -from telegram import Update +from telegram import Update, WebAppInfo, InlineKeyboardButton, InlineKeyboardMarkup, MenuButtonWebApp from telegram.constants import ParseMode from telegram.ext import ( + Application, ApplicationBuilder, CommandHandler, ContextTypes, @@ -47,6 +48,9 @@ def _load_token() -> str: BOT_TOKEN = _load_token() +# Mini App URL — set automatically by start.py via localtunnel +WEBAPP_URL = os.environ.get("WEBAPP_URL", "") + # ── Helpers ────────────────────────────────────────────────────────────────── @@ -74,7 +78,7 @@ def extract_timestamp(update: Update) -> tuple[datetime, bool]: async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE): - await update.message.reply_text( + text = ( "💪 Fitness Tracker Bot\n\n" "Send me your workout and I'll save it!\n\n" "Format:\n" @@ -87,10 +91,21 @@ async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE): "I'll use the original timestamp.\n\n" "Commands:\n" "/history — view recent workouts\n" - "/stats — quick summary", - parse_mode=ParseMode.HTML, + "/stats — quick summary" ) + 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 @@ -191,10 +206,27 @@ async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE): # ── 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() - app = ApplicationBuilder().token(BOT_TOKEN).build() + 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)) diff --git a/flake.nix b/flake.nix index d20eb4d..78aaba1 100644 --- a/flake.nix +++ b/flake.nix @@ -1,5 +1,5 @@ { - description = "BigBiggerBiggestBot — Telegram fitness tracker"; + description = "BigBiggerBiggestBot — Telegram fitness tracker with Mini App"; inputs = { nixpkgs.url = "github:NixOS/nixpkgs/nixpkgs-unstable"; @@ -16,23 +16,38 @@ pythonEnv = python.withPackages (ps: with ps; [ python-telegram-bot python-dotenv + aiohttp ]); + + localtunnel = pkgs.buildNpmPackage { + pname = "localtunnel"; + version = "2.0.2"; + src = pkgs.fetchFromGitHub { + owner = "localtunnel"; + repo = "localtunnel"; + rev = "v2.0.2"; + hash = "sha256-6gEK1VjF25Kbe2drxbxUKDNJGqZ+OXgkulPkAkMR2+k="; + }; + npmDepsHash = "sha256-R9FYkEe93oGF+dR7i1MxwzEW3EM3SasH/B6LLC2CNXM="; + dontNpmBuild = true; + }; in { - # `nix develop` — drop into a shell with everything available devShells.default = pkgs.mkShell { - packages = [ pythonEnv ]; + packages = [ pythonEnv localtunnel ]; shellHook = '' echo "💪 BigBiggerBiggestBot dev shell" - echo " Run: python bot.py" + echo " Run: python start.py (server + tunnel + bot)" + echo " Run: python bot.py (bot only, no mini app)" ''; }; - # `nix run` — start the bot from the current directory + # `nix run` — start everything via start.py apps.default = { type = "app"; - program = toString (pkgs.writeShellScript "run-bot" '' - exec ${pythonEnv}/bin/python "$PWD/bot.py" + program = toString (pkgs.writeShellScript "run-fitness-bot" '' + export PATH="${pkgs.lib.makeBinPath [ pythonEnv localtunnel ]}:$PATH" + exec ${pythonEnv}/bin/python "$PWD/start.py" ''); }; } diff --git a/server.py b/server.py new file mode 100644 index 0000000..10623a1 --- /dev/null +++ b/server.py @@ -0,0 +1,215 @@ +""" +API + static file server for the Telegram Mini App. +Serves webapp/ and REST endpoints, using the existing db.py layer. +""" +import hashlib +import hmac +import json +import logging +import os +from urllib.parse import parse_qs + +from aiohttp import web + +from db import init_db, get_db, save_workout, get_workouts, get_workout_count +from parser import parse_workout, format_workout + +logging.basicConfig( + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", + level=logging.INFO, +) +logger = logging.getLogger(__name__) + + +# ── Token (injected by start.py via env) ───────────────────────── + +def _get_bot_token() -> str: + return os.environ.get("BOT_TOKEN", "") + + +# ── Telegram initData validation ───────────────────────────────── + +def validate_init_data(init_data: str) -> dict | None: + """Validate Telegram WebApp initData. Returns user dict if valid.""" + if not init_data: + return None + + bot_token = _get_bot_token() + if not bot_token: + return None + + parsed = parse_qs(init_data, keep_blank_values=True) + received_hash = parsed.get("hash", [None])[0] + if not received_hash: + return None + + 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) + + 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 + + user_json = parsed.get("user", [None])[0] + if not user_json: + return None + + try: + return json.loads(user_json) + except json.JSONDecodeError: + return None + + +# ── Auth middleware ─────────────────────────────────────────────── + +def get_user_id(request: web.Request) -> int | None: + init_data = request.headers.get("X-Telegram-Init-Data", "") + user = validate_init_data(init_data) + if user: + return user["id"] + return None + + +def require_auth(handler): + 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 ─────────────────────────────────────────────────── + +@require_auth +async def api_get_workouts(request: web.Request): + """Return recent workouts with exercises.""" + limit = int(request.query.get("limit", "20")) + offset = int(request.query.get("offset", "0")) + workouts = get_workouts(request["user_id"], limit=limit, offset=offset) + total = get_workout_count(request["user_id"]) + return web.json_response({"workouts": workouts, "total": total}) + + +@require_auth +async def api_save_workout(request: web.Request): + """Save a workout from structured JSON or raw text.""" + body = await request.json() + raw_text = body.get("raw_text", "") + superset_groups = body.get("superset_groups") + + if superset_groups: + # Structured input from the Mini App UI + from datetime import datetime, timezone + workout_id = save_workout( + user_id=request["user_id"], + timestamp=datetime.now(timezone.utc), + superset_groups=superset_groups, + raw_text=raw_text or None, + ) + elif raw_text: + # Text-based input (same format as sending a message to the bot) + groups = parse_workout(raw_text) + if not groups: + return web.json_response({"error": "Could not parse workout text"}, status=400) + from datetime import datetime, timezone + superset_dicts = [[ex.to_dict() for ex in group] for group in groups] + workout_id = save_workout( + user_id=request["user_id"], + timestamp=datetime.now(timezone.utc), + superset_groups=superset_dicts, + raw_text=raw_text, + ) + else: + return web.json_response( + {"error": "Provide superset_groups or raw_text"}, status=400 + ) + + return web.json_response({"workout_id": workout_id}, status=201) + + +@require_auth +async def api_get_exercise_names(request: web.Request): + """Return unique exercise names this user has logged (for autocomplete).""" + with get_db() as conn: + rows = conn.execute( + """SELECT DISTINCT e.name + FROM exercises e + JOIN superset_groups sg ON sg.id = e.superset_group_id + JOIN workouts w ON w.id = sg.workout_id + WHERE w.user_id = ? + ORDER BY e.name""", + (request["user_id"],), + ).fetchall() + return web.json_response({"exercises": [r["name"] for r in rows]}) + + +@require_auth +async def api_get_stats(request: web.Request): + """Return summary stats for the user.""" + user_id = request["user_id"] + total = get_workout_count(user_id) + if total == 0: + return web.json_response({ + "total_workouts": 0, "unique_exercises": 0, + "total_sets": 0, "total_volume": 0, + }) + + workouts = get_workouts(user_id, limit=10000) + 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"] + + return web.json_response({ + "total_workouts": total, + "unique_exercises": len(exercise_names), + "total_sets": total_sets, + "total_volume": round(total_volume, 1), + }) + + +# ── App setup ──────────────────────────────────────────────────── + +def create_app() -> web.Application: + init_db() + + app = web.Application() + + app.router.add_get("/api/workouts", api_get_workouts) + app.router.add_post("/api/workouts", api_save_workout) + app.router.add_get("/api/exercises", api_get_exercise_names) + app.router.add_get("/api/stats", api_get_stats) + + # Serve the webapp/ folder + import pathlib + webapp_dir = pathlib.Path(__file__).parent / "webapp" + app.router.add_static("/", webapp_dir, show_index=True) + + return app + + +if __name__ == "__main__": + port = int(os.environ.get("API_PORT", "8080")) + host = os.environ.get("API_HOST", "0.0.0.0") + app = create_app() + logger.info("Server starting on %s:%s", host, port) + web.run_app(app, host=host, port=port) diff --git a/start.py b/start.py new file mode 100644 index 0000000..d507858 --- /dev/null +++ b/start.py @@ -0,0 +1,175 @@ +#!/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 +""" +import os +import re +import signal +import subprocess +import sys +import time +import pathlib + +SCRIPT_DIR = pathlib.Path(__file__).resolve().parent +SECRETS_FILE = pathlib.Path.home() / ".secrets" / "bigbiggerbiggestbot" + + +def load_token() -> str: + """Load bot token: secrets file → .env → BOT_TOKEN env var.""" + # 1. Secrets file (same path as bot.py uses) + if SECRETS_FILE.is_file(): + token = SECRETS_FILE.read_text().strip() + if token: + print(f" Token loaded from {SECRETS_FILE}") + return token + + # 2. .env in working directory + env_file = pathlib.Path.cwd() / ".env" + if env_file.exists(): + for line in env_file.read_text().splitlines(): + line = line.strip() + if line.startswith("BOT_TOKEN="): + token = line.split("=", 1)[1].strip().strip("\"'") + if token: + print(f" Token loaded from {env_file}") + return token + + # 3. Already in environment + token = os.environ.get("BOT_TOKEN", "").strip() + if token: + print(" Token loaded from BOT_TOKEN env var") + return token + + print("\n No bot token found!") + print(f" Put it in {SECRETS_FILE}") + print(" Or create a .env file with: BOT_TOKEN=your-token\n") + sys.exit(1) + + +def start_server(port: int, bot_token: str) -> subprocess.Popen: + env = {**os.environ, "API_PORT": str(port), "BOT_TOKEN": bot_token} + return subprocess.Popen( + [sys.executable, str(SCRIPT_DIR / "server.py")], + env=env, + ) + + +def start_tunnel(port: int) -> tuple[subprocess.Popen, str]: + print(f" Starting tunnel to port {port}...") + proc = subprocess.Popen( + ["lt", "--port", str(port)], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + ) + + 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(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] = [] + + 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(" BigBiggerBiggest — Fitness Tracker") + print(" ==========================================") + print() + + # 1. Load token + bot_token = load_token() + masked = bot_token[:5] + "..." + bot_token[-4:] + print(f" BOT_TOKEN: {masked}") + + # 2. Start API server + print(f"\n Starting API server on port {port}...") + server = start_server(port, bot_token) + procs.append(server) + time.sleep(1) + + if server.poll() is not None: + print(" Server failed to start!") + sys.exit(1) + + # 3. Start tunnel + 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(f" Mini App: {webapp_url}") + print(f" API: http://localhost:{port}") + print(f" Press Ctrl+C to stop") + print(" ==========================================") + print() + + 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/webapp/app.js b/webapp/app.js new file mode 100644 index 0000000..44479ac --- /dev/null +++ b/webapp/app.js @@ -0,0 +1,242 @@ +// ── 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 workouts yet
No workouts yet
+Loading…
+