bigbiggerbiggestbot/telegram-fitness-bot/server.py

"""
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)