from __future__ import annotations import os import calendar import re import random from datetime import datetime, date, timedelta from pathlib import Path from typing import Optional, List from urllib.parse import urlparse, parse_qs, unquote from zoneinfo import ZoneInfo from fastapi import FastAPI, HTTPException, Request from fastapi.responses import JSONResponse from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, EmailStr, Field from jose import jwt, JWTError from sqlalchemy import ( create_engine, String, Text, Integer, Boolean, DateTime, ForeignKey, UniqueConstraint, Float, Date, text, and_, or_, func, ) from sqlalchemy.exc import IntegrityError from sqlalchemy.dialects.postgresql import insert from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker, relationship import holidays from openpyxl import load_workbook # ====================== # CONFIG # ====================== DATABASE_URL = os.getenv( "DATABASE_URL", "postgresql+psycopg://postgres:postgres@db:5432/postgres", ) SEED_PT_EMAIL = os.getenv("SEED_PT_EMAIL", "mpizarro@camo.local") AUTH_LOGIN_EMAIL = os.getenv("AUTH_LOGIN_EMAIL", SEED_PT_EMAIL) AUTH_LOGIN_PASSWORD = os.getenv("AUTH_LOGIN_PASSWORD", "camo1234") AUTH_LOGIN_DEFAULT_DOMAIN = os.getenv("AUTH_LOGIN_DEFAULT_DOMAIN", "camo.local") JWT_SECRET = os.getenv("JWT_SECRET", "change_me_long_random_string") JWT_ALGORITHM = "HS256" JWT_EXPIRE_MIN = int(os.getenv("JWT_EXPIRE_MIN", "60")) BASE_PLAN_PRICES = { "Entrenamiento Personalizado 2 días por semana": 57500, "Entrenamiento Personalizado 3 días por semana": 77500, "Entrenamiento Personalizado 4 días por semana": 97500, "Evaluación Kinésica": 20000, "10 Sesiones Kinesiología": 100000, } ADDITIONAL_SERVICE_PRICES = { "service_recovery_60": 27500, "service_presoterapia_30": 9750, "service_presoterapia_10": 4750, "service_planificacion_online_4s": 27500, } WEEKDAY_CODES = {"mon", "tue", "wed", "thu", "fri", "sat", "sun"} WEEKDAY_TO_INDEX = { "mon": 0, "tue": 1, "wed": 2, "thu": 3, "fri": 4, "sat": 5, "sun": 6, } INDEX_TO_WEEKDAY = {v: k for k, v in WEEKDAY_TO_INDEX.items()} AGENDA_DAY_ORDER = {"mon": 0, "tue": 1, "wed": 2, "thu": 3, "fri": 4, "sat": 5} AGENDA_ALLOWED_TIMES_BY_DAY = { "mon": {"06:30", "07:30", "08:30", "09:30", "10:30", "18:30", "19:30", "20:30"}, "tue": {"06:30", "07:30", "08:30", "09:30", "10:30", "18:30", "19:30", "20:30"}, "wed": {"06:30", "07:30", "08:30", "09:30", "10:30", "18:30", "19:30", "20:30"}, "thu": {"06:30", "07:30", "08:30", "09:30", "10:30", "18:30", "19:30", "20:30"}, "fri": {"06:30", "07:30", "08:30", "09:30", "10:30", "18:30", "19:30", "20:30"}, "sat": {"09:30", "10:30"}, } MAX_STUDENTS_PER_SLOT = 6 CHECKIN_QR_SECRET = os.getenv("CHECKIN_QR_SECRET", "CAMOFIT_CHECKIN_2026") CHECKIN_TZ = os.getenv("CHECKIN_TZ", "America/Santiago") CHECKIN_EARLY_MIN = int(os.getenv("CHECKIN_EARLY_MIN", "20")) CHECKIN_LATE_MIN = int(os.getenv("CHECKIN_LATE_MIN", "59")) DEFAULT_EXERCISE_CATALOG_EN = [ "BACK SQUATS", "DB PRESS", "CHEST FLYS", "LEG CURLS 2\" PAUSE", "KNEE SPLIT LUNGES", "METCON", "BENCH PRESS (80)", "PULL DOWS", "LEG EXTENSION", "GLUTE REVERSE LUNGES", "PAUSE 90° BB CURLS", "HEX DEADLIFT", "GOOD MORNING", "REAR DELTS", "KNEELING ONE ARM DB PRESS", "EXC PULL UPS + BANDED PULL", "CABLE TRICEPS EXTENSION", "ABS WORKOUT", "BANDED EXT ROT", "HOLLOW HOLDS", "CLAMHSELL", "GLUTE PLANKS", "ISO EXT LUMB", "PAUSE BENCH PRESS", "LEG CURLS", "AUSTRALIAN PULL UPS", "BENCH PRESS", "AIR SQUATS", "CLAMSHELLS", "ISO EXT ROTX", "BARBELL CURLS", "AIR BIKE", "ISO EXT ROT", "CAL ROW", "ONE ARM DB PRESS", "REVERSE LUNGES", ] RANDOM_ROUTINE_POOLS = { "FULL_BODY_A": [ "Sentadilla Goblet", "Press banca con mancuernas", "Remo con mancuerna", "Peso muerto rumano", "Press militar sentado", "Plancha frontal", "Hip thrust", "Jalon al pecho", ], "FULL_BODY_B": [ "Sentadilla hack", "Press inclinado con mancuernas", "Remo en polea baja", "Zancadas caminando", "Elevaciones laterales", "Curl femoral tumbado", "Pallof press", "Ab wheel", ], "PUSH": [ "Press banca plano", "Press inclinado con barra", "Press militar de pie", "Fondos en paralelas", "Elevaciones laterales", "Extension de triceps en polea", "Press cerrado", "Aperturas en maquina", ], "PULL": [ "Dominadas asistidas", "Jalon al pecho", "Remo con barra", "Remo unilateral mancuerna", "Face pull", "Curl biceps con barra", "Curl martillo", "Remo en maquina", ], "LEGS": [ "Sentadilla trasera", "Prensa 45 grados", "Peso muerto rumano", "Curl femoral", "Extension de cuadriceps", "Hip thrust", "Elevacion de talones", "Zancada bulgara", ], "UPPER_A": [ "Press banca plano", "Remo con barra", "Press militar", "Jalon al pecho", "Fondos asistidos", "Face pull", "Curl biceps alterno", "Extension de triceps cuerda", ], "LOWER_A": [ "Sentadilla frontal", "Prensa 45 grados", "Peso muerto rumano", "Curl femoral sentado", "Hip thrust", "Abduccion en maquina", "Elevacion de talones sentado", "Plancha lateral", ], "UPPER_B": [ "Press inclinado mancuernas", "Remo en maquina", "Landmine press", "Dominadas asistidas", "Elevaciones laterales", "Pec deck", "Curl predicador", "Extension triceps overhead", ], "LOWER_B": [ "Sentadilla goblet", "Peso muerto sumo", "Step up con mancuernas", "Curl femoral tumbado", "Extension de cuadriceps", "Hip thrust con pausa", "Elevacion de talones de pie", "Dead bug", ], "REHAB": [ "Movilidad toracica", "Puente gluteo", "Bird dog", "Sentadilla a caja", "Remo con banda", "Press pecho con banda", "Plancha con apoyo de rodillas", "Respiracion diafragmatica", ], } RANDOM_ROUTINE_DAY_TEMPLATES = { 1: ["REHAB"], 2: ["FULL_BODY_A", "FULL_BODY_B"], 3: ["PUSH", "PULL", "LEGS"], 4: ["UPPER_A", "LOWER_A", "UPPER_B", "LOWER_B"], } engine = create_engine(DATABASE_URL, future=True) SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False) class Base(DeclarativeBase): pass # ====================== # MODELS # ====================== class Trainer(Base): __tablename__ = "trainers" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) email: Mapped[str] = mapped_column(String(255), unique=True) first_name: Mapped[Optional[str]] = mapped_column(String(120), nullable=True) paternal_last_name: Mapped[Optional[str]] = mapped_column(String(120), nullable=True) maternal_last_name: Mapped[Optional[str]] = mapped_column(String(120), nullable=True) contact_email: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) phone: Mapped[Optional[str]] = mapped_column(String(60), nullable=True) profile_photo_data: Mapped[Optional[str]] = mapped_column(Text, nullable=True) clients: Mapped[List["Client"]] = relationship( back_populates="trainer", cascade="all, delete-orphan", ) class AuthUser(Base): __tablename__ = "auth_users" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) username: Mapped[str] = mapped_column(String(120), unique=True, index=True) password: Mapped[str] = mapped_column(String(255)) role: Mapped[str] = mapped_column(String(20), default="coach") student_client_id: Mapped[Optional[int]] = mapped_column(ForeignKey("clients.id"), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) class Client(Base): __tablename__ = "clients" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) trainer_id: Mapped[int] = mapped_column(ForeignKey("trainers.id"), index=True) name: Mapped[str] = mapped_column(String(255)) email: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) status: Mapped[str] = mapped_column(String(32), default="active") phone: Mapped[Optional[str]] = mapped_column(String(50), nullable=True) rut: Mapped[Optional[str]] = mapped_column(String(32), nullable=True) birth_date: Mapped[Optional[date]] = mapped_column(Date, nullable=True) enrollment_date: Mapped[Optional[date]] = mapped_column(Date, nullable=True) plan: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) training_days: Mapped[Optional[str]] = mapped_column(String(32), nullable=True) agenda_schedule: Mapped[Optional[str]] = mapped_column(String(500), nullable=True) service_recovery_60: Mapped[bool] = mapped_column(Boolean, default=False) service_presoterapia_30: Mapped[bool] = mapped_column(Boolean, default=False) service_presoterapia_10: Mapped[bool] = mapped_column(Boolean, default=False) service_planificacion_online_4s: Mapped[bool] = mapped_column(Boolean, default=False) address: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) occupation: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) nutrition_tracking: Mapped[bool] = mapped_column(Boolean, default=False) sleep_tracking: Mapped[bool] = mapped_column(Boolean, default=False) habits_tracking: Mapped[bool] = mapped_column(Boolean, default=False) nutrition_notes: Mapped[Optional[str]] = mapped_column(String(2000), nullable=True) sleep_notes: Mapped[Optional[str]] = mapped_column(String(2000), nullable=True) habits_notes: Mapped[Optional[str]] = mapped_column(String(2000), nullable=True) injuries_notes: Mapped[Optional[str]] = mapped_column(String(2000), nullable=True) notes: Mapped[Optional[str]] = mapped_column(String(2000), nullable=True) goal_1: Mapped[Optional[str]] = mapped_column(String(500), nullable=True) goal_2: Mapped[Optional[str]] = mapped_column(String(500), nullable=True) goal_3: Mapped[Optional[str]] = mapped_column(String(500), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) trainer: Mapped["Trainer"] = relationship(back_populates="clients") payments: Mapped[List["Payment"]] = relationship( back_populates="client", cascade="all, delete-orphan", ) monthly_payment_history: Mapped[List["ClientMonthlyPaymentHistory"]] = relationship( back_populates="client", cascade="all, delete-orphan", ) plan_change_history: Mapped[List["ClientPlanChangeHistory"]] = relationship( back_populates="client", cascade="all, delete-orphan", order_by="desc(ClientPlanChangeHistory.changed_at)", ) measurements: Mapped[List["NutritionalMeasurement"]] = relationship( back_populates="client", cascade="all, delete-orphan", order_by="desc(NutritionalMeasurement.measured_at)", ) agenda_exceptions: Mapped[List["ClientAgendaException"]] = relationship( back_populates="client", cascade="all, delete-orphan", ) agenda_change_requests: Mapped[List["ClientAgendaChangeRequest"]] = relationship( back_populates="client", cascade="all, delete-orphan", ) routine_exercises: Mapped[List["ClientRoutineExercise"]] = relationship( back_populates="client", cascade="all, delete-orphan", order_by="ClientRoutineExercise.id", ) class ClientAgendaException(Base): __tablename__ = "client_agenda_exceptions" __table_args__ = ( UniqueConstraint("client_id", "source_date", name="uq_client_agenda_exception_source_date"), ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) client_id: Mapped[int] = mapped_column(ForeignKey("clients.id"), index=True) source_date: Mapped[date] = mapped_column(Date, index=True) target_date: Mapped[date] = mapped_column(Date, index=True) time_slot: Mapped[str] = mapped_column(String(5)) updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) client: Mapped["Client"] = relationship(back_populates="agenda_exceptions") class ClientAgendaChangeRequest(Base): __tablename__ = "client_agenda_change_requests" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) client_id: Mapped[int] = mapped_column(ForeignKey("clients.id"), index=True) source_date: Mapped[date] = mapped_column(Date, index=True) source_time: Mapped[Optional[str]] = mapped_column(String(5), nullable=True) target_date: Mapped[date] = mapped_column(Date, index=True) target_time: Mapped[str] = mapped_column(String(5)) reason: Mapped[Optional[str]] = mapped_column(String(1000), nullable=True) status: Mapped[str] = mapped_column(String(16), default="pending", index=True) trainer_note: Mapped[Optional[str]] = mapped_column(String(1000), nullable=True) reviewed_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) client: Mapped["Client"] = relationship(back_populates="agenda_change_requests") class Payment(Base): __tablename__ = "payments" __table_args__ = ( UniqueConstraint("client_id", "year", "month", name="uq_payment_client_month"), ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) client_id: Mapped[int] = mapped_column(ForeignKey("clients.id"), index=True) year: Mapped[int] = mapped_column(Integer) month: Mapped[int] = mapped_column(Integer) status: Mapped[str] = mapped_column(String(16)) amount: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) client: Mapped["Client"] = relationship(back_populates="payments") class ClientMonthlyPaymentHistory(Base): __tablename__ = "client_monthly_payment_history" __table_args__ = ( UniqueConstraint("client_id", "year", "month", name="uq_client_payment_history_month"), ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) client_id: Mapped[int] = mapped_column(ForeignKey("clients.id"), index=True) year: Mapped[int] = mapped_column(Integer) month: Mapped[int] = mapped_column(Integer) plan_name: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) base_amount: Mapped[int] = mapped_column(Integer, default=0) total_amount: Mapped[int] = mapped_column(Integer, default=0) is_prorated: Mapped[bool] = mapped_column(Boolean, default=False) billing_status: Mapped[str] = mapped_column(String(16), default="due") is_paid: Mapped[bool] = mapped_column(Boolean, default=False) frozen_reason: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) frozen_note: Mapped[Optional[str]] = mapped_column(String(500), nullable=True) frozen_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) paid_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) due_day: Mapped[int] = mapped_column(Integer, default=5) updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) client: Mapped["Client"] = relationship(back_populates="monthly_payment_history") class ClientPlanChangeHistory(Base): __tablename__ = "client_plan_change_history" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) client_id: Mapped[int] = mapped_column(ForeignKey("clients.id"), index=True) previous_plan_name: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) previous_total_amount: Mapped[int] = mapped_column(Integer, default=0) new_plan_name: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) new_total_amount: Mapped[int] = mapped_column(Integer, default=0) changed_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) client: Mapped["Client"] = relationship(back_populates="plan_change_history") class NutritionalMeasurement(Base): __tablename__ = "nutritional_measurements" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) client_id: Mapped[int] = mapped_column(ForeignKey("clients.id"), index=True) measured_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True) weight_kg: Mapped[float] = mapped_column(Float, default=0.0) body_fat_pct: Mapped[float] = mapped_column(Float, default=0.0) muscle_pct: Mapped[float] = mapped_column(Float, default=0.0) water_pct: Mapped[float] = mapped_column(Float, default=0.0) notes: Mapped[str] = mapped_column(String(500), default="") client: Mapped["Client"] = relationship(back_populates="measurements") class ClientRoutineExercise(Base): __tablename__ = "client_routine_exercises" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) client_id: Mapped[int] = mapped_column(ForeignKey("clients.id"), index=True) week_label: Mapped[str] = mapped_column(String(255)) day_label: Mapped[Optional[str]] = mapped_column(String(32), nullable=True) exercise: Mapped[str] = mapped_column(String(255)) series: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) reps: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) weight: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) tonnage: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) previous: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) rpe: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) progress: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) comment: Mapped[Optional[str]] = mapped_column(String(500), nullable=True) source_sheet: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) source_row: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) client: Mapped["Client"] = relationship(back_populates="routine_exercises") class ClientRoutineDraft(Base): __tablename__ = "client_routine_drafts" __table_args__ = ( UniqueConstraint( "client_id", "week_label", "day_label", "row_index", name="uq_client_routine_draft_key", ), ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) client_id: Mapped[int] = mapped_column(ForeignKey("clients.id"), index=True) week_label: Mapped[str] = mapped_column(String(64), index=True) day_label: Mapped[str] = mapped_column(String(16), index=True) row_index: Mapped[int] = mapped_column(Integer) exercise: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) series: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) reps: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) weight: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) previous: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) rpe: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) progress: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) semana: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) comment: Mapped[Optional[str]] = mapped_column(String(500), nullable=True) completed: Mapped[bool] = mapped_column(Boolean, default=False) updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) class ClientRoutineDraftNote(Base): __tablename__ = "client_routine_draft_notes" __table_args__ = ( UniqueConstraint( "client_id", "week_label", "day_label", name="uq_client_routine_draft_note_key", ), ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) client_id: Mapped[int] = mapped_column(ForeignKey("clients.id"), index=True) week_label: Mapped[str] = mapped_column(String(64), index=True) day_label: Mapped[str] = mapped_column(String(16), index=True) note: Mapped[Optional[str]] = mapped_column(String(2000), nullable=True) updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) class ClientRoutineWeekNote(Base): __tablename__ = "client_routine_week_notes" __table_args__ = ( UniqueConstraint( "client_id", "week_label", name="uq_client_routine_week_note_key", ), ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) client_id: Mapped[int] = mapped_column(ForeignKey("clients.id"), index=True) week_label: Mapped[str] = mapped_column(String(64), index=True) note: Mapped[Optional[str]] = mapped_column(String(2000), nullable=True) updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) class ClientRoutineEditNotification(Base): __tablename__ = "client_routine_edit_notifications" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) client_id: Mapped[int] = mapped_column(ForeignKey("clients.id"), index=True) week_label: Mapped[str] = mapped_column(String(64), index=True) day_label: Mapped[str] = mapped_column(String(16), index=True) changed_fields: Mapped[str] = mapped_column(String(120)) message: Mapped[str] = mapped_column(String(500)) is_read: Mapped[bool] = mapped_column(Boolean, default=False, index=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True) read_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) class ClientAttendanceCheckin(Base): __tablename__ = "client_attendance_checkins" __table_args__ = ( UniqueConstraint("client_id", "attendance_date", name="uq_client_attendance_checkin_day"), ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) client_id: Mapped[int] = mapped_column(ForeignKey("clients.id"), index=True) attendance_date: Mapped[date] = mapped_column(Date, index=True) scheduled_time: Mapped[Optional[str]] = mapped_column(String(5), nullable=True) checkin_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True) qr_code: Mapped[str] = mapped_column(String(255)) class TrainerAnnouncement(Base): __tablename__ = "trainer_announcements" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) trainer_id: Mapped[int] = mapped_column(ForeignKey("trainers.id"), index=True) client_id: Mapped[Optional[int]] = mapped_column(ForeignKey("clients.id"), index=True, nullable=True) message: Mapped[str] = mapped_column(String(1000)) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True) class TrainerAbsencePeriod(Base): __tablename__ = "trainer_absence_periods" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) trainer_id: Mapped[int] = mapped_column(ForeignKey("trainers.id"), index=True) start_date: Mapped[date] = mapped_column(Date, index=True) end_date: Mapped[date] = mapped_column(Date, index=True) note: Mapped[Optional[str]] = mapped_column(String(1000), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True) class ClientClassCredit(Base): __tablename__ = "client_class_credits" __table_args__ = ( UniqueConstraint( "absence_id", "client_id", "class_date", "class_time", name="uq_client_class_credit_absence_slot", ), ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) absence_id: Mapped[int] = mapped_column(ForeignKey("trainer_absence_periods.id"), index=True) trainer_id: Mapped[int] = mapped_column(ForeignKey("trainers.id"), index=True) client_id: Mapped[int] = mapped_column(ForeignKey("clients.id"), index=True) class_date: Mapped[date] = mapped_column(Date, index=True) class_time: Mapped[str] = mapped_column(String(5)) status: Mapped[str] = mapped_column(String(16), default="pending", index=True) decision_note: Mapped[Optional[str]] = mapped_column(String(1000), nullable=True) decided_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True) # ====================== # SCHEMAS # ====================== class ClientIn(BaseModel): name: str email: Optional[EmailStr] = None status: str = "active" phone: Optional[str] = None rut: Optional[str] = None birth_date: Optional[date] = None enrollment_date: Optional[date] = None plan: Optional[str] = None training_days: Optional[str] = None service_recovery_60: bool = False service_presoterapia_30: bool = False service_presoterapia_10: bool = False service_planificacion_online_4s: bool = False address: Optional[str] = None occupation: Optional[str] = None nutrition_tracking: bool = False sleep_tracking: bool = False habits_tracking: bool = False nutrition_notes: Optional[str] = None sleep_notes: Optional[str] = None habits_notes: Optional[str] = None injuries_notes: Optional[str] = None notes: Optional[str] = None goal_1: Optional[str] = None goal_2: Optional[str] = None goal_3: Optional[str] = None class ClientOut(ClientIn): id: int class Config: from_attributes = True class ClientMeasurementsCountOut(BaseModel): client_id: int measurements_count: int class MeasurementIn(BaseModel): measured_at: Optional[datetime] = None weight_kg: Optional[float] = 0 body_fat_pct: Optional[float] = 0 muscle_pct: Optional[float] = 0 water_pct: Optional[float] = 0 notes: Optional[str] = "" class MeasurementOut(BaseModel): id: int measured_at: datetime weight_kg: float body_fat_pct: float muscle_pct: float water_pct: float notes: str class Config: from_attributes = True class RoutineImportIn(BaseModel): file_path: str replace_existing: bool = True class RoutineExerciseOut(BaseModel): id: int client_id: int week_label: str day_label: Optional[str] = None exercise: str series: Optional[str] = None reps: Optional[str] = None weight: Optional[str] = None tonnage: Optional[str] = None previous: Optional[str] = None rpe: Optional[str] = None progress: Optional[str] = None comment: Optional[str] = None source_sheet: Optional[str] = None source_row: Optional[int] = None class Config: from_attributes = True class RoutineImportResultOut(BaseModel): imported_count: int sheets_processed: int class RandomRoutineAssignIn(BaseModel): replace_existing: bool = True week_label: Optional[str] = None from_enrollment_date: bool = False class RandomRoutineAssignOut(BaseModel): clients_updated: int rows_created: int week_label: Optional[str] = None weeks_generated: int from_enrollment_date: bool class RoutineDraftRowIn(BaseModel): completed: bool = False exercise: Optional[str] = None series: Optional[str] = None reps: Optional[str] = None weight: Optional[str] = None previous: Optional[str] = None rpe: Optional[str] = None progress: Optional[str] = None semana: Optional[str] = None comment: Optional[str] = None class RoutineDraftUpsertIn(BaseModel): week_label: str day_label: str propagate_exercise_changes: bool = True rows: list[RoutineDraftRowIn] = Field(default_factory=list) class RoutineDraftRowOut(RoutineDraftRowIn): row_index: int class RoutineDraftNoteIn(BaseModel): week_label: str day_label: str note: Optional[str] = None class RoutineDraftNoteOut(BaseModel): week_label: str day_label: str note: Optional[str] = None class RoutineWeekNoteIn(BaseModel): week_label: str note: Optional[str] = None class RoutineWeekNoteOut(BaseModel): week_label: str note: Optional[str] = None class RoutineProgressPointOut(BaseModel): week_label: str day_label: str session_label: str max_weight: Optional[float] = None tonnage: Optional[float] = None average_weight: Optional[float] = None rpe: Optional[float] = None progress_value: Optional[float] = None class RoutineProgressSeriesOut(BaseModel): exercise: str points: list[RoutineProgressPointOut] class RoutineEditNotificationOut(BaseModel): id: int client_id: int client_name: str week_label: str day_label: str changed_fields: list[str] message: str is_read: bool created_at: datetime read_at: Optional[datetime] = None class RoutineEditNotificationReadIn(BaseModel): ids: list[int] = Field(default_factory=list) class AttendanceCheckinScanIn(BaseModel): qr_code: str class AttendanceCheckinScanOut(BaseModel): status: str message: str attendance_date: date checkin_at: datetime scheduled_time: Optional[str] = None class AttendanceQrValueOut(BaseModel): qr_value: str class AttendanceHistoryItemOut(BaseModel): id: int client_id: int client_name: str attendance_date: date scheduled_time: Optional[str] = None checkin_at: datetime class TrainerAnnouncementCreateIn(BaseModel): client_id: Optional[int] = None message: str = Field(min_length=1, max_length=1000) class TrainerAnnouncementOut(BaseModel): id: int client_id: Optional[int] = None client_name: Optional[str] = None message: str created_at: datetime is_general: bool = False class TrainerAbsenceCreateIn(BaseModel): start_date: date end_date: date note: Optional[str] = None class TrainerAbsenceOut(BaseModel): id: int start_date: date end_date: date note: Optional[str] = None created_at: datetime generated_credits: int = 0 class ClassCreditDecisionIn(BaseModel): decision: str note: Optional[str] = None class ClassCreditOut(BaseModel): id: int absence_id: int client_id: int client_name: str class_date: date class_time: str status: str decision_note: Optional[str] = None decided_at: Optional[datetime] = None created_at: datetime class MonthlyPaymentHistoryOut(BaseModel): id: int client_id: int year: int month: int plan_name: Optional[str] = None base_amount: int total_amount: int is_prorated: bool billing_status: str = "due" is_paid: bool frozen_reason: Optional[str] = None frozen_note: Optional[str] = None frozen_at: Optional[datetime] = None paid_at: Optional[datetime] = None due_day: int updated_at: datetime class Config: from_attributes = True class PlanChangeHistoryOut(BaseModel): id: int client_id: int previous_plan_name: Optional[str] = None previous_total_amount: int new_plan_name: Optional[str] = None new_total_amount: int changed_at: datetime class Config: from_attributes = True class BillingHistoryOut(BaseModel): monthly_payments: list[MonthlyPaymentHistoryOut] plan_changes: list[PlanChangeHistoryOut] class PaymentStatusIn(BaseModel): is_paid: bool class MonthlyBillingStatusIn(BaseModel): billing_status: str frozen_reason: Optional[str] = None frozen_note: Optional[str] = None class AgendaEntryIn(BaseModel): day: str time: str class AgendaScheduleIn(BaseModel): entries: list[AgendaEntryIn] = Field(default_factory=list) class AgendaEntryOut(BaseModel): day: str time: str class ClientAgendaOut(BaseModel): client_id: int client_name: str entries: list[AgendaEntryOut] class AgendaOverrideIn(BaseModel): source_date: date target_date: Optional[date] = None time: Optional[str] = None class AgendaOverrideOut(BaseModel): client_id: int source_date: date target_date: date time: str updated_at: datetime class Config: from_attributes = True class AgendaChangeRequestIn(BaseModel): source_date: date source_time: Optional[str] = None target_date: date target_time: str reason: Optional[str] = None class AgendaChangeRequestDecisionIn(BaseModel): action: str note: Optional[str] = None class AgendaChangeRequestOut(BaseModel): id: int client_id: int client_name: str source_date: date source_time: Optional[str] = None target_date: date target_time: str reason: Optional[str] = None status: str trainer_note: Optional[str] = None reviewed_at: Optional[datetime] = None created_at: datetime updated_at: datetime class ClientMonthlyStatusOut(BaseModel): client_id: int year: int month: int plan_name: Optional[str] = None total_amount: int is_prorated: bool billing_status: str = "due" is_paid: bool has_overdue_debt: bool = False paid_at: Optional[datetime] = None due_day: int class PricingPreviewIn(BaseModel): enrollment_date: Optional[date] = None plan: Optional[str] = None training_days: Optional[str] = None service_recovery_60: bool = False service_presoterapia_30: bool = False service_presoterapia_10: bool = False service_planificacion_online_4s: bool = False class PricingPreviewOut(BaseModel): base_amount: int first_month_amount: int is_prorated: bool class TrainerProfileIn(BaseModel): first_name: str = Field(min_length=1, max_length=120) paternal_last_name: str = Field(min_length=1, max_length=120) maternal_last_name: str = Field(min_length=1, max_length=120) contact_email: str = Field(min_length=1, max_length=255) phone: str = Field(min_length=1, max_length=60) profile_photo_data: Optional[str] = None class TrainerProfileOut(BaseModel): first_name: str = "" paternal_last_name: str = "" maternal_last_name: str = "" contact_email: str = "" phone: str = "" profile_photo_data: Optional[str] = None is_complete: bool = False class LoginIn(BaseModel): email: str password: str access_role: Optional[str] = None class StudentEmailIn(BaseModel): email: str class StudentEmailCheckOut(BaseModel): message: str student_client_id: int student_name: str class StudentPasswordResetIn(BaseModel): email: str password: str = Field(min_length=4, max_length=255) class StudentPasswordResetOut(BaseModel): message: str student_client_id: int student_name: str class StudentSignupIn(BaseModel): email: str username: str = Field(min_length=3, max_length=120) password: str = Field(min_length=4, max_length=255) class StudentSignupOut(BaseModel): message: str student_client_id: int student_name: str username: str class LoginOut(BaseModel): access_token: str token_type: str = "bearer" expires_in: int user_email: str access_role: str student_client_id: Optional[int] = None student_name: Optional[str] = None auth_user_role: str = "coach" # ====================== # APP # ====================== app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) Base.metadata.create_all(bind=engine) def build_access_token( username: str, role: str, student_client_id: Optional[int] = None, auth_user_role: str = "coach", ) -> str: expire_at = datetime.utcnow() + timedelta(minutes=JWT_EXPIRE_MIN) payload = { "sub": username, "role": role, "student_client_id": student_client_id, "auth_user_role": auth_user_role, "exp": expire_at, } return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM) def get_auth_context_from_token(token: str) -> Optional[dict]: try: payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM]) except JWTError: return None username = payload.get("sub") role = payload.get("role") auth_user_role = str(payload.get("auth_user_role") or "").strip().lower() is_admin = auth_user_role == "admin" if not username or role not in {"coach", "student"}: return None student_client_id = payload.get("student_client_id") if role == "student": try: parsed_client_id = int(student_client_id) except (TypeError, ValueError): parsed_client_id = 0 if parsed_client_id <= 0 and not is_admin: return None student_client_id = parsed_client_id if parsed_client_id > 0 else None else: student_client_id = None return { "username": str(username), "role": str(role), "student_client_id": student_client_id, "auth_user_role": auth_user_role if auth_user_role in {"admin", "coach", "student"} else "coach", "is_admin": is_admin, } def _extract_client_id_from_path(path: str) -> Optional[int]: m = re.search(r"/trainer/clients/(\d+)(?:/|$)", path) if not m: return None return int(m.group(1)) def _is_student_scoped_request(request: Request) -> bool: return ( getattr(request.state, "access_role", "coach") == "student" and not bool(getattr(request.state, "is_admin", False)) ) def _student_route_allowed(method: str, path: str, student_client_id: int) -> bool: if method == "GET" and path == "/trainer/profile": return True if method == "GET" and path == "/trainer/clients": return True if method == "GET" and path == "/trainer/agenda": return True if method == "GET" and path == "/trainer/agenda/overrides": return True if method == "GET" and path == "/trainer/agenda/change-requests": return True if method == "GET" and path == "/trainer/clients/monthly-status": return True if method == "GET" and path == "/trainer/announcements": return True if method == "GET" and path == "/trainer/class-credits": return True if method == "POST" and re.fullmatch(r"/trainer/clients/\d+/agenda/change-requests", path): return _extract_client_id_from_path(path) == student_client_id if method == "GET" and re.fullmatch(r"/trainer/clients/\d+/measurements", path): return _extract_client_id_from_path(path) == student_client_id if method == "GET" and re.fullmatch(r"/trainer/clients/\d+/routine-progress", path): return _extract_client_id_from_path(path) == student_client_id if method == "GET" and re.fullmatch(r"/trainer/clients/\d+/routines", path): return _extract_client_id_from_path(path) == student_client_id if method == "GET" and re.fullmatch(r"/trainer/clients/\d+/routine-drafts/weeks", path): return _extract_client_id_from_path(path) == student_client_id if method == "GET" and re.fullmatch(r"/trainer/clients/\d+/routine-drafts", path): return _extract_client_id_from_path(path) == student_client_id if method == "PUT" and re.fullmatch(r"/trainer/clients/\d+/routine-drafts", path): return _extract_client_id_from_path(path) == student_client_id if method == "GET" and re.fullmatch(r"/trainer/clients/\d+/billing-history", path): return _extract_client_id_from_path(path) == student_client_id if method == "PUT" and re.fullmatch(r"/trainer/class-credits/\d+/decision", path): return True if method == "POST" and path == "/trainer/checkin/scan": return True if method == "GET" and path == "/trainer/checkin/history": return True return False @app.middleware("http") async def authenticate_trainer_routes(request: Request, call_next): if request.url.path.startswith("/trainer/"): auth_header = request.headers.get("authorization", "") scheme, _, token = auth_header.partition(" ") if scheme.lower() != "bearer" or not token: return JSONResponse(status_code=401, content={"detail": "No autenticado"}) auth_context = get_auth_context_from_token(token.strip()) if not auth_context: return JSONResponse(status_code=401, content={"detail": "Token invalido o expirado"}) request.state.username = auth_context["username"] request.state.access_role = auth_context["role"] request.state.student_client_id = auth_context["student_client_id"] request.state.auth_user_role = auth_context["auth_user_role"] request.state.is_admin = bool(auth_context.get("is_admin")) if auth_context["role"] == "student": student_client_id = auth_context["student_client_id"] if not auth_context.get("is_admin"): if not isinstance(student_client_id, int) or student_client_id <= 0: return JSONResponse(status_code=401, content={"detail": "Sesion de alumno invalida"}) if not _student_route_allowed( request.method.upper(), request.url.path, int(student_client_id), ): return JSONResponse(status_code=403, content={"detail": "Sin permisos para esta ruta"}) return await call_next(request) def ensure_client_columns(): with engine.begin() as conn: conn.execute(text("ALTER TABLE clients ADD COLUMN IF NOT EXISTS birth_date DATE")) conn.execute(text("ALTER TABLE clients ADD COLUMN IF NOT EXISTS enrollment_date DATE")) conn.execute(text("ALTER TABLE clients ADD COLUMN IF NOT EXISTS plan VARCHAR(255)")) conn.execute(text("ALTER TABLE clients ADD COLUMN IF NOT EXISTS training_days VARCHAR(32)")) conn.execute(text("ALTER TABLE clients ADD COLUMN IF NOT EXISTS agenda_schedule VARCHAR(500)")) conn.execute( text( "ALTER TABLE clients ADD COLUMN IF NOT EXISTS service_recovery_60 BOOLEAN NOT NULL DEFAULT FALSE" ) ) conn.execute( text( "ALTER TABLE clients ADD COLUMN IF NOT EXISTS service_presoterapia_30 BOOLEAN NOT NULL DEFAULT FALSE" ) ) conn.execute( text( "ALTER TABLE clients ADD COLUMN IF NOT EXISTS service_presoterapia_10 BOOLEAN NOT NULL DEFAULT FALSE" ) ) conn.execute( text( "ALTER TABLE clients ADD COLUMN IF NOT EXISTS service_planificacion_online_4s BOOLEAN NOT NULL DEFAULT FALSE" ) ) conn.execute(text("ALTER TABLE clients ADD COLUMN IF NOT EXISTS occupation VARCHAR(255)")) conn.execute( text("ALTER TABLE clients ADD COLUMN IF NOT EXISTS nutrition_tracking BOOLEAN NOT NULL DEFAULT FALSE") ) conn.execute( text("ALTER TABLE clients ADD COLUMN IF NOT EXISTS sleep_tracking BOOLEAN NOT NULL DEFAULT FALSE") ) conn.execute( text("ALTER TABLE clients ADD COLUMN IF NOT EXISTS habits_tracking BOOLEAN NOT NULL DEFAULT FALSE") ) conn.execute(text("ALTER TABLE clients ADD COLUMN IF NOT EXISTS nutrition_notes VARCHAR(2000)")) conn.execute(text("ALTER TABLE clients ADD COLUMN IF NOT EXISTS sleep_notes VARCHAR(2000)")) conn.execute(text("ALTER TABLE clients ADD COLUMN IF NOT EXISTS habits_notes VARCHAR(2000)")) conn.execute(text("ALTER TABLE clients ADD COLUMN IF NOT EXISTS injuries_notes VARCHAR(2000)")) ensure_client_columns() def ensure_trainer_columns(): with engine.begin() as conn: conn.execute(text("ALTER TABLE trainers ADD COLUMN IF NOT EXISTS first_name VARCHAR(120)")) conn.execute(text("ALTER TABLE trainers ADD COLUMN IF NOT EXISTS paternal_last_name VARCHAR(120)")) conn.execute(text("ALTER TABLE trainers ADD COLUMN IF NOT EXISTS maternal_last_name VARCHAR(120)")) conn.execute(text("ALTER TABLE trainers ADD COLUMN IF NOT EXISTS contact_email VARCHAR(255)")) conn.execute(text("ALTER TABLE trainers ADD COLUMN IF NOT EXISTS phone VARCHAR(60)")) conn.execute(text("ALTER TABLE trainers ADD COLUMN IF NOT EXISTS profile_photo_data TEXT")) ensure_trainer_columns() def ensure_routine_draft_columns(): with engine.begin() as conn: conn.execute( text( "ALTER TABLE client_routine_drafts " "ADD COLUMN IF NOT EXISTS completed BOOLEAN NOT NULL DEFAULT FALSE" ) ) ensure_routine_draft_columns() def ensure_monthly_history_columns(): with engine.begin() as conn: conn.execute( text( "ALTER TABLE client_monthly_payment_history " "ADD COLUMN IF NOT EXISTS base_amount INTEGER NOT NULL DEFAULT 0" ) ) conn.execute( text( "ALTER TABLE client_monthly_payment_history " "ADD COLUMN IF NOT EXISTS is_prorated BOOLEAN NOT NULL DEFAULT FALSE" ) ) conn.execute( text( "ALTER TABLE client_monthly_payment_history " "ADD COLUMN IF NOT EXISTS is_paid BOOLEAN NOT NULL DEFAULT FALSE" ) ) conn.execute( text( "ALTER TABLE client_monthly_payment_history " "ADD COLUMN IF NOT EXISTS paid_at TIMESTAMP NULL" ) ) conn.execute( text( "ALTER TABLE client_monthly_payment_history " "ADD COLUMN IF NOT EXISTS due_day INTEGER NOT NULL DEFAULT 5" ) ) conn.execute( text( "ALTER TABLE client_monthly_payment_history " "ADD COLUMN IF NOT EXISTS billing_status VARCHAR(16) NOT NULL DEFAULT 'due'" ) ) conn.execute( text( "ALTER TABLE client_monthly_payment_history " "ADD COLUMN IF NOT EXISTS frozen_reason VARCHAR(255) NULL" ) ) conn.execute( text( "ALTER TABLE client_monthly_payment_history " "ADD COLUMN IF NOT EXISTS frozen_note VARCHAR(500) NULL" ) ) conn.execute( text( "ALTER TABLE client_monthly_payment_history " "ADD COLUMN IF NOT EXISTS frozen_at TIMESTAMP NULL" ) ) conn.execute( text( "UPDATE client_monthly_payment_history " "SET billing_status = CASE WHEN is_paid THEN 'paid' ELSE 'due' END " "WHERE billing_status IS NULL OR billing_status = ''" ) ) conn.execute( text( "UPDATE client_monthly_payment_history " "SET billing_status = 'paid' " "WHERE is_paid = TRUE AND billing_status <> 'paid'" ) ) conn.execute( text( "UPDATE client_monthly_payment_history " "SET billing_status = 'due' " "WHERE is_paid = FALSE AND (billing_status IS NULL OR billing_status = '' OR billing_status = 'paid')" ) ) ensure_monthly_history_columns() def _trainer_profile_is_complete(t: Trainer) -> bool: required = [ str(t.first_name or "").strip(), str(t.paternal_last_name or "").strip(), str(t.maternal_last_name or "").strip(), str(t.contact_email or "").strip(), str(t.phone or "").strip(), ] return all(required) def serialize_trainer_profile(t: Trainer) -> TrainerProfileOut: return TrainerProfileOut( first_name=str(t.first_name or "").strip(), paternal_last_name=str(t.paternal_last_name or "").strip(), maternal_last_name=str(t.maternal_last_name or "").strip(), contact_email=str(t.contact_email or "").strip(), phone=str(t.phone or "").strip(), profile_photo_data=t.profile_photo_data, is_complete=_trainer_profile_is_complete(t), ) def calculate_total_amount_from_values( plan: Optional[str], service_recovery_60: bool, service_presoterapia_30: bool, service_presoterapia_10: bool, service_planificacion_online_4s: bool, ) -> int: total = BASE_PLAN_PRICES.get(plan or "", 0) if service_recovery_60: total += ADDITIONAL_SERVICE_PRICES["service_recovery_60"] if service_presoterapia_30: total += ADDITIONAL_SERVICE_PRICES["service_presoterapia_30"] if service_presoterapia_10: total += ADDITIONAL_SERVICE_PRICES["service_presoterapia_10"] if service_planificacion_online_4s: total += ADDITIONAL_SERVICE_PRICES["service_planificacion_online_4s"] return total def calculate_total_amount_from_client(c: Client) -> int: return calculate_total_amount_from_values( c.plan, c.service_recovery_60, c.service_presoterapia_30, c.service_presoterapia_10, c.service_planificacion_online_4s, ) def calculate_total_amount_from_client_in(c: ClientIn) -> int: return calculate_total_amount_from_values( c.plan, c.service_recovery_60, c.service_presoterapia_30, c.service_presoterapia_10, c.service_planificacion_online_4s, ) def normalize_training_days(training_days: Optional[str]) -> list[str]: if not training_days: return [] parts = [p.strip().lower() for p in training_days.split(",")] out: list[str] = [] for p in parts: if p in WEEKDAY_CODES and p not in out: out.append(p) return out def serialize_training_days(days: list[str]) -> Optional[str]: if not days: return None return ",".join(days) def parse_agenda_schedule(value: Optional[str]) -> list[dict[str, str]]: if not value: return [] out: list[dict[str, str]] = [] for raw_entry in value.split(";"): item = raw_entry.strip() if not item or "@" not in item: continue day, time_value = item.split("@", 1) day_norm = day.strip().lower() time_norm = time_value.strip() if day_norm not in AGENDA_ALLOWED_TIMES_BY_DAY: continue if time_norm not in AGENDA_ALLOWED_TIMES_BY_DAY[day_norm]: continue out.append({"day": day_norm, "time": time_norm}) return out def normalize_agenda_entries(entries: list[AgendaEntryIn]) -> list[dict[str, str]]: by_day: dict[str, str] = {} for e in entries: day = str(e.day or "").strip().lower() time_value = str(e.time or "").strip() if day not in AGENDA_ALLOWED_TIMES_BY_DAY: raise HTTPException(status_code=400, detail=f"Dia invalido: {day}") if time_value not in AGENDA_ALLOWED_TIMES_BY_DAY[day]: raise HTTPException(status_code=400, detail=f"Horario invalido para {day}: {time_value}") by_day[day] = time_value out = [{"day": day, "time": time_value} for day, time_value in by_day.items()] out.sort(key=lambda x: (AGENDA_DAY_ORDER[x["day"]], x["time"])) return out def serialize_agenda_entries(entries: list[dict[str, str]]) -> Optional[str]: if not entries: return None return ";".join([f'{e["day"]}@{e["time"]}' for e in entries]) def day_code_from_date(training_date: date) -> str: return INDEX_TO_WEEKDAY[training_date.weekday()] def validate_time_for_date(training_date: date, time_value: str): day_code = day_code_from_date(training_date) if day_code not in AGENDA_ALLOWED_TIMES_BY_DAY: raise HTTPException(status_code=400, detail="No se permiten clases para ese dia") if time_value not in AGENDA_ALLOWED_TIMES_BY_DAY[day_code]: raise HTTPException(status_code=400, detail="Horario invalido para la fecha indicada") def regular_time_for_client_on_date(c: Client, target_date: date) -> Optional[str]: day_code = day_code_from_date(target_date) schedule = parse_agenda_schedule(c.agenda_schedule) for e in schedule: if e["day"] == day_code: return e["time"] return None def client_assigned_times_for_date(db, c: Client, target_date: date) -> list[str]: times: list[str] = [] regular_time = regular_time_for_client_on_date(c, target_date) source_override = ( db.query(ClientAgendaException) .filter( ClientAgendaException.client_id == c.id, ClientAgendaException.source_date == target_date, ) .first() ) if source_override: if source_override.target_date == target_date and source_override.time_slot: times.append(str(source_override.time_slot)) elif regular_time: times.append(str(regular_time)) incoming_overrides = ( db.query(ClientAgendaException) .filter( ClientAgendaException.client_id == c.id, ClientAgendaException.target_date == target_date, ClientAgendaException.source_date != target_date, ) .all() ) for row in incoming_overrides: if row.time_slot: times.append(str(row.time_slot)) return sorted({x.strip() for x in times if str(x).strip()}) def parse_hhmm_to_minutes(value: str) -> Optional[int]: raw = str(value or "").strip() m = re.fullmatch(r"(\d{2}):(\d{2})", raw) if not m: return None hh = int(m.group(1)) mm = int(m.group(2)) if hh < 0 or hh > 23 or mm < 0 or mm > 59: return None return hh * 60 + mm def resolve_checkin_time_for_now(assigned_times: list[str], now_hhmm: str) -> Optional[str]: now_min = parse_hhmm_to_minutes(now_hhmm) if now_min is None: return None for time_value in assigned_times: slot_min = parse_hhmm_to_minutes(time_value) if slot_min is None: continue start = slot_min - max(0, CHECKIN_EARLY_MIN) end = slot_min + max(0, CHECKIN_LATE_MIN) if start <= now_min <= end: return time_value return None def list_dates_in_month(year: int, month: int) -> list[date]: days = calendar.monthrange(year, month)[1] return [date(year, month, d) for d in range(1, days + 1)] def build_agenda_occurrences_for_month( db, trainer_id: int, year: int, month: int, patch_exception: Optional[tuple[int, date, Optional[date], Optional[str]]] = None, ) -> list[dict]: clients = db.query(Client).filter(Client.trainer_id == trainer_id).all() clients_by_id = {c.id: c for c in clients} exceptions = ( db.query(ClientAgendaException) .join(Client, Client.id == ClientAgendaException.client_id) .filter(Client.trainer_id == trainer_id) .all() ) ex_map: dict[tuple[int, date], tuple[date, str]] = {} for ex in exceptions: ex_map[(ex.client_id, ex.source_date)] = (ex.target_date, ex.time_slot) if patch_exception: patch_client_id, patch_source_date, patch_target_date, patch_time = patch_exception key = (patch_client_id, patch_source_date) if patch_target_date is None or patch_time is None: ex_map.pop(key, None) else: ex_map[key] = (patch_target_date, patch_time) out: list[dict] = [] month_dates = list_dates_in_month(year, month) for c in clients: for current in month_dates: regular_time = regular_time_for_client_on_date(c, current) if not regular_time: continue ex = ex_map.get((c.id, current)) if ex: target_date, target_time = ex if target_date.year == year and target_date.month == month: out.append( { "client_id": c.id, "client_name": c.name, "source_date": current, "target_date": target_date, "time": target_time, } ) else: out.append( { "client_id": c.id, "client_name": c.name, "source_date": current, "target_date": current, "time": regular_time, } ) for (client_id, source_date), (target_date, target_time) in ex_map.items(): if source_date.year == year and source_date.month == month: continue if target_date.year != year or target_date.month != month: continue c = clients_by_id.get(client_id) if not c: continue regular_on_source = regular_time_for_client_on_date(c, source_date) if not regular_on_source: continue out.append( { "client_id": c.id, "client_name": c.name, "source_date": source_date, "target_date": target_date, "time": target_time, } ) return out def expected_sessions_per_week(plan: Optional[str]) -> Optional[int]: if not plan: return None if "2 días por semana" in plan: return 2 if "3 días por semana" in plan: return 3 if "4 días por semana" in plan: return 4 return None def random_routine_days_for_plan(plan: Optional[str]) -> int: expected = expected_sessions_per_week(plan) if expected: return expected lower = (plan or "").strip().lower() if "evaluación kinésica" in lower or "kinesiología" in lower: return 2 return 2 def default_week_label_for_routines() -> str: today = date.today() monday = today - timedelta(days=today.weekday()) return monday.isoformat() def week_start_monday(d: date) -> date: return d - timedelta(days=d.weekday()) def week_labels_between(start_date: date, end_date: date) -> list[str]: start_week = week_start_monday(start_date) end_week = week_start_monday(end_date) if start_week > end_week: return [] out: list[str] = [] current = start_week while current <= end_week: out.append(current.isoformat()) current += timedelta(days=7) return out def build_random_rows_for_focus(focus: str, rng: random.Random, max_rows: int = 6) -> list[dict]: pool = RANDOM_ROUTINE_POOLS.get(focus) or RANDOM_ROUTINE_POOLS["FULL_BODY_A"] count = min(max_rows, len(pool)) selected = rng.sample(pool, k=count) def pick_base_weight(exercise_name: str) -> int: name = exercise_name.lower() if any(k in name for k in ["sentadilla", "prensa", "peso muerto", "hip thrust"]): return rng.randint(40, 90) if any(k in name for k in ["press", "remo", "jalon", "dominadas"]): return rng.randint(20, 55) if any(k in name for k in ["curl", "triceps", "elevaciones", "face pull", "abduccion"]): return rng.randint(8, 25) if any(k in name for k in ["plancha", "bird dog", "dead bug", "respiracion", "movilidad"]): return rng.randint(5, 15) return rng.randint(12, 35) def build_weight_for_series(series_count: int, base_weight: int) -> str: parts: list[str] = [] for idx in range(series_count): bump = idx * rng.randint(0, 3) parts.append(str(max(1, base_weight + bump))) return "|".join(parts) rows: list[dict] = [] for exercise in selected: series_value = str(rng.choice([3, 4])) reps_value = str(rng.choice([8, 10, 12, 15])) rpe_value = str(rng.choice([6, 7, 8])) series_count = int(series_value) weight_value = build_weight_for_series(series_count, pick_base_weight(exercise)) rows.append( { "exercise": exercise, "series": series_value, "reps": reps_value, "weight": weight_value, "previous": weight_value, "rpe": rpe_value, "progress": "0", "semana": "", "comment": "", } ) return rows def count_trainable_sessions( year: int, month: int, days: list[str], start_date: Optional[date] = None ) -> int: if not days: return 0 holiday_set = holidays.country_holidays("CL", years=year) weekday_indexes = {WEEKDAY_TO_INDEX[d] for d in days} month_days = calendar.monthrange(year, month)[1] start_day = 1 if start_date and start_date.year == year and start_date.month == month: start_day = max(1, start_date.day) count = 0 for d in range(start_day, month_days + 1): current = date(year, month, d) if current.weekday() not in weekday_indexes: continue if current in holiday_set: continue count += 1 return count def calculate_prorated_amount(monthly_amount: int, enrollment_date: date) -> int: if monthly_amount <= 0: return 0 days_in_month = calendar.monthrange(enrollment_date.year, enrollment_date.month)[1] remaining_days = max(1, days_in_month - enrollment_date.day + 1) return round(monthly_amount * (remaining_days / days_in_month)) def calculate_prorated_amount_by_sessions( monthly_amount: int, enrollment_date: date, training_days: list[str], ) -> int: if monthly_amount <= 0: return 0 total_sessions = count_trainable_sessions( enrollment_date.year, enrollment_date.month, training_days ) if total_sessions <= 0: return 0 remaining_sessions = count_trainable_sessions( enrollment_date.year, enrollment_date.month, training_days, enrollment_date ) return round(monthly_amount * (remaining_sessions / total_sessions)) def calculate_total_for_period_from_client( c: Client, year: int, month: int ) -> tuple[int, bool]: base_total = calculate_total_amount_from_client(c) if c.enrollment_date and c.enrollment_date.year == year and c.enrollment_date.month == month: weekly_sessions = expected_sessions_per_week(c.plan) normalized_days = normalize_training_days(c.training_days) if weekly_sessions and len(normalized_days) == weekly_sessions: prorated = calculate_prorated_amount_by_sessions( base_total, c.enrollment_date, normalized_days ) return prorated, prorated < base_total prorated = calculate_prorated_amount(base_total, c.enrollment_date) return prorated, prorated < base_total return base_total, False def calculate_total_for_period_from_client_in( c: ClientIn, year: int, month: int ) -> tuple[int, bool]: base_total = calculate_total_amount_from_client_in(c) if c.enrollment_date and c.enrollment_date.year == year and c.enrollment_date.month == month: weekly_sessions = expected_sessions_per_week(c.plan) normalized_days = normalize_training_days(c.training_days) if weekly_sessions and len(normalized_days) == weekly_sessions: prorated = calculate_prorated_amount_by_sessions( base_total, c.enrollment_date, normalized_days ) return prorated, prorated < base_total prorated = calculate_prorated_amount(base_total, c.enrollment_date) return prorated, prorated < base_total return base_total, False def normalize_billing_status(value: Optional[str]) -> str: v = (value or "").strip().lower() if v in {"paid", "due", "frozen"}: return v return "paid" if v in {"pagado"} else "due" def upsert_monthly_payment_history( db, client_id: int, plan_name: Optional[str], base_amount: int, total_amount: int, is_prorated: bool, year: int, month: int, ): row = ( db.query(ClientMonthlyPaymentHistory) .filter( ClientMonthlyPaymentHistory.client_id == client_id, ClientMonthlyPaymentHistory.year == year, ClientMonthlyPaymentHistory.month == month, ) .first() ) if row: has_amount_or_plan_changes = row.plan_name != plan_name or row.total_amount != total_amount row.plan_name = plan_name row.base_amount = base_amount row.total_amount = 0 if normalize_billing_status(row.billing_status) == "frozen" else total_amount row.is_prorated = is_prorated row.due_day = 5 row.updated_at = datetime.utcnow() if normalize_billing_status(row.billing_status) == "frozen": row.is_paid = False row.paid_at = None if has_amount_or_plan_changes: if normalize_billing_status(row.billing_status) != "frozen": row.billing_status = "due" row.is_paid = False row.paid_at = None return row new_row = ClientMonthlyPaymentHistory( client_id=client_id, year=year, month=month, plan_name=plan_name, base_amount=base_amount, total_amount=total_amount, is_prorated=is_prorated, billing_status="due", is_paid=False, paid_at=None, due_day=5, updated_at=datetime.utcnow(), ) db.add(new_row) return new_row def get_or_create_monthly_payment_history_for_period( db, c: Client, year: int, month: int, ): row = ( db.query(ClientMonthlyPaymentHistory) .filter( ClientMonthlyPaymentHistory.client_id == c.id, ClientMonthlyPaymentHistory.year == year, ClientMonthlyPaymentHistory.month == month, ) .first() ) if row: row.billing_status = normalize_billing_status(row.billing_status) if row.billing_status == "frozen" and row.total_amount != 0: row.total_amount = 0 row.updated_at = datetime.utcnow() base_amount_now = calculate_total_amount_from_client(c) if row.base_amount <= 0 and base_amount_now > 0: row.base_amount = base_amount_now row.updated_at = datetime.utcnow() if row.due_day != 5: row.due_day = 5 row.updated_at = datetime.utcnow() return row base_amount = calculate_total_amount_from_client(c) total_amount, is_prorated = calculate_total_for_period_from_client(c, year, month) return upsert_monthly_payment_history( db, client_id=c.id, plan_name=c.plan, base_amount=base_amount, total_amount=total_amount, is_prorated=is_prorated, year=year, month=month, ) def as_clean_str(v) -> Optional[str]: if v is None: return None s = str(v).strip() return s if s else None def canonicalize_exercise_name(v: Optional[str]) -> Optional[str]: cleaned = as_clean_str(v) if not cleaned: return None exercise = re.sub(r"\s+", " ", cleaned).strip() exercise = re.sub(r"\bPUASE\b", "PAUSE", exercise, flags=re.IGNORECASE) if exercise.upper().startswith("LEG CURLS"): return "LEG CURLS" return exercise def normalize_exercise_name(v: Optional[str], strip_parenthesized_suffix: bool = True) -> Optional[str]: exercise = canonicalize_exercise_name(v) if not exercise: return None if strip_parenthesized_suffix: exercise = re.sub(r"\s*\([^)]*\)\s*$", "", exercise).strip() exercise = canonicalize_exercise_name(exercise) or exercise return exercise.upper() if exercise else None def normalize_week_label(v: Optional[str]) -> Optional[str]: cleaned = as_clean_str(v) if not cleaned: return None return cleaned def normalize_day_label(v: Optional[str]) -> Optional[str]: cleaned = as_clean_str(v) if not cleaned: return None upper = cleaned.upper() if upper not in {"DIA 1", "DIA 2", "DIA 3", "DIA 4"}: return None return upper def week_label_to_date(week_label: Optional[str]) -> Optional[date]: cleaned = as_clean_str(week_label) if not cleaned: return None try: return date.fromisoformat(cleaned) except ValueError: return None def _parse_numeric(v: Optional[str]) -> Optional[float]: if v is None: return None raw = str(v).strip() if not raw: return None if not re.match(r"^[-+]?\d+([.,]\d+)?$", raw): return None parsed = float(raw.replace(",", ".")) if parsed != parsed: # NaN check return None return parsed def _parse_series_count(v: Optional[str]) -> int: parsed = _parse_numeric(v) if parsed is None: return 0 rounded = int(round(parsed)) if abs(parsed - rounded) > 0.000001: return 0 if rounded <= 0: return 0 return min(rounded, 20) def _split_weight_parts(v: Optional[str]) -> list[str]: raw = as_clean_str(v) if not raw: return [] if "|" in raw: return [x.strip() for x in raw.split("|")] if "," in raw: return [x.strip() for x in raw.split(",")] return [raw] def _max_numeric_weight_from_weight_field(v: Optional[str]) -> Optional[float]: parts = _split_weight_parts(v) max_value: Optional[float] = None for part in parts: parsed = _parse_numeric(part) if parsed is None: continue if max_value is None or parsed > max_value: max_value = parsed return max_value def _format_rm_snapshot(v: Optional[float]) -> Optional[str]: if v is None: return None if abs(v - round(v)) < 0.000001: return str(int(round(v))) return f"{v:.2f}".rstrip("0").rstrip(".") def _calc_tonnage(series: Optional[str], reps: Optional[str], weight: Optional[str]) -> Optional[float]: reps_value = _parse_numeric(reps) series_count = _parse_series_count(series) if reps_value is None or series_count <= 0: return None parts = _split_weight_parts(weight) if len(parts) < series_count: return None total_weight = 0.0 for i in range(series_count): weight_part = _parse_numeric(parts[i]) if weight_part is None: return None total_weight += weight_part return reps_value * total_weight def _calc_average_weight(series: Optional[str], weight: Optional[str]) -> Optional[float]: series_count = _parse_series_count(series) if series_count <= 0: return None parts = _split_weight_parts(weight) if len(parts) < series_count: return None values: list[float] = [] for i in range(series_count): parsed = _parse_numeric(parts[i]) if parsed is None: return None values.append(parsed) if not values: return None return sum(values) / len(values) def _calc_max_weight(weight: Optional[str]) -> Optional[float]: parts = _split_weight_parts(weight) if not parts: return None values: list[float] = [] for part in parts: parsed = _parse_numeric(part) if parsed is None: continue values.append(parsed) if not values: return None return max(values) def _parse_week_date(week_label: str) -> Optional[date]: cleaned = as_clean_str(week_label) if not cleaned: return None try: return datetime.strptime(cleaned, "%Y-%m-%d").date() except ValueError: return None def _parse_routine_rows_from_workbook(file_path: Path) -> list[dict]: wb = load_workbook(file_path, data_only=True) parsed: list[dict] = [] for ws in wb.worksheets: week_label = as_clean_str(ws.title) if not week_label: continue has_training_header = False for row_idx in range(1, min(ws.max_row, 40) + 1): col_b = as_clean_str(ws.cell(row_idx, 2).value) col_c = as_clean_str(ws.cell(row_idx, 3).value) if col_c == "EJERCICIO" and col_b and col_b.upper().startswith("DIA"): has_training_header = True break if not has_training_header: continue current_day = None for row_idx in range(1, ws.max_row + 1): col_b = as_clean_str(ws.cell(row_idx, 2).value) exercise = canonicalize_exercise_name(ws.cell(row_idx, 3).value) if col_b and col_b.upper().startswith("DIA"): current_day = col_b.upper().replace(" ", " ") continue if not exercise: continue if exercise.upper() == "EJERCICIO": continue parsed.append( { "week_label": week_label, "day_label": current_day, "exercise": exercise, "series": as_clean_str(ws.cell(row_idx, 4).value), "reps": as_clean_str(ws.cell(row_idx, 5).value), "weight": as_clean_str(ws.cell(row_idx, 6).value), "tonnage": as_clean_str(ws.cell(row_idx, 7).value), "previous": as_clean_str(ws.cell(row_idx, 8).value), "rpe": as_clean_str(ws.cell(row_idx, 9).value), "progress": as_clean_str(ws.cell(row_idx, 10).value), "comment": as_clean_str(ws.cell(row_idx, 12).value), "source_sheet": ws.title, "source_row": row_idx, } ) return parsed # ====================== # AUTH ENDPOINTS # ====================== @app.post("/auth/login", response_model=LoginOut) def login(body: LoginIn): username = str(body.email).strip() if not username: raise HTTPException(status_code=401, detail="Credenciales invalidas") requested_role = str(body.access_role or "").strip().lower() if requested_role not in {"coach", "student"}: requested_role = "coach" with SessionLocal() as db: user = ( db.query(AuthUser) .filter(func.lower(AuthUser.username) == username.lower()) .first() ) if not user or user.password != body.password: raise HTTPException(status_code=401, detail="Credenciales invalidas") student_name: Optional[str] = None student_client_id: Optional[int] = None raw_user_role = str(user.role or "").strip().lower() if raw_user_role == "admin": role = requested_role elif raw_user_role == "student": role = "student" else: role = "coach" if role == "student": linked_student_id = user.student_client_id if raw_user_role == "admin" and not linked_student_id: trainer = get_or_create_trainer(db) fallback_student = ( db.query(Client) .filter(Client.trainer_id == trainer.id) .order_by(Client.id.asc()) .first() ) linked_student_id = fallback_student.id if fallback_student else None if not linked_student_id and raw_user_role != "admin": raise HTTPException(status_code=403, detail="Usuario alumno sin perfil asociado") if linked_student_id: client = db.query(Client).filter(Client.id == linked_student_id).first() if not client and raw_user_role != "admin": raise HTTPException(status_code=403, detail="Perfil de alumno no encontrado") if client: student_client_id = client.id student_name = client.name token = build_access_token( username=user.username, role=role, student_client_id=student_client_id, auth_user_role=raw_user_role if raw_user_role in {"admin", "coach", "student"} else "coach", ) return LoginOut( access_token=token, expires_in=JWT_EXPIRE_MIN * 60, user_email=user.username, access_role=role, student_client_id=student_client_id, student_name=student_name, auth_user_role=raw_user_role if raw_user_role in {"admin", "coach", "student"} else "coach", ) def _get_student_client_by_email(db, email: str) -> Client: trainer = get_or_create_trainer(db) client = ( db.query(Client) .filter( Client.trainer_id == trainer.id, Client.email.isnot(None), func.lower(Client.email) == email.lower(), ) .first() ) if not client: raise HTTPException( status_code=404, detail="El correo no coincide con ningun alumno creado por el entrenador", ) return client @app.post("/auth/student-recovery/verify-email", response_model=StudentEmailCheckOut) def verify_student_recovery_email(body: StudentEmailIn): email = str(body.email).strip().lower() if not email: raise HTTPException(status_code=400, detail="Debes ingresar un correo valido") with SessionLocal() as db: client = _get_student_client_by_email(db, email) student_user = db.query(AuthUser).filter(AuthUser.student_client_id == client.id).first() if not student_user: raise HTTPException( status_code=409, detail="El alumno aun no tiene usuario. Usa la opcion 'Nuevo alumno'.", ) return StudentEmailCheckOut( message="Correo verificado. Ahora crea tu nueva contrasena.", student_client_id=int(client.id), student_name=str(client.name), ) @app.post("/auth/student-recovery/reset-password", response_model=StudentPasswordResetOut) def reset_student_password(body: StudentPasswordResetIn): email = str(body.email).strip().lower() password = str(body.password) if not email: raise HTTPException(status_code=400, detail="Debes ingresar un correo valido") if not password: raise HTTPException(status_code=400, detail="Debes ingresar una contrasena") with SessionLocal() as db: client = _get_student_client_by_email(db, email) student_user = db.query(AuthUser).filter(AuthUser.student_client_id == client.id).first() if not student_user: raise HTTPException( status_code=409, detail="El alumno aun no tiene usuario. Usa la opcion 'Nuevo alumno'.", ) student_user.password = password student_user.role = "student" student_user.student_client_id = client.id db.commit() return StudentPasswordResetOut( message="Contrasena actualizada correctamente. Ya puedes iniciar sesion.", student_client_id=int(client.id), student_name=str(client.name), ) @app.post("/auth/student-signup", response_model=StudentSignupOut) def signup_student_access(body: StudentSignupIn): email = str(body.email).strip().lower() username = str(body.username).strip() password = str(body.password) if not email: raise HTTPException(status_code=400, detail="Debes ingresar un correo valido") if not username: raise HTTPException(status_code=400, detail="Debes ingresar un username") if not password: raise HTTPException(status_code=400, detail="Debes ingresar una contrasena") with SessionLocal() as db: client = _get_student_client_by_email(db, email) existing_username_owner = ( db.query(AuthUser) .filter(func.lower(AuthUser.username) == username.lower()) .first() ) if existing_username_owner: raise HTTPException(status_code=409, detail="username ocupado") existing_student_user = ( db.query(AuthUser) .filter(AuthUser.student_client_id == client.id) .first() ) if existing_student_user: raise HTTPException( status_code=409, detail="El alumno ya tiene usuario. Usa la opcion 'Olvide mi contrasena'.", ) db.add( AuthUser( username=username, password=password, role="student", student_client_id=client.id, ) ) db.commit() return StudentSignupOut( message="Usuario alumno creado correctamente. Ya puedes iniciar sesion.", student_client_id=int(client.id), student_name=str(client.name), username=username, ) # ====================== # SEED TRAINER # ====================== def get_or_create_trainer(db): t = db.query(Trainer).filter(Trainer.email == SEED_PT_EMAIL).first() if not t: t = Trainer(email=SEED_PT_EMAIL) db.add(t) db.commit() db.refresh(t) # Compatibilidad: algunas instalaciones tienen un perfil legado con datos # (por ejemplo pt@camo.local) y el perfil seed actual vacío. # Si el perfil seed está completamente vacío, se completa automáticamente # con el perfil más completo disponible para que el alumno vea esos datos. def is_blank(value: Optional[str]) -> bool: return not str(value or "").strip() is_seed_profile_empty = all( [ is_blank(t.first_name), is_blank(t.paternal_last_name), is_blank(t.maternal_last_name), is_blank(t.contact_email), is_blank(t.phone), is_blank(t.profile_photo_data), ] ) if is_seed_profile_empty: legacy_candidates = db.query(Trainer).filter(Trainer.id != t.id).all() def profile_score(row: Trainer) -> int: values = [ row.first_name, row.paternal_last_name, row.maternal_last_name, row.contact_email, row.phone, row.profile_photo_data, ] return sum(1 for value in values if not is_blank(value)) legacy = max(legacy_candidates, key=profile_score, default=None) if legacy and profile_score(legacy) > 0: t.first_name = str(legacy.first_name or "").strip() t.paternal_last_name = str(legacy.paternal_last_name or "").strip() t.maternal_last_name = str(legacy.maternal_last_name or "").strip() t.contact_email = str(legacy.contact_email or "").strip() t.phone = str(legacy.phone or "").strip() t.profile_photo_data = str(legacy.profile_photo_data or "").strip() or None db.commit() db.refresh(t) return t def ensure_default_auth_users(): internal_student_emails = {"dgarciab@camo.local", "dgarciab@example.com"} internal_student_names = {"dgarciab"} with SessionLocal() as db: trainer = get_or_create_trainer(db) trainer_clients = ( db.query(Client) .filter(Client.trainer_id == trainer.id) .order_by(Client.id.asc()) .all() ) def is_internal_student(c: Client) -> bool: name_lower = str(c.name or "").strip().lower() email_lower = str(c.email or "").strip().lower() return name_lower in internal_student_names or email_lower in internal_student_emails preferred_student = next((c for c in trainer_clients if not is_internal_student(c)), None) seed_student = next((c for c in trainer_clients if is_internal_student(c)), None) if seed_student: seed_student.status = "internal" if preferred_student and preferred_student.id != seed_student.id: ( db.query(AuthUser) .filter(AuthUser.student_client_id == seed_student.id) .update({"student_client_id": preferred_student.id}, synchronize_session=False) ) linked_student = preferred_student or seed_student coach_user = ( db.query(AuthUser) .filter(func.lower(AuthUser.username) == "mpizarro") .first() ) if not coach_user: coach_user = AuthUser( username="MPizarro", password="StrictLab.2k26", role="coach", student_client_id=None, ) db.add(coach_user) else: coach_user.password = "StrictLab.2k26" coach_user.role = "coach" coach_user.student_client_id = None # Limpia el usuario coach legado para que no reaparezca en el sistema. ( db.query(AuthUser) .filter(func.lower(AuthUser.username) == "entrenador1") .delete(synchronize_session=False) ) student_user = ( db.query(AuthUser) .filter(func.lower(AuthUser.username) == "dgarciab") .first() ) if not student_user: student_user = AuthUser( username="DGarciaB", password="admin123", role="student", student_client_id=(linked_student.id if linked_student else None), ) db.add(student_user) else: # Keep existing student mapping in real environments. # This seed helper should only fill missing linkage. student_user.password = "admin123" student_user.role = "student" if student_user.student_client_id is None and linked_student: student_user.student_client_id = linked_student.id admin_user = ( db.query(AuthUser) .filter(func.lower(AuthUser.username) == "admin") .first() ) if not admin_user: admin_user = AuthUser( username="Admin", password="Admin123", role="admin", student_client_id=(linked_student.id if linked_student else None), ) db.add(admin_user) else: admin_user.password = "Admin123" admin_user.role = "admin" if admin_user.student_client_id is None and linked_student: admin_user.student_client_id = linked_student.id db.commit() ensure_default_auth_users() # ====================== # CLIENT ENDPOINTS # ====================== @app.get("/trainer/profile", response_model=TrainerProfileOut) def get_trainer_profile(request: Request): with SessionLocal() as db: t = get_or_create_trainer(db) return serialize_trainer_profile(t) @app.get("/trainer/announcements", response_model=list[TrainerAnnouncementOut]) def list_trainer_announcements(request: Request, limit: int = 20): with SessionLocal() as db: t = get_or_create_trainer(db) safe_limit = max(1, min(int(limit or 20), 100)) q = ( db.query(TrainerAnnouncement, Client.name) .outerjoin(Client, Client.id == TrainerAnnouncement.client_id) .filter(TrainerAnnouncement.trainer_id == t.id) .order_by(TrainerAnnouncement.created_at.desc(), TrainerAnnouncement.id.desc()) ) if _is_student_scoped_request(request): student_client_id = int(getattr(request.state, "student_client_id", 0) or 0) q = q.filter( or_( TrainerAnnouncement.client_id.is_(None), TrainerAnnouncement.client_id == student_client_id, ) ) rows = q.limit(safe_limit).all() out: list[TrainerAnnouncementOut] = [] for ann, client_name in rows: out.append( TrainerAnnouncementOut( id=int(ann.id), client_id=int(ann.client_id) if ann.client_id else None, client_name=str(client_name or "").strip() or None, message=str(ann.message or "").strip(), created_at=ann.created_at, is_general=ann.client_id is None, ) ) return out @app.post("/trainer/announcements", response_model=TrainerAnnouncementOut) def create_trainer_announcement(request: Request, body: TrainerAnnouncementCreateIn): if _is_student_scoped_request(request): raise HTTPException(status_code=403, detail="Sin permisos para crear comunicados") with SessionLocal() as db: t = get_or_create_trainer(db) target_client_id = int(body.client_id) if body.client_id else None target_client_name: Optional[str] = None if target_client_id: c = db.query(Client).filter(Client.id == target_client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="Alumno no encontrado para comunicado") target_client_name = str(c.name or "").strip() or None message = str(body.message or "").strip() if not message: raise HTTPException(status_code=400, detail="El comunicado no puede estar vacío") ann = TrainerAnnouncement( trainer_id=t.id, client_id=target_client_id, message=message, created_at=datetime.utcnow(), ) db.add(ann) db.commit() db.refresh(ann) return TrainerAnnouncementOut( id=int(ann.id), client_id=int(ann.client_id) if ann.client_id else None, client_name=target_client_name, message=str(ann.message or "").strip(), created_at=ann.created_at, is_general=ann.client_id is None, ) @app.post("/trainer/absences", response_model=TrainerAbsenceOut) def create_trainer_absence_period(request: Request, body: TrainerAbsenceCreateIn): if _is_student_scoped_request(request): raise HTTPException(status_code=403, detail="Sin permisos para registrar ausencias") start = body.start_date end = body.end_date if end < start: raise HTTPException(status_code=400, detail="Rango de fechas invalido") if (end - start).days > 90: raise HTTPException(status_code=400, detail="El periodo maximo de ausencia es de 90 dias") with SessionLocal() as db: t = get_or_create_trainer(db) absence = TrainerAbsencePeriod( trainer_id=t.id, start_date=start, end_date=end, note=str(body.note or "").strip() or None, created_at=datetime.utcnow(), ) db.add(absence) db.flush() clients = ( db.query(Client) .filter( Client.trainer_id == t.id, or_(Client.status.is_(None), func.lower(Client.status) != "internal"), ) .all() ) generated = 0 day_count = (end - start).days + 1 for offset in range(day_count): current_day = start + timedelta(days=offset) for c in clients: # Si el alumno asistio ese dia, no corresponde generar saldo. attendance_exists = ( db.query(ClientAttendanceCheckin.id) .filter( ClientAttendanceCheckin.client_id == c.id, ClientAttendanceCheckin.attendance_date == current_day, ) .first() ) if attendance_exists: continue assigned_times = client_assigned_times_for_date(db, c, current_day) for time_value in assigned_times: credit = ClientClassCredit( absence_id=absence.id, trainer_id=t.id, client_id=c.id, class_date=current_day, class_time=time_value, status="pending", created_at=datetime.utcnow(), ) db.add(credit) generated += 1 db.commit() db.refresh(absence) return TrainerAbsenceOut( id=int(absence.id), start_date=absence.start_date, end_date=absence.end_date, note=absence.note, created_at=absence.created_at, generated_credits=generated, ) @app.get("/trainer/class-credits", response_model=list[ClassCreditOut]) def list_class_credits( request: Request, status: Optional[str] = None, client_id: Optional[int] = None, limit: int = 200, ): with SessionLocal() as db: t = get_or_create_trainer(db) safe_limit = max(1, min(int(limit or 200), 500)) q = ( db.query(ClientClassCredit, Client.name) .join(Client, Client.id == ClientClassCredit.client_id) .filter(ClientClassCredit.trainer_id == t.id) ) if _is_student_scoped_request(request): student_client_id = int(getattr(request.state, "student_client_id", 0) or 0) q = q.filter(ClientClassCredit.client_id == student_client_id) elif client_id: q = q.filter(ClientClassCredit.client_id == int(client_id)) if status: status_norm = str(status).strip().lower() if status_norm not in {"pending", "recover", "refund", "discount"}: raise HTTPException(status_code=400, detail="Estado invalido") q = q.filter(ClientClassCredit.status == status_norm) rows = q.order_by(ClientClassCredit.class_date.desc(), ClientClassCredit.class_time.asc()).limit(safe_limit).all() out: list[ClassCreditOut] = [] for credit, client_name in rows: out.append( ClassCreditOut( id=int(credit.id), absence_id=int(credit.absence_id), client_id=int(credit.client_id), client_name=str(client_name or "").strip(), class_date=credit.class_date, class_time=str(credit.class_time or ""), status=str(credit.status or "pending"), decision_note=str(credit.decision_note or "").strip() or None, decided_at=credit.decided_at, created_at=credit.created_at, ) ) return out @app.put("/trainer/class-credits/{credit_id}/decision", response_model=ClassCreditOut) def decide_class_credit(credit_id: int, body: ClassCreditDecisionIn, request: Request): decision = str(body.decision or "").strip().lower() if decision not in {"recover", "refund", "discount"}: raise HTTPException(status_code=400, detail="Decision invalida") with SessionLocal() as db: t = get_or_create_trainer(db) pair = ( db.query(ClientClassCredit, Client.name) .join(Client, Client.id == ClientClassCredit.client_id) .filter(ClientClassCredit.id == credit_id, ClientClassCredit.trainer_id == t.id) .first() ) if not pair: raise HTTPException(status_code=404, detail="Saldo no encontrado") credit, client_name = pair if _is_student_scoped_request(request): student_client_id = int(getattr(request.state, "student_client_id", 0) or 0) if credit.client_id != student_client_id: raise HTTPException(status_code=403, detail="Sin permisos para decidir este saldo") credit.status = decision credit.decision_note = str(body.note or "").strip() or None credit.decided_at = datetime.utcnow() db.commit() db.refresh(credit) return ClassCreditOut( id=int(credit.id), absence_id=int(credit.absence_id), client_id=int(credit.client_id), client_name=str(client_name or "").strip(), class_date=credit.class_date, class_time=str(credit.class_time or ""), status=str(credit.status or "pending"), decision_note=str(credit.decision_note or "").strip() or None, decided_at=credit.decided_at, created_at=credit.created_at, ) @app.put("/trainer/profile", response_model=TrainerProfileOut) def update_trainer_profile(request: Request, body: TrainerProfileIn): if _is_student_scoped_request(request): raise HTTPException(status_code=403, detail="Sin permisos para editar perfil de entrenador") with SessionLocal() as db: t = get_or_create_trainer(db) t.first_name = str(body.first_name or "").strip() t.paternal_last_name = str(body.paternal_last_name or "").strip() t.maternal_last_name = str(body.maternal_last_name or "").strip() t.contact_email = str(body.contact_email or "").strip() t.phone = str(body.phone or "").strip() raw_photo = str(body.profile_photo_data or "").strip() t.profile_photo_data = raw_photo or None db.commit() db.refresh(t) return serialize_trainer_profile(t) @app.get("/trainer/clients", response_model=list[ClientOut]) def list_clients(request: Request): with SessionLocal() as db: t = get_or_create_trainer(db) q = db.query(Client).filter( Client.trainer_id == t.id, or_(Client.status.is_(None), func.lower(Client.status) != "internal"), ) if _is_student_scoped_request(request): student_client_id = getattr(request.state, "student_client_id", None) q = q.filter(Client.id == student_client_id) return q.all() @app.get("/trainer/clients/measurements-count", response_model=list[ClientMeasurementsCountOut]) def list_clients_measurements_count(request: Request): with SessionLocal() as db: t = get_or_create_trainer(db) q = ( db.query( Client.id.label("client_id"), func.count(NutritionalMeasurement.id).label("measurements_count"), ) .outerjoin(NutritionalMeasurement, NutritionalMeasurement.client_id == Client.id) .filter( Client.trainer_id == t.id, or_(Client.status.is_(None), func.lower(Client.status) != "internal"), ) .group_by(Client.id) .order_by(Client.id) ) if _is_student_scoped_request(request): student_client_id = getattr(request.state, "student_client_id", None) q = q.filter(Client.id == student_client_id) rows = q.all() return [ ClientMeasurementsCountOut(client_id=int(row.client_id), measurements_count=int(row.measurements_count or 0)) for row in rows ] @app.post("/trainer/clients", response_model=ClientOut) def create_client(c: ClientIn): with SessionLocal() as db: t = get_or_create_trainer(db) normalized_training_days = serialize_training_days(normalize_training_days(c.training_days)) payload = c.dict(exclude={"training_days"}) new = Client( trainer_id=t.id, training_days=normalized_training_days, **payload, ) db.add(new) db.flush() now = datetime.utcnow() base_amount = calculate_total_amount_from_client_in(c) total_amount, is_prorated = calculate_total_for_period_from_client_in(c, now.year, now.month) upsert_monthly_payment_history( db, client_id=new.id, plan_name=new.plan, base_amount=base_amount, total_amount=total_amount, is_prorated=is_prorated, year=now.year, month=now.month, ) db.commit() db.refresh(new) return new @app.put("/trainer/clients/{client_id}", response_model=ClientOut) def update_client(client_id: int, c: ClientIn): with SessionLocal() as db: t = get_or_create_trainer(db) obj = ( db.query(Client) .filter(Client.id == client_id, Client.trainer_id == t.id) .first() ) if not obj: raise HTTPException(status_code=404, detail="No existe") now = datetime.utcnow() old_plan = obj.plan old_total_amount, _ = calculate_total_for_period_from_client(obj, now.year, now.month) obj.name = c.name.strip() obj.email = str(c.email).strip() if c.email else None obj.status = c.status obj.phone = c.phone.strip() if c.phone else None obj.rut = c.rut.strip() if c.rut else None obj.birth_date = c.birth_date obj.enrollment_date = c.enrollment_date obj.plan = c.plan.strip() if c.plan else None obj.training_days = serialize_training_days(normalize_training_days(c.training_days)) obj.service_recovery_60 = c.service_recovery_60 obj.service_presoterapia_30 = c.service_presoterapia_30 obj.service_presoterapia_10 = c.service_presoterapia_10 obj.service_planificacion_online_4s = c.service_planificacion_online_4s obj.address = c.address.strip() if c.address else None obj.occupation = c.occupation.strip() if c.occupation else None obj.nutrition_tracking = c.nutrition_tracking obj.sleep_tracking = c.sleep_tracking obj.habits_tracking = c.habits_tracking obj.nutrition_notes = c.nutrition_notes.strip() if c.nutrition_notes else None obj.sleep_notes = c.sleep_notes.strip() if c.sleep_notes else None obj.habits_notes = c.habits_notes.strip() if c.habits_notes else None obj.injuries_notes = c.injuries_notes.strip() if c.injuries_notes else None obj.notes = c.notes.strip() if c.notes else None obj.goal_1 = c.goal_1.strip() if c.goal_1 else None obj.goal_2 = c.goal_2.strip() if c.goal_2 else None obj.goal_3 = c.goal_3.strip() if c.goal_3 else None base_amount = calculate_total_amount_from_client(obj) new_total_amount, is_prorated = calculate_total_for_period_from_client(obj, now.year, now.month) plan_changed = old_plan != obj.plan total_changed = old_total_amount != new_total_amount if plan_changed: db.add( ClientPlanChangeHistory( client_id=obj.id, previous_plan_name=old_plan, previous_total_amount=old_total_amount, new_plan_name=obj.plan, new_total_amount=new_total_amount, changed_at=datetime.utcnow(), ) ) if plan_changed or total_changed: upsert_monthly_payment_history( db, client_id=obj.id, plan_name=obj.plan, base_amount=base_amount, total_amount=new_total_amount, is_prorated=is_prorated, year=now.year, month=now.month, ) db.commit() db.refresh(obj) return obj @app.delete("/trainer/clients/{client_id}") def delete_client(client_id: int): with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") try: # Si existe usuario alumno asociado, se elimina; para otros roles se desvincula. db.query(AuthUser).filter( AuthUser.student_client_id == c.id, func.lower(AuthUser.role) == "student", ).delete(synchronize_session=False) db.query(AuthUser).filter( AuthUser.student_client_id == c.id, or_(AuthUser.role.is_(None), func.lower(AuthUser.role) != "student"), ).update({AuthUser.student_client_id: None}, synchronize_session=False) # Eliminacion total: borra en todas las tablas con FK a clients.id. fk_rows = db.execute( text( """ SELECT tc.table_name, kcu.column_name FROM information_schema.table_constraints tc JOIN information_schema.key_column_usage kcu ON tc.constraint_name = kcu.constraint_name AND tc.constraint_schema = kcu.constraint_schema JOIN information_schema.constraint_column_usage ccu ON tc.constraint_name = ccu.constraint_name AND tc.constraint_schema = ccu.constraint_schema WHERE tc.constraint_type = 'FOREIGN KEY' AND tc.constraint_schema = 'public' AND ccu.table_name = 'clients' AND ccu.column_name = 'id' """ ) ).all() def _safe_ident(ident: str) -> str: if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", ident or ""): raise HTTPException(status_code=500, detail="Identificador SQL invalido") return f'"{ident}"' for table_name, column_name in fk_rows: if table_name == "clients": continue if table_name == "auth_users": # Ya tratado arriba para no borrar usuarios que no sean alumno. continue table_sql = _safe_ident(str(table_name)) col_sql = _safe_ident(str(column_name)) db.execute( text(f"DELETE FROM {table_sql} WHERE {col_sql} = :client_id"), {"client_id": c.id}, ) db.delete(c) db.commit() return {"ok": True} except IntegrityError: db.rollback() raise HTTPException( status_code=409, detail="No se pudo eliminar el alumno porque tiene datos relacionados activos.", ) @app.get("/trainer/agenda", response_model=list[ClientAgendaOut]) def list_client_agendas(request: Request): with SessionLocal() as db: t = get_or_create_trainer(db) q = db.query(Client).filter( Client.trainer_id == t.id, or_(Client.status.is_(None), func.lower(Client.status) != "internal"), ) if _is_student_scoped_request(request): student_client_id = getattr(request.state, "student_client_id", None) q = q.filter(Client.id == student_client_id) clients = q.order_by(Client.name.asc()).all() return [ ClientAgendaOut( client_id=c.id, client_name=c.name, entries=parse_agenda_schedule(c.agenda_schedule), ) for c in clients ] @app.put("/trainer/clients/{client_id}/agenda", response_model=ClientAgendaOut) def update_client_agenda(client_id: int, body: AgendaScheduleIn): with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") normalized = normalize_agenda_entries(body.entries or []) c.agenda_schedule = serialize_agenda_entries(normalized) db.commit() db.refresh(c) return ClientAgendaOut( client_id=c.id, client_name=c.name, entries=parse_agenda_schedule(c.agenda_schedule), ) @app.get("/trainer/agenda/overrides", response_model=list[AgendaOverrideOut]) def list_agenda_overrides(year: int, month: int, request: Request): if month < 1 or month > 12: raise HTTPException(status_code=400, detail="Mes invalido") start = date(year, month, 1) next_month = date(year + (1 if month == 12 else 0), 1 if month == 12 else month + 1, 1) with SessionLocal() as db: t = get_or_create_trainer(db) rows = ( db.query(ClientAgendaException) .join(Client, Client.id == ClientAgendaException.client_id) .filter( Client.trainer_id == t.id, ( (ClientAgendaException.source_date >= start) & (ClientAgendaException.source_date < next_month) ) | ( (ClientAgendaException.target_date >= start) & (ClientAgendaException.target_date < next_month) ), ) .order_by(ClientAgendaException.source_date.asc(), ClientAgendaException.client_id.asc()) ) if _is_student_scoped_request(request): student_client_id = getattr(request.state, "student_client_id", None) rows = rows.filter(ClientAgendaException.client_id == student_client_id) rows = rows.all() return [ AgendaOverrideOut( client_id=row.client_id, source_date=row.source_date, target_date=row.target_date, time=row.time_slot, updated_at=row.updated_at, ) for row in rows ] @app.put("/trainer/clients/{client_id}/agenda/override", response_model=Optional[AgendaOverrideOut]) def set_client_agenda_override(client_id: int, body: AgendaOverrideIn): with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") regular_on_source = regular_time_for_client_on_date(c, body.source_date) if not regular_on_source: raise HTTPException(status_code=400, detail="El alumno no tiene clase habitual ese dia") row = ( db.query(ClientAgendaException) .filter( ClientAgendaException.client_id == c.id, ClientAgendaException.source_date == body.source_date, ) .first() ) if body.target_date is None or body.time is None: if row: db.delete(row) db.commit() return None time_value = body.time.strip() validate_time_for_date(body.target_date, time_value) if not row: row = ClientAgendaException( client_id=c.id, source_date=body.source_date, target_date=body.target_date, time_slot=time_value, updated_at=datetime.utcnow(), ) db.add(row) else: row.target_date = body.target_date row.time_slot = time_value row.updated_at = datetime.utcnow() db.commit() db.refresh(row) return AgendaOverrideOut( client_id=row.client_id, source_date=row.source_date, target_date=row.target_date, time=row.time_slot, updated_at=row.updated_at, ) def serialize_agenda_change_request(row: ClientAgendaChangeRequest, client_name: str) -> AgendaChangeRequestOut: return AgendaChangeRequestOut( id=row.id, client_id=row.client_id, client_name=client_name, source_date=row.source_date, source_time=row.source_time, target_date=row.target_date, target_time=row.target_time, reason=row.reason, status=row.status, trainer_note=row.trainer_note, reviewed_at=row.reviewed_at, created_at=row.created_at, updated_at=row.updated_at, ) def serialize_routine_edit_notification( row: ClientRoutineEditNotification, client_name: str, ) -> RoutineEditNotificationOut: fields = [ as_clean_str(part).lower() for part in str(row.changed_fields or "").split(",") if as_clean_str(part) ] return RoutineEditNotificationOut( id=row.id, client_id=row.client_id, client_name=client_name, week_label=row.week_label, day_label=row.day_label, changed_fields=fields, message=row.message, is_read=bool(row.is_read), created_at=row.created_at, read_at=row.read_at, ) @app.get("/trainer/agenda/change-requests", response_model=list[AgendaChangeRequestOut]) def list_agenda_change_requests( request: Request, status: Optional[str] = None, client_id: Optional[int] = None, ): with SessionLocal() as db: t = get_or_create_trainer(db) q = ( db.query(ClientAgendaChangeRequest, Client.name) .join(Client, Client.id == ClientAgendaChangeRequest.client_id) .filter(Client.trainer_id == t.id) ) if status: status_value = status.strip().lower() if status_value not in {"pending", "approved", "rejected"}: raise HTTPException(status_code=400, detail="Estado invalido") q = q.filter(ClientAgendaChangeRequest.status == status_value) if _is_student_scoped_request(request): client_id = getattr(request.state, "student_client_id", None) if client_id: q = q.filter(ClientAgendaChangeRequest.client_id == client_id) rows = q.order_by(ClientAgendaChangeRequest.created_at.desc()).all() return [serialize_agenda_change_request(row, client_name) for row, client_name in rows] @app.post("/trainer/clients/{client_id}/agenda/change-requests", response_model=AgendaChangeRequestOut) def create_agenda_change_request(client_id: int, body: AgendaChangeRequestIn): with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") regular_on_source = regular_time_for_client_on_date(c, body.source_date) if not regular_on_source: raise HTTPException(status_code=400, detail="El alumno no tiene clase habitual ese dia") target_time = str(body.target_time or "").strip() validate_time_for_date(body.target_date, target_time) duplicate_pending = ( db.query(ClientAgendaChangeRequest) .filter( ClientAgendaChangeRequest.client_id == c.id, ClientAgendaChangeRequest.source_date == body.source_date, ClientAgendaChangeRequest.status == "pending", ) .first() ) if duplicate_pending: raise HTTPException(status_code=409, detail="Ya existe una solicitud pendiente para esa clase") source_time = str(body.source_time or "").strip() or None if source_time and len(source_time) != 5: source_time = regular_on_source row = ClientAgendaChangeRequest( client_id=c.id, source_date=body.source_date, source_time=source_time or regular_on_source, target_date=body.target_date, target_time=target_time, reason=(str(body.reason or "").strip() or None), status="pending", created_at=datetime.utcnow(), updated_at=datetime.utcnow(), ) db.add(row) db.commit() db.refresh(row) return serialize_agenda_change_request(row, c.name) @app.put("/trainer/agenda/change-requests/{request_id}/decision", response_model=AgendaChangeRequestOut) def decide_agenda_change_request(request_id: int, body: AgendaChangeRequestDecisionIn): with SessionLocal() as db: t = get_or_create_trainer(db) pair = ( db.query(ClientAgendaChangeRequest, Client) .join(Client, Client.id == ClientAgendaChangeRequest.client_id) .filter(ClientAgendaChangeRequest.id == request_id, Client.trainer_id == t.id) .first() ) if not pair: raise HTTPException(status_code=404, detail="No existe") req, c = pair if req.status != "pending": raise HTTPException(status_code=409, detail="La solicitud ya fue procesada") action = str(body.action or "").strip().lower() if action not in {"approved", "rejected"}: raise HTTPException(status_code=400, detail="Accion invalida") now_utc = datetime.utcnow() req.status = action req.trainer_note = str(body.note or "").strip() or None req.reviewed_at = now_utc req.updated_at = now_utc if action == "approved": regular_on_source = regular_time_for_client_on_date(c, req.source_date) if not regular_on_source: raise HTTPException(status_code=400, detail="El alumno no tiene clase habitual ese dia") validate_time_for_date(req.target_date, req.target_time) ex = ( db.query(ClientAgendaException) .filter( ClientAgendaException.client_id == c.id, ClientAgendaException.source_date == req.source_date, ) .first() ) if not ex: ex = ClientAgendaException( client_id=c.id, source_date=req.source_date, target_date=req.target_date, time_slot=req.target_time, updated_at=now_utc, ) db.add(ex) else: ex.target_date = req.target_date ex.time_slot = req.target_time ex.updated_at = now_utc db.commit() db.refresh(req) return serialize_agenda_change_request(req, c.name) @app.get("/trainer/routine-edit-notifications", response_model=list[RoutineEditNotificationOut]) def list_routine_edit_notifications(request: Request, unread_only: bool = True): if getattr(request.state, "access_role", "coach") == "student": raise HTTPException(status_code=403, detail="Sin permisos para ver notificaciones de rutina") with SessionLocal() as db: t = get_or_create_trainer(db) q = ( db.query(ClientRoutineEditNotification, Client.name) .join(Client, Client.id == ClientRoutineEditNotification.client_id) .filter(Client.trainer_id == t.id) ) if unread_only: q = q.filter(ClientRoutineEditNotification.is_read.is_(False)) rows = q.order_by(ClientRoutineEditNotification.created_at.desc()).limit(50).all() return [serialize_routine_edit_notification(row, client_name) for row, client_name in rows] @app.put("/trainer/routine-edit-notifications/mark-read") def mark_routine_edit_notifications_read(request: Request, body: RoutineEditNotificationReadIn): if getattr(request.state, "access_role", "coach") == "student": raise HTTPException(status_code=403, detail="Sin permisos para editar notificaciones de rutina") ids = [int(x) for x in (body.ids or []) if isinstance(x, int) and int(x) > 0] if not ids: return {"updated": 0} with SessionLocal() as db: t = get_or_create_trainer(db) rows = ( db.query(ClientRoutineEditNotification) .join(Client, Client.id == ClientRoutineEditNotification.client_id) .filter( Client.trainer_id == t.id, ClientRoutineEditNotification.id.in_(ids), ) .all() ) now_ts = datetime.utcnow() updated = 0 for row in rows: if row.is_read: continue row.is_read = True row.read_at = now_ts updated += 1 db.commit() return {"updated": updated} @app.post("/trainer/checkin/scan", response_model=AttendanceCheckinScanOut) def student_scan_checkin(request: Request, body: AttendanceCheckinScanIn): if getattr(request.state, "access_role", "coach") != "student": raise HTTPException(status_code=403, detail="Solo alumnos pueden marcar asistencia con QR") raw_qr_code = str(body.qr_code or "").strip() if not raw_qr_code: raise HTTPException(status_code=400, detail="Codigo QR invalido") def _normalize_code(value: str) -> str: return unquote(str(value or "").strip()) extracted_code = _normalize_code(raw_qr_code) try: parsed = urlparse(extracted_code) if parsed.scheme and (parsed.netloc or parsed.path): params = parse_qs(parsed.query or "") for key in ("checkin_code", "qr_code", "code", "qr"): values = params.get(key) or [] if values and str(values[0]).strip(): extracted_code = _normalize_code(values[0]) break except Exception: pass if extracted_code.lower() != str(CHECKIN_QR_SECRET).strip().lower(): raise HTTPException(status_code=400, detail="QR invalido o vencido") student_client_id = int(getattr(request.state, "student_client_id", 0) or 0) if student_client_id <= 0: raise HTTPException(status_code=401, detail="Sesion de alumno invalida") with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == student_client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="Alumno no encontrado") now_local = datetime.now(ZoneInfo(CHECKIN_TZ)) today = now_local.date() now_hhmm = now_local.strftime("%H:%M") assigned_times = client_assigned_times_for_date(db, c, today) if not assigned_times: raise HTTPException(status_code=400, detail="No tienes clase agendada para hoy.") matched_time = resolve_checkin_time_for_now(assigned_times, now_hhmm) if not matched_time: raise HTTPException(status_code=400, detail="No tienes clase agendada en este horario") existing = ( db.query(ClientAttendanceCheckin) .filter( ClientAttendanceCheckin.client_id == c.id, ClientAttendanceCheckin.attendance_date == today, ) .first() ) if existing: return AttendanceCheckinScanOut( status="already_marked", message="Tu asistencia de hoy ya estaba registrada.", attendance_date=existing.attendance_date, checkin_at=existing.checkin_at, scheduled_time=existing.scheduled_time, ) row = ClientAttendanceCheckin( client_id=c.id, attendance_date=today, scheduled_time=matched_time, checkin_at=datetime.utcnow(), qr_code=extracted_code, ) db.add(row) db.commit() db.refresh(row) return AttendanceCheckinScanOut( status="ok", message="Asistencia registrada correctamente.", attendance_date=row.attendance_date, checkin_at=row.checkin_at, scheduled_time=row.scheduled_time, ) @app.get("/trainer/checkin/qr-value", response_model=AttendanceQrValueOut) def get_checkin_qr_value(request: Request): if _is_student_scoped_request(request): raise HTTPException(status_code=403, detail="Sin permisos para ver QR de check-in") return AttendanceQrValueOut(qr_value=CHECKIN_QR_SECRET) @app.get("/trainer/checkin/history", response_model=list[AttendanceHistoryItemOut]) def list_checkin_history(request: Request, limit: int = 30, client_id: Optional[int] = None): safe_limit = max(1, min(int(limit or 30), 100)) with SessionLocal() as db: t = get_or_create_trainer(db) q = ( db.query(ClientAttendanceCheckin, Client.name) .join(Client, Client.id == ClientAttendanceCheckin.client_id) .filter(Client.trainer_id == t.id) ) if _is_student_scoped_request(request): student_client_id = int(getattr(request.state, "student_client_id", 0) or 0) q = q.filter(ClientAttendanceCheckin.client_id == student_client_id) elif client_id: q = q.filter(ClientAttendanceCheckin.client_id == int(client_id)) rows = ( q.order_by( ClientAttendanceCheckin.attendance_date.desc(), ClientAttendanceCheckin.checkin_at.desc(), ) .limit(safe_limit) .all() ) return [ AttendanceHistoryItemOut( id=row.id, client_id=row.client_id, client_name=client_name, attendance_date=row.attendance_date, scheduled_time=row.scheduled_time, checkin_at=row.checkin_at, ) for row, client_name in rows ] # ====================== # MEASUREMENTS # ====================== @app.get("/trainer/clients/{client_id}/measurements", response_model=list[MeasurementOut]) def list_measurements(client_id: int): with SessionLocal() as db: return ( db.query(NutritionalMeasurement) .filter(NutritionalMeasurement.client_id == client_id) .order_by(NutritionalMeasurement.measured_at.desc()) .all() ) @app.post("/trainer/clients/{client_id}/measurements", response_model=MeasurementOut) def create_measurement(client_id: int, m: MeasurementIn): with SessionLocal() as db: new = NutritionalMeasurement( client_id=client_id, measured_at=m.measured_at or datetime.utcnow(), weight_kg=m.weight_kg or 0, body_fat_pct=m.body_fat_pct or 0, muscle_pct=m.muscle_pct or 0, water_pct=m.water_pct or 0, notes=m.notes or "", ) db.add(new) db.commit() db.refresh(new) return new @app.put("/trainer/clients/{client_id}/measurements/{measurement_id}", response_model=MeasurementOut) def update_measurement(client_id: int, measurement_id: int, m: MeasurementIn): with SessionLocal() as db: row = ( db.query(NutritionalMeasurement) .filter( NutritionalMeasurement.id == measurement_id, NutritionalMeasurement.client_id == client_id, ) .first() ) if not row: raise HTTPException(status_code=404, detail="Medición no encontrada") if m.measured_at is not None: row.measured_at = m.measured_at if m.weight_kg is not None: row.weight_kg = m.weight_kg if m.body_fat_pct is not None: row.body_fat_pct = m.body_fat_pct if m.muscle_pct is not None: row.muscle_pct = m.muscle_pct if m.water_pct is not None: row.water_pct = m.water_pct if m.notes is not None: row.notes = m.notes db.commit() db.refresh(row) return row @app.delete("/trainer/clients/{client_id}/measurements/{measurement_id}") def delete_measurement(client_id: int, measurement_id: int): with SessionLocal() as db: row = ( db.query(NutritionalMeasurement) .filter( NutritionalMeasurement.id == measurement_id, NutritionalMeasurement.client_id == client_id, ) .first() ) if not row: raise HTTPException(status_code=404, detail="Medición no encontrada") db.delete(row) db.commit() return {"ok": True} @app.get("/trainer/routines/exercise-catalog", response_model=list[str]) def list_routine_exercise_catalog(): with SessionLocal() as db: t = get_or_create_trainer(db) rows_from_routines = ( db.query(ClientRoutineExercise.exercise) .join(Client, Client.id == ClientRoutineExercise.client_id) .filter(Client.trainer_id == t.id) .all() ) rows_from_drafts = ( db.query(ClientRoutineDraft.exercise) .join(Client, Client.id == ClientRoutineDraft.client_id) .filter(Client.trainer_id == t.id) .all() ) names: set[str] = set() for exercise_name in DEFAULT_EXERCISE_CATALOG_EN: normalized = normalize_exercise_name(exercise_name) if normalized: names.add(normalized) for (exercise_name,) in rows_from_routines: normalized = normalize_exercise_name(exercise_name) if normalized: names.add(normalized) for (exercise_name,) in rows_from_drafts: normalized = normalize_exercise_name(exercise_name) if normalized: names.add(normalized) return sorted(names, key=lambda x: x.lower()) @app.get("/trainer/clients/{client_id}/routine-drafts/weeks", response_model=list[str]) def list_client_routine_draft_weeks(client_id: int): with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") rows = ( db.query(ClientRoutineDraft.week_label) .filter(ClientRoutineDraft.client_id == client_id) .distinct() .all() ) weeks = sorted({as_clean_str(x[0]) for x in rows if as_clean_str(x[0])}, key=lambda x: x.lower()) return weeks @app.get("/trainer/clients/{client_id}/routine-drafts", response_model=list[RoutineDraftRowOut]) def list_client_routine_draft_rows(client_id: int, week_label: str, day_label: str): with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") week = normalize_week_label(week_label) day = normalize_day_label(day_label) if not week or not day: raise HTTPException(status_code=400, detail="Semana o dia invalido") rows = ( db.query(ClientRoutineDraft) .filter( ClientRoutineDraft.client_id == client_id, ClientRoutineDraft.week_label == week, ClientRoutineDraft.day_label == day, ) .order_by(ClientRoutineDraft.row_index.asc()) .all() ) # Si la semana consultada aún no tiene rutina guardada, hereda la última semana previa. # Regla: heredar ejercicios/estructura, pero NO pesos. if not rows: target_week_date = week_label_to_date(week) if target_week_date: candidates = ( db.query(ClientRoutineDraft) .filter( ClientRoutineDraft.client_id == client_id, ClientRoutineDraft.day_label == day, ) .all() ) latest_date: Optional[date] = None for row in candidates: d = week_label_to_date(row.week_label) if not d or d >= target_week_date: continue if latest_date is None or d > latest_date: latest_date = d if latest_date: latest_label = latest_date.isoformat() rows = ( db.query(ClientRoutineDraft) .filter( ClientRoutineDraft.client_id == client_id, ClientRoutineDraft.day_label == day, ClientRoutineDraft.week_label == latest_label, ) .order_by(ClientRoutineDraft.row_index.asc()) .all() ) return [ RoutineDraftRowOut( row_index=row.row_index, completed=False, exercise=row.exercise, series=row.series, reps=None, weight=None, previous=row.previous, rpe=row.rpe, progress=row.progress, semana=row.semana, comment=row.comment, ) for row in rows ] return [ RoutineDraftRowOut( row_index=row.row_index, completed=bool(row.completed), exercise=row.exercise, series=row.series, reps=row.reps, weight=row.weight, previous=row.previous, rpe=row.rpe, progress=row.progress, semana=row.semana, comment=row.comment, ) for row in rows ] @app.get("/trainer/clients/{client_id}/routine-draft-note", response_model=RoutineDraftNoteOut) def get_client_routine_draft_note(client_id: int, week_label: str, day_label: str): with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") week = normalize_week_label(week_label) day = normalize_day_label(day_label) if not week or not day: raise HTTPException(status_code=400, detail="Semana o dia invalido") row = ( db.query(ClientRoutineDraftNote) .filter( ClientRoutineDraftNote.client_id == client_id, ClientRoutineDraftNote.week_label == week, ClientRoutineDraftNote.day_label == day, ) .first() ) return RoutineDraftNoteOut( week_label=week, day_label=day, note=(row.note if row else None), ) @app.put("/trainer/clients/{client_id}/routine-draft-note", response_model=RoutineDraftNoteOut) def upsert_client_routine_draft_note(client_id: int, body: RoutineDraftNoteIn): with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") week = normalize_week_label(body.week_label) day = normalize_day_label(body.day_label) if not week or not day: raise HTTPException(status_code=400, detail="Semana o dia invalido") cleaned_note = as_clean_str(body.note) now_ts = datetime.utcnow() stmt = insert(ClientRoutineDraftNote).values( client_id=client_id, week_label=week, day_label=day, note=cleaned_note, updated_at=now_ts, ) stmt = stmt.on_conflict_do_update( constraint="uq_client_routine_draft_note_key", set_={ "note": cleaned_note, "updated_at": now_ts, }, ) db.execute(stmt) db.commit() return RoutineDraftNoteOut( week_label=week, day_label=day, note=cleaned_note, ) @app.get("/trainer/clients/{client_id}/routine-week-note", response_model=RoutineWeekNoteOut) def get_client_routine_week_note(client_id: int, week_label: str): with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") week = normalize_week_label(week_label) if not week: raise HTTPException(status_code=400, detail="Semana invalida") row = ( db.query(ClientRoutineWeekNote) .filter( ClientRoutineWeekNote.client_id == client_id, ClientRoutineWeekNote.week_label == week, ) .first() ) return RoutineWeekNoteOut( week_label=week, note=(row.note if row else None), ) @app.put("/trainer/clients/{client_id}/routine-week-note", response_model=RoutineWeekNoteOut) def upsert_client_routine_week_note(client_id: int, body: RoutineWeekNoteIn): with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") week = normalize_week_label(body.week_label) if not week: raise HTTPException(status_code=400, detail="Semana invalida") cleaned_note = as_clean_str(body.note) now_ts = datetime.utcnow() stmt = insert(ClientRoutineWeekNote).values( client_id=client_id, week_label=week, note=cleaned_note, updated_at=now_ts, ) stmt = stmt.on_conflict_do_update( constraint="uq_client_routine_week_note_key", set_={ "note": cleaned_note, "updated_at": now_ts, }, ) db.execute(stmt) db.commit() return RoutineWeekNoteOut( week_label=week, note=cleaned_note, ) @app.get("/trainer/clients/{client_id}/routine-progress", response_model=list[RoutineProgressSeriesOut]) def list_client_routine_progress(client_id: int): day_order = {"DIA 1": 1, "DIA 2": 2, "DIA 3": 3, "DIA 4": 4} with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") # Progreso considera todas las filas de rutina (verdes y rojas). rows = ( db.query(ClientRoutineDraft) .filter(ClientRoutineDraft.client_id == client_id) .order_by( ClientRoutineDraft.week_label.asc(), ClientRoutineDraft.day_label.asc(), ClientRoutineDraft.row_index.asc(), ) .all() ) exercise_sessions: dict[str, dict[str, dict]] = {} for row in rows: exercise = canonicalize_exercise_name(row.exercise) week = as_clean_str(row.week_label) day = normalize_day_label(row.day_label) if not exercise or not week or not day: continue session_key = f"{week}|{day}" session_label = f"{week} - {day}" parsed_week_date = _parse_week_date(week) max_weight = _calc_max_weight(row.weight) tonnage = _calc_tonnage(row.series, row.reps, row.weight) average_weight = _calc_average_weight(row.series, row.weight) rpe = _parse_numeric(row.rpe) progress_value = _parse_numeric(row.progress) sessions_map = exercise_sessions.setdefault(exercise, {}) item = sessions_map.get(session_key) if not item: item = { "week_label": week, "day_label": day, "session_label": session_label, "week_date": parsed_week_date, "day_idx": day_order.get(day, 99), "max_weight_value": None, "tonnage_total": 0.0, "tonnage_count": 0, "avg_weight_total": 0.0, "avg_weight_count": 0, "rpe_total": 0.0, "rpe_count": 0, "progress_total": 0.0, "progress_count": 0, } sessions_map[session_key] = item if max_weight is not None: current_max = item["max_weight_value"] item["max_weight_value"] = ( max_weight if current_max is None else max(current_max, max_weight) ) if tonnage is not None: item["tonnage_total"] += tonnage item["tonnage_count"] += 1 if average_weight is not None: item["avg_weight_total"] += average_weight item["avg_weight_count"] += 1 if rpe is not None: item["rpe_total"] += rpe item["rpe_count"] += 1 if progress_value is not None: item["progress_total"] += progress_value item["progress_count"] += 1 output: list[RoutineProgressSeriesOut] = [] for exercise in sorted(exercise_sessions.keys(), key=lambda x: x.lower()): session_values = list(exercise_sessions[exercise].values()) session_values.sort( key=lambda item: ( item["week_date"] or date.max, item["day_idx"], item["session_label"], ) ) points: list[RoutineProgressPointOut] = [] for session in session_values: max_weight = session["max_weight_value"] tonnage = ( session["tonnage_total"] if session["tonnage_count"] > 0 else None ) average_weight = ( (session["avg_weight_total"] / session["avg_weight_count"]) if session["avg_weight_count"] > 0 else None ) rpe = ( (session["rpe_total"] / session["rpe_count"]) if session["rpe_count"] > 0 else None ) progress_value = ( (session["progress_total"] / session["progress_count"]) if session["progress_count"] > 0 else None ) points.append( RoutineProgressPointOut( week_label=session["week_label"], day_label=session["day_label"], session_label=session["session_label"], max_weight=max_weight, tonnage=tonnage, average_weight=average_weight, rpe=rpe, progress_value=progress_value, ) ) if points: output.append(RoutineProgressSeriesOut(exercise=exercise, points=points)) return output @app.put("/trainer/clients/{client_id}/routine-drafts", response_model=list[RoutineDraftRowOut]) def upsert_client_routine_draft_rows(client_id: int, body: RoutineDraftUpsertIn, request: Request): with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") week = normalize_week_label(body.week_label) day = normalize_day_label(body.day_label) if not week or not day: raise HTTPException(status_code=400, detail="Semana o dia invalido") target_week_date = week_label_to_date(week) day_order = {"DIA 1": 1, "DIA 2": 2, "DIA 3": 3, "DIA 4": 4} target_day_rank = day_order.get(day, 99) existing_current_rows = ( db.query(ClientRoutineDraft) .filter( ClientRoutineDraft.client_id == client_id, ClientRoutineDraft.week_label == week, ClientRoutineDraft.day_label == day, ) .all() ) old_exercise_by_row_index = {row.row_index: as_clean_str(row.exercise) for row in existing_current_rows} old_series_by_row_index = {row.row_index: as_clean_str(row.series) for row in existing_current_rows} old_reps_by_row_index = {row.row_index: as_clean_str(row.reps) for row in existing_current_rows} old_weight_by_row_index = {row.row_index: as_clean_str(row.weight) for row in existing_current_rows} db.query(ClientRoutineDraft).filter( ClientRoutineDraft.client_id == client_id, ClientRoutineDraft.week_label == week, ClientRoutineDraft.day_label == day, ).delete() # Snapshot de RM por ejercicio: se calcula al momento de guardar y no # reescribe semanas anteriores. rm_by_exercise_key: dict[str, float] = {} historical_rows = ( db.query( ClientRoutineDraft.week_label, ClientRoutineDraft.day_label, ClientRoutineDraft.exercise, ClientRoutineDraft.weight, ) .filter(ClientRoutineDraft.client_id == client_id) .all() ) for week_label_value, day_label_value, ex_value, weight_value in historical_rows: ex_clean = as_clean_str(ex_value) if not ex_clean: continue # Regla temporal: para calcular RM solo se considera historial previo # al día/semana que se está guardando ahora. if target_week_date: row_week_date = week_label_to_date(week_label_value) if row_week_date: if row_week_date > target_week_date: continue if row_week_date == target_week_date: row_day_rank = day_order.get(as_clean_str(day_label_value) or "", 99) if row_day_rank >= target_day_rank: continue ex_key = ex_clean.upper() historical_max = _max_numeric_weight_from_weight_field(weight_value) if historical_max is None: continue previous_max = rm_by_exercise_key.get(ex_key) if previous_max is None or historical_max > previous_max: rm_by_exercise_key[ex_key] = historical_max out: list[RoutineDraftRowOut] = [] new_exercise_by_row_index: dict[int, Optional[str]] = {} changed_fields_for_notification: set[str] = set() for idx, row in enumerate(body.rows[:30]): cleaned_completed = bool(row.completed) cleaned_exercise = canonicalize_exercise_name(row.exercise) cleaned_series = as_clean_str(row.series) cleaned_reps = as_clean_str(row.reps) cleaned_weight = as_clean_str(row.weight) cleaned_previous = as_clean_str(row.previous) cleaned_rpe = as_clean_str(row.rpe) cleaned_progress = as_clean_str(row.progress) cleaned_comment = as_clean_str(row.comment) now_ts = datetime.utcnow() new_exercise_by_row_index[idx] = cleaned_exercise if old_series_by_row_index.get(idx, "") != cleaned_series: changed_fields_for_notification.add("series") if old_reps_by_row_index.get(idx, "") != cleaned_reps: changed_fields_for_notification.add("repeticiones") if old_weight_by_row_index.get(idx, "") != cleaned_weight: changed_fields_for_notification.add("peso") cleaned_semana: Optional[str] = None if cleaned_exercise: ex_key = cleaned_exercise.upper() current_rm = rm_by_exercise_key.get(ex_key) row_max = _max_numeric_weight_from_weight_field(cleaned_weight) if row_max is not None and (current_rm is None or row_max > current_rm): current_rm = row_max rm_by_exercise_key[ex_key] = current_rm cleaned_semana = _format_rm_snapshot(current_rm) stmt = insert(ClientRoutineDraft).values( client_id=client_id, week_label=week, day_label=day, row_index=idx, exercise=cleaned_exercise, series=cleaned_series, reps=cleaned_reps, weight=cleaned_weight, previous=cleaned_previous, rpe=cleaned_rpe, progress=cleaned_progress, semana=cleaned_semana, comment=cleaned_comment, completed=cleaned_completed, updated_at=now_ts, ) stmt = stmt.on_conflict_do_update( constraint="uq_client_routine_draft_key", set_={ "exercise": cleaned_exercise, "series": cleaned_series, "reps": cleaned_reps, "weight": cleaned_weight, "previous": cleaned_previous, "rpe": cleaned_rpe, "progress": cleaned_progress, "semana": cleaned_semana, "comment": cleaned_comment, "completed": cleaned_completed, "updated_at": now_ts, }, ) db.execute(stmt) out.append( RoutineDraftRowOut( row_index=idx, completed=cleaned_completed, exercise=cleaned_exercise, series=cleaned_series, reps=cleaned_reps, weight=cleaned_weight, previous=cleaned_previous, rpe=cleaned_rpe, progress=cleaned_progress, semana=cleaned_semana, comment=cleaned_comment, ) ) # Si cambia el ejercicio en una fila, propágalo a semanas futuras del mismo día/fila. # Los pesos futuros se limpian para forzar asignación manual. if target_week_date and body.propagate_exercise_changes: changed_row_indexes = set() for row_index, new_ex in new_exercise_by_row_index.items(): old_ex = old_exercise_by_row_index.get(row_index) if old_ex != new_ex: changed_row_indexes.add(row_index) if changed_row_indexes: future_rows = ( db.query(ClientRoutineDraft) .filter( ClientRoutineDraft.client_id == client_id, ClientRoutineDraft.day_label == day, ) .all() ) for future in future_rows: future_week_date = week_label_to_date(future.week_label) if not future_week_date or future_week_date <= target_week_date: continue if future.row_index not in changed_row_indexes: continue future.exercise = new_exercise_by_row_index.get(future.row_index) future.reps = None future.weight = None future.updated_at = datetime.utcnow() is_student_editor = _is_student_scoped_request(request) if is_student_editor and changed_fields_for_notification: ordered_fields = [x for x in ["series", "repeticiones", "peso"] if x in changed_fields_for_notification] if len(ordered_fields) == 1: suffix = ordered_fields[0] elif len(ordered_fields) == 2: suffix = f"{ordered_fields[0]} y {ordered_fields[1]}" else: suffix = f"{ordered_fields[0]}, {ordered_fields[1]} y {ordered_fields[2]}" db.add( ClientRoutineEditNotification( client_id=c.id, week_label=week, day_label=day, changed_fields=",".join(ordered_fields), message=f"{c.name} ha modificado {suffix}.", is_read=False, created_at=datetime.utcnow(), read_at=None, ) ) db.commit() return out @app.get("/trainer/clients/{client_id}/routines", response_model=list[RoutineExerciseOut]) def list_client_routines(client_id: int): with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") rows = ( db.query(ClientRoutineExercise) .filter(ClientRoutineExercise.client_id == client_id) .order_by( ClientRoutineExercise.week_label.asc(), ClientRoutineExercise.day_label.asc(), ClientRoutineExercise.id.asc(), ) .all() ) return rows @app.post( "/trainer/clients/{client_id}/routines/import-server-file", response_model=RoutineImportResultOut, ) def import_client_routines_from_server_file(client_id: int, body: RoutineImportIn): with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") file_path = Path(body.file_path).expanduser().resolve() if not file_path.exists() or not file_path.is_file(): raise HTTPException(status_code=404, detail="Archivo no encontrado") if file_path.suffix.lower() != ".xlsx": raise HTTPException(status_code=400, detail="Solo se permite .xlsx") try: parsed_rows = _parse_routine_rows_from_workbook(file_path) except Exception as e: raise HTTPException(status_code=400, detail=f"No se pudo leer el Excel: {e}") from e if body.replace_existing: db.query(ClientRoutineExercise).filter( ClientRoutineExercise.client_id == client_id ).delete() for row in parsed_rows: cleaned_exercise = canonicalize_exercise_name(row["exercise"]) db.add( ClientRoutineExercise( client_id=client_id, week_label=row["week_label"], day_label=row["day_label"], exercise=cleaned_exercise, series=row["series"], reps=row["reps"], weight=row["weight"], tonnage=row["tonnage"], previous=row["previous"], rpe=row["rpe"], progress=row["progress"], comment=row["comment"], source_sheet=row["source_sheet"], source_row=row["source_row"], ) ) db.commit() processed_sheets = len({x["source_sheet"] for x in parsed_rows if x.get("source_sheet")}) return RoutineImportResultOut( imported_count=len(parsed_rows), sheets_processed=processed_sheets, ) @app.post("/trainer/routines/assign-random-by-plan", response_model=RandomRoutineAssignOut) def assign_random_routines_by_plan(body: RandomRoutineAssignIn): with SessionLocal() as db: t = get_or_create_trainer(db) clients = ( db.query(Client) .filter(Client.trainer_id == t.id) .order_by(Client.id.asc()) .all() ) if body.from_enrollment_date and body.week_label: raise HTTPException( status_code=400, detail="No combines week_label con from_enrollment_date=true", ) single_week_label: Optional[str] = None today = date.today() current_week_monday = week_start_monday(today) if not body.from_enrollment_date: if body.week_label: single_week_label = normalize_week_label(body.week_label) if not single_week_label: raise HTTPException(status_code=400, detail="week_label invalido. Usa formato YYYY-MM-DD") else: single_week_label = default_week_label_for_routines() if not clients: return RandomRoutineAssignOut( clients_updated=0, rows_created=0, week_label=single_week_label, weeks_generated=0, from_enrollment_date=body.from_enrollment_date, ) rng = random.Random() total_rows_created = 0 generated_weeks: set[str] = set() for client in clients: day_count = max(1, min(4, random_routine_days_for_plan(client.plan))) day_templates = RANDOM_ROUTINE_DAY_TEMPLATES.get(day_count, RANDOM_ROUTINE_DAY_TEMPLATES[2]) if body.from_enrollment_date: base_date = client.enrollment_date or ( client.created_at.date() if client.created_at else today ) week_labels = week_labels_between(base_date, current_week_monday) if not week_labels: week_labels = [current_week_monday.isoformat()] else: week_labels = [single_week_label or default_week_label_for_routines()] if body.replace_existing: db.query(ClientRoutineExercise).filter( ClientRoutineExercise.client_id == client.id ).delete() db.query(ClientRoutineDraft).filter( ClientRoutineDraft.client_id == client.id ).delete() else: db.query(ClientRoutineExercise).filter( ClientRoutineExercise.client_id == client.id, ClientRoutineExercise.week_label.in_(week_labels), ).delete(synchronize_session=False) db.query(ClientRoutineDraft).filter( ClientRoutineDraft.client_id == client.id, ClientRoutineDraft.week_label.in_(week_labels), ).delete(synchronize_session=False) for week_label in week_labels: generated_weeks.add(week_label) for day_idx in range(day_count): day_label = f"DIA {day_idx + 1}" focus = day_templates[day_idx % len(day_templates)] generated_rows = build_random_rows_for_focus(focus, rng, max_rows=6) for row_idx, row in enumerate(generated_rows): cleaned_exercise = canonicalize_exercise_name(row["exercise"]) db.add( ClientRoutineExercise( client_id=client.id, week_label=week_label, day_label=day_label, exercise=cleaned_exercise, series=row["series"], reps=row["reps"], weight=row["weight"], tonnage="", previous=row["previous"], rpe=row["rpe"], progress=row["progress"], comment=row["comment"], source_sheet="AUTO_RANDOM", source_row=row_idx + 1, ) ) db.add( ClientRoutineDraft( client_id=client.id, week_label=week_label, day_label=day_label, row_index=row_idx, exercise=cleaned_exercise, series=row["series"], reps=row["reps"], weight=row["weight"], previous=row["previous"], rpe=row["rpe"], progress=row["progress"], semana=row["semana"], comment=row["comment"], updated_at=datetime.utcnow(), ) ) total_rows_created += 1 db.commit() return RandomRoutineAssignOut( clients_updated=len(clients), rows_created=total_rows_created, week_label=single_week_label, weeks_generated=len(generated_weeks), from_enrollment_date=body.from_enrollment_date, ) @app.get("/trainer/clients/{client_id}/billing-history", response_model=BillingHistoryOut) def get_client_billing_history(client_id: int): with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") monthly = ( db.query(ClientMonthlyPaymentHistory) .filter(ClientMonthlyPaymentHistory.client_id == client_id) .order_by( ClientMonthlyPaymentHistory.year.desc(), ClientMonthlyPaymentHistory.month.desc(), ) .all() ) changes = ( db.query(ClientPlanChangeHistory) .filter(ClientPlanChangeHistory.client_id == client_id) .order_by(ClientPlanChangeHistory.changed_at.desc()) .all() ) return BillingHistoryOut(monthly_payments=monthly, plan_changes=changes) @app.get("/trainer/clients/monthly-status", response_model=list[ClientMonthlyStatusOut]) def list_clients_monthly_status( request: Request, year: Optional[int] = None, month: Optional[int] = None, ): with SessionLocal() as db: t = get_or_create_trainer(db) now = datetime.utcnow() target_year = year or now.year target_month = month or now.month q = db.query(Client).filter(Client.trainer_id == t.id) if _is_student_scoped_request(request): student_client_id = getattr(request.state, "student_client_id", None) q = q.filter(Client.id == student_client_id) clients = q.all() client_ids = [c.id for c in clients] overdue_by_client: set[int] = set() if client_ids: overdue_rows = ( db.query(ClientMonthlyPaymentHistory.client_id) .filter( ClientMonthlyPaymentHistory.client_id.in_(client_ids), ClientMonthlyPaymentHistory.billing_status == "due", or_( ClientMonthlyPaymentHistory.year < target_year, and_( ClientMonthlyPaymentHistory.year == target_year, ClientMonthlyPaymentHistory.month < target_month, ), ), ) .distinct() .all() ) overdue_by_client = {row.client_id for row in overdue_rows} out: list[ClientMonthlyStatusOut] = [] for c in clients: row = get_or_create_monthly_payment_history_for_period(db, c, target_year, target_month) out.append( ClientMonthlyStatusOut( client_id=c.id, year=target_year, month=target_month, plan_name=row.plan_name, total_amount=row.total_amount, is_prorated=row.is_prorated, billing_status=normalize_billing_status(row.billing_status), is_paid=row.is_paid, has_overdue_debt=c.id in overdue_by_client, paid_at=row.paid_at, due_day=row.due_day, ) ) db.commit() return out @app.put( "/trainer/clients/{client_id}/billing-history/{year}/{month}/payment-status", response_model=MonthlyPaymentHistoryOut, ) def set_monthly_payment_status(client_id: int, year: int, month: int, body: PaymentStatusIn): with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") if month < 1 or month > 12: raise HTTPException(status_code=400, detail="Mes invalido") row = get_or_create_monthly_payment_history_for_period(db, c, year, month) row.billing_status = "paid" if body.is_paid else "due" row.is_paid = body.is_paid row.frozen_reason = None row.frozen_note = None row.frozen_at = None if body.is_paid: row.paid_at = datetime.utcnow() else: row.paid_at = None base_amount = calculate_total_amount_from_client(c) total_amount, is_prorated = calculate_total_for_period_from_client(c, year, month) row.base_amount = base_amount row.total_amount = total_amount row.is_prorated = is_prorated row.updated_at = datetime.utcnow() db.commit() db.refresh(row) return row @app.put( "/trainer/clients/{client_id}/billing-history/{year}/{month}/status", response_model=MonthlyPaymentHistoryOut, ) def set_monthly_billing_status(client_id: int, year: int, month: int, body: MonthlyBillingStatusIn): with SessionLocal() as db: t = get_or_create_trainer(db) c = db.query(Client).filter(Client.id == client_id, Client.trainer_id == t.id).first() if not c: raise HTTPException(status_code=404, detail="No existe") if month < 1 or month > 12: raise HTTPException(status_code=400, detail="Mes invalido") raw_status = (body.billing_status or "").strip().lower() status = "paid" if raw_status == "pagado" else raw_status if status not in {"paid", "due", "frozen"}: raise HTTPException(status_code=400, detail="Estado invalido") row = get_or_create_monthly_payment_history_for_period(db, c, year, month) row.billing_status = status row.updated_at = datetime.utcnow() if status == "paid": base_amount = calculate_total_amount_from_client(c) total_amount, is_prorated = calculate_total_for_period_from_client(c, year, month) row.base_amount = base_amount row.total_amount = total_amount row.is_prorated = is_prorated row.is_paid = True row.paid_at = datetime.utcnow() row.frozen_reason = None row.frozen_note = None row.frozen_at = None elif status == "due": base_amount = calculate_total_amount_from_client(c) total_amount, is_prorated = calculate_total_for_period_from_client(c, year, month) row.base_amount = base_amount row.total_amount = total_amount row.is_prorated = is_prorated row.is_paid = False row.paid_at = None row.frozen_reason = None row.frozen_note = None row.frozen_at = None else: row.is_paid = False row.paid_at = None row.total_amount = 0 row.frozen_reason = (body.frozen_reason or "").strip() or None row.frozen_note = (body.frozen_note or "").strip() or None row.frozen_at = datetime.utcnow() db.commit() db.refresh(row) return row @app.post("/trainer/clients/pricing-preview", response_model=PricingPreviewOut) def pricing_preview(body: PricingPreviewIn): base_amount = calculate_total_amount_from_values( body.plan, body.service_recovery_60, body.service_presoterapia_30, body.service_presoterapia_10, body.service_planificacion_online_4s, ) if not body.enrollment_date: return PricingPreviewOut( base_amount=base_amount, first_month_amount=base_amount, is_prorated=False, ) c = ClientIn( name="preview", plan=body.plan, training_days=body.training_days, enrollment_date=body.enrollment_date, service_recovery_60=body.service_recovery_60, service_presoterapia_30=body.service_presoterapia_30, service_presoterapia_10=body.service_presoterapia_10, service_planificacion_online_4s=body.service_planificacion_online_4s, ) first_month_amount, is_prorated = calculate_total_for_period_from_client_in( c, body.enrollment_date.year, body.enrollment_date.month ) return PricingPreviewOut( base_amount=base_amount, first_month_amount=first_month_amount, is_prorated=is_prorated, ) # ====================== # PAYMENTS (para Dashboard) # ====================== class PaymentOut(BaseModel): id: int client_id: int year: int month: int status: str amount: Optional[int] = None updated_at: datetime class Config: from_attributes = True @app.get("/trainer/payments", response_model=list[PaymentOut]) def list_payments(): with SessionLocal() as db: t = get_or_create_trainer(db) # IDs de clientes del trainer client_ids = [row[0] for row in db.query(Client.id).filter(Client.trainer_id == t.id).all()] if not client_ids: return [] pays = ( db.query(Payment) .filter(Payment.client_id.in_(client_ids)) .order_by(Payment.updated_at.desc()) .all() ) return pays