"""Telegram bot for uploading .FIT files.""" import logging import uuid from pathlib import Path from aiogram import Bot, Dispatcher, Router, F from aiogram.types import Message, InlineKeyboardMarkup, InlineKeyboardButton, WebAppInfo from aiogram.filters import CommandStart from sqlalchemy import select from backend.app.core.config import settings from backend.app.core.database import async_session from backend.app.models.rider import Rider from backend.app.models.activity import Activity from backend.app.models.fitness import PowerCurve from backend.app.services.fit_parser import parse_fit_file from backend.app.services.metrics import calculate_metrics from backend.app.services.intervals import detect_intervals from backend.app.services.power_curve import calculate_power_curve from backend.app.services.coaching import link_activity_to_plan logger = logging.getLogger(__name__) router = Router() async def get_or_create_rider(telegram_id: int, name: str, username: str | None) -> Rider: """Find rider by telegram_id or create a new one.""" async with async_session() as session: query = select(Rider).where(Rider.telegram_id == telegram_id) result = await session.execute(query) rider = result.scalar_one_or_none() if not rider: rider = Rider( telegram_id=telegram_id, telegram_username=username, name=name, ) session.add(rider) await session.commit() await session.refresh(rider) return rider async def process_fit_upload(content: bytes, rider: Rider) -> Activity: """Parse FIT file and save activity with all related data.""" upload_dir = Path(settings.UPLOAD_DIR) upload_dir.mkdir(parents=True, exist_ok=True) file_id = uuid.uuid4() file_path = upload_dir / f"{file_id}.fit" file_path.write_bytes(content) async with async_session() as session: # Re-attach rider to this session rider = await session.get(Rider, rider.id) activity, data_points, exercise_sets = parse_fit_file(content, rider.id, str(file_path)) if exercise_sets: activity.exercise_sets = exercise_sets # Auto-link to training plan await link_activity_to_plan(activity, rider.id, session) session.add(activity) await session.flush() if data_points: for dp in data_points: dp.activity_id = activity.id session.add_all(data_points) metrics = calculate_metrics(data_points, activity, ftp=rider.ftp) if metrics: session.add(metrics) intervals = detect_intervals(data_points, ftp=rider.ftp) for interval in intervals: interval.activity_id = activity.id session.add_all(intervals) curve_data = calculate_power_curve(data_points) if curve_data: pc = PowerCurve(activity_id=activity.id, curve_data=curve_data) session.add(pc) await session.commit() await session.refresh(activity) return activity def format_duration(seconds: int) -> str: h = seconds // 3600 m = (seconds % 3600) // 60 return f"{h}h {m}m" if h > 0 else f"{m}m" @router.message(CommandStart()) async def cmd_start(message: Message): user = message.from_user await get_or_create_rider( telegram_id=user.id, name=user.full_name, username=user.username, ) await message.answer( f"Welcome to VeloBrain, {user.first_name}!\n\n" "Send me a .FIT file from your bike computer and I'll analyze your ride.\n\n" "I'll calculate:\n" "- Power metrics (NP, TSS, IF)\n" "- Power zones & HR zones\n" "- Power curve\n" "- Interval detection\n\n" "Just drag & drop your .FIT file here!" ) @router.message(F.document) async def handle_document(message: Message, bot: Bot): doc = message.document if not doc.file_name or not doc.file_name.lower().endswith(".fit"): await message.answer("Please send a .FIT file.") return user = message.from_user rider = await get_or_create_rider( telegram_id=user.id, name=user.full_name, username=user.username, ) status_msg = await message.answer("Processing your .FIT file...") try: file = await bot.download(doc) content = file.read() activity = await process_fit_upload(content, rider) m = activity.metrics lines = [ f"*{activity.name or 'Workout'}*", f"Duration: {format_duration(activity.duration)}", ] if activity.distance: lines.append(f"Distance: {activity.distance / 1000:.1f} km") if activity.elevation_gain: lines.append(f"Elevation: {activity.elevation_gain:.0f} m") if m: if m.avg_power: lines.append(f"Avg Power: {m.avg_power:.0f} W") if m.normalized_power: lines.append(f"NP: {m.normalized_power:.0f} W") if m.tss: lines.append(f"TSS: {m.tss:.0f}") if m.intensity_factor: lines.append(f"IF: {m.intensity_factor:.2f}") if m.avg_hr: lines.append(f"Avg HR: {m.avg_hr} bpm") if m.avg_cadence: lines.append(f"Avg Cadence: {m.avg_cadence} rpm") if m.calories: lines.append(f"Calories: {m.calories} kcal") # Exercise sets for strength workouts if activity.exercise_sets: exercises: dict[str, list] = {} for s in activity.exercise_sets: name = s.get("exercise_name", "Unknown") exercises.setdefault(name, []).append(s) lines.append("") for name, sets in exercises.items(): reps_str = " / ".join( f"{s.get('repetitions', '?')}x{s.get('weight', 0):.0f}kg" if s.get('weight') else f"{s.get('repetitions', '?')} reps" for s in sets ) lines.append(f" {name}: {reps_str}") intervals_count = len(activity.intervals or []) if intervals_count > 0: work = [i for i in activity.intervals if i.interval_type == "work"] lines.append(f"Intervals: {len(work)} work / {intervals_count - len(work)} rest") lines.append(f"\nView details in the web app!") keyboard = InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="Открыть в WebApp", web_app=WebAppInfo(url=f"https://sport.luminic.space//activities/{activity.id}"))]]) await status_msg.edit_text("\n".join(lines), parse_mode="Markdown", reply_markup=keyboard) except Exception as e: logger.exception("Error processing FIT file") await status_msg.edit_text(f"Error processing file: {str(e)}") @router.message() async def handle_other(message: Message): await message.answer( "Send me a .FIT file to analyze your ride!\n" "Use /start to see what I can do." ) def create_bot() -> tuple[Bot, Dispatcher]: bot = Bot(token=settings.TELEGRAM_BOT_TOKEN) dp = Dispatcher() dp.include_router(router) return bot, dp