"""Detect fully logged workout sessions (one POST /api/log batch).""" from __future__ import annotations from datetime import datetime from sqlmodel import Session, select from tracker.models import ProgramExercise, WorkoutLog def session_batch_is_complete( day_label: str, batch_logs: list[WorkoutLog], expected: list[ProgramExercise], ) -> bool: """True when this batch has exactly the expected per-exercise set counts for the day.""" if not batch_logs or not expected: return False if any(row.day_label != day_label for row in batch_logs): return False by_exercise: dict[int, int] = {} for row in batch_logs: by_exercise[row.exercise_id] = by_exercise.get(row.exercise_id, 0) + 1 want_ids = {pe.exercise_id for pe in expected} if set(by_exercise.keys()) != want_ids: return False for pe in expected: if by_exercise.get(pe.exercise_id, 0) != pe.target_sets: return False return True def find_last_completed_workout( session: Session, program_id: int, ) -> tuple[str | None, str | None]: """Most recent day_label and date (YYYY-MM-DD) where a full batch was logged.""" logs = session.exec( select(WorkoutLog) .where(WorkoutLog.program_id == program_id) .order_by(WorkoutLog.logged_at.desc()), ).all() idx = 0 while idx < len(logs): anchor = logs[idx].logged_at day_label = logs[idx].day_label batch: list[WorkoutLog] = [] while idx < len(logs) and logs[idx].logged_at == anchor: batch.append(logs[idx]) idx += 1 same_day = [row for row in batch if row.day_label == day_label] expected = session.exec( select(ProgramExercise) .where(ProgramExercise.program_id == program_id) .where(ProgramExercise.day_label == day_label) .order_by(ProgramExercise.order_index), ).all() if session_batch_is_complete(day_label, same_day, list(expected)): logged_at = anchor date_str = logged_at.strftime("%Y-%m-%d") if isinstance(logged_at, datetime) else str(logged_at)[:10] return day_label, date_str return None, None