from __future__ import annotations from datetime import date, datetime, timedelta from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session, joinedload from app.database import get_db from app.models import AccessoryState, Profile, User, Workout, WorkoutExercise from app.routers.auth import require_auth from app.schemas import ( CalendarResponse, ExportResponse, ImportRequest, MessageResponse, OnboardingRequest, ProbeRequest, ProfileResponse, StartRestRequest, TrainingDaysRequest, WorkoutDetail, WorkoutSummary, ) from app.services.reschedule import plan_weekday_change from app.services.auth import get_or_create_profile from app.services.schedule import clear_cycle_workouts, generate_schedule from app.services.workout import ( load_workout, profile_response, start_rest_phase, submit_probe, workout_detail, workout_summary, ) from app.services.weights import round_to_plate router = APIRouter(prefix="/api", tags=["app"]) @router.get("/profile", response_model=ProfileResponse) def get_profile(_: User = Depends(require_auth), db: Session = Depends(get_db)) -> ProfileResponse: return profile_response(get_or_create_profile(db)) @router.post("/onboarding", response_model=ProfileResponse) def onboarding( body: OnboardingRequest, _: User = Depends(require_auth), db: Session = Depends(get_db), ) -> ProfileResponse: profile = get_or_create_profile(db) if profile.onboarded: raise HTTPException(400, "Онбординг уже пройден. Используйте сброс в настройках.") one_rm = round_to_plate(body.one_rm_kg) profile.one_rm_kg = one_rm profile.training_weekdays = body.training_weekdays profile.start_date = body.start_date profile.onboarded = True profile.phase = "active" profile.cycle_number = 1 clear_cycle_workouts(db, 1) generate_schedule( db, one_rm_kg=one_rm, training_weekdays=body.training_weekdays, start_date=body.start_date, cycle_number=1, ) db.commit() db.refresh(profile) return profile_response(profile) @router.get("/calendar", response_model=CalendarResponse) def calendar( year: int, month: int, _: User = Depends(require_auth), db: Session = Depends(get_db), ) -> CalendarResponse: if month < 1 or month > 12: raise HTTPException(400, "Неверный месяц") profile = get_or_create_profile(db) workouts = ( db.query(Workout) .options(joinedload(Workout.exercises)) .filter( Workout.cycle_number == profile.cycle_number, ) .order_by(Workout.scheduled_date) .all() ) # filter by month for display but also include all cycle for week strip — return month filtered month_workouts = [w for w in workouts if w.scheduled_date.year == year and w.scheduled_date.month == month] return CalendarResponse( year=year, month=month, workouts=[workout_summary(w) for w in month_workouts], profile=profile_response(profile), ) @router.post("/training-days", response_model=ProfileResponse) def update_training_days( body: TrainingDaysRequest, _: User = Depends(require_auth), db: Session = Depends(get_db), ) -> ProfileResponse: profile = get_or_create_profile(db) if not profile.onboarded: raise HTTPException(400, "Сначала создайте программу") rows = db.query(Workout).filter(Workout.cycle_number == profile.cycle_number).all() done = [w.scheduled_date for w in rows if w.status == "completed"] start = date.today() if done: start = max(start, max(done) + timedelta(days=1)) assigned = plan_weekday_change(rows, list(body.training_weekdays), start) for row in rows: if row.id in assigned: row.scheduled_date = date(2099, 1, 1) + timedelta(days=row.id % 300) db.flush() by_id = {row.id: row for row in rows} for workout_row_id, scheduled in assigned.items(): by_id[workout_row_id].scheduled_date = scheduled profile.training_weekdays = list(body.training_weekdays) db.commit() db.refresh(profile) return profile_response(profile) @router.get("/workouts/upcoming", response_model=list[WorkoutSummary]) def upcoming( _: User = Depends(require_auth), db: Session = Depends(get_db), ) -> list[WorkoutSummary]: profile = get_or_create_profile(db) workouts = ( db.query(Workout) .options(joinedload(Workout.exercises)) .filter(Workout.cycle_number == profile.cycle_number) .order_by(Workout.scheduled_date) .all() ) return [workout_summary(w) for w in workouts] @router.get("/workouts/{workout_id}", response_model=WorkoutDetail) def get_workout( workout_id: int, _: User = Depends(require_auth), db: Session = Depends(get_db), ) -> WorkoutDetail: return workout_detail(db, load_workout(db, workout_id)) @router.post("/rest/start", response_model=ProfileResponse) def rest_start( body: StartRestRequest, _: User = Depends(require_auth), db: Session = Depends(get_db), ) -> ProfileResponse: profile = get_or_create_profile(db) return profile_response(start_rest_phase(db, profile, body.rest_days)) @router.post("/probe", response_model=ProfileResponse) def probe( body: ProbeRequest, _: User = Depends(require_auth), db: Session = Depends(get_db), ) -> ProfileResponse: profile = get_or_create_profile(db) return profile_response( submit_probe( db, profile, one_rm_kg=body.one_rm_kg, start_date=body.start_date, training_weekdays=body.training_weekdays, ) ) @router.get("/export", response_model=ExportResponse) def export_data(_: User = Depends(require_auth), db: Session = Depends(get_db)) -> ExportResponse: profile = get_or_create_profile(db) workouts = ( db.query(Workout) .options(joinedload(Workout.exercises).joinedload(WorkoutExercise.sets)) .order_by(Workout.scheduled_date) .all() ) accessories = db.query(AccessoryState).all() def dump_workout(w: Workout) -> dict: return { "cycle_number": w.cycle_number, "week_number": w.week_number, "day_index": w.day_index, "scheduled_date": w.scheduled_date.isoformat(), "status": w.status, "started_at": w.started_at.isoformat() if w.started_at else None, "completed_at": w.completed_at.isoformat() if w.completed_at else None, "current_exercise_index": w.current_exercise_index, "current_set_index": w.current_set_index, "exercises": [ { "order_index": e.order_index, "name": e.name, "kind": e.kind, "prescribed_weight_kg": e.prescribed_weight_kg, "weight_locked": e.weight_locked, "needs_weight_input": e.needs_weight_input, "pending_prompt": e.pending_prompt, "sets": [ { "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, "completed_at": s.completed_at.isoformat() if s.completed_at else None, } for s in sorted(e.sets, key=lambda x: x.set_index) ], } for e in sorted(w.exercises, key=lambda x: x.order_index) ], } return ExportResponse( exported_at=datetime.utcnow(), data={ "profile": { "one_rm_kg": profile.one_rm_kg, "training_weekdays": profile.training_weekdays, "start_date": profile.start_date.isoformat() if profile.start_date else None, "onboarded": profile.onboarded, "phase": profile.phase, "rest_until": profile.rest_until.isoformat() if profile.rest_until else None, "probe_available_at": profile.probe_available_at.isoformat() if profile.probe_available_at else None, "cycle_number": profile.cycle_number, }, "workouts": [dump_workout(w) for w in workouts], "accessories": [ { "cycle_number": a.cycle_number, "exercise_name": a.exercise_name, "weight_kg": a.weight_kg, "locked": a.locked, "pending_weight_kg": a.pending_weight_kg, } for a in accessories ], }, ) @router.post("/import", response_model=MessageResponse) def import_data( body: ImportRequest, _: User = Depends(require_auth), db: Session = Depends(get_db), ) -> MessageResponse: if body.version != 1: raise HTTPException(400, "Неподдерживаемая версия экспорта") data = body.data # full replace of progress db.query(Workout).delete() db.query(AccessoryState).delete() profile = get_or_create_profile(db) p = data.get("profile", {}) profile.one_rm_kg = p.get("one_rm_kg") profile.training_weekdays = p.get("training_weekdays") profile.start_date = datetime.fromisoformat(p["start_date"]).date() if p.get("start_date") else None profile.onboarded = bool(p.get("onboarded")) profile.phase = p.get("phase") or "active" profile.rest_until = datetime.fromisoformat(p["rest_until"]) if p.get("rest_until") else None profile.probe_available_at = ( datetime.fromisoformat(p["probe_available_at"]) if p.get("probe_available_at") else None ) profile.cycle_number = int(p.get("cycle_number") or 1) from app.models import WorkoutSet for a in data.get("accessories", []): db.add( AccessoryState( cycle_number=a["cycle_number"], exercise_name=a["exercise_name"], weight_kg=a.get("weight_kg"), locked=bool(a.get("locked")), pending_weight_kg=a.get("pending_weight_kg"), ) ) for w in data.get("workouts", []): workout = Workout( cycle_number=w["cycle_number"], week_number=w["week_number"], day_index=w["day_index"], scheduled_date=datetime.fromisoformat(w["scheduled_date"]).date(), status=w.get("status") or "planned", started_at=datetime.fromisoformat(w["started_at"]) if w.get("started_at") else None, completed_at=datetime.fromisoformat(w["completed_at"]) if w.get("completed_at") else None, current_exercise_index=w.get("current_exercise_index") or 0, current_set_index=w.get("current_set_index") or 0, ) db.add(workout) db.flush() for e in w.get("exercises", []): ex = WorkoutExercise( workout_id=workout.id, order_index=e["order_index"], name=e["name"], kind=e["kind"], prescribed_weight_kg=e.get("prescribed_weight_kg"), weight_locked=bool(e.get("weight_locked")), needs_weight_input=bool(e.get("needs_weight_input")), pending_prompt=e.get("pending_prompt"), ) db.add(ex) db.flush() for s in e.get("sets", []): db.add( WorkoutSet( exercise_id=ex.id, set_index=s["set_index"], target_percent=s.get("target_percent"), target_reps=s.get("target_reps"), target_weight_kg=s.get("target_weight_kg"), is_failure=bool(s.get("is_failure")), completed=bool(s.get("completed")), actual_reps=s.get("actual_reps"), actual_weight_kg=s.get("actual_weight_kg"), completed_at=datetime.fromisoformat(s["completed_at"]) if s.get("completed_at") else None, ) ) db.commit() return MessageResponse(message="Данные импортированы") @router.post("/reset", response_model=MessageResponse) def reset_all(_: User = Depends(require_auth), db: Session = Depends(get_db)) -> MessageResponse: db.query(Workout).delete() db.query(AccessoryState).delete() profile = get_or_create_profile(db) profile.one_rm_kg = None profile.training_weekdays = None profile.start_date = None profile.onboarded = False profile.phase = "active" profile.rest_until = None profile.probe_available_at = None profile.cycle_number = 1 db.commit() return MessageResponse(message="Прогресс удалён. Пройдите онбординг заново.")