from __future__ import annotations from datetime import date, datetime, timedelta from typing import Optional from fastapi import HTTPException from sqlalchemy.orm import Session, joinedload from app.config import settings from app.models import AccessoryState, Profile, Workout, WorkoutExercise, WorkoutSet from app.schemas import ( ExerciseResponse, ProfileResponse, SetResponse, WorkoutDetail, WorkoutSummary, ) from app.services.schedule import apply_accessory_weight_to_future, generate_schedule from app.services.weights import round_exercise_weight, round_to_plate, weight_step_for def profile_response(profile: Profile, now: Optional[datetime] = None) -> ProfileResponse: now = now or datetime.utcnow() rest_countdown = None can_probe = False if profile.phase == "rest" and profile.probe_available_at: delta = (profile.probe_available_at - now).total_seconds() if delta > 0: rest_countdown = int(delta) else: can_probe = True elif profile.phase == "probe": can_probe = True return ProfileResponse( one_rm_kg=profile.one_rm_kg, training_weekdays=profile.training_weekdays, start_date=profile.start_date, onboarded=profile.onboarded, phase=profile.phase if not (profile.phase == "rest" and can_probe) else "probe", rest_until=profile.rest_until, probe_available_at=profile.probe_available_at, cycle_number=profile.cycle_number, rest_countdown_seconds=rest_countdown, can_probe=can_probe, ) def set_response(s: WorkoutSet) -> SetResponse: return SetResponse( id=s.id, set_index=s.set_index, target_percent=s.target_percent, target_reps=s.target_reps, target_weight_kg=s.target_weight_kg, is_failure=s.is_failure, completed=s.completed, actual_reps=s.actual_reps, actual_weight_kg=s.actual_weight_kg, ) def exercise_response(ex: WorkoutExercise) -> ExerciseResponse: return ExerciseResponse( id=ex.id, order_index=ex.order_index, name=ex.name, kind=ex.kind, prescribed_weight_kg=ex.prescribed_weight_kg, weight_locked=ex.weight_locked, needs_weight_input=ex.needs_weight_input, pending_prompt=ex.pending_prompt, weight_step_kg=weight_step_for(ex.name, kind=ex.kind), sets=[set_response(s) for s in sorted(ex.sets, key=lambda x: x.set_index)], ) def workout_summary(w: Workout) -> WorkoutSummary: return WorkoutSummary( id=w.id, cycle_number=w.cycle_number, week_number=w.week_number, day_index=w.day_index, scheduled_date=w.scheduled_date, status=w.status, exercise_count=len(w.exercises) if w.exercises is not None else 0, ) def maybe_expire_pause(db: Session, workout: Workout, now: Optional[datetime] = None) -> Workout: now = now or datetime.utcnow() if workout.status == "paused" and workout.paused_at: limit = workout.paused_at + timedelta(minutes=settings.pause_limit_minutes) if now > limit: workout.status = "abandoned" workout.paused_at = None workout.rest_started_at = None db.commit() db.refresh(workout) return workout def workout_detail(db: Session, workout: Workout, today: Optional[date] = None) -> WorkoutDetail: today = today or date.today() now = datetime.utcnow() workout = maybe_expire_pause(db, workout, now) active = ( db.query(Workout) .filter( Workout.cycle_number == workout.cycle_number, Workout.id != workout.id, Workout.status.in_(["in_progress", "paused"]), ) .first() ) can_start = workout.status == "planned" and workout.scheduled_date == today and active is None can_resume = False pause_expires_at = None if workout.status == "paused" and workout.paused_at: pause_expires_at = workout.paused_at + timedelta(minutes=settings.pause_limit_minutes) can_resume = now <= pause_expires_at and workout.scheduled_date == today rest_elapsed = 0 if workout.status == "in_progress" and workout.rest_started_at: rest_elapsed = max(0, int((now - workout.rest_started_at).total_seconds())) return WorkoutDetail( id=workout.id, cycle_number=workout.cycle_number, week_number=workout.week_number, day_index=workout.day_index, scheduled_date=workout.scheduled_date, status=workout.status, started_at=workout.started_at, paused_at=workout.paused_at, completed_at=workout.completed_at, current_exercise_index=workout.current_exercise_index, current_set_index=workout.current_set_index, rest_started_at=workout.rest_started_at, rest_elapsed_seconds=rest_elapsed, can_start=can_start, can_resume=can_resume, pause_expires_at=pause_expires_at, exercises=[exercise_response(e) for e in sorted(workout.exercises, key=lambda x: x.order_index)], ) def load_workout(db: Session, workout_id: int) -> Workout: def _fetch() -> Workout | None: return ( db.query(Workout) .options(joinedload(Workout.exercises).joinedload(WorkoutExercise.sets)) .filter(Workout.id == workout_id) .first() ) workout = _fetch() if not workout: raise HTTPException(404, "Тренировка не найдена") if sync_accessory_weights(db, workout): db.commit() workout = _fetch() assert workout is not None return workout def sync_accessory_weights(db: Session, workout: Workout) -> bool: """Подтянуть сохранённый вес каждого вспомогательного из AccessoryState.""" changed = False for ex in workout.exercises: if ex.kind != "accessory": continue if any(s.completed for s in ex.sets): continue state = ( db.query(AccessoryState) .filter( AccessoryState.cycle_number == workout.cycle_number, AccessoryState.exercise_name == ex.name, ) .first() ) if not state: continue weight = state.pending_weight_kg if state.pending_weight_kg is not None else state.weight_kg if weight is None: continue if ( ex.prescribed_weight_kg != weight or ex.needs_weight_input or ex.weight_locked != bool(state.locked) ): ex.prescribed_weight_kg = weight ex.needs_weight_input = False ex.weight_locked = bool(state.locked) for s in ex.sets: if not s.completed: s.target_weight_kg = weight changed = True return changed def start_workout(db: Session, workout: Workout, today: Optional[date] = None) -> Workout: today = today or date.today() if workout.scheduled_date != today: raise HTTPException(400, "Начать можно только сегодняшнюю тренировку") if workout.status not in ("planned",): raise HTTPException(400, "Тренировку нельзя начать") active = ( db.query(Workout) .filter( Workout.cycle_number == workout.cycle_number, Workout.status.in_(["in_progress", "paused"]), ) .first() ) if active: raise HTTPException(400, "Уже есть активная тренировка") workout.status = "in_progress" workout.started_at = datetime.utcnow() workout.current_exercise_index = 0 workout.current_set_index = 0 db.commit() db.refresh(workout) return load_workout(db, workout.id) def pause_workout(db: Session, workout: Workout) -> Workout: if workout.status != "in_progress": raise HTTPException(400, "Пауза доступна только во время тренировки") workout.status = "paused" workout.paused_at = datetime.utcnow() workout.rest_started_at = None db.commit() return load_workout(db, workout.id) def resume_workout(db: Session, workout: Workout, today: Optional[date] = None) -> Workout: today = today or date.today() workout = maybe_expire_pause(db, workout) if workout.status == "abandoned": raise HTTPException(400, "Тренировка остановлена: пауза дольше 15 минут") if workout.status != "paused": raise HTTPException(400, "Тренировка не на паузе") if workout.scheduled_date != today: raise HTTPException(400, "Продолжить можно только в тот же день") assert workout.paused_at if datetime.utcnow() > workout.paused_at + timedelta(minutes=settings.pause_limit_minutes): workout.status = "abandoned" db.commit() raise HTTPException(400, "Тренировка остановлена: пауза дольше 15 минут") workout.status = "in_progress" workout.paused_at = None db.commit() return load_workout(db, workout.id) def set_exercise_weight(db: Session, workout: Workout, exercise: WorkoutExercise, weight_kg: float) -> Workout: if exercise.kind != "accessory": raise HTTPException(400, "Вес задаётся только для вспомогательных") if workout.status == "completed": raise HTTPException(400, "Выполненную тренировку менять нельзя") if exercise.sets and all(s.completed for s in exercise.sets): raise HTTPException(400, "Это упражнение уже выполнено") weight = round_exercise_weight(weight_kg, exercise.name, kind=exercise.kind) exercise.prescribed_weight_kg = weight exercise.needs_weight_input = False for s in exercise.sets: if not s.completed: s.target_weight_kg = weight apply_accessory_weight_to_future( db, cycle_number=workout.cycle_number, exercise_name=exercise.name, weight_kg=weight, from_date=workout.scheduled_date, lock=False, ) db.commit() return load_workout(db, workout.id) def adjust_completed_set( db: Session, workout: Workout, exercise: WorkoutExercise, workout_set: WorkoutSet, *, actual_weight_kg: float, actual_reps: int, ) -> Workout: if exercise.kind != "accessory": raise HTTPException(400, "Править можно только вспомогательные") if not workout_set.completed: raise HTTPException(400, "Можно править только выполненный подход") workout_set.actual_weight_kg = round_exercise_weight(actual_weight_kg, exercise.name, kind=exercise.kind) workout_set.actual_reps = actual_reps db.commit() return load_workout(db, workout.id) def _advance_pointer(workout: Workout, exercises: list[WorkoutExercise]) -> bool: """Advance to next incomplete set. Returns True if workout completed.""" for ei, ex in enumerate(exercises): for s in sorted(ex.sets, key=lambda x: x.set_index): if not s.completed: workout.current_exercise_index = ei workout.current_set_index = s.set_index return False return True def _analyze_accessory(db: Session, workout: Workout, exercise: WorkoutExercise) -> None: if exercise.kind != "accessory": return if exercise.weight_locked: return if any(not s.completed for s in exercise.sets): return all_hit = all( (s.actual_reps or 0) >= (s.target_reps or 0) for s in exercise.sets if s.target_reps is not None ) if not all_hit: exercise.pending_prompt = "choose_weight_missed" else: exercise.pending_prompt = "confirm_easy" def complete_set( db: Session, workout: Workout, *, actual_reps: Optional[int] = None, actual_weight_kg: Optional[float] = None, ) -> Workout: if workout.status != "in_progress": raise HTTPException(400, "Тренировка не активна") exercises = sorted(workout.exercises, key=lambda x: x.order_index) if workout.current_exercise_index >= len(exercises): raise HTTPException(400, "Нет текущего упражнения") exercise = exercises[workout.current_exercise_index] if exercise.pending_prompt: raise HTTPException(400, "Сначала ответьте на вопрос по упражнению") if exercise.kind == "accessory" and exercise.needs_weight_input: raise HTTPException(400, "Сначала введите рабочий вес") sets = sorted(exercise.sets, key=lambda x: x.set_index) current = next((s for s in sets if s.set_index == workout.current_set_index), None) if not current or current.completed: # find first incomplete current = next((s for s in sets if not s.completed), None) if not current: raise HTTPException(400, "Нет активного подхода") if exercise.kind == "bench" or exercise.kind == "failure": current.completed = True current.completed_at = datetime.utcnow() current.actual_weight_kg = current.target_weight_kg current.actual_reps = current.target_reps else: if actual_reps is None: raise HTTPException(400, "Укажите выполненные повторения") weight = actual_weight_kg if actual_weight_kg is not None else exercise.prescribed_weight_kg if weight is None: raise HTTPException(400, "Не задан вес") current.completed = True current.completed_at = datetime.utcnow() current.actual_reps = actual_reps current.actual_weight_kg = round_exercise_weight(weight, exercise.name, kind=exercise.kind) # Check if exercise finished → maybe prompt if all(s.completed for s in sets): _analyze_accessory(db, workout, exercise) if exercise.pending_prompt: workout.rest_started_at = None db.commit() return load_workout(db, workout.id) done = _advance_pointer(workout, exercises) if done: workout.status = "completed" workout.completed_at = datetime.utcnow() workout.rest_started_at = None _maybe_finish_cycle(db, workout) else: workout.rest_started_at = datetime.utcnow() db.commit() return load_workout(db, workout.id) def resolve_prompt( db: Session, workout: Workout, exercise: WorkoutExercise, *, choice: str, weight_kg: Optional[float] = None, increase_kg: Optional[float] = None, decrease_kg: Optional[float] = None, ) -> Workout: if workout.status != "in_progress": raise HTTPException(400, "Тренировка не активна") if not exercise.pending_prompt: raise HTTPException(400, "Нет активного вопроса") prompt = exercise.pending_prompt today = workout.scheduled_date current_weight = exercise.prescribed_weight_kg step = weight_step_for(exercise.name, kind=exercise.kind) if prompt == "confirm_easy": if choice == "easy_no": if current_weight is None: raise HTTPException(400, "Вес не задан") apply_accessory_weight_to_future( db, cycle_number=workout.cycle_number, exercise_name=exercise.name, weight_kg=current_weight, from_date=today, lock=True, ) exercise.weight_locked = True exercise.pending_prompt = None elif choice == "easy_yes": exercise.pending_prompt = "choose_weight_easy" db.commit() return load_workout(db, workout.id) else: raise HTTPException(400, "Неверный выбор") elif prompt == "choose_weight_easy": if current_weight is None: raise HTTPException(400, "Текущий вес не задан") if increase_kg is None or increase_kg <= 0: raise HTTPException(400, "Укажите, на сколько поднять вес") bump = round_to_plate(increase_kg, step) if bump < step: bump = step weight = round_to_plate(current_weight + bump, step) if weight <= current_weight: weight = round_to_plate(current_weight + step, step) apply_accessory_weight_to_future( db, cycle_number=workout.cycle_number, exercise_name=exercise.name, weight_kg=weight, from_date=today + timedelta(days=1), lock=False, ) exercise.pending_prompt = None elif prompt == "choose_weight_missed": if current_weight is None: raise HTTPException(400, "Текущий вес не задан") if decrease_kg is None or decrease_kg <= 0: raise HTTPException(400, "Укажите, на сколько снизить вес") drop = round_to_plate(decrease_kg, step) if drop < step: drop = step weight = round_to_plate(current_weight - drop, step) if weight >= current_weight: weight = round_to_plate(current_weight - step, step) if weight < step: weight = step apply_accessory_weight_to_future( db, cycle_number=workout.cycle_number, exercise_name=exercise.name, weight_kg=weight, from_date=today + timedelta(days=1), lock=False, ) exercise.pending_prompt = None else: raise HTTPException(400, "Неизвестный вопрос") exercises = sorted(workout.exercises, key=lambda x: x.order_index) done = _advance_pointer(workout, exercises) if done: workout.status = "completed" workout.completed_at = datetime.utcnow() workout.rest_started_at = None _maybe_finish_cycle(db, workout) else: workout.rest_started_at = datetime.utcnow() db.commit() return load_workout(db, workout.id) def _maybe_finish_cycle(db: Session, workout: Workout) -> None: remaining = ( db.query(Workout) .filter( Workout.cycle_number == workout.cycle_number, Workout.status.in_(["planned", "in_progress", "paused"]), Workout.id != workout.id, ) .count() ) # also count if this one not yet marked - it's being completed unfinished_others = ( db.query(Workout) .filter( Workout.cycle_number == workout.cycle_number, Workout.id != workout.id, Workout.status.notin_(["completed", "abandoned"]), ) .count() ) if unfinished_others == 0: profile = db.query(Profile).first() if profile: profile.phase = "rest" # wait for user to choose rest days via API; temporarily set probe far profile.probe_available_at = None profile.rest_until = None def start_rest_phase(db: Session, profile: Profile, rest_days: int) -> Profile: if profile.phase not in ("rest", "probe"): # allow if cycle just finished unfinished = ( db.query(Workout) .filter( Workout.cycle_number == profile.cycle_number, Workout.status.notin_(["completed", "abandoned"]), ) .count() ) if unfinished > 0 and profile.phase == "active": raise HTTPException(400, "Цикл ещё не завершён") if rest_days < settings.rest_min_days or rest_days > settings.rest_max_days: raise HTTPException(400, f"Отдых от {settings.rest_min_days} до {settings.rest_max_days} дней") now = datetime.utcnow() profile.phase = "rest" profile.probe_available_at = now + timedelta(days=rest_days) profile.rest_until = profile.probe_available_at db.commit() db.refresh(profile) return profile def submit_probe( db: Session, profile: Profile, *, one_rm_kg: float, start_date: date, training_weekdays: Optional[list[int]] = None, ) -> Profile: now = datetime.utcnow() if profile.phase == "rest" and profile.probe_available_at and now < profile.probe_available_at: raise HTTPException(400, "Ещё рано для проходки") profile.phase = "probe" if profile.phase == "rest" else profile.phase weekdays = training_weekdays or profile.training_weekdays if not weekdays: raise HTTPException(400, "Не заданы дни тренировок") one_rm = round_to_plate(one_rm_kg) new_cycle = profile.cycle_number + 1 profile.one_rm_kg = one_rm profile.training_weekdays = weekdays profile.start_date = start_date profile.cycle_number = new_cycle profile.phase = "active" profile.rest_until = None profile.probe_available_at = None profile.onboarded = True generate_schedule( db, one_rm_kg=one_rm, training_weekdays=weekdays, start_date=start_date, cycle_number=new_cycle, ) db.commit() db.refresh(profile) return profile