app/services/progress.py - master - pa1ch/FitAssistant

app/services/progress.py
228 строк8 KB
Pavel Chugaev
feat(exercises): поиск и ранжирование в выборе упражнений
14 июл 2026, 03:09
14 июл 2026, 03:0920890b1
"""Progress analytics: stats, exercise history, measurement trends."""

from dataclasses import dataclass
from datetime import date, timedelta

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.exercise import Exercise
from app.models.measurement import BodyMeasurement
from app.models.workout import COMPLETED, ExerciseSet, Workout, WorkoutExercise


@dataclass
class WorkoutStats:
    total_workouts: int
    workouts_this_month: int
    workouts_last_30_days: int
    total_sets: int
    most_frequent_exercise: str | None
    avg_workouts_per_week: float


async def get_workout_stats(session: AsyncSession, athlete_id: int) -> WorkoutStats:
    total = (
        await session.scalar(
            select(func.count(Workout.id)).where(
                Workout.athlete_id == athlete_id,
                Workout.status == COMPLETED,
            )
        )
        or 0
    )

    today = date.today()
    first_of_month = today.replace(day=1)
    this_month = (
        await session.scalar(
            select(func.count(Workout.id)).where(
                Workout.athlete_id == athlete_id,
                Workout.status == COMPLETED,
                Workout.date >= first_of_month,
            )
        )
        or 0
    )

    thirty_days_ago = today - timedelta(days=30)
    last_30 = (
        await session.scalar(
            select(func.count(Workout.id)).where(
                Workout.athlete_id == athlete_id,
                Workout.status == COMPLETED,
                Workout.date >= thirty_days_ago,
            )
        )
        or 0
    )

    total_sets = (
        await session.scalar(
            select(func.count(ExerciseSet.id))
            .join(WorkoutExercise, ExerciseSet.workout_exercise_id == WorkoutExercise.id)
            .join(Workout, WorkoutExercise.workout_id == Workout.id)
            .where(
                Workout.athlete_id == athlete_id,
                Workout.status == COMPLETED,
            )
        )
        or 0
    )

    # Most frequent exercise
    freq_row = (
        await session.execute(
            select(Exercise.name, func.count(WorkoutExercise.id).label("cnt"))
            .join(WorkoutExercise, Exercise.id == WorkoutExercise.exercise_id)
            .join(Workout, WorkoutExercise.workout_id == Workout.id)
            .where(
                Workout.athlete_id == athlete_id,
                Workout.status == COMPLETED,
            )
            .group_by(Exercise.name)
            .order_by(func.count(WorkoutExercise.id).desc())
            .limit(1)
        )
    ).first()
    most_freq = freq_row[0] if freq_row else None

    # Avg per week
    first_workout_date = await session.scalar(
        select(func.min(Workout.date)).where(
            Workout.athlete_id == athlete_id,
            Workout.status == COMPLETED,
        )
    )
    if first_workout_date and total > 0:
        weeks = max((today - first_workout_date).days / 7, 1)
        avg_per_week = round(total / weeks, 1)
    else:
        avg_per_week = 0

    return WorkoutStats(
        total_workouts=total,
        workouts_this_month=this_month,
        workouts_last_30_days=last_30,
        total_sets=total_sets,
        most_frequent_exercise=most_freq,
        avg_workouts_per_week=avg_per_week,
    )


@dataclass
class ExerciseProgress:
    date: date
    max_weight: float  # 0 when the exercise has no added weight (bodyweight)
    total_volume: float  # sum(weight * reps); 0 for bodyweight
    max_reps: int  # best single-set reps that day
    total_reps: int  # sum of reps across all sets that day
    total_sets: int
    # Estimated one-rep max (Epley: weight * (1 + reps/30)), best single set that
    # day. This is the headline "strength" number: it normalises heavy-low-rep and
    # light-high-rep days onto one comparable curve. 0 for bodyweight exercises
    # (no added weight), where strength progress is read from reps instead.
    est_1rm: float


def is_weighted(points: list[ExerciseProgress]) -> bool:
    """Whether an exercise uses added weight (so progress is shown by weight).

    Bodyweight movements (pull-ups, dips, push-ups) carry no added weight on any
    set — for those, ``False``, and callers chart reps instead. See the
    bodyweight-exercises design.
    """
    return any(p.max_weight > 0 for p in points)


async def get_exercise_progress(
    session: AsyncSession,
    athlete_id: int,
    exercise_id: int,
) -> list[ExerciseProgress]:
    """Get per-workout progress for a specific exercise.

    Reports both weight metrics (max weight, volume — meaningful for weighted
    exercises) and rep metrics (max/total reps — used for bodyweight ones). A
    ``weight_kg`` of NULL means a bodyweight set and contributes 0 to weight/volume.
    """
    stmt = (
        select(
            Workout.date,
            func.max(ExerciseSet.weight_kg).label("max_weight"),
            func.sum(ExerciseSet.weight_kg * ExerciseSet.reps).label("volume"),
            func.max(ExerciseSet.reps).label("max_reps"),
            func.sum(ExerciseSet.reps).label("total_reps"),
            func.count(ExerciseSet.id).label("sets"),
            # Epley per set; NULL weight (bodyweight) yields NULL and is ignored by max().
            func.max(ExerciseSet.weight_kg * (1.0 + ExerciseSet.reps / 30.0)).label("est_1rm"),
        )
        .join(WorkoutExercise, WorkoutExercise.workout_id == Workout.id)
        .join(ExerciseSet, ExerciseSet.workout_exercise_id == WorkoutExercise.id)
        .where(
            Workout.athlete_id == athlete_id,
            Workout.status == COMPLETED,
            WorkoutExercise.exercise_id == exercise_id,
        )
        .group_by(Workout.date)
        .order_by(Workout.date)
    )
    result = await session.execute(stmt)
    return [
        ExerciseProgress(
            date=row.date,
            max_weight=float(row.max_weight or 0),
            total_volume=float(row.volume or 0),
            max_reps=int(row.max_reps or 0),
            total_reps=int(row.total_reps or 0),
            total_sets=row.sets,
            est_1rm=round(float(row.est_1rm or 0), 1),
        )
        for row in result
    ]


@dataclass
class WeeklyTonnage:
    week_start: date  # Monday of the ISO week
    tonnage: float  # sum(weight * reps) lifted that week; 0 for bodyweight
    total_reps: int  # sum of reps that week (the bodyweight equivalent of tonnage)
    total_sets: int


async def get_exercise_weekly_tonnage(
    session: AsyncSession,
    athlete_id: int,
    exercise_id: int,
) -> list[WeeklyTonnage]:
    """Weekly training volume for one exercise: how much was lifted each week.

    Buckets the per-workout points into ISO weeks (Monday start) in Python rather
    than in SQL — this stays dialect-safe (the unit tests run on SQLite, which has
    no ``date_trunc``) and reuses the exact numbers already shown per workout.
    """
    points = await get_exercise_progress(session, athlete_id, exercise_id)
    weeks: dict[date, WeeklyTonnage] = {}
    for p in points:
        monday = p.date - timedelta(days=p.date.weekday())
        bucket = weeks.get(monday)
        if bucket is None:
            bucket = WeeklyTonnage(week_start=monday, tonnage=0.0, total_reps=0, total_sets=0)
            weeks[monday] = bucket
        bucket.tonnage += p.total_volume
        bucket.total_reps += p.total_reps
        bucket.total_sets += p.total_sets
    return [weeks[k] for k in sorted(weeks)]


async def get_measurement_history(
    session: AsyncSession,
    athlete_id: int,
) -> list[BodyMeasurement]:
    stmt = (
        select(BodyMeasurement)
        .where(BodyMeasurement.athlete_id == athlete_id)
        .order_by(BodyMeasurement.date)
    )
    result = await session.execute(stmt)
    return list(result.scalars())